From 6a1f95f75cc87390ae99df3f9498c0038105ecb3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 07:10:04 -0700 Subject: [PATCH 1/2] Add tests for turn-resume catch-up (CL-6380) Covers reconstructing a running turn's committed text from turn_part rows, the turns route attaching that snapshot only while a turn is running, and useStreamingReply hydrating from a fetched snapshot on a fresh mount without ever clobbering a live event that already opened the reply. --- packages/chat-ui/src/streaming-reply.test.ts | 19 ++++ .../chat-ui/test/use-streaming-reply.test.tsx | 54 ++++++++++++ packages/chat/test/agent-turns-routes.test.ts | 86 +++++++++++++++++++ .../insights/src/turn-text-snapshot.test.ts | 45 ++++++++++ 4 files changed, 204 insertions(+) create mode 100644 packages/insights/src/turn-text-snapshot.test.ts 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/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/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/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(""); + }); +}); From 172c790b2aab781de7d924422bd47ad19e3ad15b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 07:10:13 -0700 Subject: [PATCH 2/2] CL-6380: an in-flight reply survives navigation, and resumes on return MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A turn already ran entirely server-side — closing the SSE EventSource on unmount never touched it — but the client had no way to catch up: useStreamingReply/useTurnActivity reset to null on every mount, so returning to a workbench mid-reply showed nothing until the next token happened to arrive. Adds a read path for the catch-up: turnTextSnapshot reconstructs a running turn's visible text from the platform's own inference_turn/ turn_part rows (no new storage, same tables trace-reader.ts already reads), wired into GET /workbenches/:id/turns/:turnId behind an optional dep so a deployment without it just serves no snapshot. The client fetches this once per mount and hands it to useStreamingReply's new resumeFromTurn, which hydrates the reply immediately and is guarded to never override state a live event already produced. --- apps/hub/src/index.ts | 3 + packages/chat-ui/src/api.ts | 75 ++++++++++++++++++ packages/chat-ui/src/chat-workspace.tsx | 30 +++++++ packages/chat-ui/src/streaming-reply.ts | 42 ++++++++++ packages/chat/src/routes.ts | 25 +++++- packages/insights/src/index.ts | 5 ++ packages/insights/src/turn-text-snapshot.ts | 88 +++++++++++++++++++++ 7 files changed, 267 insertions(+), 1 deletion(-) create mode 100644 packages/insights/src/turn-text-snapshot.ts 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.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/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/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.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, + ); + }, + }; +}