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..06297ff21572 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,47 @@ 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. + 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", + 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;