From bbfa6417e86c6648c0974f6c9b7ff675306bc6ca Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 13 Nov 2026 16:37:34 -0300 Subject: [PATCH 1/2] perf(server): sending to a long compacted thread skips its old prompts A send in a compacted thread still read every prompt and error of the thread. Each prompt is a separate random read, so a thread of about 8,000 runs spent most of its read on them. The command read now loads the failed errors of every run and the prompts of the recent runs, of fresh-context runs and of the newest runs back to the one a model switch is measured from. Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 10 +-- .../orchestration-v2/ProjectionStore.test.ts | 80 ++++++++++++++--- .../src/orchestration-v2/ProjectionStore.ts | 87 +++++++++++++++++-- .../src/orchestration-v2/modelSwitch.ts | 35 +++++--- 4 files changed, 177 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a6f412e7cedd..30fab183efae 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1288,7 +1288,7 @@ function isScheduledFreshRun( run: Pick, turnItems: OrchestrationV2ThreadProjection["turnItems"], ): boolean { - // Turn items, unlike messages, are not narrowed to the runs since the latest compaction. + // A compacted thread's command read loads these prompts for every fresh-context run. return ( run.freshContext === true && turnItems.some( @@ -6522,9 +6522,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, effects: Ref.Ref>, ) { - // A compacted thread reads every run and turn item, but attempts, nodes, messages and provider - // turns only for its recent runs and the runs this command names. An idle compaction reads - // in full: its busy check below looks for a delegated result's message under any old run. + // A compacted thread reads every run, but attempts, nodes, messages, provider turns and prompts + // only for its recent runs and the runs this command names. An idle compaction reads in full: + // its busy check below looks for a delegated result's message under any old run. let projection = yield* getProjectionWithPendingEvents( command.threadId, events, @@ -6796,7 +6796,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } const userMessages = projection.messages.filter((message) => message.role === "user"); // The recent runs of a compacted thread can hold only a `/compact` while an older prompt - // exists. Every prompt's turn item is loaded; without one, every user message is checked. + // exists, so when every loaded prompt is one, every user message is checked. const onlyMaintenanceHistory = [ ...userMessages, ...projection.turnItems.flatMap((item) => (item.type === "user_message" ? [item] : [])), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 6886912fc4cc..f8b5190513db 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -798,7 +798,7 @@ it.effect("idle compaction filters keep the latest run, live work and the newest // One thread holds every way a run can still matter after a compaction. A run is requested at // its ordinal in minutes and ends 30 seconds later, unless the table says otherwise. -it.effect("a compacted thread reads every run, and run records only for its recent runs", () => +it.effect("a compacted thread reads every run, but records and prompts only for recent ones", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const threadId = yield* addRolledBackRecoveryCandidate("compacted"); @@ -844,9 +844,10 @@ it.effect("a compacted thread reads every run, and run records only for its rece driver, }, }); - for (const [ordinal, status, endedAt] of [ + for (const [ordinal, status, endedAt, fresh] of [ [2, "queued", null], - [3, "completed"], + // A fresh-context run starts its own conversation, so its prompt is read for the schedule check. + [3, "completed", undefined, true], // A held queue ran this old run after the compactions. [4, "completed", at(11.5)], [5, "completed"], @@ -890,6 +891,7 @@ it.effect("a compacted thread reads every run, and run records only for its rece completedAt: ended, checkpointId: null, contextHandoffId: null, + ...(fresh === true ? { freshContext: true } : {}), }, }); yield* store.apply({ @@ -1025,19 +1027,44 @@ it.effect("a compacted thread reads every run, and run records only for its rece })), ); const all = [2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]; - // Every run and every turn item is read, so a model switch is still measured from the last - // prompt. Run 1, the helper's, holds only a command item. - const expected = (recent: ReadonlyArray) => ({ + // Every run is read. Turn items are the failed errors of every run, and the prompts of the + // recent runs, of the fresh-context run 3 and of the run a model switch is measured from. + // Run 1, the helper's, holds only a command item. + const expected = ( + recent: ReadonlyArray, + prompts: ReadonlyArray = [...recent, 3], + ) => ({ runs: 15, attempts: recent, nodes: recent, providerTurns: recent, messages: recent, - turnItems: [1, ...all], + turnItems: [...new Set([5, ...prompts])].toSorted((a, b) => a - b), }); - // Nothing is compacted yet, so the whole history is current. - assert.deepEqual(yield* read(), expected(all)); + // Run 5 failed, and nothing is compacted yet, so the whole history is current. + yield* store.apply({ + id: EventId.make("event:compacted:error-5"), + type: "turn-item.updated", + threadId, + runId: runId(5), + nodeId: rootNodeId(5), + driver, + occurredAt: at(5), + payload: { + ...item(5, 52, "failed"), + id: TurnItemId.make("item:compacted:error-5"), + type: "error", + failure: { + class: "usage_limit", + message: "Plan limit reached.", + resetAt: "2099-01-01T00:00:00.000Z", + code: "usageLimitExceeded", + retryable: null, + }, + }, + }); + assert.deepEqual(yield* read(), { ...expected(all), turnItems: [1, ...all] }); // The window starts at the compaction's run. Queued run 2 and held run 4 are older but had // not ended by then. @@ -1207,11 +1234,12 @@ it.effect("a compacted thread reads every run, and run records only for its rece assert.deepEqual(unsettled.providerThreads, ["compacted", "old-pending", "recent-pending"]); assert.lengthOf(unsettled.rootNodes, all.length); assert.lengthOf(unsettled.runs, 15); - // Together: every run, every turn item and the provider thread the runs use stay, and the - // windowed records are read for the recent and named runs, less each settled task's node. + // Together: every run and the provider thread the runs use stay, and the windowed records are + // read for the recent and named runs, less each settled task's node. const recentRuns = ["11", "12", "13", "14", "15", "2", "3", "4", "6"]; assert.deepEqual(both, { ...unsettled, + turnItems: windowed.turnItems, attempts: recentRuns, rootNodes: recentRuns, taskNodes: ["recent-pending"], @@ -1228,6 +1256,36 @@ it.effect("a compacted thread reads every run, and run records only for its rece ), both, ); + + // A prompt that clients do not show cannot anchor a model switch. With the recent ones, and + // run 4's, folded into async answers, the switch is measured from run 10, so its prompt is + // read, with run 9's from the same batch. + for (const ordinal of [4, 11, 12, 14, 15]) { + yield* store.apply({ + id: EventId.make(`event:compacted:async-${ordinal}`), + type: "turn-item.updated", + threadId, + runId: runId(ordinal), + nodeId: rootNodeId(ordinal), + driver, + occurredAt: at(ordinal), + payload: { + createdBy: "user", + creationSource: "web", + ...item(ordinal, ordinal * 10, "completed"), + id: TurnItemId.make(`item:compacted:prompt-${ordinal}`), + type: "user_message", + messageId: MessageId.make(`async-answer:${ordinal}`), + inputIntent: "turn_start", + text: `Answer ${ordinal}`, + attachments: [], + }, + }); + } + assert.deepEqual( + (yield* read()).turnItems, + expected([2, 4, 11, 12, 13, 14, 15], [2, 3, 4, 9, 10, 11, 12, 13, 14, 15]).turnItems, + ); }).pipe(Effect.provide(layerTest)), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index e90a82786953..e44e7e342a8d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -14,10 +14,16 @@ import type { import { latestRootProviderFailure, latestUnheldRun, + runRanAfter, threadErrorSummary, usageLimitRunPresentedAsLatest, } from "@t3tools/shared/orchestrationV2ThreadError"; import { threadPullRequestsOf } from "../sourceControl/threadPullRequests.ts"; +import { + anchorsModelSwitch, + isModelSwitchCandidate, + isShownUserMessageItem, +} from "./modelSwitch.ts"; import type { OrchestrationV2AppThread, OrchestrationV2Checkpoint, @@ -382,9 +388,9 @@ export interface ProjectionRecordFilter { */ readonly omitSettledSubagentRecords?: boolean; /** - * For a compacted thread: reads attempts, nodes, messages and provider turns only for runs from - * the latest completed compaction's run on, runs that had not ended when that run was - * requested, and the runs named here. Runs and turn items are not narrowed. + * For a compacted thread, reads attempts, nodes, messages and provider turns of the recent runs + * and the runs named here. Turn items are the failed errors of every run and the prompts of the + * recent, fresh-context and model-switch runs. `turnItemTypes` does not apply. Runs stay whole. */ readonly sinceLastCompaction?: { readonly keepRunIds: ReadonlyArray; @@ -3511,7 +3517,7 @@ export const layer: Layer.Layer = const threadColumn = (ids: ReadonlyArray | undefined, column = "thread_id") => sql.literal(ids === undefined ? column : `+${column}`); - /** The runs whose records a new message reads. Null when nothing was compacted. */ + /** The ids of the runs whose records a new message reads. Null when nothing was compacted. */ const runIdsSinceLastCompaction = Effect.fn("ProjectionStore.runIdsSinceLastCompaction")( function* ( threadId: ThreadId, @@ -3550,10 +3556,67 @@ export const layer: Layer.Layer = // The compaction's own run is always recent, so no rows means no compaction. return recent.length === 0 ? null - : encodeIdList([...recent.map((row) => row.run_id), ...keep.keepRunIds]); + : [...recent.map((row) => row.run_id), ...keep.keepRunIds]; }, ); + /** + * What a command reads of a compacted thread's prompts and errors: the failed errors of every + * run, and the prompts of the recent runs, of fresh-context runs, and of the newest runs back + * to the one `modelSwitchSince` measures from. That one can be older than the compaction. + */ + const commandTurnItems = Effect.fn("ProjectionStore.commandTurnItems")(function* ( + threadId: ThreadId, + runs: ReadonlyArray, + recentRunIds: ReadonlyArray, + ) { + const errors = yield* sql` + SELECT payload_json + FROM orchestration_v2_projection_turn_items + INDEXED BY orchestration_v2_projection_turn_items_failed_error_idx + WHERE thread_id = ${threadId} AND type = 'error' AND status = 'failed' + `.pipe(Effect.flatMap(decodeRows(decodeTurnItemPayload, threadId))); + const readPrompts = (runIds: ReadonlyArray) => + sql` + SELECT payload_json + FROM orchestration_v2_projection_turn_items + INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx + WHERE thread_id = ${threadId} AND type = 'user_message' + AND run_id IN (SELECT value FROM json_each(${encodeIdList(runIds)})) + `.pipe(Effect.flatMap(decodeRows(decodeTurnItemPayload, threadId))); + const read = new Set(recentRunIds); + for (const run of runs) if (run.freshContext === true) read.add(run.id); + const prompts = yield* readPrompts([...read]); + const shownMessageIds = new Set(); + const showPrompts = (items: ReadonlyArray) => { + for (const item of items) { + if (item.type === "user_message" && isShownUserMessageItem(item)) { + shownMessageIds.add(item.messageId); + } + } + }; + showPrompts(prompts); + const newestFirst = runs + .filter(isModelSwitchCandidate) + .toSorted((left, right) => (runRanAfter(left, right) ? -1 : 1)); + // The batches double, so a thread with no such run takes few queries. + for (let start = 0, size = 1; start < newestFirst.length; start += size, size *= 2) { + const batch = newestFirst.slice(start, start + size); + const unread = batch.filter((run) => !read.has(run.id)); + if (unread.length > 0) { + for (const run of unread) read.add(run.id); + const more = yield* readPrompts(unread.map((run) => run.id)); + showPrompts(more); + prompts.push(...more); + } + if (batch.some((run) => anchorsModelSwitch(run, shownMessageIds))) break; + } + return [...errors, ...prompts].toSorted( + (left, right) => + left.ordinal - right.ordinal || (left.id < right.id ? -1 : left.id > right.id ? 1 : 0), + ); + }); + const readCanonicalProjection = Effect.fn("ProjectionStore.readCanonicalProjection")( function* ( threadId: ThreadId, @@ -3571,10 +3634,11 @@ export const layer: Layer.Layer = if (!threadRow) { return yield* new ProjectionStoreThreadNotFoundError({ threadId }); } - const recentRunIds = + const recentRuns = window === undefined && filter?.sinceLastCompaction !== undefined ? yield* runIdsSinceLastCompaction(threadId, filter.sinceLastCompaction) : null; + const recentRunIds = recentRuns === null ? null : encodeIdList(recentRuns); const inRecentRuns = (column: string) => recentRunIds === null ? sql`` @@ -3592,6 +3656,8 @@ export const layer: Layer.Layer = }; const boundedTurnItemRows: ReadonlyArray = (fields !== undefined && !fields.includes("turnItems")) || + // A compacted thread reads its items once its runs are known. + recentRuns !== null || (window === undefined && matchesNothing( filter?.turnItemRequestIds, @@ -3641,7 +3707,7 @@ export const layer: Layer.Layer = decodeTurnItemPayload, threadId, )(boundedTurnItemRows); - const turnItems = decodedTurnItems.map((item, index) => { + const storedTurnItems = decodedTurnItems.map((item, index) => { const row = boundedTurnItemRows[index]!; return row.work_last_id == null || row.work_item_count == null || @@ -3664,7 +3730,7 @@ export const layer: Layer.Layer = }); const cohortJson = (field: string) => JSON.stringify( - turnItems.flatMap((payload) => { + storedTurnItems.flatMap((payload) => { const value = nullableStringField(payload, field); return value === null ? [] : [value]; }), @@ -4275,12 +4341,17 @@ export const layer: Layer.Layer = decodeRows(decodeContextHandoffPayload, threadId)(contextHandoffRows), decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows), ]); + const turnItems = + recentRuns === null || (fields !== undefined && !fields.includes("turnItems")) + ? storedTurnItems + : yield* commandTurnItems(threadId, runs, recentRuns); const messageOrdinals = new Map(); // The literal `type = 'user_message'` lets SQLite pick the partial user-message index. const userOnly = messages.every((message) => message.role === "user"); // The loaded items already order the messages when they hold every item the lookup reads. const loadsEveryItemOf = (type: OrchestrationV2TurnItem["type"]) => window === undefined && + recentRuns === null && (fields === undefined || fields.includes("turnItems")) && filter?.turnItemRequestIds === undefined && filter?.turnItemStatuses === undefined && diff --git a/apps/server/src/orchestration-v2/modelSwitch.ts b/apps/server/src/orchestration-v2/modelSwitch.ts index 5b893cd0026d..9b904500fc66 100644 --- a/apps/server/src/orchestration-v2/modelSwitch.ts +++ b/apps/server/src/orchestration-v2/modelSwitch.ts @@ -7,6 +7,27 @@ import { import { modelSelectionsEqual } from "@t3tools/shared/model"; import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; +/** + * Clients hide pending queued messages, fold async answers into their question, and replace + * `/compact` with a compaction row. + */ +export const isShownUserMessageItem = (item: OrchestrationV2TurnItem): boolean => + item.type === "user_message" && + !(item.inputIntent === "queued_turn" && item.status === "pending") && + !item.messageId.startsWith("async-answer:") && + !isCompactionCommandMessage({ role: "user", ...item }); + +type ModelSwitchRun = OrchestrationV2ThreadProjection["runs"][number]; + +/** A run that has not been undone and has started or can start. */ +export const isModelSwitchCandidate = (run: ModelSwitchRun): boolean => + run.status !== "rolled_back" && run.status !== "queued"; + +/** A run `modelSwitchSince` can measure from: its prompt is shown, or its handoff row shows it. */ +export const anchorsModelSwitch = (run: ModelSwitchRun, shownMessageIds: ReadonlySet) => + isModelSwitchCandidate(run) && + (shownMessageIds.has(run.userMessageId) || run.contextHandoffId !== null); + /** * The model or effort change a new run brings, measured from the latest run the timeline shows * through its message or its handoff row, so consecutive dividers always chain. `baseline` stands @@ -18,23 +39,15 @@ export function modelSwitchSince( to: ModelSelection, baseline?: ModelSelection, ): Pick, "modelSwitch"> { - // Clients hide pending queued messages, fold async answers into their question, and replace - // `/compact` with a compaction row. const shown = new Set( projection.turnItems.flatMap((item) => - item.type === "user_message" && - !(item.inputIntent === "queued_turn" && item.status === "pending") && - !item.messageId.startsWith("async-answer:") && - !isCompactionCommandMessage({ role: "user", ...item }) - ? [item.messageId] - : [], + item.type === "user_message" && isShownUserMessageItem(item) ? [item.messageId] : [], ), ); - let previous: OrchestrationV2ThreadProjection["runs"][number] | undefined; + let previous: ModelSwitchRun | undefined; for (const run of projection.runs) { - if (run.status === "rolled_back" || run.status === "queued") continue; // A wake shows as a notification, but its context handoff row still shows the swap. - if (!shown.has(run.userMessageId) && run.contextHandoffId === null) continue; + if (!anchorsModelSwitch(run, shown)) continue; if (previous === undefined || runRanAfter(run, previous)) previous = run; } const from = previous?.modelSelection ?? baseline; From 227f683fd59f112db2c86c299970393915bc8856 Mon Sep 17 00:00:00 2001 From: Braulio Oliveira Date: Fri, 13 Nov 2026 16:48:58 -0300 Subject: [PATCH 2/2] test(server): the async-answer case anchors runs it should not show Co-Authored-By: Claude Sonnet 5.5 --- .../src/orchestration-v2/ProjectionStore.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f8b5190513db..06297ff21572 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1260,7 +1260,18 @@ it.effect("a compacted thread reads every run, but records and prompts only for // A prompt that clients do not show cannot anchor a model switch. With the recent ones, and // run 4's, folded into async answers, the switch is measured from run 10, so its prompt is // read, with run 9's from the same batch. + const storedRuns = (yield* store.getThreadRecords(threadId, ["runs"])).runs; for (const ordinal of [4, 11, 12, 14, 15]) { + yield* store.apply({ + id: EventId.make(`event:compacted:async-run-${ordinal}`), + type: "run.updated", + threadId, + occurredAt: at(ordinal), + payload: { + ...storedRuns.find((run) => run.id === runId(ordinal))!, + userMessageId: MessageId.make(`async-answer:${ordinal}`), + }, + }); yield* store.apply({ id: EventId.make(`event:compacted:async-${ordinal}`), type: "turn-item.updated",