Compare commits

...
7 changed files with 119 additions and 53 deletions
+16 -12
View File
@@ -371,12 +371,13 @@ const lowerAssistantMessage = Effect.fn("OpenAIChat.lowerAssistantMessage")(func
return text
})()
const cached = message.content.findLast((part) => "cache" in part && part.cache !== undefined)
const cacheControl = options.cacheControl?.(cached && "cache" in cached ? cached.cache : undefined)
const result = {
role: "assistant" as const,
content: content.length === 0 ? null : ProviderShared.joinText(content),
tool_calls: toolCalls.length === 0 ? undefined : toolCalls,
reasoning_details: details,
cache_control: options.cacheControl?.(cached && "cache" in cached ? cached.cache : undefined),
content: content.length > 0 ? content.map((part) => part.text).join("") : toolCalls.length > 0 ? null : "",
...(toolCalls.length > 0 ? { tool_calls: toolCalls } : {}),
...(details !== undefined ? { reasoning_details: details } : {}),
...(cacheControl !== undefined ? { cache_control: cacheControl } : {}),
}
if (field === undefined || reasoningText === undefined) return result
return { ...result, [field]: reasoningText }
@@ -716,14 +717,13 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
return [{ ...state, usage }, events] as const
}
const reasoningField = state.reasoningField ?? (!state.lifecycle.text.has("text-0") ? reasoning?.field : undefined)
const reasoningField = state.reasoningField ?? reasoning?.field
const detailDelta = Array.isArray(delta?.reasoning_details) ? delta.reasoning_details : undefined
if (detailDelta !== undefined) appendReasoningDetails(state.reasoningDetails, detailDelta)
const reasoningDetailsObserved = state.reasoningDetailsObserved || detailDelta !== undefined
const deltaMetadata = reasoningMetadata(reasoningField)
const text = detailDelta?.length ? (detailText(detailDelta) ?? reasoning?.text) : reasoning?.text
if (!state.lifecycle.text.has("text-0") && text !== undefined)
lifecycle = Lifecycle.reasoningDelta(lifecycle, events, "reasoning-0", text, deltaMetadata)
if (text !== undefined) lifecycle = Lifecycle.reasoningDelta(lifecycle, events, "reasoning-0", text, deltaMetadata)
else if (
reasoningDetailsObserved &&
!lifecycle.reasoning.has("reasoning-0") &&
@@ -812,14 +812,18 @@ const step = (state: ParserState, event: OpenAIChatEvent) =>
const finishEvents = (state: ParserState): ReadonlyArray<LLMEvent> => {
const events: LLMEvent[] = []
const hasToolCalls = state.toolCallEvents.length > 0
const toolCallEvents =
state.finishReason === undefined && Object.keys(state.tools).length > 0
? Effect.runSync(ToolStream.finishAll(ADAPTER, state.tools)).events
: state.toolCallEvents
const hasToolCalls = toolCallEvents.length > 0
const reason = state.finishReason
? {
...state.finishReason,
normalized:
state.finishReason.normalized === "stop" && hasToolCalls ? "tool-calls" : state.finishReason.normalized,
}
: undefined
: { normalized: hasToolCalls ? ("tool-calls" as const) : ("unknown" as const) }
const metadata = reasoningMetadata(
state.reasoningField,
state.reasoningDetailsObserved ? state.reasoningDetails : undefined,
@@ -829,9 +833,9 @@ const finishEvents = (state: ParserState): ReadonlyArray<LLMEvent> => {
? Lifecycle.reasoningStart(state.lifecycle, events, "reasoning-0", reasoningMetadata(state.reasoningField))
: state.lifecycle
const ended = Lifecycle.reasoningEnd(started, events, "reasoning-0", metadata)
const lifecycle = state.toolCallEvents.length ? Lifecycle.stepStart(ended, events) : ended
events.push(...state.toolCallEvents)
if (reason) Lifecycle.finish(lifecycle, events, { reason, usage: state.usage })
const lifecycle = toolCallEvents.length ? Lifecycle.stepStart(ended, events) : ended
events.push(...toolCallEvents)
Lifecycle.finish(lifecycle, events, { reason, usage: state.usage })
return events
}
+48 -26
View File
@@ -102,6 +102,24 @@ describe("OpenAI Chat route", () => {
}),
)
it.effect("concatenates assistant text parts without adding separators", () =>
Effect.gen(function* () {
const prepared = yield* compileRequest(
LLM.request({
model,
messages: [
Message.assistant([
{ type: "text", text: "Hello" },
{ type: "text", text: " world" },
]),
],
}),
)
expect(prepared.body.messages).toEqual([{ role: "assistant", content: "Hello world" }])
}),
)
it.effect("writes reasoning to a configured custom field on every assistant message", () =>
Effect.gen(function* () {
const prepared = yield* compileRequest(
@@ -578,7 +596,7 @@ describe("OpenAI Chat route", () => {
}),
)
expect(prepared.body.messages).toEqual([{ role: "assistant", content: null, reasoning_content: "hidden" }])
expect(prepared.body.messages).toEqual([{ role: "assistant", content: "", reasoning_content: "hidden" }])
}),
)
@@ -827,7 +845,7 @@ describe("OpenAI Chat route", () => {
}),
)
it.effect("ignores scalar reasoning after content starts", () =>
it.effect("preserves scalar reasoning after content starts", () =>
Effect.gen(function* () {
const details = [{ type: "reasoning.text", text: "detail", format: "unknown", index: 0 }]
const response = yield* LLMClient.generate(request).pipe(
@@ -843,11 +861,11 @@ describe("OpenAI Chat route", () => {
),
)
expect(response.reasoning).toBe("detail")
expect(response.events.filter(LLMEvent.is.reasoningStart)).toHaveLength(1)
expect(response.events.filter(LLMEvent.is.reasoningEnd)).toHaveLength(1)
expect(response.reasoning).toBe("detailscalar")
expect(response.events.filter(LLMEvent.is.reasoningStart)).toHaveLength(2)
expect(response.events.filter(LLMEvent.is.reasoningEnd)).toHaveLength(2)
expect(response.message.content.find((part) => part.type === "reasoning")?.providerMetadata).toEqual({
openai: { reasoningDetails: details },
openai: { reasoningField: "reasoning", reasoningDetails: details },
})
}),
)
@@ -947,7 +965,7 @@ describe("OpenAI Chat route", () => {
expect(response.events.filter(LLMEvent.is.reasoningEnd)).toHaveLength(1)
const replay = yield* compileRequest(LLM.request({ model, messages: [response.message] }))
expect(replay.body.messages).toEqual([{ role: "assistant", content: null, reasoning_details: details }])
expect(replay.body.messages).toEqual([{ role: "assistant", content: "", reasoning_details: details }])
}),
)
@@ -997,7 +1015,7 @@ describe("OpenAI Chat route", () => {
)
expect(replay.body.messages).toEqual([
{ role: "assistant", content: null, reasoning: "firstsecond", reasoning_details: [first, second] },
{ role: "assistant", content: "", reasoning: "firstsecond", reasoning_details: [first, second] },
])
}),
)
@@ -1022,7 +1040,7 @@ describe("OpenAI Chat route", () => {
)
expect(replay.body.messages).toEqual([
{ role: "assistant", content: null, reasoning_content: "AB", reasoning_details: [detail] },
{ role: "assistant", content: "", reasoning_content: "AB", reasoning_details: [detail] },
])
}),
)
@@ -1044,7 +1062,7 @@ describe("OpenAI Chat route", () => {
)
expect(replay.body.messages).toEqual([
{ role: "assistant", content: null, reasoning_content: "thinking", reasoning_details: details },
{ role: "assistant", content: "", reasoning_content: "thinking", reasoning_details: details },
])
}),
)
@@ -1152,7 +1170,7 @@ describe("OpenAI Chat route", () => {
}),
)
it.effect("fails a streamed tool call when the provider ends without a finish reason", () =>
it.effect("finalizes a streamed tool call when the provider ends without a finish reason", () =>
Effect.gen(function* () {
const body = sseEvents(
deltaChunk({
@@ -1164,27 +1182,31 @@ describe("OpenAI Chat route", () => {
const input = LLMRequest.update(request, {
tools: [ToolDefinition.make({ name: "lookup", description: "Lookup data", inputSchema: { type: "object" } })],
})
const events: LLMEvent[] = []
const streamError = yield* LLMClient.stream(input).pipe(
Stream.runForEach((event) => Effect.sync(() => events.push(event))),
Effect.flip,
Effect.provide(fixedResponse(body)),
)
const error = yield* LLMClient.generate(input).pipe(Effect.provide(fixedResponse(body)), Effect.flip)
const response = yield* LLMClient.generate(input).pipe(Effect.provide(fixedResponse(body)))
expect(events).toEqual([
expect(response.events).toEqual([
{ type: "step-start", index: 0 },
{ type: "tool-input-start", id: "call_1", name: "lookup", providerMetadata: undefined },
{ type: "tool-input-delta", id: "call_1", name: "lookup", text: '{"query"' },
{ type: "tool-input-delta", id: "call_1", name: "lookup", text: ':"weather"}' },
{ type: "tool-input-end", id: "call_1", name: "lookup", providerMetadata: undefined },
{
type: "tool-call",
id: "call_1",
name: "lookup",
input: { query: "weather" },
providerExecuted: undefined,
providerMetadata: undefined,
},
{
type: "step-finish",
index: 0,
reason: { normalized: "tool-calls" },
usage: undefined,
providerMetadata: undefined,
},
{ type: "finish", reason: { normalized: "tool-calls" }, usage: undefined },
])
expect(events.filter(LLMEvent.is.toolCall)).toEqual([])
expect(streamError.reason).toMatchObject({
_tag: "InvalidProviderOutput",
classification: "incomplete-stream",
})
expect(streamError.message).toContain("The provider response ended unexpectedly.")
expect(error.message).toContain("The provider response ended unexpectedly.")
}),
)
+3 -1
View File
@@ -645,7 +645,9 @@ const layer = Layer.effect(
yield* execution.awaitIdle(input.sessionID)
const started = yield* Effect.gen(function* () {
const shell = yield* Shell.Service
return yield* shell.create({ command: input.command, cwd: session.location.directory, timeout: 0 })
return yield* shell
.create({ command: input.command, cwd: session.location.directory, timeout: 0 })
.pipe(Effect.orDie)
}).pipe(Effect.provide(locations.get(session.location)))
yield* bus.publish(
SessionEvent.Shell.Started,
+17 -12
View File
@@ -5,6 +5,7 @@ import { Context, Deferred, Duration, Effect, Fiber, Layer, Schema, Stream } fro
import { ChildProcess } from "effect/unstable/process"
import { produce } from "immer"
import { Shell } from "@opencode-ai/schema/shell"
import { AppProcess } from "@opencode-ai/util/process"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Config } from "./config"
import { Bus } from "./bus"
@@ -50,7 +51,7 @@ export interface Interface {
readonly create: <E = never, R = never>(
input: Shell.CreateInput,
before?: (input: ShellCreateBefore) => Effect.Effect<void, E, R>,
) => Effect.Effect<Shell.Info, E, R>
) => Effect.Effect<Shell.Info, E | AppProcess.AppProcessError, R>
// Currently running commands only; exited shells are retained for get/output but excluded here.
readonly list: () => Effect.Effect<Shell.Info[]>
readonly get: (id: Shell.ID) => Effect.Effect<Shell.Info, NotFoundError>
@@ -213,19 +214,23 @@ export const layer = (options?: ShellSelect.Options) =>
// Spawn through the Environment and stream combined output to the file. The handle is scope-bound, so
// the managing fiber keeps its scope open until the command terminates (it awaits `done` at the
// end). `create` returns once `ready` resolves with the registered session.
const ready = Deferred.makeUnsafe<Active>()
const ready = Deferred.makeUnsafe<Active, AppProcess.AppProcessError>()
runFork(
Effect.scoped(
Effect.gen(function* () {
const handle = yield* environment.spawner.spawn(
ChildProcess.make(invocation.shell, args, {
cwd: invocation.cwd,
env: invocation.env,
stdin: "ignore",
detached: process.platform !== "win32",
forceKillAfter: Duration.seconds(3),
}),
)
const handle = yield* environment.spawner
.spawn(
ChildProcess.make(invocation.shell, args, {
cwd: invocation.cwd,
env: invocation.env,
stdin: "ignore",
detached: process.platform !== "win32",
forceKillAfter: Duration.seconds(3),
}),
)
.pipe(
Effect.mapError((cause) => new AppProcess.AppProcessError({ command: invocation.command, cause })),
)
const session: Active = {
info: produce(info, (draft) => {
draft.pid = handle.pid
@@ -327,7 +332,7 @@ export const layer = (options?: ShellSelect.Options) =>
// release (kill) the process before its exit is observed.
yield* Deferred.await(session.done).pipe(Effect.catch(() => Effect.void))
}),
).pipe(Effect.catch(() => Effect.void)),
).pipe(Effect.catchTag("AppProcessError", (error) => Deferred.fail(ready, error))),
)
const session = yield* Deferred.await(ready)
+25
View File
@@ -286,6 +286,31 @@ describe("ShellTool", () => {
),
)
it.live(
"reports a command that fails to spawn",
() =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) => {
reset()
const command = "printf before\0after"
return withSession(tmp.path, (registry) =>
executeTool(registry, call({ command })).pipe(
Effect.andThen((settled) =>
Effect.sync(() => {
expect(settled.status).toBe("error")
if (settled.status !== "error") return
expect(settled.error?.message).toContain("Command failed")
}),
),
),
)
},
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]().then(() => undefined)),
),
{ timeout: 2_000 },
)
it.live("permissions compound commands separately", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
+3 -1
View File
@@ -21,7 +21,9 @@ export const ShellHandler = HttpApiBuilder.group(Api, "server.shell", (handlers)
Effect.fn(function* (ctx) {
const shell = yield* Shell.Service
const location = yield* Location.Service
return yield* response(shell.create({ ...ctx.payload, cwd: ctx.payload.cwd || location.directory }))
return yield* response(
shell.create({ ...ctx.payload, cwd: ctx.payload.cwd || location.directory }).pipe(Effect.orDie),
)
}),
)
.handle(
+7 -1
View File
@@ -258,7 +258,13 @@ const makeCrossSpawnSpawner = Effect.gen(function* () {
const spawn = (command: ChildProcess.StandardCommand, opts: NodeChildProcess.SpawnOptions) =>
Effect.callback<readonly [NodeChildProcess.ChildProcess, ExitSignal], PlatformError.PlatformError>((resume) => {
const signal = Deferred.makeUnsafe<readonly [code: number | null, signal: NodeJS.Signals | null]>()
const proc = launch(command.command, command.args, opts)
let proc: NodeChildProcess.ChildProcess
try {
proc = launch(command.command, command.args, opts)
} catch (err) {
resume(Effect.fail(toPlatformError("spawn", toError(err), command)))
return Effect.void
}
let end = false
let exit: readonly [code: number | null, signal: NodeJS.Signals | null] | undefined
proc.on("error", (err) => {