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..ebf6285457 --- /dev/null +++ b/app/src/lib/copilot/quiet-run.ts @@ -0,0 +1,163 @@ +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. 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 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 + * 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(); +});