diff --git a/packages/opencode/src/session/llm-native-events.ts b/packages/opencode/src/session/llm-native-events.ts index 0003c2a2a7..34dc5f02c8 100644 --- a/packages/opencode/src/session/llm-native-events.ts +++ b/packages/opencode/src/session/llm-native-events.ts @@ -35,6 +35,37 @@ const stringifyResult = (result: ToolResultValue) => { return JSON.stringify(result.value) } +// Recognize the opencode `Tool.ExecuteResult` shape inside a `tool-result` +// event's `result.value`. Native-path tool dispatchers wrap their handler +// output in this shape so the AI-SDK-shaped session event carries the +// real `title`, `metadata`, and `output` fields rather than the JSON +// encoding of the whole record. Provider-executed tools (Anthropic +// `web_search` etc.) and synthetic results that don't follow the shape +// still go through `stringifyResult` below. +type ExecuteShape = { + readonly title?: unknown + readonly metadata?: unknown + readonly output?: unknown +} + +const isExecuteResult = (value: unknown): value is ExecuteShape => { + if (typeof value !== "object" || value === null || Array.isArray(value)) return false + const v = value as ExecuteShape + return typeof v.output === "string" +} + +const toolResultOutput = (result: ToolResultValue) => { + if (result.type !== "json" || !isExecuteResult(result.value)) { + return { title: "", metadata: {}, output: stringifyResult(result) } + } + const value = result.value + return { + title: typeof value.title === "string" ? value.title : "", + metadata: typeof value.metadata === "object" && value.metadata !== null ? (value.metadata as Record) : {}, + output: typeof value.output === "string" ? value.output : "", + } +} + const response = () => ({ id: "", timestamp: new Date(0), modelId: "" }) const finishReason = (reason: Extract["reason"]) => @@ -147,7 +178,7 @@ export const mapper = () => { toolCallId: event.id, toolName: event.name, input: state.toolInputs.get(event.id) ?? {}, - output: { title: "", metadata: {}, output: stringifyResult(event.result) }, + output: toolResultOutput(event.result), }, ] case "tool-error": diff --git a/packages/opencode/src/session/llm-native-tools.ts b/packages/opencode/src/session/llm-native-tools.ts new file mode 100644 index 0000000000..2e58197dd3 --- /dev/null +++ b/packages/opencode/src/session/llm-native-tools.ts @@ -0,0 +1,248 @@ +import { + LLM, + type LLMClient, + type LLMError, + type LLMEvent, + type LLMRequest, + type FinishReason, + type ContentPart, + type RequestExecutor, +} from "@opencode-ai/llm" +import { Cause, Deferred, Effect, FiberSet, Queue, Stream, type Scope } from "effect" +import type { Tool, ToolExecutionOptions } from "ai" + +// Maximum number of model rounds before the streaming-dispatch loop stops. +// Mirrors `ToolRuntime.run`'s default; tweak via `maxSteps` if a caller needs +// a different ceiling. +export const DEFAULT_MAX_STEPS = 10 + +// What we care about from the round's events to (a) decide whether to start +// another round and (b) build the continuation request's message history. +interface RoundState { + finishReason: FinishReason | undefined + // Echoed back as the next round's assistant message — text deltas merged + // into a single text part, reasoning deltas into a single reasoning part, + // tool calls appended in order. Provider-executed tool results are also + // appended here so the provider sees the full hosted-tool round-trip. + assistantContent: ContentPart[] + // Client-side tool dispatches. One entry per `tool-call` event we forked + // a handler for, populated when the handler completes. + toolResults: Array<{ id: string; name: string; result: unknown }> +} + +const appendStreamingText = (state: RoundState, type: "text" | "reasoning", text: string) => { + const last = state.assistantContent.at(-1) + if (last?.type === type) { + state.assistantContent[state.assistantContent.length - 1] = { ...last, text: `${last.text}${text}` } + return + } + state.assistantContent.push({ type, text }) +} + +const accumulate = (state: RoundState, event: LLMEvent) => { + if (event.type === "text-delta") return appendStreamingText(state, "text", event.text) + if (event.type === "reasoning-delta") return appendStreamingText(state, "reasoning", event.text) + if (event.type === "tool-call") { + state.assistantContent.push( + LLM.toolCall({ + id: event.id, + name: event.name, + input: event.input, + providerExecuted: event.providerExecuted, + }), + ) + return + } + if (event.type === "tool-result" && event.providerExecuted) { + state.assistantContent.push( + LLM.toolResult({ + id: event.id, + name: event.name, + result: event.result, + providerExecuted: true, + }), + ) + return + } + if (event.type === "request-finish") { + state.finishReason = event.reason + } +} + +// Dispatch a single client-side tool call. Returns the synthetic LLMEvent +// that should be injected back into the round's stream — either a +// `tool-result` (success) or `tool-error` (handler threw / unknown tool). +// Errors from the AI SDK execute handler are caught and turned into +// `tool-error` so the round survives and the model can self-correct on +// the next step. +const dispatchTool = ( + call: { readonly id: string; readonly name: string; readonly input: unknown }, + tools: Record, + abort: AbortSignal, +): Effect.Effect => + Effect.gen(function* () { + const tool = tools[call.name] + if (!tool || typeof tool.execute !== "function") { + return { + type: "tool-error", + id: call.id, + name: call.name, + message: `Unknown tool: ${call.name}`, + } satisfies LLMEvent + } + const options: ToolExecutionOptions = { + toolCallId: call.id, + messages: [], + abortSignal: abort, + } + return yield* Effect.tryPromise({ + try: () => Promise.resolve(tool.execute!(call.input as never, options)), + catch: (err) => err, + }).pipe( + Effect.map( + (result): LLMEvent => ({ + type: "tool-result", + id: call.id, + name: call.name, + result: { type: "json", value: result }, + }), + ), + Effect.catch( + (err): Effect.Effect => + Effect.succeed({ + type: "tool-error", + id: call.id, + name: call.name, + message: err instanceof Error ? err.message : String(err), + }), + ), + ) + }) + +// Drive one model round. Streams every LLM event in real time; each +// non-provider-executed `tool-call` event forks a dispatcher fiber that +// pushes the resulting `tool-result` (or `tool-error`) event back into the +// same stream as soon as the handler completes. The round ends when: +// 1. the LLM stream completes, AND +// 2. every forked dispatcher has finished. +// At that point the queue is closed (consumers see end-of-stream) and +// `done` resolves with the accumulated state so the multi-round driver can +// decide whether to recurse. +const runOneRound = ( + client: LLMClient, + request: LLMRequest, + tools: Record, + abort: AbortSignal, +): Effect.Effect< + { + readonly events: Stream.Stream + readonly done: Deferred.Deferred + }, + never, + Scope.Scope | RequestExecutor.Service +> => + Effect.gen(function* () { + const queue = yield* Queue.unbounded() + const fiberSet = yield* FiberSet.make() + const state: RoundState = { finishReason: undefined, assistantContent: [], toolResults: [] } + const done = yield* Deferred.make() + + yield* Effect.forkScoped( + Effect.gen(function* () { + yield* client.stream(request).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + accumulate(state, event) + yield* Queue.offer(queue, event) + if (event.type === "tool-call" && !event.providerExecuted) { + yield* FiberSet.run( + fiberSet, + dispatchTool(event, tools, abort).pipe( + Effect.flatMap((resultEvent) => + Effect.gen(function* () { + if (resultEvent.type === "tool-result") { + state.toolResults.push({ + id: resultEvent.id, + name: resultEvent.name, + result: (resultEvent.result as { readonly value: unknown }).value, + }) + } + yield* Queue.offer(queue, resultEvent) + }), + ), + ), + ) + } + }), + ), + Effect.catchCause((cause) => + Effect.gen(function* () { + yield* Queue.failCause(queue, cause) + yield* Deferred.succeed(done, state) + }), + ), + ) + yield* FiberSet.awaitEmpty(fiberSet) + yield* Queue.end(queue) + yield* Deferred.succeed(done, state) + }), + ) + + return { events: Stream.fromQueue(queue), done } + }) + +// Build the next round's `LLMRequest` by appending the assistant message that +// echoes everything the round produced (text, reasoning, tool calls, hosted +// tool results) plus a `tool` role message per dispatched result. Lowering +// of these LLM-shaped messages back to the provider wire format is handled +// inside the existing adapter `prepare` step. +const continuationRequest = (request: LLMRequest, state: RoundState): LLMRequest => { + const assistant = LLM.message({ role: "assistant", content: state.assistantContent }) + const toolMessages = state.toolResults.map((entry) => + LLM.toolMessage({ id: entry.id, name: entry.name, result: entry.result }), + ) + return LLM.updateRequest(request, { + messages: [...request.messages, assistant, ...toolMessages], + }) +} + +/** + * Run a multi-round model+tool stream with streaming dispatch within each + * round. As each `tool-call` event arrives, the matching AI SDK tool's + * `execute` runs in a forked fiber and its result is injected back into the + * stream as a synthetic `tool-result` event. This matches the AI SDK's + * `streamText` UX: long-running tools don't block subsequent tool-call + * streaming, and consumers see results land as they complete. + * + * Stops when the model finishes a round with anything other than + * `tool-calls`, when `maxSteps` is reached, or when the underlying scope is + * interrupted (e.g. via the abort signal). + */ +export const runWithTools = (input: { + readonly client: LLMClient + readonly request: LLMRequest + readonly tools: Record + readonly abort: AbortSignal + readonly maxSteps?: number +}): Stream.Stream => { + const maxSteps = input.maxSteps ?? DEFAULT_MAX_STEPS + const round = (request: LLMRequest, step: number): Stream.Stream => + Stream.unwrap( + Effect.gen(function* () { + const { events, done } = yield* runOneRound(input.client, request, input.tools, input.abort) + const continuation = Stream.unwrap( + Effect.gen(function* () { + const state = yield* Deferred.await(done) + if (state.finishReason !== "tool-calls") return Stream.empty + if (state.toolResults.length === 0) return Stream.empty + if (step + 1 >= maxSteps) return Stream.empty + return round(continuationRequest(request, state), step + 1) + }), + ) + return events.pipe(Stream.concat(continuation)) + }), + ) + return round(input.request, 0) +} + +export * as LLMNativeTools from "./llm-native-tools" diff --git a/packages/opencode/src/session/llm.ts b/packages/opencode/src/session/llm.ts index 49bb014327..988b5bbac7 100644 --- a/packages/opencode/src/session/llm.ts +++ b/packages/opencode/src/session/llm.ts @@ -39,6 +39,7 @@ import * as Option from "effect/Option" import * as OtelTracer from "@effect/opentelemetry/Tracer" import { LLMNative } from "./llm-native" import { LLMNativeEvents } from "./llm-native-events" +import { LLMNativeTools } from "./llm-native-tools" const log = Log.create({ service: "llm" }) export const OUTPUT_TOKEN_MAX = ProviderTransform.OUTPUT_TOKEN_MAX @@ -517,10 +518,30 @@ const live: Layer.Layer< // per-element, `map.flush()` emits the remaining `*-end` events for any // text/reasoning/tool-input parts left open at stream close. The flush // stream is built lazily (`Stream.unwrap(Effect.sync(...))`) so it - // observes the mapper's final state after `mapConcat` has consumed every + // observes the mapper's final state after `flatMap` has consumed every // upstream event. + // + // The upstream source is one of two paths: + // + // - When `nativeTools` is unset (zero-tool sessions), call the LLM + // client directly. One model round, single stream, no dispatch. + // - When `nativeTools` is set, hand both the request and the matching + // AI SDK `tools` record to `LLMNativeTools.runWithTools`, which + // drives the multi-round loop with streaming dispatch: each + // `tool-call` event forks a tool handler fiber, and the + // handler's result is injected back into the same stream as a + // synthetic `tool-result` event. Long-running tools don't block + // subsequent tool-call streaming. const map = LLMNativeEvents.mapper() - return nativeClient.stream(llmRequest).pipe( + const upstream = input.nativeTools && input.nativeTools.length > 0 + ? LLMNativeTools.runWithTools({ + client: nativeClient, + request: llmRequest, + tools: input.tools, + abort: input.abort, + }) + : nativeClient.stream(llmRequest) + return upstream.pipe( Stream.flatMap((event) => Stream.fromIterable(map.map(event))), Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))), Stream.provideService(RequestExecutor.Service, executor), diff --git a/packages/opencode/test/session/llm-native-stream.test.ts b/packages/opencode/test/session/llm-native-stream.test.ts index 5809abb503..44dce8ec70 100644 --- a/packages/opencode/test/session/llm-native-stream.test.ts +++ b/packages/opencode/test/session/llm-native-stream.test.ts @@ -10,12 +10,14 @@ import { ProviderPatch, RequestExecutor, } from "@opencode-ai/llm" -import { Effect, Layer, Schema, Stream } from "effect" +import { Effect, Layer, Ref, Schema, Stream } from "effect" import { HttpClient, HttpClientResponse } from "effect/unstable/http" +import { tool, jsonSchema } from "ai" import { ModelID, ProviderID } from "../../src/provider/schema" import { MessageID, PartID, SessionID } from "../../src/session/schema" import { LLMNative } from "../../src/session/llm-native" import { LLMNativeEvents } from "../../src/session/llm-native-events" +import { LLMNativeTools } from "../../src/session/llm-native-tools" import { ProviderTest } from "../fake/provider" import { testEffect } from "../lib/effect" import type { MessageV2 } from "../../src/session/message-v2" @@ -37,6 +39,30 @@ const fixedResponse = (body: BodyInit, init: ResponseInit = { headers: { "conten ), ) +// Scripted multi-response HTTP layer. Each request consumes the next body in +// order; the final body repeats if more requests arrive. Mirrors the +// `scriptedResponses` helper in `packages/llm/test/lib/http.ts`. +const scriptedResponses = (bodies: ReadonlyArray, init: ResponseInit = { headers: { "content-type": "text/event-stream" } }) => + RequestExecutor.layer.pipe( + Layer.provide( + Layer.unwrap( + Effect.gen(function* () { + const cursor = yield* Ref.make(0) + return Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.gen(function* () { + const index = yield* Ref.getAndUpdate(cursor, (n) => n + 1) + const body = bodies[index] ?? bodies[bodies.length - 1] + return HttpClientResponse.fromWeb(request, new Response(body, init)) + }), + ), + ) + }), + ), + ), + ) + // Encode an Anthropic SSE body. Each event becomes a `data:` line; the codec // also expects `event:` lines but the package's SSE framing only reads the // data field. @@ -150,6 +176,122 @@ describe("LLMNative stream wire-up (audit gap #4 phase 1)", () => { }), ) + // Phase 2 step 2b: drives the streaming-dispatch loop end-to-end. The + // scripted Anthropic backend replies in two rounds — round 1 is a tool + // call, round 2 is text after the tool result feeds back. Asserts that + // `runWithTools` (a) forks the AI SDK execute when the `tool-call` event + // arrives, (b) injects a synthetic `tool-result` event into the same + // stream, (c) issues a continuation request with the tool result in + // history, and (d) the stream concludes with the second-round text. + it.effect("dispatches a tool call mid-stream and continues the conversation", () => + Effect.gen(function* () { + const mdl = anthropicModel() + const lookupParameters = Schema.Struct({ + query: Schema.String.annotate({ description: "Search query" }), + }) + const lookupTool: Tool.Def = { + id: "lookup", + description: "Lookup project data", + parameters: lookupParameters, + execute: () => Effect.succeed({ title: "Weather lookup", metadata: {}, output: '{"forecast":"sunny"}' }), + } + + // AI SDK side: the same tool wrapped so `tool.execute(args, opts)` + // resolves with the same opencode `ExecuteResult` shape the live + // `prompt.ts:resolveTools` would produce. The dispatcher inside + // `runWithTools` calls this; the synthetic `tool-result` LLM event + // carries the result back into the stream. + const aiTool = tool({ + description: "Lookup project data", + inputSchema: jsonSchema({ + type: "object", + properties: { query: { type: "string", description: "Search query" } }, + required: ["query"], + }), + execute: async () => ({ + title: "Weather lookup", + metadata: {}, + output: '{"forecast":"sunny"}', + }), + }) + + const userID = MessageID.ascending() + const llmRequest = yield* LLMNative.request({ + id: "smoke-tool-loop", + provider: ProviderTest.info({ id: ProviderID.make("anthropic"), key: "anthropic-key" }, mdl), + model: mdl, + system: ["Be concise."], + messages: [userMessage(mdl, userID, [userPart(userID, "What is the weather?")])], + tools: [lookupTool], + }) + + // Round 1: model issues `lookup` tool call. + const round1 = sseBody([ + { type: "message_start", message: { usage: { input_tokens: 5 } } }, + { type: "content_block_start", index: 0, content_block: { type: "tool_use", id: "call_1", name: "lookup" } }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: '{"query"' } }, + { type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: ':"weather"}' } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "tool_use" }, usage: { output_tokens: 1 } }, + { type: "message_stop" }, + ]) + // Round 2: model replies with text after seeing the tool result. + const round2 = sseBody([ + { type: "message_start", message: { usage: { input_tokens: 12 } } }, + { type: "content_block_start", index: 0, content_block: { type: "text", text: "" } }, + { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "It is sunny." } }, + { type: "content_block_stop", index: 0 }, + { type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } }, + { type: "message_stop" }, + ]) + + const client = LLMClient.make({ adapters, patches: ProviderPatch.defaults }) + const map = LLMNativeEvents.mapper() + + const events = yield* LLMNativeTools.runWithTools({ + client, + request: llmRequest, + tools: { lookup: aiTool }, + abort: new AbortController().signal, + }).pipe( + Stream.flatMap((event) => Stream.fromIterable(map.map(event))), + Stream.concat(Stream.unwrap(Effect.sync(() => Stream.fromIterable(map.flush())))), + Stream.runCollect, + Effect.provide(scriptedResponses([round1, round2])), + ) + + const collected = Array.from(events) + + // Round 1: tool call streams, dispatcher fires, synthetic tool-result lands. + const toolCall = collected.find((event) => event.type === "tool-call") + expect(toolCall).toMatchObject({ + type: "tool-call", + toolCallId: "call_1", + toolName: "lookup", + input: { query: "weather" }, + }) + + const toolResult = collected.find((event) => event.type === "tool-result") + expect(toolResult).toMatchObject({ + type: "tool-result", + toolCallId: "call_1", + toolName: "lookup", + output: { title: "Weather lookup", output: '{"forecast":"sunny"}' }, + }) + + // Round 2: text-delta arrives after the tool result. + const round2Text = collected.find((event) => event.type === "text-delta") + expect(round2Text).toMatchObject({ type: "text-delta", text: "It is sunny." }) + + // Final finish should be `stop`, not `tool-calls` (tool loop terminated). + const finalFinish = [...collected].reverse().find((event) => event.type === "finish") + expect(finalFinish).toMatchObject({ finishReason: "stop" }) + + // No errors leaked through. + expect(collected.some((event) => event.type === "error")).toBe(false) + }), + ) + // Phase 2 step 2a: verifies a tool-bearing `nativeTools` array reaches the // wire as Anthropic `tools[]` blocks. The model in this fixture answers with // plain text instead of issuing a tool call (we don't yet have dispatch).