From 065c309ef0979d1168aff1776af3efffb818381c Mon Sep 17 00:00:00 2001 From: Jerel John Velarde Date: Wed, 19 Aug 2026 13:52:34 -0700 Subject: [PATCH 1/2] Notice when a Bot stops talking, instead of spinning forever A Bot is any AG-UI endpoint, so it is infrastructure this deployment does not run: it gets redeployed mid-answer, its own upstream times out, and it will happily hold a connection open and write nothing more. Nothing here noticed. The channel stayed busy, the composer stayed locked, and the only way out was to reload the page or restart the server, which is a poor thing to ask of somebody who is looking at a Bot that appears to be thinking and never was. The watch is on activity, not on duration. A turn may legitimately run for an hour while events keep arriving; a turn whose stream has produced nothing at all for the configured timeout is wedged. A duration limit would cap how much work a Bot is allowed to do, which nobody asked for. This caps how long a person is asked to watch a spinner, which is the actual complaint. It sits on the Bot's own response body rather than on the reply to the browser, because in Intelligence mode that reply is a JSON envelope with a fixed length and the AG-UI events reach the browser over a WebSocket to the gateway. The Bot's response is the stream that stalls and the only place a stall can be seen. The wrapper is a pass-through transform that counts chunks and does nothing else: it must not buffer, must not delay a chunk and must not read the bytes, because a watchdog that can misread a working run into a broken one is worse than the failure it was added for. On a stall it writes one RUN_ERROR into the same stream and closes it. Both surfaces already understand that event, so nobody downstream had to learn a new one; the sentence it carries names the Bot and says what happened. The channel now draws that sentence at the end of the transcript rather than above the composer, which is where the missing answer was going to be and where the person is already looking. No exemption exists for a frontend tool call, and none is needed. The run ends before the browser executes one: the Bot emits its tool calls and RUN_FINISHED, its stream closes, the browser runs the tool and a second run carries the result back. A browser tool that takes ten minutes holds no stream open. AGENT_STALL_TIMEOUT_MS configures it, and zero or an unset variable leaves every stream alone. A turn that is ended is a turn somebody loses, so an existing deployment does not acquire that behaviour without asking for it; .env.example ships two minutes, which is already what a Bot in this repository treats as the outer bound of a quiet connection. Each stall writes an agent.stream_stalled row, because one hung turn reads as a bad afternoon and the same Bot hanging twice a day for a month is a fact about an endpoint that only becomes visible when somebody can count it. --- .env.example | 17 + app/src/components/channels/channel-chat.tsx | 34 +- .../components/channels/chat-transcript.tsx | 47 ++- .../components/channels/conversation-view.tsx | 4 + app/src/routes/_authed/admin/audit.tsx | 2 + server/src/audit.ts | 14 + server/src/channels/stall-guard.ts | 357 ++++++++++++++++++ server/src/channels/turn-watchdog.ts | 183 +++++++++ server/src/config.ts | 35 ++ server/src/copilot.ts | 36 +- server/src/index.ts | 16 + server/tests/config.test.ts | 33 ++ server/tests/copilot.test.ts | 64 ++++ server/tests/stall-guard.test.ts | 216 +++++++++++ server/tests/turn-watchdog.test.ts | 190 ++++++++++ 15 files changed, 1224 insertions(+), 24 deletions(-) create mode 100644 server/src/channels/stall-guard.ts create mode 100644 server/src/channels/turn-watchdog.ts create mode 100644 server/tests/stall-guard.test.ts create mode 100644 server/tests/turn-watchdog.test.ts diff --git a/.env.example b/.env.example index 89d8a39..4a5563c 100644 --- a/.env.example +++ b/.env.example @@ -49,6 +49,23 @@ INTELLIGENCE_GATEWAY_WS_URL=wss://realtime.intelligence.copilotkit.ai INTELLIGENCE_API_KEY= COPILOTKIT_LICENSE_TOKEN= +# How long a Bot's stream may say nothing before this deployment gives up on the turn, in +# milliseconds. A Bot is any AG-UI endpoint, which means it will be redeployed mid-answer, its own +# upstream will time out, and it will sometimes accept a connection and then write nothing at all. +# Without this the channel stays busy, the composer stays locked, and the only way out is a reload. +# +# Silence, not duration. A turn may legitimately run for an hour while events keep arriving; what is +# measured here is the gap between them. Nothing about how long a Bot is allowed to work changes. +# +# Two minutes because that is already what a Bot in this repository treats as the outer bound of a +# quiet connection (`agent-bot` sets `idleTimeout: 120`), and because it is far longer than any real +# silence inside a run: the longest legitimate one is the wait for a model's first token, which is +# seconds. Browser tool calls do not need allowing for. They run between turns, not during one, the +# run ends before the browser executes the tool and a second run carries the result back. +# +# 0, or leaving this unset, switches the watchdog off. Nothing is watched and no turn is ever ended. +AGENT_STALL_TIMEOUT_MS=120000 + # Model key. Required by the proof-of-concept Bot, which speaks OpenAI's API directly, and by the # framework Bot unless you point it at another provider below. OPENAI_API_KEY= diff --git a/app/src/components/channels/channel-chat.tsx b/app/src/components/channels/channel-chat.tsx index 8f4546a..c586f6b 100644 --- a/app/src/components/channels/channel-chat.tsx +++ b/app/src/components/channels/channel-chat.tsx @@ -289,23 +289,12 @@ export function ChannelChat({ disabled={!channel.active} messages={transcriptMessages(agent.messages, seed)} notice={ - <> - {runError ? ( -

- {runError} -

- ) : null} - {channel.active ? null : ( -

- This coworker has been deleted. The conversation stays readable, - but it can no longer reply. -

- )} - + channel.active ? null : ( +

+ This coworker has been deleted. The conversation stays readable, + but it can no longer reply. +

+ ) } onSubmit={async (draft) => { // `draft.agentId` carries the @mentioned coworker, but nothing routes on it yet: this @@ -336,6 +325,17 @@ export function ChannelChat({ copilotkit.stopAgent({ agent }); }} pending={agent.isRunning} + /* + * At the END OF THE TRANSCRIPT rather than above the composer, which is where this used to + * be. A turn that ends without an answer leaves a gap exactly where the reply was going to + * appear, and the person is already looking at it; an explanation in the composer area is a + * different part of the screen from the thing it explains. + * + * `runError` carries whatever ended the turn, in that thing's own words. A Bot that stopped + * streaming says so, because the deployment's stall watchdog writes that sentence into the + * run before closing it; see server/src/channels/stall-guard.ts. + */ + stopped={runError ?? undefined} /> ); diff --git a/app/src/components/channels/chat-transcript.tsx b/app/src/components/channels/chat-transcript.tsx index 20d4c0c..33541f6 100644 --- a/app/src/components/channels/chat-transcript.tsx +++ b/app/src/components/channels/chat-transcript.tsx @@ -25,6 +25,14 @@ type ChatTranscriptProps = { /** Comma-separated `/` command names, used to tell a skill chip from a leading slash. */ commandNames?: string; messages: ReadonlyArray>; + /** + * Why the last turn ended without an answer, if it did. + * + * A sentence rather than a flag, because the reasons are not interchangeable: a Bot that refused, + * a Bot whose endpoint is down and a Bot that simply stopped talking are three different things to + * be told, and only the thing that ended the turn knows which one happened. + */ + stopped?: string; }; /** @@ -75,6 +83,31 @@ function Thinking() { ); } +/** + * The turn ended and no answer came. + * + * In the same slot as `Thinking`, and for the same reason it is there: the person is looking at the + * bottom of the transcript, immediately under their own message, because that is where the answer + * was going to appear. Saying so above the composer put the explanation in a different part of the + * screen from the gap it explains, and left the last thing in the conversation looking unfinished. + * + * NOT A MESSAGE, deliberately. It has no id, is never anchored, and is gone the moment the next turn + * starts. Making it a transcript row would put a sentence into the conversation that nobody said, + * and the conversation is sent back to the model on the next turn, so the Bot would then read its + * own obituary as something it had written. + */ +function Stopped({ reason }: { reason: string }) { + return ( +

+ {reason} +

+ ); +} + /** * How many of the newest turns cascade when a channel is opened, and how far apart. * @@ -341,6 +374,7 @@ export function ChatTranscript({ busy = false, commandNames = "", messages, + stopped, }: ChatTranscriptProps) { /* * NOT MEMOISED, AND THAT IS DELIBERATE. `useMemo` keyed on `messages` looks obviously right and @@ -431,11 +465,18 @@ export function ChatTranscript({ ), )} {/* - * Outside the item list, so it is not a message. It has no id, is never anchored, and - * disappears the moment the answer starts — giving it a `MessageScrollerItem` would ask + * Outside the item list, so neither of these is a message. Each has no id, is never + * anchored, and is gone by the next turn — giving one a `MessageScrollerItem` would ask * the scroller to measure and anchor something that exists for a second and a half. + * + * One or the other, never both: a turn that ended has stopped being in flight, and a + * shimmering "Thinking" under a line saying the Bot stopped would contradict it. */} - {waitingOnFirstToken ? : null} + {stopped ? ( + + ) : waitingOnFirstToken ? ( + + ) : null} diff --git a/app/src/components/channels/conversation-view.tsx b/app/src/components/channels/conversation-view.tsx index fcb5d6d..ea1c109 100644 --- a/app/src/components/channels/conversation-view.tsx +++ b/app/src/components/channels/conversation-view.tsx @@ -16,6 +16,7 @@ export function ConversationView({ commands, disabled = false, pending = false, + stopped, onSubmit, onStop, }: { @@ -30,6 +31,8 @@ export function ConversationView({ commands?: readonly CommandOption[]; disabled?: boolean; pending?: boolean; + /** Why the last turn ended without an answer. Drawn at the end of the transcript, not here. */ + stopped?: string; onSubmit: (draft: ComposerDraft) => void | Promise; /** Stop the Bot mid-answer; forwarded to turn the send button into a stop button. */ onStop?: () => void; @@ -51,6 +54,7 @@ export function ConversationView({ .map((command) => command.name) .join(",")} messages={messages} + {...(stopped ? { stopped } : {})} />
diff --git a/app/src/routes/_authed/admin/audit.tsx b/app/src/routes/_authed/admin/audit.tsx index d6f44e5..b5865fd 100644 --- a/app/src/routes/_authed/admin/audit.tsx +++ b/app/src/routes/_authed/admin/audit.tsx @@ -259,6 +259,8 @@ const NAMED_TARGETS = new Set([ const DECISIONS: Record = { "bot.declined": "The Bot declined", + // Not a refusal and not a failed action: the Bot said nothing and the turn was ended for it. + "agent.stream_stalled": "The Bot stopped responding", "computer.policy_loaded": "Boundary at start-up", "computer.isolation_loaded": "Isolation at start-up", "computer.control_taken": "A person took the wheel", diff --git a/server/src/audit.ts b/server/src/audit.ts index 0ff5794..d72250e 100644 --- a/server/src/audit.ts +++ b/server/src/audit.ts @@ -41,6 +41,20 @@ export const auditEventTypes = [ "connector.sync_failed", "knowledge.searched", "agent.invoked", + /** + * A Bot's stream stopped producing anything and the turn was ended for it. + * + * Recorded because a Bot is somebody else's infrastructure and this is the failure it has that + * nothing else in the trail can show. Every other row here is something that happened; this one is + * the absence of anything happening, which leaves no trace of its own. + * + * It is also the sort of thing nobody notices is happening repeatedly. One hung turn reads as a + * bad afternoon and gets a shrug; the same Bot hanging twice a day for a month is a fact about an + * endpoint, and it only becomes visible when somebody can count it. The row names the Bot, how + * long its stream was silent, and how many chunks it managed first, so a reader can tell an + * endpoint that dies mid-answer from one that never answers at all. + */ + "agent.stream_stalled", "mcp.call_succeeded", "mcp.call_rejected", // Every action a Bot takes on its computer, allowed or refused. Both, always: a trail that records diff --git a/server/src/channels/stall-guard.ts b/server/src/channels/stall-guard.ts new file mode 100644 index 0000000..e55e0c8 --- /dev/null +++ b/server/src/channels/stall-guard.ts @@ -0,0 +1,357 @@ +/** + * The watch on a Bot's own stream, and what happens when it stops saying anything. + * + * A Bot is any AG-UI endpoint, which makes it infrastructure this deployment does not run: it will + * be redeployed mid-run, its upstream will time out, it will accept a connection and then write + * nothing. None of that is exceptional. What was missing was anybody noticing: an open stream that + * never produces another byte leaves the channel busy, the composer locked and a person watching a + * Bot that appears to be thinking and is not. + * + * The seam is the Bot's own HTTP response, wrapped on the way in. In Intelligence mode this server's + * reply to the browser is a JSON envelope, not a stream, and the AG-UI events reach the browser over + * a WebSocket to the gateway (see the module comment on turn-watchdog.ts for what was measured). + * The Bot's response, however, is an ordinary streaming HTTP body that this process reads to the + * end, so it is both the thing that stalls and the only place a stall can be seen. + * + * A PASS-THROUGH TRANSFORM IS THE ONLY ACCEPTABLE SHAPE HERE, and the transform below does exactly + * one thing besides handing the chunk on. It must not buffer, because a Bot's answer is streamed a + * token at a time and holding chunks back to inspect them would turn a live answer into a paragraph + * that lands all at once. It must not await anything, because a chunk delayed here is a chunk + * delayed on somebody's screen. And it must not decide anything about the bytes, because the moment + * it parses them it can be wrong about them, and a watchdog that can misread a working run into a + * broken one is worse than the failure it was added for. Counting is the whole of its job. + * + * On a stall it writes one RUN_ERROR event into the same stream and closes it. RUN_ERROR is the + * event both surfaces already understand: the channel renders it through the subscriber it already + * has, and the packaged chat shows its message in its own banner, so nothing downstream needs to + * learn a new event to tell somebody what happened. AG-UI permits RUN_ERROR at any point in a + * stream, including as the very first event, which is what a Bot that never spoke produces. + */ +import { type AuditStore, recordAuditEvent } from "../audit"; +import { type StalledStream, TurnWatchdog } from "./turn-watchdog"; + +/** The fetch an `HttpAgent` uses, as @ag-ui/client 0.0.57 declares it. */ +export type AgentFetch = ( + url: string, + requestInit: RequestInit, +) => Promise; + +/** Which Bot a watched stream belongs to. The name is for the sentence a person reads. */ +export type WatchedBot = { id: string; name: string }; + +export type StallGuardOptions = { + /** Silence this long ends the turn. Zero or less leaves every stream untouched. */ + stallMs: number; + /** Absent leaves the trail without stall rows; the recovery still happens. */ + auditStore?: AuditStore; + now?: () => number; +}; + +export type StallGuard = { + /** + * Wrap a fetch so that every stream it opens for this Bot is watched. + * + * Returns the fetch it was given, unchanged, when the watchdog is off. A deployment that has not + * configured a timeout gets the code path it had before this existed, rather than a wrapper that + * happens never to fire. + */ + watch: (bot: WatchedBot, fetchImplementation?: AgentFetch) => AgentFetch; + /** Stop sweeping. For orderly shutdown and for tests that must not leave a timer behind. */ + stop: () => void; +}; + +/** + * Only a stream that says it is server-sent events gets an event written into it. + * + * This is deliberately stricter than the AG-UI client's own rule, which treats every content type + * except the protobuf one as SSE. Being stricter fails in the safe direction: a Bot answering in a + * framing this does not recognise gets a clean close and the run ends with the client's own terminal + * event, which is a worse message than ours but is still an ending. Guessing the other way would + * mean writing SSE bytes into a binary stream and corrupting a run that was merely slow. + */ +const SSE_CONTENT_TYPE = "text/event-stream"; + +const ENCODER = new TextEncoder(); + +/** + * How often to look, derived from the deadline rather than fixed. + * + * A quarter of the timeout means a stall is reported within a quarter of the deadline of it + * happening, which is proportionate: nobody watching a two-minute limit cares about thirty seconds + * of reporting lag, and nobody testing a sixty-millisecond limit can wait a second for it. Bounded + * at both ends so a very long timeout still sweeps every second and a very short one does not spin. + */ +function sweepIntervalFor(stallMs: number): number { + return Math.min(1_000, Math.max(50, Math.floor(stallMs / 4))); +} + +/** What one watched stream needs in order to be ended from outside it. */ +type OpenStream = { + bot: WatchedBot; + writer: WritableStreamDefaultWriter; + /** Cancels the Bot's side, so a wedged endpoint does not keep a socket here forever. */ + cancelUpstream: () => void; + /** Whether an event may be written into this stream. See SSE_CONTENT_TYPE. */ + sse: boolean; + /** + * The request body, held as the string it was already serialised to. + * + * Read only when a stall has to name the turn it ended, so the thread and run identifiers cost a + * JSON parse on the rare path rather than on every run. Nothing else in the request is wanted, and + * nothing from it is recorded beyond those two ids. + */ + requestBody: string | null; + /** Set by whichever of the pump or the stall gets there first. */ + finished: boolean; +}; + +export function createStallGuard(options: StallGuardOptions): StallGuard { + const streams = new Map(); + const sweepEveryMs = sweepIntervalFor(options.stallMs); + let sweeper: ReturnType | undefined; + + const watchdog = new TurnWatchdog({ + stallMs: options.stallMs, + ...(options.now ? { now: options.now } : {}), + onStall: (stalled) => { + void giveUp(stalled); + }, + }); + + /** + * The sweep runs only while something is being watched. + * + * An idle deployment therefore holds no timer at all, and the one it holds while a Bot is + * answering is unreferenced, so it can never be the reason a process refuses to exit. + */ + function arm(): void { + if (sweeper !== undefined) return; + sweeper = setInterval(() => { + watchdog.sweep(); + if (watchdog.watching === 0) disarm(); + }, sweepEveryMs); + sweeper.unref(); + } + + function disarm(): void { + if (sweeper === undefined) return; + clearInterval(sweeper); + sweeper = undefined; + } + + /** + * Take a stream off the watch, exactly once. + * + * Both the pump and the stall want to be the one that finishes a stream, and which of them gets + * there first depends on timing nobody controls. Returning the stream only to the first caller is + * what stops a stalled stream also being closed by its own pump a moment later, and stops a + * cancelled read being reported as a stall. + */ + function release(id: string): OpenStream | undefined { + const stream = streams.get(id); + if (!stream || stream.finished) return undefined; + stream.finished = true; + streams.delete(id); + watchdog.close(id); + return stream; + } + + /** + * End a turn whose Bot has stopped talking. + * + * The row is written before the stream is touched. A wedged Bot may also have a consumer that has + * stopped reading, in which case the writes below never settle, and the record of what happened + * must not depend on a promise that a broken stream owes us. + */ + async function giveUp(stalled: StalledStream): Promise { + const stream = release(stalled.id); + if (!stream) return; + + const turn = turnOf(stream.requestBody); + console.error( + JSON.stringify({ + type: "agent-stream-stalled", + bot: stream.bot.id, + silentForMs: stalled.silentForMs, + chunks: stalled.chunks, + ...(turn ? { thread: turn.threadId, run: turn.runId } : {}), + note: "The Bot's stream produced nothing for the configured timeout, so the turn was ended.", + }), + ); + + if (options.auditStore) { + await recordAuditEvent(options.auditStore, { + eventType: "agent.stream_stalled", + targetType: "agent", + targetId: stream.bot.id, + payload: { + bot: stream.bot.id, + silentForMs: stalled.silentForMs, + // Chunks, not events. One chunk can carry several AG-UI events and the boundaries are the + // network's, so saying "events" here would be a number that looks precise and is not. Zero + // is the fact worth reading: the Bot answered the request and then said nothing at all. + chunks: stalled.chunks, + ...(turn ? { thread: turn.threadId, run: turn.runId } : {}), + }, + }).catch(() => undefined); + } + + // The Bot's side goes first, so a socket into an endpoint that will never answer is released + // whatever the browser's side of the stream is doing. + stream.cancelUpstream(); + + if (stream.sse) { + // Deliberately not awaited: see the note above about a consumer that has stopped reading. The + // write is queued ahead of the close, so it either lands in order or neither does. + void stream.writer + .write(stalledEvent(stream.bot.name, options.stallMs)) + .catch(() => undefined); + } + void stream.writer.close().catch(() => undefined); + } + + function watch( + bot: WatchedBot, + fetchImplementation?: AgentFetch, + ): AgentFetch { + const inner: AgentFetch = + fetchImplementation ?? ((url, requestInit) => fetch(url, requestInit)); + if (!watchdog.enabled) return inner; + + return async (url, requestInit) => { + const response = await inner(url, requestInit); + const body = response.body; + // A refusal or an empty body is not a stream and has already ended. Passing it through + // untouched keeps every error path exactly as it was, which is the point of the pass-through. + if (!response.ok || body === null) return response; + + const id = crypto.randomUUID(); + const relay = new TransformStream({ + transform(chunk, controller) { + watchdog.record(id); + controller.enqueue(chunk); + }, + }); + const reader = body.getReader(); + const writer = relay.writable.getWriter(); + + streams.set(id, { + bot, + writer, + cancelUpstream: () => { + void reader.cancel().catch(() => undefined); + }, + sse: (response.headers.get("content-type") ?? "").includes( + SSE_CONTENT_TYPE, + ), + requestBody: + typeof requestInit.body === "string" ? requestInit.body : null, + finished: false, + }); + watchdog.open({ id, botId: bot.id }); + arm(); + + void pump(id, reader, writer); + + // The same response, with the body replaced. Status and headers are carried over because the + // AG-UI client reads the content type off this response to choose its parser. + return new Response(relay.readable, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + }; + } + + /** + * Move bytes across, and end the stream the way the Bot ended it. + * + * A read that throws is passed on as an abort rather than as a clean close. A broken connection + * carries a real reason, and relabelling it as an ending would file a transport failure as a Bot + * that merely stopped, which is exactly the distinction this file exists to be able to draw. + */ + async function pump( + id: string, + reader: ReadableStreamDefaultReader, + writer: WritableStreamDefaultWriter, + ): Promise { + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + await writer.write(value); + } + if (release(id)) await writer.close().catch(() => undefined); + } catch (error) { + if (release(id)) await writer.abort(error).catch(() => undefined); + } + } + + return { + watch, + stop: disarm, + }; +} + +/** + * The turn, read from the request the stalled stream was answering. + * + * Returns null rather than throwing for anything unexpected. A stall is already a failure, and a + * malformed body is no reason to lose the row that records it; the Bot is named either way, and the + * Bot is what a deployment counts these by. + */ +function turnOf( + body: string | null, +): { threadId: string; runId: string } | null { + if (!body) return null; + try { + const parsed: unknown = JSON.parse(body); + if (!parsed || typeof parsed !== "object") return null; + const { threadId, runId } = parsed as { + threadId?: unknown; + runId?: unknown; + }; + return typeof threadId === "string" && typeof runId === "string" + ? { threadId, runId } + : null; + } catch { + return null; + } +} + +/** + * The one event a stalled stream gets, as a server-sent event. + * + * Written by hand rather than through an encoder because the framing is two lines and adding a + * dependency to produce them would be the larger change. The AG-UI client parses `data:` lines as + * JSON and validates them against its own schemas, so this is the same shape a Bot would have sent. + * + * The wording avoids " - " and a trailing three-digit number. The packaged chat's banner truncates a + * message at the first of those and strips the second, so a sentence containing either arrives at a + * person cut in half. + */ +function stalledEvent(botName: string, stallMs: number): Uint8Array { + const event = { + type: "RUN_ERROR", + message: + `${botName} stopped responding. Nothing arrived from it for ${inWords(stallMs)}, ` + + "so this turn was ended. Ask again, or check that the Bot is running.", + code: "AGENT_STREAM_STALLED", + }; + return ENCODER.encode(`data: ${JSON.stringify(event)}\n\n`); +} + +/** + * The timeout as a person would say it, because the sentence above is read by one. + * + * Floored at a second. A timeout below one is only ever a test's, and "nothing arrived from it for + * 0 seconds" is a sentence that makes a reader doubt everything else on the screen. + */ +function inWords(ms: number): string { + if (ms >= 60_000 && ms % 60_000 === 0) { + const minutes = ms / 60_000; + return minutes === 1 ? "a minute" : `${minutes} minutes`; + } + const seconds = Math.max(1, Math.round(ms / 1_000)); + return seconds === 1 ? "a second" : `${seconds} seconds`; +} diff --git a/server/src/channels/turn-watchdog.ts b/server/src/channels/turn-watchdog.ts new file mode 100644 index 0000000..9805f5e --- /dev/null +++ b/server/src/channels/turn-watchdog.ts @@ -0,0 +1,183 @@ +/** + * Whether a turn has gone quiet, which is not the same question as whether it has taken a long time. + * + * A turn may legitimately run for an hour. A Bot reading a long document, or a model writing slowly, + * produces events the whole way and there is nothing wrong with it. A turn whose stream has produced + * nothing at all for `stallMs` is wedged. Telling those apart by activity rather than by duration is + * the whole design, and the difference matters: a duration limit puts a ceiling on how much work a + * Bot is allowed to do, which nobody asked for, whereas a silence limit puts a ceiling on how long a + * person is asked to watch a spinner, which is the actual complaint. + * + * WHAT THE STREAM ACTUALLY DOES IN THIS PRODUCT, measured before any of this was designed, because + * the answer decides whether a long tool call would be reported as a stall. + * + * OpenBot runs in Intelligence mode and has no other mode, so `POST /api/copilotkit/agent/:id/run` + * does not stream at all. It answers `Content-Type: application/json` with a fixed Content-Length in + * about a second, carrying a join token, and the AG-UI events reach the browser over a WebSocket to + * the Intelligence gateway. Wrapping that response body would watch a JSON envelope go past. The + * stream that can actually stall is the one on the other side of this server: the Bot's own AG-UI + * response, read here, event by event, and republished to the gateway. That is the stream this + * watchdog is pointed at, and stall-guard.ts is where it is wrapped. + * + * The computer tools are frontend tools, executed in the browser, and the run ENDS before the + * browser runs one. The Bot emits TOOL_CALL_START, ARGS and END and then RUN_FINISHED, its stream + * closes, the browser executes the tool, and the CopilotKit client opens a SECOND run carrying the + * result. Verified in @copilotkit/core 1.67.1, whose `processAgentResult` executes frontend tools + * only after `agent.runAgent` has resolved and then re-enters `runAgent`, and against `agent-bot` in + * this repository, which writes its tool calls followed by RUN_FINISHED and closes the stream. + * + * So no exemption for tool calls exists here, and none is needed. A browser tool that takes ten + * minutes holds no stream open, because there is no stream open while it runs. An exemption would be + * a hole in the only thing this watchdog does, opened for a case that does not occur. + * + * It holds no timer of its own. The clock is injected and `sweep` is called by whoever owns the + * schedule, so a test can move time by a minute without waiting one, and so the decision about when + * to look is made in one place rather than hidden in here. + */ + +/** Milliseconds, from whatever source the caller trusts. Injected so tests need not sleep. */ +export type Clock = () => number; + +/** One stream under watch. */ +export type WatchedStream = { + /** + * How the caller finds this stream again. Minted per stream rather than taken from the run, + * because the id has to exist before the first byte arrives and the run identifiers are inside the + * request the stream is answering. + */ + id: string; + /** The Bot on the far end, so a stall names what went quiet rather than which socket it was. */ + botId: string; +}; + +export type StalledStream = WatchedStream & { + /** How long it had been silent when it was given up on. Never less than `stallMs`. */ + silentForMs: number; + /** Chunks that arrived before the silence. Zero means the Bot never said anything at all. */ + chunks: number; +}; + +export type TurnWatchdogOptions = { + /** + * Silence this long means the stream is wedged. + * + * Zero, or anything below it, switches the watchdog off completely: `open` records nothing and + * `sweep` has nothing to find. A deployment that has not asked for a watchdog gets the behaviour + * it had before there was one, rather than a watchdog with a surprising number in it. + */ + stallMs: number; + /** Called once per stalled stream, after the stream has been taken off the watch. */ + onStall: (stream: StalledStream) => void; + now?: Clock; +}; + +type OpenStream = WatchedStream & { + lastChunkAt: number; + chunks: number; +}; + +export class TurnWatchdog { + private readonly stallMs: number; + private readonly onStall: (stream: StalledStream) => void; + private readonly now: Clock; + private readonly streams = new Map(); + + constructor(options: TurnWatchdogOptions) { + this.stallMs = options.stallMs; + this.onStall = options.onStall; + this.now = options.now ?? (() => Date.now()); + } + + /** Whether this deployment asked for a watchdog at all. */ + get enabled(): boolean { + return this.stallMs > 0; + } + + /** How many streams are currently being watched. */ + get watching(): number { + return this.streams.size; + } + + /** + * Start watching a stream. + * + * The clock starts here rather than at the first chunk, so a Bot that accepts a connection and + * then says nothing at all is caught by the same rule as one that stops halfway. That case is the + * common one: an endpoint that has been redeployed answers the request and never writes. + * + * Opening an id that is already open restarts its clock rather than raising. A second open for the + * same stream can only mean the caller believes it is alive, and the safe reading of that is to + * believe them. + */ + open(stream: WatchedStream): void { + if (!this.enabled) return; + this.streams.set(stream.id, { + ...stream, + lastChunkAt: this.now(), + chunks: 0, + }); + } + + /** + * Note that something arrived. + * + * An id that is not being watched is ignored rather than started, and that is what makes the + * callback fire exactly once per stream: `sweep` removes a stream as it reports it, so a chunk + * that arrives from a Bot after it was given up on cannot quietly begin a second watch which would + * stall again and report the same wedged turn twice. + */ + record(id: string): void { + const stream = this.streams.get(id); + if (!stream) return; + stream.lastChunkAt = this.now(); + stream.chunks += 1; + } + + /** Stop watching. Unknown ids are ignored, so closing a stream that already stalled is harmless. */ + close(id: string): void { + this.streams.delete(id); + } + + /** + * Report every stream that has gone quiet, and returns how many there were. + * + * Each one is removed before its callback runs, so a callback that throws still leaves the + * watchdog consistent, and a slow callback cannot be entered twice for the same stream. + */ + sweep(): number { + if (!this.enabled) return 0; + const now = this.now(); + let stalled = 0; + + for (const [id, stream] of this.streams) { + const silentForMs = now - stream.lastChunkAt; + if (silentForMs < this.stallMs) continue; + + this.streams.delete(id); + stalled += 1; + try { + this.onStall({ + id, + botId: stream.botId, + silentForMs, + chunks: stream.chunks, + }); + } catch (error) { + // One Bot's failure must not stop the others being closed. This is the same blast-radius + // argument the process-level rejection handler in index.ts makes: a wedged Bot is somebody + // else's infrastructure, and the sweep it happens to be in is holding every other person's + // stuck turn. Logged loudly, because a watchdog that fails silently is worse than none. + console.error( + JSON.stringify({ + type: "turn-watchdog-callback-error", + stream: id, + bot: stream.botId, + error: String(error), + }), + ); + } + } + + return stalled; + } +} diff --git a/server/src/config.ts b/server/src/config.ts index 7d2c1a7..9e9c2c8 100644 --- a/server/src/config.ts +++ b/server/src/config.ts @@ -34,6 +34,15 @@ export type DeploymentConfig = { deploymentId: string | undefined; tenantPackageDirectory: string; runtime: RuntimeCapabilities; + /** + * How long a Bot's stream may say nothing before this deployment ends the turn, in milliseconds. + * + * Zero means no watchdog, and an unset variable means zero. A turn that is ended is a turn + * somebody loses, so a deployment that has not said it wants that gets the behaviour it already + * had. `.env.example` ships a value, so a new clone starts with the watch on and an upgraded + * deployment does not acquire it without being asked. + */ + agentStallTimeoutMs: number; oauth: { google?: { clientId: string; clientSecret: string }; }; @@ -314,6 +323,31 @@ function actionPolicy(environment: Environment): ActionPolicy | undefined { return result.policy; } +/** + * How long silence on a Bot's stream is allowed to last. + * + * Refuses to start on anything that is not a whole number of milliseconds, rather than falling back + * to the default. Same reasoning as the action policy above it: an operator who meant to write a + * two-minute timeout and typed something else would otherwise get a running deployment with a + * silently different boundary, and no indication that anything was wrong. + * + * Zero is a legitimate value and means off. It is not the same as a malformed one. + */ +function agentStallTimeoutMs(environment: Environment): number { + const raw = optional(environment, "AGENT_STALL_TIMEOUT_MS"); + if (!raw) { + return 0; + } + + const milliseconds = Number(raw); + if (!Number.isInteger(milliseconds) || milliseconds < 0) { + throw new Error( + "AGENT_STALL_TIMEOUT_MS must be a whole number of milliseconds, or 0 to switch the watchdog off", + ); + } + return milliseconds; +} + export function loadConfig( environment: Environment = process.env, ): DeploymentConfig { @@ -330,6 +364,7 @@ export function loadConfig( tenantPackageDirectory: optional(environment, "TENANT_PACKAGE_DIR") ?? "../examples/fintech", runtime: runtimeCapabilities(environment), + agentStallTimeoutMs: agentStallTimeoutMs(environment), oauth: { google }, auth: authConfig(environment, google), devNoAuth: devAuthEnabled(environment), diff --git a/server/src/copilot.ts b/server/src/copilot.ts index 5c00e36..185db75 100644 --- a/server/src/copilot.ts +++ b/server/src/copilot.ts @@ -7,6 +7,7 @@ import { } from "@copilotkit/runtime/v2"; import { createCopilotHonoHandler } from "@copilotkit/runtime/v2/hono"; import type { AgentActor } from "./agents/profile-types"; +import type { StallGuard } from "./channels/stall-guard"; import type { DeploymentConfig } from "./config"; /** @@ -193,9 +194,14 @@ export function buildAgents( agents: RegisteredAgent[], model: RuntimeModel, apiKey: string | null, + /** Absent leaves every stream unwatched, which is what an unconfigured timeout means. */ + stallGuard?: StallGuard, ): Record { return Object.fromEntries( - agents.map((agent) => [agent.id, buildAgent(agent, model, apiKey)]), + agents.map((agent) => [ + agent.id, + buildAgent(agent, model, apiKey, stallGuard), + ]), ); } @@ -203,6 +209,7 @@ function buildAgent( agent: RegisteredAgent, model: RuntimeModel, apiKey: string | null, + stallGuard?: StallGuard, ): AbstractAgent { if (agent.type === "built_in") { return new BuiltInAgent(builtInAgentConfiguration(agent, model, apiKey)); @@ -210,7 +217,7 @@ function buildAgent( if (agent.type === "unavailable") { return new UnavailableAgent(agent); } - return remoteAgentWithStandingRole(agent); + return remoteAgentWithStandingRole(agent, stallGuard); } /** @@ -220,14 +227,24 @@ function buildAgent( * so the same coworker works against any endpoint that speaks the protocol. Any copy of the standing * message already in the conversation is dropped: the endpoint must receive exactly one, first, * however many times the thread has been replayed. + * + * The stall watch goes on the fetch rather than into that middleware, because the middleware works + * in AG-UI events and a stall is the absence of one. The thing that has to be watched is the + * response body, and the fetch is where this deployment still holds it. */ -function remoteAgentWithStandingRole(agent: RegisteredRemoteAgent) { +function remoteAgentWithStandingRole( + agent: RegisteredRemoteAgent, + stallGuard?: StallGuard, +) { const remote = new HttpAgent({ url: agent.endpoint, agentId: agent.id, // The customer's own key, if their agent sits behind one. `HttpAgentConfig` is // `{ url, headers?, fetch? }`, verified against @ag-ui/client 0.0.57. ...(agent.headers ? { headers: agent.headers } : {}), + ...(stallGuard + ? { fetch: stallGuard.watch({ id: agent.id, name: agent.name }) } + : {}), }); remote.use((input, next) => next.run({ @@ -262,6 +279,7 @@ export async function resolveRuntimeAgents( loadAgents: () => Promise, model: RuntimeModel, resolveModelApiKey: () => Promise, + stallGuard?: StallGuard, ): Promise> { const registered = await loadAgents(); if (registered.length === 0) { @@ -273,7 +291,7 @@ export async function resolveRuntimeAgents( const apiKey = registered.some((agent) => agent.type === "built_in") ? await resolveModelApiKey() : null; - return buildAgents(registered, model, apiKey); + return buildAgents(registered, model, apiKey, stallGuard); } /** Who is asking. Agent visibility is decided per person, so a run has to know this first. */ @@ -296,6 +314,12 @@ export function createRequestAgents( loadAgents: LoadAgentsForActor, model: RuntimeModel, resolveModelApiKey: () => Promise, + /** + * Shared across every request rather than built per run, because it is the thing that has to + * outlive one: the sweep that notices a silent stream has to still be running after the request + * that opened it has been answered. + */ + stallGuard?: StallGuard, ) { return async ({ request }: { request: Request }) => { const actor = await identifyActor(request); @@ -303,6 +327,7 @@ export function createRequestAgents( () => loadAgents(actor), model, resolveModelApiKey, + stallGuard, ); }; } @@ -321,6 +346,8 @@ export function mountCopilotRuntime( resolveModelApiKey: () => Promise, identifyUser: IdentifyUser, identifyActor: IdentifyActor, + /** The watch on Bot streams. Absent means the deployment has not configured a timeout. */ + stallGuard?: StallGuard, basePath = "/api/copilotkit", ) { const { intelligence } = config.runtime; @@ -345,6 +372,7 @@ export function mountCopilotRuntime( loadAgents, model, resolveModelApiKey, + stallGuard, ) as never, }); diff --git a/server/src/index.ts b/server/src/index.ts index 2d72bca..3fc067a 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -12,6 +12,7 @@ import { startChannelActivityListener, } from "./channels/events"; import { createChannelStore } from "./channels/routes"; +import { createStallGuard } from "./channels/stall-guard"; import { createThreadIdentity } from "./channels/thread-identity"; import { websocket as channelSocket } from "./channels/socket"; import { createSandboxedStore } from "./components/sandboxed"; @@ -281,6 +282,20 @@ process.on("unhandledRejection", (reason) => { ); }); +/** + * The watch on Bot streams, built once and shared by every run. + * + * It has to outlive the request that opens a stream: the sweep that notices a silent one is still + * running long after the run request has been answered, because in Intelligence mode that request is + * answered in about a second and the Bot keeps writing for as long as it has something to say. + * + * The same audit store as everything else, so a Bot that hangs is recorded beside what Bots do. + */ +const stallGuard = createStallGuard({ + stallMs: config.agentStallTimeoutMs, + auditStore: bootAuditStore, +}); + const app = createApp( config, auth, @@ -317,6 +332,7 @@ const app = createApp( }), identifyUser, identifyActor, + stallGuard, ), computerClient, // The only path to an acting call. diff --git a/server/tests/config.test.ts b/server/tests/config.test.ts index 722a5c2..326ad5c 100644 --- a/server/tests/config.test.ts +++ b/server/tests/config.test.ts @@ -153,4 +153,37 @@ describe("deployment configuration", () => { }), ).toThrow("Google authentication requires BETTER_AUTH_SECRET"); }); + + // A turn that is ended is a turn somebody loses, so an unset variable leaves every stream alone + // rather than acquiring a timeout the deployment never asked for. `.env.example` ships a value. + test("leaves the stall watchdog off when nothing is configured", () => { + expect(loadConfig(baseEnvironment).agentStallTimeoutMs).toBe(0); + }); + + test("takes a timeout in milliseconds, and zero as switching it off", () => { + expect( + loadConfig({ ...baseEnvironment, AGENT_STALL_TIMEOUT_MS: "120000" }) + .agentStallTimeoutMs, + ).toBe(120_000); + expect( + loadConfig({ ...baseEnvironment, AGENT_STALL_TIMEOUT_MS: "0" }) + .agentStallTimeoutMs, + ).toBe(0); + }); + + // Refused rather than defaulted, for the same reason a malformed policy is: an operator who meant + // to write a boundary and mistyped it would otherwise get a deployment enforcing something else. + test.each(["two minutes", "-1", "1.5", ""])( + "refuses to start on AGENT_STALL_TIMEOUT_MS=%p", + (value) => { + const attempt = () => + loadConfig({ ...baseEnvironment, AGENT_STALL_TIMEOUT_MS: value }); + if (value === "") { + // An empty value is an absent one, which is the off case rather than a malformed one. + expect(attempt().agentStallTimeoutMs).toBe(0); + return; + } + expect(attempt).toThrow("AGENT_STALL_TIMEOUT_MS"); + }, + ); }); diff --git a/server/tests/copilot.test.ts b/server/tests/copilot.test.ts index a35764d..5063c8c 100644 --- a/server/tests/copilot.test.ts +++ b/server/tests/copilot.test.ts @@ -163,6 +163,70 @@ describe("registered Copilot agents", () => { expect(agents.risk).toBeInstanceOf(HttpAgent); }); + /* + * The watch goes on the fetch of a remote Bot and nowhere else. + * + * A built-in agent talks to a model provider through the AI SDK rather than over an AG-UI stream, + * so there is no response body here to watch and nothing for the guard to be given. Asserting the + * Bot's own name reaches it matters because that name is what the person is shown when its stream + * goes quiet, and a guard handed the wrong one would say so convincingly. + */ + test("hands a remote Bot's fetch to the stall guard, and a built-in Bot none", () => { + const watched: { id: string; name: string }[] = []; + const stallGuard = { + watch: (bot: { id: string; name: string }) => { + watched.push(bot); + return async () => new Response(null); + }, + stop: () => undefined, + }; + + const agents = buildAgents( + [ + { + id: "general-assistant", + name: "General Assistant", + type: "built_in", + systemPrompt: "Be helpful.", + }, + { + id: "risk", + name: "Risk", + type: "remote_ag_ui", + endpoint: "http://risk.internal/ag-ui", + }, + ], + { provider: "openai", defaultModel: "gpt-4.1" }, + "openai-secret", + stallGuard, + ); + + expect(watched).toEqual([{ id: "risk", name: "Risk" }]); + expect(agents.risk).toBeInstanceOf(HttpAgent); + }); + + test("leaves a remote Bot's fetch alone when no timeout is configured", () => { + const agents = buildAgents( + [ + { + id: "risk", + name: "Risk", + type: "remote_ag_ui", + endpoint: "http://risk.internal/ag-ui", + }, + ], + { provider: "openai", defaultModel: "gpt-4.1" }, + null, + ); + + const remote = agents.risk; + if (!(remote instanceof HttpAgent)) { + throw new Error("Expected the remote agent"); + } + // @ag-ui/client defaults this to its own fetch when the config does not carry one. + expect(typeof remote.fetch).toBe("function"); + }); + test("resolves fresh built-in agents and credentials for every request", async () => { const registered = [ { diff --git a/server/tests/stall-guard.test.ts b/server/tests/stall-guard.test.ts new file mode 100644 index 0000000..8cfb46e --- /dev/null +++ b/server/tests/stall-guard.test.ts @@ -0,0 +1,216 @@ +import { describe, expect, test } from "bun:test"; +import type { AuditEventInput } from "../src/audit"; +import { createStallGuard } from "../src/channels/stall-guard"; + +/** + * These drive real streams, because the properties that matter are properties of a stream. + * + * The timeouts are tens of milliseconds rather than the two minutes a deployment runs, since the + * sweep interval is derived from the timeout: a test asserting the rule should not have to wait the + * length of the rule. Nothing else is faked. The bodies below are `ReadableStream`s and the wrapper + * under test is the same one an `HttpAgent` is handed. + */ + +const BOT = { id: "risk-analyst", name: "Risk Analyst" }; + +/** A body that arrives and closes, like a Bot that answered. */ +function speaks(frames: string[]): ReadableStream { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + for (const frame of frames) controller.enqueue(encoder.encode(frame)); + controller.close(); + }, + }); +} + +/** A body that is accepted and never written to, like a Bot whose own upstream has hung. */ +function saysNothing(): ReadableStream { + return new ReadableStream({ + start() { + // Deliberately empty. The whole point is that nothing ever arrives and nothing ever closes. + }, + }); +} + +function sse(body: ReadableStream, status = 200): Response { + return new Response(body, { + status, + headers: { "content-type": "text/event-stream" }, + }); +} + +const RUN_REQUEST: RequestInit = { + method: "POST", + body: JSON.stringify({ threadId: "thread-7", runId: "run-9" }), +}; + +function collecting(): { + store: { insert: (event: AuditEventInput) => Promise }; + rows: AuditEventInput[]; +} { + const rows: AuditEventInput[] = []; + return { + rows, + store: { + insert: async (event) => { + rows.push(event); + }, + }, + }; +} + +describe("a Bot that stops streaming", () => { + test("is told to the person in the stream they are already reading, and the stream ends", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); + const watched = guard.watch(BOT, async () => sse(saysNothing())); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + // Reading to the end returns only because the guard closed it. A stream nobody ends never does. + const body = await new Response(response.body).text(); + guard.stop(); + + expect(body).toContain('"RUN_ERROR"'); + expect(body).toContain("Risk Analyst stopped responding"); + expect(body).toContain("AGENT_STREAM_STALLED"); + // The framing has to be one an AG-UI client parses, which is a `data:` line and a blank line. + expect(body.startsWith("data: ")).toBe(true); + expect(body.endsWith("\n\n")).toBe(true); + }); + + test("leaves a row naming the Bot, the turn and how long the silence was", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); + const watched = guard.watch(BOT, async () => sse(saysNothing())); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + await new Response(response.body).text(); + guard.stop(); + + const row = audit.rows.find( + (event) => event.eventType === "agent.stream_stalled", + ); + expect(row).toBeDefined(); + expect(row?.targetId).toBe("risk-analyst"); + expect(row?.payload.bot).toBe("risk-analyst"); + expect(row?.payload.thread).toBe("thread-7"); + expect(row?.payload.run).toBe("run-9"); + expect(row?.payload.chunks).toBe(0); + expect(Number(row?.payload.silentForMs)).toBeGreaterThanOrEqual(60); + }); +}); + +describe("the sentence a person is left with", () => { + test("survives the packaged chat's banner, which truncates on its own punctuation", async () => { + const guard = createStallGuard({ stallMs: 60 }); + const watched = guard.watch(BOT, async () => sse(saysNothing())); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + const body = await new Response(response.body).text(); + guard.stop(); + + const message = String( + (JSON.parse(body.slice("data: ".length)) as { message: unknown }).message, + ); + // The banner cuts a message at the first " - ", strips a trailing three-digit number and takes + // anything after "See more:" away with it. A sentence containing any of those reaches a person + // cut in half, so the wording avoids all three rather than relying on nobody noticing. + expect(message).not.toContain(" - "); + expect(message).not.toContain("See more:"); + expect(message).not.toMatch(/:\s*\d{3}$/); + // Said in words a person reads, not in the milliseconds a deployment configured, and never as + // "0 seconds": the timeout here is a test's, and a floor keeps the sentence sane at any value. + expect(message).toContain("for a second"); + }); +}); + +describe("a Bot that is answering", () => { + test("has its bytes passed through untouched and nothing added", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 5_000, auditStore: audit.store }); + const frames = [ + 'data: {"type":"RUN_STARTED"}\n\n', + 'data: {"type":"TEXT_MESSAGE_CHUNK","delta":"hello"}\n\n', + 'data: {"type":"RUN_FINISHED"}\n\n', + ]; + const watched = guard.watch(BOT, async () => sse(speaks(frames))); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + const body = await new Response(response.body).text(); + guard.stop(); + + expect(body).toBe(frames.join("")); + expect(audit.rows).toHaveLength(0); + }); + + test("keeps the status and the content type the parser chooses on", async () => { + const guard = createStallGuard({ stallMs: 5_000 }); + const watched = guard.watch(BOT, async () => sse(speaks(["data: {}\n\n"]))); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + await new Response(response.body).text(); + guard.stop(); + + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toBe("text/event-stream"); + }); +}); + +describe("what the watch refuses to touch", () => { + test("a refusal passes straight through, body and all", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); + const refusal = new Response("no", { status: 502 }); + const watched = guard.watch(BOT, async () => refusal); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + guard.stop(); + + expect(response).toBe(refusal); + expect(await response.text()).toBe("no"); + expect(audit.rows).toHaveLength(0); + }); + + test("a framing it does not recognise is closed rather than written into", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); + const watched = guard.watch( + BOT, + async () => + new Response(saysNothing(), { + headers: { "content-type": "application/vnd.ag-ui.event+proto" }, + }), + ); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + const body = await new Response(response.body).arrayBuffer(); + guard.stop(); + + // Nothing written in: SSE bytes in a binary stream would corrupt a run that was merely slow. + expect(body.byteLength).toBe(0); + expect( + audit.rows.some((event) => event.eventType === "agent.stream_stalled"), + ).toBe(true); + }); +}); + +describe("a deployment with no timeout configured", () => { + test("gets the fetch it handed in, unwrapped", () => { + const guard = createStallGuard({ stallMs: 0 }); + const inner = async () => sse(speaks([])); + expect(guard.watch(BOT, inner)).toBe(inner); + guard.stop(); + }); + + test("and its streams are never ended for it", async () => { + const audit = collecting(); + const guard = createStallGuard({ stallMs: 0, auditStore: audit.store }); + const watched = guard.watch(BOT, async () => sse(speaks(["data: {}\n\n"]))); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + expect(await new Response(response.body).text()).toBe("data: {}\n\n"); + expect(audit.rows).toHaveLength(0); + guard.stop(); + }); +}); diff --git a/server/tests/turn-watchdog.test.ts b/server/tests/turn-watchdog.test.ts new file mode 100644 index 0000000..2b45920 --- /dev/null +++ b/server/tests/turn-watchdog.test.ts @@ -0,0 +1,190 @@ +import { describe, expect, test } from "bun:test"; +import { + type StalledStream, + TurnWatchdog, +} from "../src/channels/turn-watchdog"; + +/** + * These test the distinction the watchdog exists to draw, not the plumbing around it. + * + * Every case is one a deployment can actually be in: a Bot that answers slowly and is fine, a Bot + * that answered once and then stopped, a Bot that accepted the connection and never wrote a byte. + * Time is injected rather than waited for, so a two-minute timeout is tested in no time at all and + * the assertions are about the rule rather than about how fast the machine running them is. + * + * There is no exemption for a frontend tool call and therefore no test of one. The run ends before + * the browser executes a tool and a second run carries the result back, so no stream is ever held + * open across one; see the module comment on turn-watchdog.ts for what was measured. + */ + +/** A clock a test moves by hand. */ +function clock(): { now: () => number; advance: (ms: number) => void } { + let current = 0; + return { + now: () => current, + advance: (ms: number) => { + current += ms; + }, + }; +} + +function watching(stallMs: number) { + const time = clock(); + const stalled: StalledStream[] = []; + const watchdog = new TurnWatchdog({ + stallMs, + now: time.now, + onStall: (stream) => stalled.push(stream), + }); + return { time, stalled, watchdog }; +} + +describe("a turn is judged on activity, not on how long it has run", () => { + test("a stream that has said nothing for the timeout is given up on", () => { + const { time, stalled, watchdog } = watching(60_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + time.advance(59_999); + expect(watchdog.sweep()).toBe(0); + + time.advance(1); + expect(watchdog.sweep()).toBe(1); + expect(stalled).toHaveLength(1); + expect(stalled[0]?.botId).toBe("risk-analyst"); + expect(stalled[0]?.silentForMs).toBe(60_000); + // Zero chunks is the fact worth reading: the Bot answered the request and then said nothing. + expect(stalled[0]?.chunks).toBe(0); + }); + + test("a stream that keeps talking is never given up on, however long it runs", () => { + const { time, stalled, watchdog } = watching(60_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + // An hour of work, a chunk every half minute. A duration limit would have ended this repeatedly. + for (let minute = 0; minute < 120; minute++) { + time.advance(30_000); + watchdog.record("stream-1"); + expect(watchdog.sweep()).toBe(0); + } + + expect(stalled).toHaveLength(0); + expect(watchdog.watching).toBe(1); + }); + + test("a stream that spoke and then stopped is given up on from its last chunk", () => { + const { time, stalled, watchdog } = watching(60_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + time.advance(50_000); + watchdog.record("stream-1"); + watchdog.record("stream-1"); + time.advance(50_000); + // A hundred seconds since it opened, fifty since it last said anything. Silence is what counts. + expect(watchdog.sweep()).toBe(0); + + time.advance(10_000); + expect(watchdog.sweep()).toBe(1); + expect(stalled[0]?.silentForMs).toBe(60_000); + expect(stalled[0]?.chunks).toBe(2); + }); + + test("a stream that ends is forgotten and is never reported", () => { + const { time, stalled, watchdog } = watching(60_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + watchdog.close("stream-1"); + + time.advance(600_000); + expect(watchdog.sweep()).toBe(0); + expect(stalled).toHaveLength(0); + expect(watchdog.watching).toBe(0); + }); + + test("closing a stream that was never watched is harmless", () => { + const { watchdog } = watching(60_000); + expect(() => watchdog.close("never-opened")).not.toThrow(); + expect(watchdog.watching).toBe(0); + }); +}); + +describe("a stalled stream is reported exactly once", () => { + test("sweeping again finds nothing, because the entry went with the callback", () => { + const { time, stalled, watchdog } = watching(1_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + time.advance(5_000); + expect(watchdog.sweep()).toBe(1); + expect(watchdog.sweep()).toBe(0); + expect(watchdog.sweep()).toBe(0); + expect(stalled).toHaveLength(1); + }); + + test("a late chunk from a Bot already given up on does not start a second watch", () => { + const { time, stalled, watchdog } = watching(1_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + time.advance(2_000); + expect(watchdog.sweep()).toBe(1); + + // The endpoint wakes up and writes. There is no turn left to end, and reviving it here would + // stall again and report the same wedged turn twice. + watchdog.record("stream-1"); + time.advance(60_000); + expect(watchdog.sweep()).toBe(0); + expect(stalled).toHaveLength(1); + }); + + test("one wedged Bot's callback throwing does not spare the others", () => { + const time = clock(); + const reported: string[] = []; + const watchdog = new TurnWatchdog({ + stallMs: 1_000, + now: time.now, + onStall: (stream) => { + reported.push(stream.id); + if (stream.id === "stream-1") throw new Error("the surface blew up"); + }, + }); + + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + watchdog.open({ id: "stream-2", botId: "general-assistant" }); + time.advance(2_000); + + expect(watchdog.sweep()).toBe(2); + expect(reported).toEqual(["stream-1", "stream-2"]); + expect(watchdog.watching).toBe(0); + }); +}); + +describe("a deployment that has not asked for a watchdog does not get one", () => { + test("zero watches nothing, whatever is opened", () => { + const { time, stalled, watchdog } = watching(0); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + expect(watchdog.enabled).toBe(false); + expect(watchdog.watching).toBe(0); + + time.advance(3_600_000); + expect(watchdog.sweep()).toBe(0); + expect(stalled).toHaveLength(0); + }); + + test("a positive timeout is what switches it on", () => { + const { watchdog } = watching(1); + expect(watchdog.enabled).toBe(true); + }); +}); + +describe("opening a stream twice", () => { + test("restarts its clock rather than raising", () => { + const { time, stalled, watchdog } = watching(1_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + time.advance(900); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + time.advance(900); + + expect(watchdog.sweep()).toBe(0); + expect(watchdog.watching).toBe(1); + expect(stalled).toHaveLength(0); + }); +}); From 75b2140b4eda56c7f669ad2df64b644b4461486a Mon Sep 17 00:00:00 2001 From: Jerel John Velarde Date: Wed, 19 Aug 2026 14:38:49 -0700 Subject: [PATCH 2/2] Watch the Bot, not whoever is reading it The clock was kept in the wrong place. `watchdog.record` was called from the relay's transform callback, and a TransformStream runs its transform only once the readable side is being pulled, so what was being timed was the moment the CONSUMER took a chunk rather than the moment the Bot produced one. A reader that paused for longer than the timeout therefore looked exactly like a Bot that had gone silent: the run was ended, the audit row said the endpoint had sent nothing, and the person was told their Bot had stopped responding while it was streaming the whole time. In this deployment that reader is the Intelligence runner publishing every event on to the gateway over the network, which is precisely the sort of thing that pauses. The pump now times the resolution of each read from the Bot, and stops the clock across the handover to the relay, so the only quiet ever counted is quiet on the wire. The recovery no longer waits on the database. The audit row was written before the stream was touched, on the argument that a record must not depend on a promise a broken stream owes us; but the writes it was protecting are queued and unawaited anyway, while the insert it moved in front of them is a bare statement against the pool every other write shares, with no deadline of its own. A saturated pool or an unreachable Postgres is exactly the condition a Bot is most likely to hang in, and in that condition the watchdog fired and then parked, leaving the spinner and the locked composer it exists to end. The socket is now released, the sentence queued and the stream closed first, and the row written after. The direct Bot chat says something now. The channel drew the sentence at the end of its transcript; the other surface drew nothing at all, because the banner it was assumed to have belongs to a provider this app does not mount and is suppressed even there unless the dev console is on. It now watches the same runs the chat starts and draws a line under the page header. Not where the missing answer was going to be, which is where the channel puts it, but the packaged chat owns and virtualises its message list and reaching into it means taking on its scrolling. Both surfaces fall back to the same sentence from the same place, so a person who uses both is not told two different things about the same silence. A stream is now judged by the rule the client actually applies. The guard would only write into `text/event-stream`, which is stricter than @ag-ui/client, and being stricter was not the safe direction it was argued to be: a Bot serving events under any other content type had its stream closed with nothing in it, so the run ended and nobody was told anything on either surface. The client treats everything except the protobuf media type as server-sent events, and so does this now. The shipped timeout drops to a minute, because two minutes was in a race it could lose. Every Bot in this repository serves on Bun with `idleTimeout: 120`, and Bun tears a wedged streaming response down about a second past that; a watchdog set to the same two minutes lands within a second of the socket dying, and when it lost the person got a transport error instead of the designed sentence and the trail got no row. Half the Bot's own idle timeout is far enough clear to be deterministic, and is still far longer than any real silence inside a run. Three smaller things on the trail and around it. A stalled turn takes the same colour as an action that did not happen, rather than the muted one that reads as "Allowed", and it joins the "Did not happen" filter. The two numbers the row exists to carry are drawn: how long the stream was silent, and whether the Bot had managed to say anything first, which is the difference between an endpoint that dies mid-answer and one that never begins. And the test asserting an unwatched Bot's fetch was left alone was asserting nothing, since @ag-ui/client fills that field in either way; it now tells the two apart with a sentinel. --- .env.example | 18 ++- app/src/components/channels/channel-chat.tsx | 13 +- app/src/lib/audit/silence.ts | 29 +++++ app/src/lib/copilot/stopped-turn.ts | 60 +++++++++ app/src/routes/_authed/_app/bot.tsx | 27 ++++ app/src/routes/_authed/admin/audit.tsx | 27 +++- app/tests/audit-silence.test.ts | 44 +++++++ app/tests/stopped-turn.test.ts | 41 +++++++ server/src/channels/stall-guard.ts | 123 ++++++++++++------- server/src/channels/turn-watchdog.ts | 65 +++++++++- server/src/copilot.ts | 8 +- server/tests/copilot.test.ts | 43 ++++--- server/tests/stall-guard.test.ts | 80 ++++++++++-- server/tests/turn-watchdog.test.ts | 31 +++++ 14 files changed, 517 insertions(+), 92 deletions(-) create mode 100644 app/src/lib/audit/silence.ts create mode 100644 app/src/lib/copilot/stopped-turn.ts create mode 100644 app/tests/audit-silence.test.ts create mode 100644 app/tests/stopped-turn.test.ts diff --git a/.env.example b/.env.example index 4a5563c..ab4d200 100644 --- a/.env.example +++ b/.env.example @@ -57,14 +57,20 @@ COPILOTKIT_LICENSE_TOKEN= # Silence, not duration. A turn may legitimately run for an hour while events keep arriving; what is # measured here is the gap between them. Nothing about how long a Bot is allowed to work changes. # -# Two minutes because that is already what a Bot in this repository treats as the outer bound of a -# quiet connection (`agent-bot` sets `idleTimeout: 120`), and because it is far longer than any real -# silence inside a run: the longest legitimate one is the wait for a model's first token, which is -# seconds. Browser tool calls do not need allowing for. They run between turns, not during one, the -# run ends before the browser executes the tool and a second run carries the result back. +# A minute, because this deadline is in a race it has to win. Every Bot in this repository serves on +# Bun with `idleTimeout: 120`, and Bun tears a wedged streaming response down at roughly a second +# past that. A watchdog set to the same two minutes lands within a second of the socket dying, and +# whichever gets there first decides what the person sees: this deployment's sentence naming the Bot +# and saying the turn was ended, or "The socket connection was closed unexpectedly" and no audit row +# at all. Half the Bot's own idle timeout is far enough clear that the answer is always the first. +# +# A minute is still far longer than any real silence inside a run. The longest legitimate one is the +# wait for a model's first token, which is seconds. Browser tool calls do not need allowing for: +# they run between turns, not during one, because the run ends before the browser executes the tool +# and a second run carries the result back. # # 0, or leaving this unset, switches the watchdog off. Nothing is watched and no turn is ever ended. -AGENT_STALL_TIMEOUT_MS=120000 +AGENT_STALL_TIMEOUT_MS=60000 # Model key. Required by the proof-of-concept Bot, which speaks OpenAI's API directly, and by the # framework Bot unless you point it at another provider below. diff --git a/app/src/components/channels/channel-chat.tsx b/app/src/components/channels/channel-chat.tsx index c586f6b..1b2c8cf 100644 --- a/app/src/components/channels/channel-chat.tsx +++ b/app/src/components/channels/channel-chat.tsx @@ -19,6 +19,7 @@ import type { AgentChannel } from "@/lib/channels/queries"; import { useActiveBot } from "@/lib/copilot/active-bot"; import { ConversationProvider } from "@/lib/copilot/conversation"; import { repairUnansweredToolCalls } from "@/lib/copilot/repair-history"; +import { stoppedReason } from "@/lib/copilot/stopped-turn"; import { useSkillCommands } from "@/lib/plugins/skill-commands"; /** @@ -221,14 +222,10 @@ export function ChannelChat({ setRunError(message); }; const subscription = agent.subscribe?.({ - onRunErrorEvent: ({ event }) => - fail(event?.message ?? "The Bot stopped without saying why."), - onRunFailed: ({ error }) => - fail( - error instanceof Error - ? error.message - : "The Bot stopped without saying why.", - ), + // Both surfaces fall back to the same sentence, from the same place, so a person who uses + // both is not told two different things about the same silence. + onRunErrorEvent: ({ event }) => fail(stoppedReason(event?.message)), + onRunFailed: ({ error }) => fail(stoppedReason(error)), onRunFinishedEvent: () => { const wasOurs = awaitingReply.current; awaitingReply.current = false; diff --git a/app/src/lib/audit/silence.ts b/app/src/lib/audit/silence.ts new file mode 100644 index 0000000..0a1e378 --- /dev/null +++ b/app/src/lib/audit/silence.ts @@ -0,0 +1,29 @@ +/** + * How a stalled turn reads on the audit page. + * + * The row for a Bot that stopped talking carries two numbers nothing else in the trail carries: how + * long its stream had been quiet when the deployment gave up on it, and how much it had managed to + * say first. They are the whole reason that row exists. An endpoint that dies halfway through an + * answer and one that accepts a connection and never writes are different faults with different + * fixes, and on the page they are the same line unless these two are drawn. + * + * Chunks, not events, because that is what was counted: one chunk can carry several AG-UI events and + * the boundaries are the network's. Saying "events" here would be a number that looks precise and is + * not. What a reader needs from it is whether it is zero, and that it says plainly. + * + * Returns null rather than a placeholder when the payload does not carry both. An older row written + * before this was recorded should show nothing, not "0 chunks", which would be a claim about a Bot + * that nobody ever measured. + */ +export function silenceOf(payload: Record): string | null { + const silentForMs = payload.silentForMs; + const chunks = payload.chunks; + if (typeof silentForMs !== "number" || typeof chunks !== "number") { + return null; + } + + const seconds = Math.max(1, Math.round(silentForMs / 1000)); + const quiet = `Silent for ${seconds}s`; + if (chunks === 0) return `${quiet}, having said nothing at all`; + return `${quiet}, after ${chunks} ${chunks === 1 ? "chunk" : "chunks"}`; +} diff --git a/app/src/lib/copilot/stopped-turn.ts b/app/src/lib/copilot/stopped-turn.ts new file mode 100644 index 0000000..adabd35 --- /dev/null +++ b/app/src/lib/copilot/stopped-turn.ts @@ -0,0 +1,60 @@ +import { useAgent } from "@copilotkit/react-core/v2"; +import { useEffect, useState } from "react"; + +/** + * Why the last turn ended without an answer, for a surface that has to say so itself. + * + * A run can end three ways. It finishes, which needs no explanation. It fails in the browser, which + * arrives as an error. Or the Bot's own stream stops producing anything and this deployment ends the + * turn for it, which arrives as a RUN_ERROR carrying the sentence the server wrote (see + * server/src/channels/stall-guard.ts). The last two both leave the same hole on screen: the composer + * unlocks, the spinner disappears, and nothing says what happened. + * + * The reason is kept as a sentence rather than a flag because the reasons are not interchangeable. A + * Bot that refused, a Bot whose endpoint is down and a Bot that simply stopped talking are three + * different things to be told, and only the thing that ended the turn knows which one it was. + */ + +/** + * The sentence to show, in the words of whatever ended the turn. + * + * Falls back only when there is genuinely nothing to pass on. Saying "the Bot stopped without saying + * why" is honest about that; inventing a cause would not be, and this is the one moment a person has + * no other way to find out what went wrong. + */ +export function stoppedReason(reported: unknown): string { + const said = + reported instanceof Error + ? reported.message + : typeof reported === "string" + ? reported + : ""; + return said.trim() || "The Bot stopped without saying why."; +} + +/** + * Watch one Bot's runs and hold on to the reason the last one ended, if it ended badly. + * + * Bound by agent id rather than handed an agent, so a caller that only renders the packaged chat + * does not have to reach for one: `useAgent` returns the same shared instance the chat itself binds + * to, so this watches exactly the runs that chat starts. + * + * Cleared when the next run begins rather than on a timer. A sentence about a turn that is over + * should stay until there is something newer to look at, and the person deciding when that is is the + * one who sends the next message. + */ +export function useStoppedTurn(agentId: string): string | null { + const { agent } = useAgent({ agentId }); + const [stopped, setStopped] = useState(null); + + useEffect(() => { + const subscription = agent.subscribe?.({ + onRunInitialized: () => setStopped(null), + onRunErrorEvent: ({ event }) => setStopped(stoppedReason(event?.message)), + onRunFailed: ({ error }) => setStopped(stoppedReason(error)), + }); + return () => subscription?.unsubscribe(); + }, [agent]); + + return stopped; +} diff --git a/app/src/routes/_authed/_app/bot.tsx b/app/src/routes/_authed/_app/bot.tsx index a0d5677..a3dd902 100644 --- a/app/src/routes/_authed/_app/bot.tsx +++ b/app/src/routes/_authed/_app/bot.tsx @@ -2,6 +2,7 @@ import { CopilotChat } from "@copilotkit/react-core/v2"; import { createFileRoute } from "@tanstack/react-router"; import { useActiveBot } from "@/lib/copilot/active-bot"; import { useBotThread } from "@/lib/copilot/bot-thread"; +import { useStoppedTurn } from "@/lib/copilot/stopped-turn"; export const Route = createFileRoute("/_authed/_app/bot")({ component: RouteComponent, @@ -18,6 +19,15 @@ function RouteComponent() { useActiveBot(agentId); // Minted by this deployment rather than by the chat, and the same one on the next visit. const threadId = useBotThread(agentId); + /* + * A turn that ends without an answer has to be said out loud here, because the packaged chat says + * nothing. It reports a failed run to an `onError` prop and otherwise carries on as though the + * turn simply finished: the composer unlocks, the spinner goes, and the transcript keeps the + * person's own message with nothing under it. The banner that would have explained it belongs to + * a provider this app does not mount. + */ + const stopped = useStoppedTurn(agentId); + return (
@@ -26,6 +36,23 @@ function RouteComponent() { Ask it to open a page and watch it work.

+ {/* + * Under the header rather than at the end of the transcript, which is where the missing answer + * was going to be and where the channel draws its own version of this. The packaged chat owns + * that list and virtualises it, so reaching into it means replacing the whole message view and + * taking on its scrolling. The cost of putting the sentence here instead is that it is not + * beside the gap it explains; what it buys is that it is always on screen, whatever the + * transcript has been scrolled to, and that it survives the next release of the chat. + */} + {stopped ? ( +

+ {stopped} +

+ ) : null}
{/* Remount when switching Bots so chat state stays bound to the selected agent. */} {threadId ? ( diff --git a/app/src/routes/_authed/admin/audit.tsx b/app/src/routes/_authed/admin/audit.tsx index b5865fd..2a6319b 100644 --- a/app/src/routes/_authed/admin/audit.tsx +++ b/app/src/routes/_authed/admin/audit.tsx @@ -10,6 +10,7 @@ import { import { Button } from "@/components/ui/button"; import { useBotNames } from "@/lib/agents/bot-names"; import { auditEventsQueryOptions } from "@/lib/audit/queries"; +import { silenceOf } from "@/lib/audit/silence"; /** * Read surface for policy, computer, component, MCP, and credential audit events. @@ -38,7 +39,12 @@ const FILTERS = [ search: "?eventType=computer.action_refused,mcp.call_rejected,component.refused,component.function_refused", }, - { label: "Did not happen", search: "?eventType=computer.action_failed" }, + { + label: "Did not happen", + // A stalled stream belongs here. It is the same complaint as an action that was allowed and then + // did not take: nothing was refused, and nothing came of it either. + search: "?eventType=computer.action_failed,agent.stream_stalled", + }, ] as const; function AuditPage() { @@ -137,8 +143,12 @@ function Row({ event.eventType === "component.refused" || event.eventType === "component.function_refused" || event.eventType === "mcp.call_rejected"; - // Allowed by policy but not carried out. - const failed = event.eventType === "computer.action_failed"; + const stalled = event.eventType === "agent.stream_stalled"; + // Allowed by policy but not carried out. A stalled turn belongs in the same family: the Bot was + // asked and the answer never arrived. Colour is how this table is read, and a row left in the + // muted foreground reads as "Allowed", which a turn nobody ever got an answer to was not. + const failed = event.eventType === "computer.action_failed" || stalled; + const silence = stalled ? silenceOf(payload) : null; return ( @@ -228,6 +238,14 @@ function Row({ {payload.failure}
) : null} + {/* + * The two numbers the stall row is worth reading for. Without them every stalled turn looks + * the same, and the difference between an endpoint that dies halfway through an answer and + * one that never begins is the difference between a slow Bot and a dead one. + */} + {silence ? ( +
{silence}
+ ) : null} {/* Show concrete policy rules, but suppress the uninformative default `true` allow rule. */} {decision.rule && decision.rule !== "true" ? (
@@ -259,7 +277,8 @@ const NAMED_TARGETS = new Set([ const DECISIONS: Record = { "bot.declined": "The Bot declined", - // Not a refusal and not a failed action: the Bot said nothing and the turn was ended for it. + // Not a refusal, so not the refusal colour: nothing was blocked. The Bot was asked and never + // answered, which is the same complaint as an action that was allowed and then did not happen. "agent.stream_stalled": "The Bot stopped responding", "computer.policy_loaded": "Boundary at start-up", "computer.isolation_loaded": "Isolation at start-up", diff --git a/app/tests/audit-silence.test.ts b/app/tests/audit-silence.test.ts new file mode 100644 index 0000000..089f1a3 --- /dev/null +++ b/app/tests/audit-silence.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, test } from "bun:test"; +import { silenceOf } from "../src/lib/audit/silence"; + +/** + * The line that tells a stalled-turn row apart from every other stalled-turn row. + * + * The distinction the audit trail exists to draw here is between a Bot that dies partway through an + * answer and one that accepts the request and never writes, so the zero case is the one that has to + * read unmistakably. + */ + +describe("what a stalled turn says on the audit page", () => { + test("names the silence and how much came before it", () => { + expect(silenceOf({ silentForMs: 120_000, chunks: 14 })).toBe( + "Silent for 120s, after 14 chunks", + ); + }); + + test("says outright when nothing ever arrived", () => { + expect(silenceOf({ silentForMs: 60_000, chunks: 0 })).toBe( + "Silent for 60s, having said nothing at all", + ); + }); + + test("counts one chunk as one", () => { + expect(silenceOf({ silentForMs: 60_000, chunks: 1 })).toBe( + "Silent for 60s, after 1 chunk", + ); + }); + + test("never says nought seconds, whatever the deployment configured", () => { + expect(silenceOf({ silentForMs: 60, chunks: 0 })).toBe( + "Silent for 1s, having said nothing at all", + ); + }); + + test("draws nothing for a row that does not carry the numbers", () => { + // An older row, written before these were recorded. Showing "0 chunks" there would be a claim + // about a Bot nobody measured. + expect(silenceOf({ bot: "risk-analyst" })).toBeNull(); + expect(silenceOf({ silentForMs: 60_000 })).toBeNull(); + expect(silenceOf({ silentForMs: "60000", chunks: 0 })).toBeNull(); + }); +}); diff --git a/app/tests/stopped-turn.test.ts b/app/tests/stopped-turn.test.ts new file mode 100644 index 0000000..6842b94 --- /dev/null +++ b/app/tests/stopped-turn.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, test } from "bun:test"; +import { stoppedReason } from "../src/lib/copilot/stopped-turn"; + +/** + * What a person is told when a turn ends and no answer came. + * + * The cases are the three things that actually arrive: the sentence the deployment's stall watchdog + * wrote into the run, an error thrown in the browser, and nothing at all. + */ + +describe("the reason a turn ended", () => { + test("passes on what ended the turn, in its own words", () => { + expect( + stoppedReason( + "Risk Analyst stopped responding. Nothing arrived from it for 2 minutes, so this turn was ended. Ask again, or check that the Bot is running.", + ), + ).toContain("Risk Analyst stopped responding"); + }); + + test("reads an Error the same way, because a failed run carries one", () => { + expect( + stoppedReason(new Error("The endpoint refused the connection")), + ).toBe("The endpoint refused the connection"); + }); + + test("says so plainly when nothing was reported, rather than inventing a cause", () => { + // This is the one moment a person has no other way to find out what went wrong, so a guess here + // would be worse than an admission. + expect(stoppedReason(undefined)).toBe( + "The Bot stopped without saying why.", + ); + expect(stoppedReason("")).toBe("The Bot stopped without saying why."); + expect(stoppedReason(" ")).toBe("The Bot stopped without saying why."); + expect(stoppedReason(new Error(""))).toBe( + "The Bot stopped without saying why.", + ); + expect(stoppedReason({ message: "not a string or an Error" })).toBe( + "The Bot stopped without saying why.", + ); + }); +}); diff --git a/server/src/channels/stall-guard.ts b/server/src/channels/stall-guard.ts index e55e0c8..b6aca71 100644 --- a/server/src/channels/stall-guard.ts +++ b/server/src/channels/stall-guard.ts @@ -13,19 +13,33 @@ * The Bot's response, however, is an ordinary streaming HTTP body that this process reads to the * end, so it is both the thing that stalls and the only place a stall can be seen. * - * A PASS-THROUGH TRANSFORM IS THE ONLY ACCEPTABLE SHAPE HERE, and the transform below does exactly - * one thing besides handing the chunk on. It must not buffer, because a Bot's answer is streamed a + * A PASS-THROUGH RELAY IS THE ONLY ACCEPTABLE SHAPE HERE, and the relay below is an identity + * transform with nothing in it at all. It must not buffer, because a Bot's answer is streamed a * token at a time and holding chunks back to inspect them would turn a live answer into a paragraph * that lands all at once. It must not await anything, because a chunk delayed here is a chunk * delayed on somebody's screen. And it must not decide anything about the bytes, because the moment * it parses them it can be wrong about them, and a watchdog that can misread a working run into a - * broken one is worse than the failure it was added for. Counting is the whole of its job. + * broken one is worse than the failure it was added for. + * + * THE CLOCK IS KEPT BY THE PUMP AND NOT BY THE RELAY, which is the difference between watching the + * Bot and watching whoever is reading it. A `TransformStream` runs its transform only once the + * readable side is being pulled, so timestamping in there would record when the CONSUMER took a + * chunk rather than when the Bot produced one; a consumer that paused for longer than the timeout + * would then be reported as a Bot that had gone silent, on a run that was streaming the whole time. + * The pump times the resolution of each read from the Bot instead, and stops the clock entirely + * while it waits for the consumer to take delivery, so the only quiet ever counted is quiet on the + * wire. In this deployment that consumer is the Intelligence runner publishing every event on to + * the gateway over the network, which is exactly the sort of thing that pauses. * * On a stall it writes one RUN_ERROR event into the same stream and closes it. RUN_ERROR is the - * event both surfaces already understand: the channel renders it through the subscriber it already - * has, and the packaged chat shows its message in its own banner, so nothing downstream needs to - * learn a new event to tell somebody what happened. AG-UI permits RUN_ERROR at any point in a - * stream, including as the very first event, which is what a Bot that never spoke produces. + * event both surfaces already subscribe to, so nothing downstream had to learn a new one. What + * neither of them had was anywhere to put it, and both now draw the sentence themselves: see + * app/src/components/channels/chat-transcript.tsx for the channel and + * app/src/routes/_authed/_app/bot.tsx for the direct Bot chat. The packaged chat draws nothing of + * its own for a failed run — the banner that would have done it belongs to the v1 provider this app + * does not mount, and is suppressed even there unless the dev console is switched on. AG-UI permits + * RUN_ERROR at any point in a stream, including as the very first event, which is what a Bot that + * never spoke produces. */ import { type AuditStore, recordAuditEvent } from "../audit"; import { type StalledStream, TurnWatchdog } from "./turn-watchdog"; @@ -61,15 +75,21 @@ export type StallGuard = { }; /** - * Only a stream that says it is server-sent events gets an event written into it. + * The one framing a sentence must never be written into. + * + * @ag-ui/client 0.0.57 picks its parser by comparing the response's content type to exactly this + * media type: this value means protocol buffers, and every other value, including one it has never + * seen, is read as server-sent events. The rule here is that same rule, deliberately, because the + * question being asked is not "what did the Bot mean by this header" but "what will the thing at the + * other end of this relay try to parse". Testing for `text/event-stream` instead was stricter than + * the client and silently worse: a Bot serving SSE bytes under some other content type had its + * stream closed with nothing in it, so the run ended, the composer unlocked, and nobody was told + * anything on either surface. * - * This is deliberately stricter than the AG-UI client's own rule, which treats every content type - * except the protobuf one as SSE. Being stricter fails in the safe direction: a Bot answering in a - * framing this does not recognise gets a clean close and the run ends with the client's own terminal - * event, which is a worse message than ours but is still an ending. Guessing the other way would - * mean writing SSE bytes into a binary stream and corrupting a run that was merely slow. + * Guessing the other way is the failure worth avoiding, and this still avoids it. Appending SSE + * bytes to a protobuf stream would corrupt a run that was merely slow. */ -const SSE_CONTENT_TYPE = "text/event-stream"; +const PROTOBUF_CONTENT_TYPE = "application/vnd.ag-ui.event+proto"; const ENCODER = new TextEncoder(); @@ -91,7 +111,7 @@ type OpenStream = { writer: WritableStreamDefaultWriter; /** Cancels the Bot's side, so a wedged endpoint does not keep a socket here forever. */ cancelUpstream: () => void; - /** Whether an event may be written into this stream. See SSE_CONTENT_TYPE. */ + /** Whether an event may be written into this stream. See PROTOBUF_CONTENT_TYPE. */ sse: boolean; /** * The request body, held as the string it was already serialised to. @@ -159,9 +179,19 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { /** * End a turn whose Bot has stopped talking. * - * The row is written before the stream is touched. A wedged Bot may also have a consumer that has - * stopped reading, in which case the writes below never settle, and the record of what happened - * must not depend on a promise that a broken stream owes us. + * NOTHING THE PERSON NEEDS WAITS ON ANYTHING THAT CAN FAIL TO SETTLE. Every step of the recovery + * below is either synchronous or queued and deliberately unawaited, so the socket is released, the + * sentence is queued and the stream is closed within this tick whatever else in the deployment is + * unwell. A wedged Bot may also have a consumer that has stopped reading, in which case those + * writes never settle; awaiting them would leave the spinner and the locked composer this whole + * file exists to end. + * + * The audit row comes last, after the close, and the same argument is what puts it there. The + * store is a bare insert against the pool every other write in this deployment shares (see + * `createAuditStore`) with no deadline of its own, and a Bot is most likely to hang in exactly the + * conditions where that pool is saturated or Postgres is unreachable. Recovering first and + * recording second costs a row if the process dies in between, and a lost row is a smaller loss + * than a watchdog that fires and then fails open. */ async function giveUp(stalled: StalledStream): Promise { const stream = release(stalled.id); @@ -179,6 +209,18 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { }), ); + // The Bot's side goes first, so a socket into an endpoint that will never answer is released + // whatever the browser's side of the stream is doing. + stream.cancelUpstream(); + + if (stream.sse) { + // The write is queued ahead of the close, so it either lands in order or neither does. + void stream.writer + .write(stalledEvent(stream.bot.name, options.stallMs)) + .catch(() => undefined); + } + void stream.writer.close().catch(() => undefined); + if (options.auditStore) { await recordAuditEvent(options.auditStore, { eventType: "agent.stream_stalled", @@ -195,19 +237,6 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { }, }).catch(() => undefined); } - - // The Bot's side goes first, so a socket into an endpoint that will never answer is released - // whatever the browser's side of the stream is doing. - stream.cancelUpstream(); - - if (stream.sse) { - // Deliberately not awaited: see the note above about a consumer that has stopped reading. The - // write is queued ahead of the close, so it either lands in order or neither does. - void stream.writer - .write(stalledEvent(stream.bot.name, options.stallMs)) - .catch(() => undefined); - } - void stream.writer.close().catch(() => undefined); } function watch( @@ -226,12 +255,9 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { if (!response.ok || body === null) return response; const id = crypto.randomUUID(); - const relay = new TransformStream({ - transform(chunk, controller) { - watchdog.record(id); - controller.enqueue(chunk); - }, - }); + // An identity transform, with no transform in it. All it is here for is a writable end this + // process can put one last event into and close from outside the pump; see `giveUp`. + const relay = new TransformStream(); const reader = body.getReader(); const writer = relay.writable.getWriter(); @@ -241,9 +267,7 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { cancelUpstream: () => { void reader.cancel().catch(() => undefined); }, - sse: (response.headers.get("content-type") ?? "").includes( - SSE_CONTENT_TYPE, - ), + sse: response.headers.get("content-type") !== PROTOBUF_CONTENT_TYPE, requestBody: typeof requestInit.body === "string" ? requestInit.body : null, finished: false, @@ -264,7 +288,14 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { } /** - * Move bytes across, and end the stream the way the Bot ended it. + * Move bytes across, keep the clock, and end the stream the way the Bot ended it. + * + * This loop is the only place that knows which of its two waits is which, which is why the timing + * lives here. A read that resolves is the Bot having said something and is the only evidence the + * endpoint is alive. A write that has not resolved yet is the far side of the relay not having + * taken delivery, and says nothing about the Bot at all, so the clock is stopped across it and + * restarted when the handover completes. Timing the relay instead would have reported a paused + * consumer as a silent Bot, which is the one mistake this must not make. * * A read that throws is passed on as an abort rather than as a clean close. A broken connection * carries a real reason, and relabelling it as an ending would file a transport failure as a Bot @@ -279,7 +310,10 @@ export function createStallGuard(options: StallGuardOptions): StallGuard { for (;;) { const { done, value } = await reader.read(); if (done) break; + watchdog.record(id); + watchdog.pause(id); await writer.write(value); + watchdog.resume(id); } if (release(id)) await writer.close().catch(() => undefined); } catch (error) { @@ -326,9 +360,10 @@ function turnOf( * dependency to produce them would be the larger change. The AG-UI client parses `data:` lines as * JSON and validates them against its own schemas, so this is the same shape a Bot would have sent. * - * The wording avoids " - " and a trailing three-digit number. The packaged chat's banner truncates a - * message at the first of those and strips the second, so a sentence containing either arrives at a - * person cut in half. + * The wording is the whole of what a person gets. Both surfaces draw this message and nothing else + * around it, so it names the Bot, says what was observed rather than what was concluded, says the + * turn is over, and says what to do next. No identifiers and no milliseconds: a run id in a sentence + * is a thing the reader has to decide to ignore, and it is not what they came to find out. */ function stalledEvent(botName: string, stallMs: number): Uint8Array { const event = { diff --git a/server/src/channels/turn-watchdog.ts b/server/src/channels/turn-watchdog.ts index 9805f5e..72d6ce6 100644 --- a/server/src/channels/turn-watchdog.ts +++ b/server/src/channels/turn-watchdog.ts @@ -8,6 +8,14 @@ * Bot is allowed to do, which nobody asked for, whereas a silence limit puts a ceiling on how long a * person is asked to watch a spinner, which is the actual complaint. * + * SILENCE ONLY MEANS ANYTHING WHILE THIS PROCESS IS WAITING TO HEAR FROM THE BOT. The Bot's stream + * is relayed onward as it arrives, and whoever reads the relayed copy is entitled to take its time: + * in Intelligence mode that reader is publishing every event on to the gateway over the network. A + * chunk sitting in that handover is quiet that says nothing about the endpoint, and counting it + * would end a run that was streaming perfectly and put the blame on a Bot for a pause on this side + * of the wire. That is the one outcome a watchdog must never produce, so the clock is stopped for + * the handover and started again when it completes, which is what `pause` and `resume` are for. + * * WHAT THE STREAM ACTUALLY DOES IN THIS PRODUCT, measured before any of this was designed, because * the answer decides whether a long tool call would be reported as a stall. * @@ -38,7 +46,12 @@ /** Milliseconds, from whatever source the caller trusts. Injected so tests need not sleep. */ export type Clock = () => number; -/** One stream under watch. */ +/** + * What the watch is told about a stream, which is deliberately almost nothing: a way to find it + * again and the Bot on the far end. Never the bytes, because nothing in here is allowed to read + * them, and never the response, because holding one would make this a party to the stream rather + * than an observer of it. + */ export type WatchedStream = { /** * How the caller finds this stream again. Minted per stream rather than taken from the run, @@ -74,6 +87,8 @@ export type TurnWatchdogOptions = { type OpenStream = WatchedStream & { lastChunkAt: number; chunks: number; + /** While true the clock is not running, because the wait is not on the Bot. See `pause`. */ + paused: boolean; }; export class TurnWatchdog { @@ -93,7 +108,10 @@ export class TurnWatchdog { return this.stallMs > 0; } - /** How many streams are currently being watched. */ + /** + * Read by whoever owns the sweep, so a process with nothing to watch can stop looking rather than + * hold a timer that ticks for the life of the deployment. + */ get watching(): number { return this.streams.size; } @@ -115,11 +133,16 @@ export class TurnWatchdog { ...stream, lastChunkAt: this.now(), chunks: 0, + paused: false, }); } /** - * Note that something arrived. + * Note that something arrived from the Bot. + * + * Called when a read off the Bot's own body resolves, which is the only moment this process learns + * anything about the endpoint. Anywhere later in the relay would be measuring how promptly the + * next component took delivery instead. * * An id that is not being watched is ignored rather than started, and that is what makes the * callback fire exactly once per stream: `sweep` removes a stream as it reports it, so a chunk @@ -133,13 +156,46 @@ export class TurnWatchdog { stream.chunks += 1; } + /** + * Stop counting silence against a stream, because the wait has stopped being the Bot's. + * + * Between one chunk and the next, this process is not waiting on the endpoint at all: it is + * waiting for whoever reads the relayed stream to take the chunk it already has. That wait can be + * long for reasons that have nothing to do with the Bot, and a Bot that is streaming steadily must + * not be ended because the far side of the relay went to sleep. Counting only the stretches spent + * waiting on the endpoint is what keeps the number in the audit row true to its own name. + * + * The cost is that a stream whose reader stops for good is never reported. It is not silent from + * the Bot, so nothing here ends it, and it stays in the watch until the relay errors and the pump + * releases it. That is the right way round: believing a healthy Bot is broken ends somebody's + * turn, and believing a broken reader is healthy costs an entry in a map. + */ + pause(id: string): void { + const stream = this.streams.get(id); + if (!stream) return; + stream.paused = true; + } + + /** + * Start counting again, from now rather than from the last chunk. + * + * The stretch that just ended was not the Bot's, so it must not be carried forward into the next + * deadline; a handover that took most of the timeout would otherwise leave the Bot a sliver of it. + */ + resume(id: string): void { + const stream = this.streams.get(id); + if (!stream) return; + stream.lastChunkAt = this.now(); + stream.paused = false; + } + /** Stop watching. Unknown ids are ignored, so closing a stream that already stalled is harmless. */ close(id: string): void { this.streams.delete(id); } /** - * Report every stream that has gone quiet, and returns how many there were. + * Reports every stream that has gone quiet, and returns how many there were. * * Each one is removed before its callback runs, so a callback that throws still leaves the * watchdog consistent, and a slow callback cannot be entered twice for the same stream. @@ -150,6 +206,7 @@ export class TurnWatchdog { let stalled = 0; for (const [id, stream] of this.streams) { + if (stream.paused) continue; const silentForMs = now - stream.lastChunkAt; if (silentForMs < this.stallMs) continue; diff --git a/server/src/copilot.ts b/server/src/copilot.ts index 185db75..2425823 100644 --- a/server/src/copilot.ts +++ b/server/src/copilot.ts @@ -346,8 +346,12 @@ export function mountCopilotRuntime( resolveModelApiKey: () => Promise, identifyUser: IdentifyUser, identifyActor: IdentifyActor, - /** The watch on Bot streams. Absent means the deployment has not configured a timeout. */ - stallGuard?: StallGuard, + /** + * The watch on Bot streams. Not optional, unlike the parameter it forwards to: a guard built from + * a timeout of zero already watches nothing, so an unconfigured deployment has one to hand and + * there is no reason for a caller to have to say `undefined` here to reach `basePath`. + */ + stallGuard: StallGuard, basePath = "/api/copilotkit", ) { const { intelligence } = config.runtime; diff --git a/server/tests/copilot.test.ts b/server/tests/copilot.test.ts index 5063c8c..a2951ba 100644 --- a/server/tests/copilot.test.ts +++ b/server/tests/copilot.test.ts @@ -205,26 +205,37 @@ describe("registered Copilot agents", () => { expect(agents.risk).toBeInstanceOf(HttpAgent); }); + /* + * Told apart by a sentinel, because nothing else tells them apart. + * + * @ag-ui/client fills `fetch` in with a wrapper of its own whenever the config does not carry one, + * so a remote Bot always has a function there and asserting that it does asserts nothing at all. + * The same registration is built twice, with a guard whose watch returns a fetch nothing else + * could have produced and then without one, and the two are compared. + */ test("leaves a remote Bot's fetch alone when no timeout is configured", () => { - const agents = buildAgents( - [ - { - id: "risk", - name: "Risk", - type: "remote_ag_ui", - endpoint: "http://risk.internal/ag-ui", - }, - ], - { provider: "openai", defaultModel: "gpt-4.1" }, - null, - ); + const sentinel = async () => new Response(null); + const registered = [ + { + id: "risk", + name: "Risk", + type: "remote_ag_ui" as const, + endpoint: "http://risk.internal/ag-ui", + }, + ]; + const model = { provider: "openai" as const, defaultModel: "gpt-4.1" }; - const remote = agents.risk; - if (!(remote instanceof HttpAgent)) { + const guarded = buildAgents(registered, model, null, { + watch: () => sentinel, + stop: () => undefined, + }).risk; + const unguarded = buildAgents(registered, model, null).risk; + if (!(guarded instanceof HttpAgent) || !(unguarded instanceof HttpAgent)) { throw new Error("Expected the remote agent"); } - // @ag-ui/client defaults this to its own fetch when the config does not carry one. - expect(typeof remote.fetch).toBe("function"); + + expect(guarded.fetch).toBe(sentinel); + expect(unguarded.fetch).not.toBe(sentinel); }); test("resolves fresh built-in agents and credentials for every request", async () => { diff --git a/server/tests/stall-guard.test.ts b/server/tests/stall-guard.test.ts index 8cfb46e..9148db1 100644 --- a/server/tests/stall-guard.test.ts +++ b/server/tests/stall-guard.test.ts @@ -102,7 +102,7 @@ describe("a Bot that stops streaming", () => { }); describe("the sentence a person is left with", () => { - test("survives the packaged chat's banner, which truncates on its own punctuation", async () => { + test("is the whole explanation, because nothing is drawn around it", async () => { const guard = createStallGuard({ stallMs: 60 }); const watched = guard.watch(BOT, async () => sse(saysNothing())); @@ -113,15 +113,17 @@ describe("the sentence a person is left with", () => { const message = String( (JSON.parse(body.slice("data: ".length)) as { message: unknown }).message, ); - // The banner cuts a message at the first " - ", strips a trailing three-digit number and takes - // anything after "See more:" away with it. A sentence containing any of those reaches a person - // cut in half, so the wording avoids all three rather than relying on nobody noticing. - expect(message).not.toContain(" - "); - expect(message).not.toContain("See more:"); - expect(message).not.toMatch(/:\s*\d{3}$/); + // Both surfaces draw this message and nothing else: no banner, no heading, no error code beside + // it. So it has to name what went quiet, say the turn is over, and say what to do about it. + expect(message).toContain("Risk Analyst"); + expect(message).toContain("this turn was ended"); + expect(message).toContain("Ask again"); // Said in words a person reads, not in the milliseconds a deployment configured, and never as // "0 seconds": the timeout here is a test's, and a floor keeps the sentence sane at any value. expect(message).toContain("for a second"); + // No identifiers. The thread and run are in the audit row, where somebody is looking for them. + expect(message).not.toContain("thread-7"); + expect(message).not.toContain("run-9"); }); }); @@ -144,6 +146,28 @@ describe("a Bot that is answering", () => { expect(audit.rows).toHaveLength(0); }); + test("is not given up on because whoever reads the relay paused", async () => { + const audit = collecting(); + // Sixty milliseconds of silence ends a turn here, and the reader below waits five times that + // before it takes its first chunk. A watch pointed at the wrong end of the relay reports this + // as a Bot that never said anything, on a run that had already finished streaming. + const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); + const frames = [ + 'data: {"type":"RUN_STARTED"}\n\n', + 'data: {"type":"TEXT_MESSAGE_CHUNK","delta":"hello"}\n\n', + 'data: {"type":"RUN_FINISHED"}\n\n', + ]; + const watched = guard.watch(BOT, async () => sse(speaks(frames))); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + await Bun.sleep(300); + const body = await new Response(response.body).text(); + guard.stop(); + + expect(body).toBe(frames.join("")); + expect(audit.rows).toHaveLength(0); + }); + test("keeps the status and the content type the parser chooses on", async () => { const guard = createStallGuard({ stallMs: 5_000 }); const watched = guard.watch(BOT, async () => sse(speaks(["data: {}\n\n"]))); @@ -172,7 +196,7 @@ describe("what the watch refuses to touch", () => { expect(audit.rows).toHaveLength(0); }); - test("a framing it does not recognise is closed rather than written into", async () => { + test("a protobuf stream is closed rather than written into", async () => { const audit = collecting(); const guard = createStallGuard({ stallMs: 60, auditStore: audit.store }); const watched = guard.watch( @@ -193,6 +217,46 @@ describe("what the watch refuses to touch", () => { audit.rows.some((event) => event.eventType === "agent.stream_stalled"), ).toBe(true); }); + + test("a Bot serving events under some other content type is still told", async () => { + const guard = createStallGuard({ stallMs: 60 }); + // Not `text/event-stream`, and not the protobuf media type either. The AG-UI client parses + // anything but the latter as server-sent events, so this is a stream a sentence reaches, and a + // rule stricter than the client's would have ended this run in silence on both surfaces. + const watched = guard.watch( + BOT, + async () => + new Response(saysNothing(), { + headers: { "content-type": "application/json" }, + }), + ); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + const body = await new Response(response.body).text(); + guard.stop(); + + expect(body).toContain("Risk Analyst stopped responding"); + }); +}); + +describe("the recovery a person is waiting on", () => { + test("does not wait on the database it also writes a row to", async () => { + // An audit store whose insert never settles, which is what a saturated pool or an unreachable + // Postgres looks like from here. It is also the condition a Bot is most likely to hang in, so a + // recovery sequenced behind this write would fail open in exactly the case it exists for. + const guard = createStallGuard({ + stallMs: 60, + auditStore: { insert: () => new Promise(() => undefined) }, + }); + const watched = guard.watch(BOT, async () => sse(saysNothing())); + + const response = await watched("http://bot.internal/ag-ui", RUN_REQUEST); + // This returns only if the stream was closed. Nothing else ever closes it. + const body = await new Response(response.body).text(); + guard.stop(); + + expect(body).toContain("Risk Analyst stopped responding"); + }); }); describe("a deployment with no timeout configured", () => { diff --git a/server/tests/turn-watchdog.test.ts b/server/tests/turn-watchdog.test.ts index 2b45920..f3924ca 100644 --- a/server/tests/turn-watchdog.test.ts +++ b/server/tests/turn-watchdog.test.ts @@ -88,6 +88,37 @@ describe("a turn is judged on activity, not on how long it has run", () => { expect(stalled[0]?.chunks).toBe(2); }); + test("silence while the relay's reader has not taken delivery is not charged to the Bot", () => { + const { time, stalled, watchdog } = watching(60_000); + watchdog.open({ id: "stream-1", botId: "risk-analyst" }); + + // The Bot spoke and this process is now holding that chunk out to whoever reads the relayed + // stream. In this deployment that is the Intelligence runner, publishing over the network. + watchdog.record("stream-1"); + watchdog.pause("stream-1"); + time.advance(600_000); + expect(watchdog.sweep()).toBe(0); + expect(stalled).toHaveLength(0); + + // Ten minutes of it, and the Bot still gets its whole timeout from the moment it is waited on + // again. The wait that just ended was not its. + watchdog.resume("stream-1"); + time.advance(59_999); + expect(watchdog.sweep()).toBe(0); + + time.advance(1); + expect(watchdog.sweep()).toBe(1); + expect(stalled[0]?.silentForMs).toBe(60_000); + expect(stalled[0]?.chunks).toBe(1); + }); + + test("pausing or resuming a stream nobody is watching is harmless", () => { + const { watchdog } = watching(60_000); + expect(() => watchdog.pause("never-opened")).not.toThrow(); + expect(() => watchdog.resume("never-opened")).not.toThrow(); + expect(watchdog.watching).toBe(0); + }); + test("a stream that ends is forgotten and is never reported", () => { const { time, stalled, watchdog } = watching(60_000); watchdog.open({ id: "stream-1", botId: "risk-analyst" });