diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 4e5947dec..2a736d8f0 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -103,6 +103,7 @@ import { import { applyInsightsMigrations, createDrizzleRunTraceReader, + createDrizzleTurnTextSnapshotReader, createInsightsRoutes, createPostgresTurnLatencyStore, createPostgresUsageStore, @@ -1288,6 +1289,8 @@ export async function createHub(config: HubConfig) { tenancy: chatTenancy, threads: threadStore, agentTurns, + turnTextSnapshot: (input) => + createDrizzleTurnTextSnapshotReader(db).read(input), blockResponses: blockResponseStore, reactions: reactionStore, pins: pinStore, diff --git a/packages/chat-ui/src/api.ts b/packages/chat-ui/src/api.ts index 3ecfef4c1..a782ddc0d 100644 --- a/packages/chat-ui/src/api.ts +++ b/packages/chat-ui/src/api.ts @@ -1236,6 +1236,81 @@ export function patchBenchChatSettings( ); } +// The turn projection's read surface (CL-6329/CL-6380): what a client +// reattaching to a workbench (page navigation, tab refocus, a dropped SSE +// connection) uses to find whether a turn is still running and, if so, +// replay whatever text it has already committed before the live stream's +// tail resumes — see `GET /workbenches/:id/turns[/:turnId]` in +// `packages/chat/src/routes.ts`. +const AgentTurnWire = type({ + id: "string", + workbenchId: "string", + agentAddress: "string", + childRunId: "string", + status: "'running' | 'completed' | 'failed'", + "replyMessageId?": "string | null", +}); +export type AgentTurnSummary = typeof AgentTurnWire.infer; + +const AgentTurnsListWire = type({ items: AgentTurnWire.array() }); + +function turnsPath(tenantId: string, workbenchId: string): string { + return `/api/tenants/${tenantId}/chat/workbenches/${workbenchId}/turns`; +} + +export function listWorkbenchTurns( + tenantId: string, + workbenchId: string, +): Promise { + return request(turnsPath(tenantId, workbenchId), AgentTurnsListWire).then( + (body) => body.items, + ); +} + +const AgentTurnDetailWire = AgentTurnWire.and({ + "textSnapshot?": "string | null", +}); +export type AgentTurnDetail = typeof AgentTurnDetailWire.infer; + +export function getWorkbenchTurn( + tenantId: string, + workbenchId: string, + turnId: string, +): Promise { + return request( + `${turnsPath(tenantId, workbenchId)}/${turnId}`, + AgentTurnDetailWire, + ); +} + +/** + * The newest still-`running` turn for `agentAddress`, or `null` if none — + * what a remounting workbench asks on mount to know whether to hydrate its + * streaming indicator immediately rather than wait for the next live event. + * A 404 (no turn store injected on this deployment) reads the same as "no + * running turn": the feature is simply unavailable, never an error the + * caller needs to handle. + */ +export async function fetchRunningTurn( + tenantId: string, + workbenchId: string, + agentAddress: string, +): Promise { + let turns: readonly AgentTurnSummary[]; + try { + turns = await listWorkbenchTurns(tenantId, workbenchId); + } catch (cause) { + if (cause instanceof ChatApiError && cause.status === 404) return null; + throw cause; + } + const running = turns.find( + (turn) => turn.status === "running" && turn.agentAddress === agentAddress, + ); + if (running === undefined) return null; + const detail = await getWorkbenchTurn(tenantId, workbenchId, running.id); + return { ...detail, textSnapshot: detail.textSnapshot ?? null }; +} + /** * A readable name for a run, since the runs listing carries no name field: * the asset id's final path segment with any extension stripped, e.g. diff --git a/packages/chat-ui/src/chat-workspace.tsx b/packages/chat-ui/src/chat-workspace.tsx index 8874dc5ba..905c2dcda 100644 --- a/packages/chat-ui/src/chat-workspace.tsx +++ b/packages/chat-ui/src/chat-workspace.tsx @@ -28,6 +28,7 @@ import { workbenchesQueryKey, workbenchesQueryKeyPrefix, describeChatError, + fetchRunningTurn, inviteAgent, listWorkbenches, listInvitableDefinitions, @@ -626,6 +627,7 @@ function ChatWorkspaceInner({ replyTimedOut, handleStreamEvent: handleStreamingReplyEvent, noteAwaitingReply, + resumeFromTurn, } = useStreamingReply(activeWorkbenchId); const { activity: turnActivity, handleStreamEvent: handleTurnActivityEvent } = useTurnActivity(activeWorkbenchId); @@ -790,6 +792,34 @@ function ChatWorkspaceInner({ (participant) => isAgentAddress(participant.address), ); + const resumeAgentAddress = (activeWorkbench?.participants ?? []).find( + (participant) => isAgentAddress(participant.address), + )?.address; + + // CL-6380: a turn runs entirely server-side — this component mounting or + // unmounting never starts or stops it (see `useWorkbenchStream`'s own + // header: unmount only closes the `EventSource`, nothing server-side). + // So a fresh mount (first visit, or a return after navigating away while + // a reply was still streaming) asks once whether the agent has a turn + // still running and, if so, replays its committed text immediately + // rather than showing nothing until the next live token arrives. Any + // live event that beats this fetch back always wins — see + // `resumeFromTurn`'s own guard. + useEffect(() => { + if (activeWorkbenchId === null || resumeAgentAddress === undefined) { + return; + } + let cancelled = false; + fetchRunningTurn(tenantId, activeWorkbenchId, resumeAgentAddress) + .then((runningTurn) => { + if (!cancelled) resumeFromTurn(runningTurn); + }) + .catch(() => undefined); + return () => { + cancelled = true; + }; + }, [tenantId, activeWorkbenchId, resumeAgentAddress]); + const { pendingSends, handleSend, retryPendingSend, discardPendingSend } = useOptimisticSends({ tenantId, diff --git a/packages/chat-ui/src/streaming-reply.test.ts b/packages/chat-ui/src/streaming-reply.test.ts index 72d661d46..81c3d3579 100644 --- a/packages/chat-ui/src/streaming-reply.test.ts +++ b/packages/chat-ui/src/streaming-reply.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { + hydrateStreamingReplyFromTurn, nextStreamingReplyState, openPendingReply, typingAgentNames, @@ -209,3 +210,21 @@ describe("typingAgentNames", () => { expect(typingAgentNames({ text: "" }, [HUMAN])).toEqual([]); }); }); + +describe("hydrateStreamingReplyFromTurn (CL-6380: reattach snapshot)", () => { + test("no running turn resumes to nothing", () => { + expect(hydrateStreamingReplyFromTurn(null)).toBeNull(); + }); + + test("a running turn with committed text opens the reply carrying it", () => { + expect( + hydrateStreamingReplyFromTurn({ textSnapshot: "streamed so far" }), + ).toEqual({ text: "streamed so far" }); + }); + + test("a running turn with no text yet opens the same empty pending pulse as openPendingReply", () => { + expect(hydrateStreamingReplyFromTurn({ textSnapshot: null })).toEqual({ + text: "", + }); + }); +}); diff --git a/packages/chat-ui/src/streaming-reply.ts b/packages/chat-ui/src/streaming-reply.ts index bf96e2d85..a08a40250 100644 --- a/packages/chat-ui/src/streaming-reply.ts +++ b/packages/chat-ui/src/streaming-reply.ts @@ -109,6 +109,23 @@ export function openPendingReply( return current ?? { text: "" }; } +/** + * The catch-up snapshot a client reattaching mid-turn (a fresh mount after + * navigating away and back, CL-6380) hydrates its streaming reply with, + * before the live SSE tail resumes: a running turn with committed text + * opens the reply already carrying it; a running turn with none yet (still + * in its first inference call) opens the same empty pending state + * `openPendingReply` would; no running turn at all means there's nothing to + * resume. Never called once a live event has already produced state — see + * `resumeFromTurn`'s own guard below. + */ +export function hydrateStreamingReplyFromTurn( + runningTurn: { readonly textSnapshot?: string | null } | null, +): StreamingReplyState { + if (runningTurn === null) return null; + return { text: runningTurn.textSnapshot ?? "" }; +} + /** How long an empty pending reply may sit with no tokens before the * indicator clears itself — the backstop for a turn whose stream events * never arrive (agent down, SSE dropped mid-reconnect). */ @@ -133,6 +150,10 @@ export function useStreamingReply( readonly replyTimedOut: boolean; readonly handleStreamEvent: (eventType: string, data: unknown) => void; readonly noteAwaitingReply: () => void; + /** See `resumeFromTurn`'s own doc comment below. */ + readonly resumeFromTurn: ( + runningTurn: { readonly textSnapshot?: string | null } | null, + ) => void; } { const [streamingReply, setStreamingReply] = useState(null); @@ -220,11 +241,32 @@ export function useStreamingReply( ); } + /** + * Applies a fetched turn-state snapshot (see `api.ts`'s + * `fetchRunningTurn`) on a fresh mount, before any live event has + * arrived. Guarded to only ever fill an empty (`null`) state — a stream + * event that already opened or grew the reply always wins, since it is + * strictly newer than a snapshot fetched moments earlier over a separate + * request. A `null` turn (nothing running) is a no-op, not a reset: it + * must never clear a reply a fast SSE `reactor.start` already opened + * while the snapshot fetch was in flight. + */ + function resumeFromTurn( + runningTurn: { readonly textSnapshot?: string | null } | null, + ) { + if (runningTurn === null) return; + setReplyTimedOut(false); + setStreamingReply( + (current) => current ?? hydrateStreamingReplyFromTurn(runningTurn), + ); + } + return { streamingReply, replyTimedOut, handleStreamEvent, noteAwaitingReply, + resumeFromTurn, }; } diff --git a/packages/chat-ui/test/use-streaming-reply.test.tsx b/packages/chat-ui/test/use-streaming-reply.test.tsx index 443528064..aed893a01 100644 --- a/packages/chat-ui/test/use-streaming-reply.test.tsx +++ b/packages/chat-ui/test/use-streaming-reply.test.tsx @@ -26,6 +26,9 @@ function mount( let send: (eventType: string, data: unknown) => void = () => {}; let setWorkbenchId: (id: string | null) => void = () => {}; let awaitReply: () => void = () => {}; + let resume: ( + runningTurn: { readonly textSnapshot: string | null } | null, + ) => void = () => {}; function Host() { const [workbenchId, updateWorkbenchId] = useState(initialWorkbenchId); @@ -35,11 +38,13 @@ function mount( replyTimedOut, handleStreamEvent, noteAwaitingReply, + resumeFromTurn, } = useStreamingReply(workbenchId, clearMs, minVisibleMs); latestState = streamingReply; latestTimedOut = replyTimedOut; send = handleStreamEvent; awaitReply = noteAwaitingReply; + resume = resumeFromTurn; return null; } @@ -60,6 +65,12 @@ function mount( act(() => { awaitReply(); }), + resumeFromTurn: ( + runningTurn: { readonly textSnapshot: string | null } | null, + ) => + act(() => { + resume(runningTurn); + }), settle: (ms: number) => act(() => sleep(ms)), get: () => latestState, timedOut: () => latestTimedOut, @@ -186,6 +197,49 @@ describe("useStreamingReply's reply-timeout backstop (CL-6252 #6)", () => { }); }); +describe("useStreamingReply.resumeFromTurn (CL-6380: catch-up on remount)", () => { + test("hydrates the reply from a running turn's committed text on a fresh mount", () => { + const harness = mount("chan_a"); + expect(harness.get()).toBeNull(); + + harness.resumeFromTurn({ textSnapshot: "already streamed so far" }); + expect(harness.get()).toEqual({ text: "already streamed so far" }); + harness.unmount(); + }); + + test("a running turn with no text yet opens the same empty pending pulse as noteAwaitingReply", () => { + const harness = mount("chan_a"); + harness.resumeFromTurn({ textSnapshot: null }); + expect(harness.get()).toEqual({ text: "" }); + harness.unmount(); + }); + + test("no running turn is a no-op, not a reset", () => { + const harness = mount("chan_a"); + harness.send("chat.agent", { + type: "inference.start", + seq: 0, + data: { model: "x" }, + }); + harness.send("chat.agent", delta("hi")); + expect(harness.get()).toEqual({ text: "hi" }); + + harness.resumeFromTurn(null); + expect(harness.get()).toEqual({ text: "hi" }); + harness.unmount(); + }); + + test("a live event that already opened the reply wins over a slower snapshot fetch", () => { + const harness = mount("chan_a"); + harness.send("chat.agent", delta("live wins")); + expect(harness.get()).toEqual({ text: "live wins" }); + + harness.resumeFromTurn({ textSnapshot: "stale snapshot" }); + expect(harness.get()).toEqual({ text: "live wins" }); + harness.unmount(); + }); +}); + describe("useStreamingReply's typing-pulse floor", () => { test("a token arriving immediately still leaves the empty pulse up until minVisibleMs", async () => { const harness = mount("chan_a", undefined, 40); diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index cc771ba4a..e6102ce32 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -209,6 +209,19 @@ export type CreateChatRoutesDeps = { * contract `pins` and `blockResponses` already follow. */ agentTurns?: AgentTurnStore; + /** + * The visible text a still-running turn has committed to the platform's + * own `turn_part` rows so far (CL-6380) — what a client reattaching + * mid-turn replays as its catch-up snapshot before the live stream's tail + * resumes. Omitted, a running turn's detail simply carries no + * `textSnapshot` field and the client falls back to showing only the + * indicator until the next live event arrives — the same "no store, no + * feature" contract every other optional dep here follows. + */ + turnTextSnapshot?: (input: { + readonly tenantId: string; + readonly runId: string; + }) => Promise; /** * Poll/form response storage — see `./block-responses.ts`. Omitted * entirely, the response routes 404 rather than silently accepting @@ -3651,7 +3664,17 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { if (turn === undefined || turn.workbenchId !== workbenchId) { return c.json(ErrorEnvelope("not_found", "turn not found"), 404); } - return c.json(turn); + // Only a still-running turn gets a catch-up snapshot attached — a + // settled turn's reply already lives in the timeline as an ordinary + // message, and re-deriving its text here would just duplicate it. + const textSnapshot = + turn.status === "running" && deps.turnTextSnapshot !== undefined + ? await deps.turnTextSnapshot({ + tenantId: access.ownerTenantId, + runId: turn.childRunId, + }) + : null; + return c.json({ ...turn, textSnapshot }); }, ); diff --git a/packages/chat/test/agent-turns-routes.test.ts b/packages/chat/test/agent-turns-routes.test.ts index f8bcac179..3d10fdf77 100644 --- a/packages/chat/test/agent-turns-routes.test.ts +++ b/packages/chat/test/agent-turns-routes.test.ts @@ -120,6 +120,92 @@ describe("GET /workbenches/:id/turns/:turnId", () => { expect(turn.error).toBe("the agent never answered"); }); + test("a running turn carries a catch-up textSnapshot from the injected reader", async () => { + const agentTurns = createInMemoryAgentTurnStore(); + const deps = buildDeps({ + agentTurns, + turnTextSnapshot: async ({ runId }) => + runId === "turn__0" ? "streamed so far" : null, + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const workbenchId = ( + await createWorkbench(app, { kind: "workbench", name: "room" }) + ).body.id; + + const opened = await agentTurns.startTurn({ + tenantId: TENANT.id, + workbenchId, + agentAddress: "ins_echo1@acme.example", + requestMessageIds: ["msg_1"], + }); + + const res = await app.request( + `/workbenches/${workbenchId}/turns/${opened.id}`, + ); + const turn = (await res.json()) as AgentTurn & { + textSnapshot: string | null; + }; + expect(turn.status).toBe("running"); + expect(turn.textSnapshot).toBe("streamed so far"); + }); + + test("a settled turn never carries a textSnapshot — its reply is already a message", async () => { + const agentTurns = createInMemoryAgentTurnStore(); + const deps = buildDeps({ + agentTurns, + turnTextSnapshot: async () => "should never be read", + }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const workbenchId = ( + await createWorkbench(app, { kind: "workbench", name: "room" }) + ).body.id; + + const opened = await agentTurns.startTurn({ + tenantId: TENANT.id, + workbenchId, + agentAddress: "ins_echo1@acme.example", + requestMessageIds: ["msg_1"], + }); + await agentTurns.finishTurn({ + tenantId: TENANT.id, + turnId: opened.id, + status: "completed", + replyMessageId: "msg_reply", + }); + + const res = await app.request( + `/workbenches/${workbenchId}/turns/${opened.id}`, + ); + const turn = (await res.json()) as AgentTurn & { + textSnapshot: string | null; + }; + expect(turn.textSnapshot).toBeNull(); + }); + + test("a running turn with no reader injected reads back a null textSnapshot", async () => { + const agentTurns = createInMemoryAgentTurnStore(); + const deps = buildDeps({ agentTurns }); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const workbenchId = ( + await createWorkbench(app, { kind: "workbench", name: "room" }) + ).body.id; + + const opened = await agentTurns.startTurn({ + tenantId: TENANT.id, + workbenchId, + agentAddress: "ins_echo1@acme.example", + requestMessageIds: ["msg_1"], + }); + + const res = await app.request( + `/workbenches/${workbenchId}/turns/${opened.id}`, + ); + const turn = (await res.json()) as AgentTurn & { + textSnapshot: string | null; + }; + expect(turn.textSnapshot).toBeNull(); + }); + test("a turn belonging to another workbench is not found here", async () => { const agentTurns = createInMemoryAgentTurnStore(); const deps = buildDeps({ agentTurns }); diff --git a/packages/insights/src/index.ts b/packages/insights/src/index.ts index d18174cd6..852dee5d1 100644 --- a/packages/insights/src/index.ts +++ b/packages/insights/src/index.ts @@ -74,3 +74,8 @@ export { withTurnPartPersistGuard, turnPartPersistFailures, } from "./turn-part-write-guard"; +export { + createDrizzleTurnTextSnapshotReader, + snapshotTextFromParts, + type TurnTextSnapshotReader, +} from "./turn-text-snapshot"; diff --git a/packages/insights/src/turn-text-snapshot.test.ts b/packages/insights/src/turn-text-snapshot.test.ts new file mode 100644 index 000000000..844f37c79 --- /dev/null +++ b/packages/insights/src/turn-text-snapshot.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, test } from "bun:test"; + +import { snapshotTextFromParts } from "./turn-text-snapshot"; + +describe("snapshotTextFromParts", () => { + test("concatenates text parts within a turn by ordinal", () => { + const text = snapshotTextFromParts( + ["turn_1"], + [ + { turnId: "turn_1", type: "text", content: "world", ordinal: 1 }, + { turnId: "turn_1", type: "text", content: "hello ", ordinal: 0 }, + ], + ); + expect(text).toBe("hello world"); + }); + + test("concatenates across turns oldest first, ignoring turns not passed", () => { + const text = snapshotTextFromParts( + ["turn_1", "turn_2"], + [ + { turnId: "turn_2", type: "text", content: "second", ordinal: 0 }, + { turnId: "turn_1", type: "text", content: "first", ordinal: 0 }, + { turnId: "turn_3", type: "text", content: "unreferenced", ordinal: 0 }, + ], + ); + expect(text).toBe("firstsecond"); + }); + + test("skips non-text parts and null content", () => { + const text = snapshotTextFromParts( + ["turn_1"], + [ + { turnId: "turn_1", type: "tool", content: "ignored", ordinal: 0 }, + { turnId: "turn_1", type: "text", content: null, ordinal: 1 }, + { turnId: "turn_1", type: "text", content: "kept", ordinal: 2 }, + ], + ); + expect(text).toBe("kept"); + }); + + test("a turn with no text parts contributes nothing, not a gap", () => { + const text = snapshotTextFromParts(["turn_1"], []); + expect(text).toBe(""); + }); +}); diff --git a/packages/insights/src/turn-text-snapshot.ts b/packages/insights/src/turn-text-snapshot.ts new file mode 100644 index 000000000..0494ed1ba --- /dev/null +++ b/packages/insights/src/turn-text-snapshot.ts @@ -0,0 +1,88 @@ +// Reads the visible text already streamed for a chat turn straight off the +// platform's own inference_turn / turn_part tables (CL-6380) — the same +// tables `trace-reader.ts` reads, no new storage. A chat `AgentTurn`'s +// `childRunId` names the workflow_run that produced it, and `inferenceTurn` +// rows key on that same run id (one inference_turn per LLM call within the +// turn, e.g. one per tool-loop step); `turn_part` rows of type "text" only +// land once their inference call reaches `inference.done` — mid-call token +// deltas never get an intermediate row — so a turn still streaming its +// first inference call reads back an empty snapshot, not an error: the +// caller's live tail (the SSE stream reattaching) is what carries the rest. +import { and, asc, eq, inArray } from "drizzle-orm"; +import type { DB } from "@intx/db"; +import { inferenceTurn, turnPart } from "@intx/db/schema"; + +export interface TurnTextSnapshotReader { + /** The visible text this turn has committed to `turn_part` so far, in + * inference-call then ordinal order — the "part cursor" catch-up a + * client reattaching mid-turn replays before the live tail resumes. + * `null` for a run with no turns yet (nothing to reconstruct), never for + * one that simply hasn't produced text yet (that reads as `""`). */ + read(input: { + readonly tenantId: string; + readonly runId: string; + }): Promise; +} + +/** Pure so the reconstruction rule is testable without a database: text + * parts only, concatenated turn-by-turn (oldest first) and, within a turn, + * ordinal-by-ordinal — mirrors the collector's own `accumulatedText` + * (`vendor/intx/hub-sessions/src/event-collector.ts`) without depending on + * it. */ +export function snapshotTextFromParts( + turnIdsOldestFirst: readonly string[], + parts: readonly { + readonly turnId: string; + readonly content: string | null; + readonly type: string; + readonly ordinal: number; + }[], +): string { + const byTurn = new Map(); + for (const part of parts) { + if (part.type !== "text" || part.content === null) continue; + const bucket = byTurn.get(part.turnId); + const entry = { content: part.content, ordinal: part.ordinal }; + if (bucket === undefined) byTurn.set(part.turnId, [entry]); + else bucket.push(entry); + } + let text = ""; + for (const turnId of turnIdsOldestFirst) { + const bucket = byTurn.get(turnId); + if (bucket === undefined) continue; + for (const part of [...bucket].sort((a, b) => a.ordinal - b.ordinal)) { + text += part.content; + } + } + return text; +} + +export function createDrizzleTurnTextSnapshotReader( + db: DB["db"], +): TurnTextSnapshotReader { + return { + async read({ tenantId, runId }) { + const turns = await db.query.inferenceTurn.findMany({ + where: and( + eq(inferenceTurn.runId, runId), + eq(inferenceTurn.tenantId, tenantId), + ), + orderBy: asc(inferenceTurn.startedAt), + }); + if (turns.length === 0) return null; + + const parts = await db.query.turnPart.findMany({ + where: inArray( + turnPart.turnId, + turns.map((turn) => turn.id), + ), + orderBy: asc(turnPart.ordinal), + }); + + return snapshotTextFromParts( + turns.map((turn) => turn.id), + parts, + ); + }, + }; +}