End an empty reply on a line the surface receives, not one only the graph sees (#292)

A reply with no text and no tool call ends the graph — the conditional edge
sees no calls and stops. #289 tried to give that run a visible line by
substituting a fallback AIMessage into the graph's state (withVisibleReply),
so the CHANGELOG already promises "an empty reply ends on a visible line
rather than in silence".

It never reached the person. This service streams AG-UI by reading the run's
framework events, and it emits text only from on_chat_model_stream (the model's
own deltas) and tool events from the tools node. A message a graph node
returns is never one of those, so a fallback placed in state is a fallback the
surface never sees: an empty reply still ended on a bare RUN_STARTED/
RUN_FINISHED pair with nothing between them.

Proof, driving the real reader over a real graph with a fake empty model:
the graph's final state held the fallback, and the client received only
["RUN_FINISHED"].

The guard belongs on the wire, where the surface reads. streamRun now tracks
whether anything a person can see reached it — a line of prose or a tool call —
and ends a run that produced neither on the same fallback line. "Visible" is
decided by textOfChunk, the one rule the streamed deltas already use, so a
reply that is only a Responses-API reasoning summary (text the person is never
shown) counts as empty here too — the case the state-based guard got wrong,
since hasVisibleText read any `text` field as visible.

The translation is extracted to stream.ts for the same reason history.ts and
deltas.ts are: index.ts calls serve() at module scope, so importing runAgent
to test it binds a port. stream.ts imports no runtime module beyond deltas, so
its tests need no provider key and no network.

Co-authored-by: kevin9327 <kevin9327@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: kevin9327

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<Response> {
   const encoder = new EventEncoder();
   const stream = new ReadableStream<Uint8Array>({
@@ -431,169 +398,19 @@ async function runAgent(input: RunAgentInput): Promise<Response> {
         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<string, unknown> }
-        >();
-
-        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<string, unknown>,
-                });
-              }
-            }
-          }
-
-          /*
-           * 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();
     },
   });