From 27b8165cc757c1afe7947512e15f0962c1c93aff Mon Sep 17 00:00:00 2001 From: David McKay Date: Fri, 13 Nov 2026 20:30:45 -0800 Subject: [PATCH 1/2] End a turn whose run went quiet once the runtime says it has no run, or can't say With the service that carries a conversation's events stalled mid-run, the reply stopped half written and its step said "Running", with Stop up, for good: the server had given the run up, and nothing could tell the browser. After 45 seconds with nothing arriving from a run the runtime is asked whether the conversation still has one going, and again every 45 seconds while it says yes. When it says no, or cannot answer, the turn ends with a sentence saying so and what was saved is read back. --- app/src/components/channels/channel-chat.tsx | 58 ++++++- app/src/lib/copilot/quiet-run.ts | 160 ++++++++++++++++++ app/tests/quiet-run.test.ts | 168 +++++++++++++++++++ 3 files changed, 383 insertions(+), 3 deletions(-) create mode 100644 app/src/lib/copilot/quiet-run.ts create mode 100644 app/tests/quiet-run.test.ts diff --git a/app/src/components/channels/channel-chat.tsx b/app/src/components/channels/channel-chat.tsx index 2891b618c5..2bc0cc17a0 100644 --- a/app/src/components/channels/channel-chat.tsx +++ b/app/src/components/channels/channel-chat.tsx @@ -1,4 +1,4 @@ -import type { Message } from "@ag-ui/core"; +import { EventType, type Message } from "@ag-ui/core"; import { type Attachment, CopilotChatConfigurationProvider, @@ -77,6 +77,12 @@ import { offerItsOwnTools } from "@/lib/copilot/bot-offer"; import { isBrokenStream, STREAM_BROKE } from "@/lib/copilot/broken-stream"; import { ConversationProvider } from "@/lib/copilot/conversation"; import { afterMs, joinWithin } from "@/lib/copilot/join-thread"; +import { + askRunLiveness, + RUN_ENDED_UNHEARD, + RUN_UNREACHABLE, + watchQuietRun, +} from "@/lib/copilot/quiet-run"; import { repairUnansweredToolCalls, unansweredToolCalls, @@ -1308,9 +1314,55 @@ export function ChannelChat({ * says so under the turn rather than looping. */ for (let carriedOn = false; ; carriedOn = true) { - await withVoiceContext(copilotkit, target.agentId, voiceHistory, () => - whileObserved(target, () => copilotkit.runAgent({ agent: target })), + /* + * A RUN THAT GOES QUIET IS ASKED ABOUT, and ended here only on the runtime's answer: no run + * going, or no answer (`quiet-run.ts`). Its events reach this browser by a service that can + * stall, and a run the server gave up then looked like one still thinking: "Running" and + * Stop, for good. A browser step taken between this turn's runs is quiet of its own, so the + * watch covers one run at a time. + */ + const quiet = watchQuietRun({ + ask: () => askRunLiveness(runtimeAgentId, channel.threadId), + }); + const hearing = target.subscribe({ + onEvent: ({ event }) => { + // Between the server's runs the turn is this browser's (a step it takes, a card the + // person answers), and quiet there is not a run gone missing. + if ( + event.type === EventType.RUN_FINISHED || + event.type === EventType.RUN_ERROR + ) + quiet.rest(); + else quiet.heard(); + }, + }); + const run = withVoiceContext( + copilotkit, + target.agentId, + voiceHistory, + () => + whileObserved(target, () => copilotkit.runAgent({ agent: target })), ); + const ended = await Promise.race([ + run.then(() => "ran" as const), + quiet.over, + ]).finally(() => { + quiet.stop(); + hearing.unsubscribe(); + }); + if (ended !== "ran") { + // The run is let go of here; whatever it does later is not this turn's any more. + run.catch(() => undefined); + void target.detachActiveRun().catch(() => undefined); + awaitingReply.current = false; + // What the thread holds is what was saved: read back when the runtime can be asked. + if (ended === "over") { + const whole = await readBackFromThread(target, existingMessageIds); + if (whole) reportReadBack(target); + } + setRunError(ended === "over" ? RUN_ENDED_UNHEARD : RUN_UNREACHABLE); + break; + } if (!streamBroke.current || turnFailure.current !== null) break; streamBroke.current = false; awaitingReply.current = false; diff --git a/app/src/lib/copilot/quiet-run.ts b/app/src/lib/copilot/quiet-run.ts new file mode 100644 index 0000000000..7dfd1c941d --- /dev/null +++ b/app/src/lib/copilot/quiet-run.ts @@ -0,0 +1,160 @@ +import { type ClientOptions, tryClient } from "@/lib/client"; +import { newId } from "../new-id"; + +/** + * A run that has gone quiet, and whether it is still a run. + * + * FOUND IN TESTING, with the service that carries a conversation's events stalled mid-run. The Bot's + * reply stopped half written and the step under it said "Running", with Stop up, for as long as + * anybody watched. The server had given the run up after a minute. Nothing told this browser, + * because a run's events reach it by that same service, and a run that ends without its last event + * looks exactly like one that is thinking. + * + * SILENCE IS NOT THE SIGNAL. A Bot can think, or wait on an app, for minutes. After + * {@link QUIET_RUN_MS} with nothing arriving, the runtime is ASKED whether the conversation still + * has a run going, and asked again every {@link QUIET_RUN_MS} while it says yes. Only its answer + * ends the turn: no run going, or no answer at all. + */ + +/** How long nothing may arrive from a run before the runtime is asked about it. */ +export const QUIET_RUN_MS = 45_000; + +/** How long the runtime is given to answer. Past this it has not answered. */ +export const LIVENESS_ANSWER_MS = 15_000; + +/** + * What the runtime said: the conversation has a run going, it has none, or it could not say (it + * refused, failed, or did not answer in time). + */ +export type RunLiveness = "live" | "over" | "unreachable"; + +/** Said under a turn whose run ended without telling this browser. */ +export const RUN_ENDED_UNHEARD = + "The Bot stopped before it finished. What it saved is shown here. Send a message to carry on."; + +/** Said under a turn whose run could not be asked about. */ +export const RUN_UNREACHABLE = + "OpenBot lost touch with the Bot while it was working. Your message was sent. Try again in a moment, or reload to see what was saved."; + +/** + * Ask the runtime whether this conversation has a run going, by the call a conversation is joined + * with: it answers the live run's id when there is one, and none when there is not. + */ +export async function askRunLiveness( + agentId: string, + threadId: string, + options: { + request?: (path: string, options: ClientOptions) => Promise; + answerMs?: number; + } = {}, +): Promise { + const request = options.request ?? tryClient; + const answerMs = options.answerMs ?? LIVENESS_ANSWER_MS; + const giveUp = new AbortController(); + const timer = setTimeout(() => giveUp.abort(), answerMs); + try { + const response = await Promise.race([ + request(`/api/copilotkit/agent/${encodeURIComponent(agentId)}/connect`, { + method: "POST", + body: { + threadId, + runId: newId(), + messages: [], + tools: [], + context: [], + state: {}, + forwardedProps: {}, + }, + signal: giveUp.signal, + }), + // A request that ignores its signal has still not answered. + new Promise((resolve) => { + giveUp.signal.addEventListener("abort", () => resolve(null), { + once: true, + }); + }), + ]); + if (response === null) return "unreachable"; + if (response.status === 204) return "over"; + if (!response.ok) return "unreachable"; + const body: unknown = await response.json().catch(() => null); + if (body === null || typeof body !== "object") return "unreachable"; + return typeof (body as { runId?: unknown }).runId === "string" + ? "live" + : "over"; + } catch { + return "unreachable"; + } finally { + clearTimeout(timer); + } +} + +/** + * Watch a run for quiet. `heard` is called for everything that arrives from it, and `rest` when + * the server's run ends with the turn still going here. After `quietMs` + * with nothing, `ask` is asked, and again each `quietMs` while it answers "live". The first other + * answer settles `over`, once, and the watch stops. `stop` ends it with nothing settled. + */ +export function watchQuietRun(options: { + ask: () => Promise; + quietMs?: number; +}): { + heard: () => void; + /** + * The run on the server has ended and the turn is this browser's for now (a step it takes + * itself, a card the person answers): nothing is expected, so nothing is asked, until `heard`. + */ + rest: () => void; + stop: () => void; + over: Promise>; +} { + const quietMs = options.quietMs ?? QUIET_RUN_MS; + let timer: ReturnType | undefined; + let stopped = false; + let asking = false; + // Whether anything arrived while the runtime was being asked: its answer is then out of date. + let heardSince = false; + let settle: (outcome: Exclude) => void = () => {}; + const over = new Promise>((resolve) => { + settle = resolve; + }); + const wait = () => { + clearTimeout(timer); + if (stopped) return; + timer = setTimeout(() => { + asking = true; + heardSince = false; + void options + .ask() + .catch(() => "unreachable" as const) + .then((answer) => { + asking = false; + if (stopped) return; + // Resting: the answer is about a run that has since ended in the ordinary way. + if (heardSince && timer === undefined) return; + if (answer === "live" || heardSince) return wait(); + stopped = true; + settle(answer); + }); + }, quietMs); + }; + wait(); + return { + // Something arrived: the quiet starts again, and an answer on its way is about a run that + // has since spoken, so it is not taken. + heard: () => { + if (asking) heardSince = true; + else wait(); + }, + rest: () => { + heardSince = true; + clearTimeout(timer); + timer = undefined; + }, + stop: () => { + stopped = true; + clearTimeout(timer); + }, + over, + }; +} diff --git a/app/tests/quiet-run.test.ts b/app/tests/quiet-run.test.ts new file mode 100644 index 0000000000..54f0484546 --- /dev/null +++ b/app/tests/quiet-run.test.ts @@ -0,0 +1,168 @@ +import { expect, test } from "bun:test"; +import { + askRunLiveness, + type RunLiveness, + watchQuietRun, +} from "@/lib/copilot/quiet-run"; + +/* + * FOUND IN TESTING, with the service that carries a conversation's events frozen mid-run. The reply + * stopped half written and the step said "Running", with Stop up, for as long as anybody watched: + * the server had given the run up, and nothing could tell this browser. A quiet run is now asked + * about, and ended on the runtime's answer, never on the quiet alone. + */ + +const later = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +/** The runtime's connect call, answering as it does. */ +function runtime(answer: () => Promise) { + const asked: { path: string; threadId: unknown }[] = []; + return { + asked, + request: async (path: string, options: { body?: unknown }) => { + asked.push({ + path, + threadId: (options.body as { threadId?: unknown }).threadId, + }); + return answer(); + }, + }; +} + +test("the runtime's answer says whether a conversation still has a run going", async () => { + const live = runtime(async () => + Response.json({ threadId: "t1", runId: "run-7", joinToken: "j" }), + ); + expect( + await askRunLiveness("personal-assistant", "t1", { request: live.request }), + ).toBe("live"); + expect(live.asked).toEqual([ + { + path: "/api/copilotkit/agent/personal-assistant/connect", + threadId: "t1", + }, + ]); + + // Joined, with no run named: the conversation has none going. + const idle = runtime(async () => + Response.json({ threadId: "t1", joinToken: "j" }), + ); + expect( + await askRunLiveness("personal-assistant", "t1", { request: idle.request }), + ).toBe("over"); + const nothing = runtime(async () => new Response(null, { status: 204 })); + expect( + await askRunLiveness("personal-assistant", "t1", { + request: nothing.request, + }), + ).toBe("over"); + + // It failed, as it did while the service behind it answered 500. + const failing = runtime(async () => + Response.json({ error: "Connect request failed" }, { status: 500 }), + ); + expect( + await askRunLiveness("personal-assistant", "t1", { + request: failing.request, + }), + ).toBe("unreachable"); +}); + +test("a runtime that never answers, as with the service behind it frozen, has not answered", async () => { + const frozen = runtime(() => new Promise(() => {})); + const started = Date.now(); + expect( + await askRunLiveness("personal-assistant", "t1", { + request: frozen.request, + answerMs: 40, + }), + ).toBe("unreachable"); + expect(Date.now() - started).toBeLessThan(1000); +}); + +test("a quiet run is asked about, kept while the runtime says it is going, and ended when it says it is not", async () => { + const answers: RunLiveness[] = ["live", "live", "over"]; + let asked = 0; + const watch = watchQuietRun({ + quietMs: 20, + ask: async () => { + asked += 1; + return answers.shift() ?? "over"; + }, + }); + expect(await watch.over).toBe("over"); + // Alive but quiet twice over, and still waited for. + expect(asked).toBe(3); +}); + +test("a run that keeps speaking is never asked about, and nothing is asked once it stops being watched", async () => { + let asked = 0; + const watch = watchQuietRun({ + quietMs: 30, + ask: async () => { + asked += 1; + return "over"; + }, + }); + for (let index = 0; index < 8; index += 1) { + await later(10); + watch.heard(); + } + expect(asked).toBe(0); + watch.stop(); + await later(80); + expect(asked).toBe(0); +}); + +test("between a turn's runs, with the turn in this browser's hands, nothing is asked until the next run speaks", async () => { + let asked = 0; + let settled: string | undefined; + const watch = watchQuietRun({ + quietMs: 20, + ask: async () => { + asked += 1; + return "over"; + }, + }); + void watch.over.then((outcome) => { + settled = outcome; + }); + // The server's run ended on a step the browser takes: quiet here is not a run gone missing. + watch.rest(); + await later(90); + expect(asked).toBe(0); + expect(settled).toBeUndefined(); + // The next run speaks, then goes quiet for good. + watch.heard(); + expect(await watch.over).toBe("over"); + expect(asked).toBe(1); +}); + +test("an answer about a run that spoke while the runtime was being asked is not taken", async () => { + let release: (answer: RunLiveness) => void = () => {}; + let asked = 0; + let settled: string | undefined; + const watch = watchQuietRun({ + quietMs: 20, + ask: () => { + asked += 1; + return asked === 1 + ? new Promise((resolve) => { + release = resolve; + }) + : Promise.resolve("live" as const); + }, + }); + void watch.over.then((outcome) => { + settled = outcome; + }); + await later(40); + expect(asked).toBe(1); + // It speaks while the first question is still out, and the answer then says it had no run. + watch.heard(); + release("over"); + await later(10); + expect(settled).toBeUndefined(); + watch.stop(); +}); From eeb41cf84a0c9f3a1a12afd8f4fe055f0e73d656 Mon Sep 17 00:00:00 2001 From: David McKay Date: Fri, 13 Nov 2026 20:45:45 -0800 Subject: [PATCH 2/2] Don't say the message was sent when the run could not be asked about --- app/src/lib/copilot/quiet-run.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/app/src/lib/copilot/quiet-run.ts b/app/src/lib/copilot/quiet-run.ts index 7dfd1c941d..ebf6285457 100644 --- a/app/src/lib/copilot/quiet-run.ts +++ b/app/src/lib/copilot/quiet-run.ts @@ -32,9 +32,12 @@ export type RunLiveness = "live" | "over" | "unreachable"; export const RUN_ENDED_UNHEARD = "The Bot stopped before it finished. What it saved is shown here. Send a message to carry on."; -/** Said under a turn whose run could not be asked about. */ +/** + * Said under a turn whose run could not be asked about. It does not say the message was sent: a + * run that never started goes quiet the same way, and nothing here can tell the two apart. + */ export const RUN_UNREACHABLE = - "OpenBot lost touch with the Bot while it was working. Your message was sent. Try again in a moment, or reload to see what was saved."; + "OpenBot lost touch with the Bot before it finished. Try again in a moment, or reload to see what was saved."; /** * Ask the runtime whether this conversation has a run going, by the call a conversation is joined