diff --git a/agent-langgraph/src/index.ts b/agent-langgraph/src/index.ts index 9b439a17..fc7ed28f 100644 --- a/agent-langgraph/src/index.ts +++ b/agent-langgraph/src/index.ts @@ -12,9 +12,9 @@ import { import { ChatOpenAI } from "@langchain/openai"; import { serve } from "bun"; import { hasManagedAgentToken } from "../../shared/agent-authorisation"; -import { textOfChunk } from "./deltas"; import { toLangChainMessages } from "./history"; import { readReasoningEffort } from "./model-options"; +import { streamRun } from "./stream"; /** * The same Bot, on a framework. @@ -332,9 +332,7 @@ function buildGraph(input: RunAgentInput) { return new StateGraph(MessagesAnnotation) .addNode("answer", async (state) => ({ - messages: [ - withVisibleReply((await bound.invoke(state.messages)) as AIMessage), - ], + messages: [await bound.invoke(state.messages)], })) .addNode("tools", async (state) => { const last = state.messages.at(-1) as AIMessage; @@ -386,37 +384,6 @@ function buildGraph(input: RunAgentInput) { .compile(); } -/** - * A reply with nothing in it ends the run in silence, so give it a line to end on. - * - * When a model returns no text and no tool call, the conditional edge sees no calls and returns END, - * and the person is left looking at a turn that produced no answer and no reason. Strict providers do - * this on a run they will not answer. Re-asking tends to get the same empty reply, so rather than - * loop, the run ends on a visible message saying what happened. Only a genuinely empty reply is - * touched: a reply with any text, or any tool call, is returned exactly as the model produced it. - */ -function withVisibleReply(reply: AIMessage): AIMessage { - const hasCall = (reply.tool_calls ?? []).length > 0; - if (hasCall || hasVisibleText(reply.content)) return reply; - return new AIMessage({ content: EMPTY_REPLY_FALLBACK }); -} - -function hasVisibleText(content: AIMessage["content"]): boolean { - if (typeof content === "string") return content.trim().length > 0; - if (Array.isArray(content)) { - return content.some((part) => - typeof part === "string" - ? part.trim().length > 0 - : typeof (part as { text?: unknown }).text === "string" && - (part as { text: string }).text.trim().length > 0, - ); - } - return false; -} - -const EMPTY_REPLY_FALLBACK = - "The model returned an empty reply and the run ended without an answer. This can happen with a strict provider; try asking again."; - async function runAgent(input: RunAgentInput): Promise { const encoder = new EventEncoder(); const stream = new ReadableStream({ @@ -431,169 +398,19 @@ async function runAgent(input: RunAgentInput): Promise { runId: input.runId, } as BaseEvent); - /* - * One message id per stretch of prose. - * - * A run is several turns now: the Bot may speak, call a tool, read the result and speak - * again. Reusing one id reopens a message the surface has already closed, and the second half - * of the answer is dropped. - */ - let messageIndex = 0; - let messageId = `msg_${input.runId}_0`; - let textOpen = false; - const closeText = () => { - if (!textOpen) return; - send({ type: "TEXT_MESSAGE_END", messageId } as BaseEvent); - textOpen = false; - messageIndex += 1; - messageId = `msg_${input.runId}_${messageIndex}`; - }; - - try { - const graph = buildGraph(input); - const events = await graph.streamEvents( - { messages: toLangChainMessages(input) }, - { version: "v2" }, - ); - - // Accumulated rather than emitted per chunk, because a tool call's arguments arrive in - // fragments and AG-UI wants one call. The framework hands back assembled `tool_calls` on the - // final message, which is precisely the plumbing agent-bot does by hand. - /** Calls seen on the way past, so a result can be paired with the arguments it answered. */ - const pending = new Map< - string, - { name: string; args: Record } - >(); - - for await (const event of events) { - if (event.event === "on_chat_model_stream") { - const chunk = event.data?.chunk as - | { content?: unknown } - | undefined; - /* - * Both content shapes, because the API decides which one arrives. - * - * Chat completions streams a string. The Responses API streams content blocks, so - * reading only the string shape dropped every delta and the run finished having said - * nothing — the "no text at all on gpt-5.6-*" this repository documents in - * `.env.example` and `docker-compose.yml`. - */ - const text = textOfChunk(chunk?.content); - if (!text) continue; - - if (!textOpen) { - send({ - type: "TEXT_MESSAGE_START", - messageId, - role: "assistant", - } as BaseEvent); - textOpen = true; - } - send({ - type: "TEXT_MESSAGE_CONTENT", - messageId, - delta: text, - } as BaseEvent); - } - - if (event.event === "on_chat_model_end") { - const output = event.data?.output as AIMessage | undefined; - if (output) { - for (const call of output.tool_calls ?? []) { - pending.set(call.id ?? call.name, { - name: call.name, - args: (call.args ?? {}) as Record, - }); - } - } - } - - /* - * The tools node finished. Reported here, in order, rather than collected for the end: the - * surface draws a conversation, and a call arriving after the answer it informed reads as - * though the Bot spoke first and did the work afterwards. - */ - if (event.event === "on_chain_end" && event.name === "tools") { - const output = event.data?.output as - | { messages?: { tool_call_id?: string; content?: unknown }[] } - | undefined; - // Prose and tool calls cannot interleave inside one message. - closeText(); - for (const message of output?.messages ?? []) { - const id = message.tool_call_id ?? ""; - const call = pending.get(id); - if (!call) continue; - send({ - type: "TOOL_CALL_START", - toolCallId: id, - toolCallName: call.name, - } as BaseEvent); - send({ - type: "TOOL_CALL_ARGS", - toolCallId: id, - delta: JSON.stringify(call.args), - } as BaseEvent); - send({ type: "TOOL_CALL_END", toolCallId: id } as BaseEvent); - send({ - type: "TOOL_CALL_RESULT", - messageId: `${id}-result`, - toolCallId: id, - content: String(message.content ?? ""), - role: "tool", - } as BaseEvent); - pending.delete(id); - } - } - } - - closeText(); - - /* - * Calls this process did not run, which is what a tool the surface owns looks like from here. - * - * The graph ends the run on one of those rather than inventing a result, so the `tools` node - * never fires and the loop above never reports the call. Without this the run is a clean - * RUN_STARTED/RUN_FINISHED pair carrying nothing at all: the person's message sits there with - * no answer under it, the surface never learns there was a browser action to execute, and - * because an empty run is not an error by the protocol, nothing says so. No result is sent - * with them; producing it is the surface's half, and it begins the next run holding it. - */ - for (const [id, call] of pending) { - send({ - type: "TOOL_CALL_START", - toolCallId: id, - toolCallName: call.name, - } as BaseEvent); - send({ - type: "TOOL_CALL_ARGS", - toolCallId: id, - delta: JSON.stringify(call.args), - } as BaseEvent); - send({ type: "TOOL_CALL_END", toolCallId: id } as BaseEvent); - } - pending.clear(); + // The graph is built and its event stream opened inside `streamRun`, so a failure doing either + // is reported as RUN_ERROR through the same path as a failure mid-stream. + await streamRun( + async () => + buildGraph(input).streamEvents( + { messages: toLangChainMessages(input) }, + { version: "v2" }, + ), + input, + send, + ); - send({ - type: "RUN_FINISHED", - threadId: input.threadId, - runId: input.runId, - } as BaseEvent); - } catch (error) { - // A text message left open would strand the surface mid-message, so it is closed before the - // error is reported. agent-bot has the same hazard and the same ordering. - if (textOpen) { - send({ type: "TEXT_MESSAGE_END", messageId } as BaseEvent); - } - send({ - type: "RUN_ERROR", - message: - error instanceof Error - ? error.message - : "The Bot could not answer.", - } as BaseEvent); - } finally { - controller.close(); - } + controller.close(); }, }); diff --git a/agent-langgraph/src/stream.ts b/agent-langgraph/src/stream.ts new file mode 100644 index 00000000..dbc9eb7d --- /dev/null +++ b/agent-langgraph/src/stream.ts @@ -0,0 +1,237 @@ +/** + * One run's framework events, read onto the AG-UI wire. + * + * Its own module for the reason `history.ts`, `deltas.ts` and `model-options.ts` are: `index.ts` + * calls `serve()` at module scope, so importing `runAgent` to reach this logic binds a port. The + * translation is where a run's shape is decided — when a message opens and closes, when a tool call + * is reported, and what a run that produced nothing a person can see ends on — which is exactly the + * part worth testing without a model. + */ +import type { BaseEvent, RunAgentInput } from "@ag-ui/core"; +import type { AIMessage } from "@langchain/core/messages"; +import { textOfChunk } from "./deltas"; + +/** + * The line a run ends on when it produced nothing a person can see. + * + * A reply with no text and no tool call ends the graph — the conditional edge sees no calls and + * stops — and without a line the person is left looking at a RUN_STARTED/RUN_FINISHED pair carrying + * nothing between them and no reason. Strict providers do this on a run they will not answer. + * + * It is emitted HERE, on the wire, and not by substituting a message into the graph's state. Only + * the model's own stream and the tool node reach this reader; a message a node returns is never + * streamed, so a fallback put into state is a fallback the surface never sees. "Visible" is decided + * by {@link textOfChunk}, the same rule the streamed deltas use, so a reply that is only a reasoning + * summary — text the person is never meant to see — counts as empty here too. + */ +export const EMPTY_REPLY_FALLBACK = + "The model returned an empty reply and the run ended without an answer. This can happen with a strict provider; try asking again."; + +/** The framework's streamed events, in the shape this reader looks at. */ +export interface RunStreamEvent { + event: string; + name?: string; + data?: { + chunk?: { content?: unknown }; + output?: unknown; + }; +} + +/** + * Translate one run's framework events into AG-UI events, each sent through `send`. + * + * The caller opens the run with RUN_STARTED and closes the stream; this fills the middle and ends it + * with RUN_FINISHED, or RUN_ERROR if anything threw. `makeEvents` is called inside the try so a + * failure building the graph or opening the stream is reported the same way as a failure mid-stream, + * which is the single try the inline version had. + */ +export async function streamRun( + makeEvents: () => Promise>, + input: Pick, + send: (event: BaseEvent) => void, +): Promise { + /* + * One message id per stretch of prose. + * + * A run is several turns now: the Bot may speak, call a tool, read the result and speak again. + * Reusing one id reopens a message the surface has already closed, and the second half of the + * answer is dropped. + */ + let messageIndex = 0; + let messageId = `msg_${input.runId}_0`; + let textOpen = false; + const closeText = () => { + if (!textOpen) return; + send({ type: "TEXT_MESSAGE_END", messageId } as BaseEvent); + textOpen = false; + messageIndex += 1; + messageId = `msg_${input.runId}_${messageIndex}`; + }; + + /* + * Whether anything a person can see reached the wire. + * + * A run that ends having sent neither a line of prose nor a tool call is the silent empty reply, + * and gets EMPTY_REPLY_FALLBACK rather than a bare RUN_FINISHED. A tool call counts because the + * surface has something to do with it — draw it, or put it to a person — even though this process + * sent no result with it. + */ + let sentVisibleText = false; + let sentToolCall = false; + + try { + const events = await makeEvents(); + + // Accumulated rather than emitted per chunk, because a tool call's arguments arrive in + // fragments and AG-UI wants one call. The framework hands back assembled `tool_calls` on the + // final message, which is precisely the plumbing agent-bot does by hand. + /** Calls seen on the way past, so a result can be paired with the arguments it answered. */ + const pending = new Map< + string, + { name: string; args: Record } + >(); + + for await (const event of events) { + if (event.event === "on_chat_model_stream") { + /* + * Both content shapes, because the API decides which one arrives. + * + * Chat completions streams a string. The Responses API streams content blocks, so reading + * only the string shape dropped every delta and the run finished having said nothing — the + * "no text at all on gpt-5.6-*" this repository documents in `.env.example` and + * `docker-compose.yml`. + */ + const text = textOfChunk(event.data?.chunk?.content); + if (!text) continue; + + if (!textOpen) { + send({ + type: "TEXT_MESSAGE_START", + messageId, + role: "assistant", + } as BaseEvent); + textOpen = true; + } + send({ + type: "TEXT_MESSAGE_CONTENT", + messageId, + delta: text, + } as BaseEvent); + sentVisibleText = true; + } + + if (event.event === "on_chat_model_end") { + const output = event.data?.output as AIMessage | undefined; + if (output) { + for (const call of output.tool_calls ?? []) { + pending.set(call.id ?? call.name, { + name: call.name, + args: (call.args ?? {}) as Record, + }); + } + } + } + + /* + * The tools node finished. Reported here, in order, rather than collected for the end: the + * surface draws a conversation, and a call arriving after the answer it informed reads as + * though the Bot spoke first and did the work afterwards. + */ + if (event.event === "on_chain_end" && event.name === "tools") { + const output = event.data?.output as + | { messages?: { tool_call_id?: string; content?: unknown }[] } + | undefined; + // Prose and tool calls cannot interleave inside one message. + closeText(); + for (const message of output?.messages ?? []) { + const id = message.tool_call_id ?? ""; + const call = pending.get(id); + if (!call) continue; + send({ + type: "TOOL_CALL_START", + toolCallId: id, + toolCallName: call.name, + } as BaseEvent); + send({ + type: "TOOL_CALL_ARGS", + toolCallId: id, + delta: JSON.stringify(call.args), + } as BaseEvent); + send({ type: "TOOL_CALL_END", toolCallId: id } as BaseEvent); + send({ + type: "TOOL_CALL_RESULT", + messageId: `${id}-result`, + toolCallId: id, + content: String(message.content ?? ""), + role: "tool", + } as BaseEvent); + sentToolCall = true; + pending.delete(id); + } + } + } + + closeText(); + + /* + * Calls this process did not run, which is what a tool the surface owns looks like from here. + * + * The graph ends the run on one of those rather than inventing a result, so the `tools` node + * never fires and the loop above never reports the call. Without this the run is a clean + * RUN_STARTED/RUN_FINISHED pair carrying nothing at all: the person's message sits there with + * no answer under it, the surface never learns there was a browser action to execute, and + * because an empty run is not an error by the protocol, nothing says so. No result is sent + * with them; producing it is the surface's half, and it begins the next run holding it. + */ + for (const [id, call] of pending) { + send({ + type: "TOOL_CALL_START", + toolCallId: id, + toolCallName: call.name, + } as BaseEvent); + send({ + type: "TOOL_CALL_ARGS", + toolCallId: id, + delta: JSON.stringify(call.args), + } as BaseEvent); + send({ type: "TOOL_CALL_END", toolCallId: id } as BaseEvent); + sentToolCall = true; + } + pending.clear(); + + /* + * Nothing a person can see was produced: no prose, and no call for the surface to run or draw. + * End on a visible line rather than a silent RUN_FINISHED, which is the whole point of noticing. + */ + if (!sentVisibleText && !sentToolCall) { + send({ + type: "TEXT_MESSAGE_START", + messageId, + role: "assistant", + } as BaseEvent); + send({ + type: "TEXT_MESSAGE_CONTENT", + messageId, + delta: EMPTY_REPLY_FALLBACK, + } as BaseEvent); + send({ type: "TEXT_MESSAGE_END", messageId } as BaseEvent); + } + + send({ + type: "RUN_FINISHED", + threadId: input.threadId, + runId: input.runId, + } as BaseEvent); + } catch (error) { + // A text message left open would strand the surface mid-message, so it is closed before the + // error is reported. agent-bot has the same hazard and the same ordering. + if (textOpen) { + send({ type: "TEXT_MESSAGE_END", messageId } as BaseEvent); + } + send({ + type: "RUN_ERROR", + message: + error instanceof Error ? error.message : "The Bot could not answer.", + } as BaseEvent); + } +} diff --git a/agent-langgraph/tests/stream.test.ts b/agent-langgraph/tests/stream.test.ts new file mode 100644 index 00000000..5b749767 --- /dev/null +++ b/agent-langgraph/tests/stream.test.ts @@ -0,0 +1,249 @@ +import { describe, expect, test } from "bun:test"; +import { + EMPTY_REPLY_FALLBACK, + type RunStreamEvent, + streamRun, +} from "../src/stream"; + +/** + * The reader that turns a run's framework events into the AG-UI wire. + * + * Driven with hand-built event streams rather than a real graph, because the thing worth pinning is + * the translation: which AG-UI events a given sequence of framework events produces, and — the case + * this suite exists for — that a run which produced nothing a person can see still ends on a visible + * line rather than a silent RUN_FINISHED. The empty-reply guard used to live in the graph's state, + * where this reader never looks, so it never reached the surface; the assertions below are on what + * the surface actually receives. + */ + +const RUN = { runId: "run_1", threadId: "thread_1" }; + +/** Collect what streamRun sends, as plain `{type, ...}` records. */ +async function collect( + events: RunStreamEvent[] | (() => Promise>), +): Promise>> { + const sent: Array> = []; + const makeEvents = + typeof events === "function" + ? events + : async () => + (async function* () { + for (const event of events) yield event; + })(); + await streamRun(makeEvents, RUN, (event) => + sent.push(event as unknown as Record), + ); + return sent; +} + +const types = (sent: Array>) => + sent.map((event) => event.type); + +const modelStream = (content: unknown): RunStreamEvent => ({ + event: "on_chat_model_stream", + data: { chunk: { content } }, +}); + +describe("a run that produces nothing a person can see", () => { + test("an empty reply ends on the fallback line, not a silent RUN_FINISHED", async () => { + // No stream deltas, no tool calls: the shape a strict provider returns on a run it will not + // answer, and the shape the graph ends immediately. + const sent = await collect([]); + + expect(types(sent)).toEqual([ + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_CONTENT", + "TEXT_MESSAGE_END", + "RUN_FINISHED", + ]); + expect(sent[1]).toMatchObject({ delta: EMPTY_REPLY_FALLBACK }); + // The message the fallback is drawn as has to be a real, closed message the surface can render. + expect(sent[0]).toMatchObject({ + role: "assistant", + messageId: "msg_run_1_0", + }); + }); + + test("a reply that is only a reasoning summary counts as empty", async () => { + // The Responses API puts a reasoning model's private working in a content block with a `text` + // field under a non-text type. It is never shown, so a reply carrying only that is silent to the + // person and must reach the same fallback — the case the old state-based guard got wrong, because + // it read any `text` field as visible. + const sent = await collect([ + modelStream([{ type: "reasoning", text: "thinking it over" }]), + ]); + + expect(types(sent)).toEqual([ + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_CONTENT", + "TEXT_MESSAGE_END", + "RUN_FINISHED", + ]); + expect(sent[1]).toMatchObject({ delta: EMPTY_REPLY_FALLBACK }); + }); +}); + +describe("a run that does produce something is left alone", () => { + test("streamed prose is forwarded and no fallback is added", async () => { + const sent = await collect([modelStream("Hel"), modelStream("lo")]); + + expect(types(sent)).toEqual([ + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_CONTENT", + "TEXT_MESSAGE_CONTENT", + "TEXT_MESSAGE_END", + "RUN_FINISHED", + ]); + expect(sent[1]).toMatchObject({ delta: "Hel" }); + expect(sent[2]).toMatchObject({ delta: "lo" }); + // The fallback text appears nowhere. + expect(sent.some((event) => event.delta === EMPTY_REPLY_FALLBACK)).toBe( + false, + ); + }); + + test("responses-API content blocks are forwarded as text", async () => { + const sent = await collect([ + modelStream([ + { type: "reasoning", text: "private" }, + { type: "text", text: "shown" }, + ]), + ]); + + expect(types(sent)).toContain("TEXT_MESSAGE_CONTENT"); + expect(sent.find((e) => e.type === "TEXT_MESSAGE_CONTENT")).toMatchObject({ + delta: "shown", + }); + expect(sent.some((event) => event.delta === EMPTY_REPLY_FALLBACK)).toBe( + false, + ); + }); + + test("a tool the surface owns is reported with no result and no fallback", async () => { + // The model asked for a browser-side tool; the graph ends the run without running it, so the + // tools node never fires and the call is drained at the end with no result. It is still a thing + // the surface has to do, so the run is not silent and gets no fallback. + const sent = await collect([ + { + event: "on_chat_model_end", + data: { + output: { + tool_calls: [ + { id: "call_1", name: "show_chart", args: { kind: "bar" } }, + ], + }, + }, + }, + ]); + + expect(types(sent)).toEqual([ + "TOOL_CALL_START", + "TOOL_CALL_ARGS", + "TOOL_CALL_END", + "RUN_FINISHED", + ]); + expect(sent[0]).toMatchObject({ + toolCallId: "call_1", + toolCallName: "show_chart", + }); + expect(sent[1]).toMatchObject({ delta: JSON.stringify({ kind: "bar" }) }); + // A drained surface call carries no result — that is the surface's half. + expect(types(sent)).not.toContain("TOOL_CALL_RESULT"); + }); + + test("a tool this deployment ran is reported with its result and no fallback", async () => { + const sent = await collect([ + { + event: "on_chat_model_end", + data: { + output: { + tool_calls: [{ id: "call_2", name: "search", args: { q: "cats" } }], + }, + }, + }, + { + event: "on_chain_end", + name: "tools", + data: { + output: { + messages: [{ tool_call_id: "call_2", content: "found 3" }], + }, + }, + }, + ]); + + expect(types(sent)).toEqual([ + "TOOL_CALL_START", + "TOOL_CALL_ARGS", + "TOOL_CALL_END", + "TOOL_CALL_RESULT", + "RUN_FINISHED", + ]); + expect(sent[3]).toMatchObject({ toolCallId: "call_2", content: "found 3" }); + expect(sent.some((event) => event.delta === EMPTY_REPLY_FALLBACK)).toBe( + false, + ); + }); +}); + +describe("message boundaries across a multi-turn run", () => { + test("prose after a tool result opens a new message id", async () => { + const sent = await collect([ + modelStream("Let me check"), + { + event: "on_chat_model_end", + data: { + output: { tool_calls: [{ id: "call_3", name: "search", args: {} }] }, + }, + }, + { + event: "on_chain_end", + name: "tools", + data: { + output: { messages: [{ tool_call_id: "call_3", content: "done" }] }, + }, + }, + modelStream("Here it is"), + ]); + + const firstText = sent.find((e) => e.type === "TEXT_MESSAGE_START"); + const lastText = [...sent] + .reverse() + .find((e) => e.type === "TEXT_MESSAGE_START"); + expect(firstText).toMatchObject({ messageId: "msg_run_1_0" }); + // Reusing one id reopens a message the surface already closed and drops the second half; the + // close between turns has to advance it. + expect(lastText).toMatchObject({ messageId: "msg_run_1_1" }); + expect(firstText?.messageId).not.toBe(lastText?.messageId); + }); +}); + +describe("a failure mid-stream", () => { + test("an open message is closed and the run ends on RUN_ERROR", async () => { + const sent = await collect(async () => + (async function* () { + yield modelStream("half a sentence"); + throw new Error("provider hung up"); + })(), + ); + + expect(types(sent)).toEqual([ + "TEXT_MESSAGE_START", + "TEXT_MESSAGE_CONTENT", + "TEXT_MESSAGE_END", + "RUN_ERROR", + ]); + expect(sent.at(-1)).toMatchObject({ message: "provider hung up" }); + // A run that errored is not also finished. + expect(types(sent)).not.toContain("RUN_FINISHED"); + }); + + test("a failure opening the stream is reported as RUN_ERROR", async () => { + const sent = await collect(async () => { + throw new Error("could not build graph"); + }); + + expect(types(sent)).toEqual(["RUN_ERROR"]); + expect(sent[0]).toMatchObject({ message: "could not build graph" }); + }); +});