diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e793548ea50d954d363bcf24d639f86edd3664d9..6f4e99814696572efcb51112b1dab9ab3d3b7d15 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -940,6 +940,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Command decisions need control records, not historical assistant/tool output. // Handoff preparation explicitly reads its history after choosing a strategy. + // This runs inside the write transaction for every command, so finished leaf + // nodes of finished runs stay unread: busy threads hold thousands of them. const readCommandProjection = (threadId: ThreadId) => projectionStore .getThreadRecords( @@ -958,7 +960,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "checkpointScopes", "contextTransfers", ], - { turnItemTypes: ["user_message", "error"], messageRoles: ["user"] }, + { + turnItemTypes: ["user_message", "error"], + messageRoles: ["user"], + omitSettledLeafNodes: true, + }, ) .pipe( Effect.map((records): OrchestrationV2ThreadProjection => ({ diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index d2bd4a13ffecda56e7a53c2a828f8f54b23bcd95..0729b0a93f1624850923de97efe4afc580b1d5fd 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -382,6 +382,89 @@ const threadWorktreePathsReadActiveThreads = Effect.gen(function* () { ]); }); +const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = yield* addRolledBackRecoveryCandidate("settled-leaf-nodes"); + const settledRun = (yield* store.getThreadProjection(threadId)).runs[0]!; + const activeRun = { + ...settledRun, + id: RunId.make("run:settled-leaf-nodes:active"), + ordinal: 2, + userMessageId: MessageId.make("message:settled-leaf-nodes:active"), + rootNodeId: NodeId.make("node:settled-leaf-nodes:active-root"), + status: "running" as const, + completedAt: null, + }; + yield* store.apply({ + id: EventId.make("event:settled-leaf-nodes:active-run"), + type: "run.created", + threadId, + runId: activeRun.id, + nodeId: activeRun.rootNodeId, + driver, + providerInstanceId, + occurredAt: now, + payload: activeRun, + }); + const nodes = [ + ["settled-root", settledRun, "root_turn", "completed"], + ["settled-tool", settledRun, "tool_call", "completed"], + ["settled-message", settledRun, "assistant_message", "failed"], + ["settled-reasoning", settledRun, "reasoning", "cancelled"], + ["settled-subagent", settledRun, "subagent", "completed"], + ["settled-run-live-tool", settledRun, "tool_call", "running"], + ["active-run-done-tool", activeRun, "tool_call", "completed"], + ] as const; + for (const [name, run, kind, status] of nodes) { + const id = NodeId.make(`node:settled-leaf-nodes:${name}`); + yield* store.apply({ + id: EventId.make(`event:settled-leaf-nodes:${name}`), + type: "node.updated", + threadId, + runId: run.id, + nodeId: id, + driver, + occurredAt: now, + payload: { + id, + threadId, + runId: run.id, + rootNodeId: kind === "root_turn" ? id : run.rootNodeId!, + parentNodeId: kind === "root_turn" ? null : run.rootNodeId, + kind, + status, + countsForRun: kind === "root_turn", + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: status === "running" ? null : now, + }, + }); + } + const nodeNames = (records: { readonly nodes: ReadonlyArray<{ readonly id: string }> }) => + records.nodes + .map((node) => node.id.replace("node:settled-leaf-nodes:", "")) + .filter((name) => nodes.some(([candidate]) => candidate === name)) + .toSorted(); + + assert.deepEqual( + nodeNames(yield* store.getThreadRecords(threadId, ["nodes"], { omitSettledLeafNodes: true })), + ["active-run-done-tool", "settled-root", "settled-run-live-tool", "settled-subagent"], + ); + assert.deepEqual( + nodeNames(yield* store.getThreadRecords(threadId, ["nodes"])), + nodes.map(([name]) => name).toSorted(), + ); +}); + +it.effect("memory command reads skip finished leaf nodes of finished runs", () => + commandReadsSkipSettledLeafNodes.pipe(Effect.provide(ProjectionStore.layerMemory)), +); + it.effect("memory worktree path lookup reads only active threads", () => threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), ); @@ -657,6 +740,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { assert.deepEqual(recovery?.restartCancelledBackgroundWork, work); }), ); + it.effect( + "command reads skip finished leaf nodes of finished runs", + () => commandReadsSkipSettledLeafNodes, + ); it.effect("limits turn-start history to the requested runs, including an empty selection", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 7443eb7aab8d102065bc0a0278048bf8bfb16e26..ad92e25c8516a51d154706f4f6d97e7f2cbe05e7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -304,7 +304,39 @@ export interface ProjectionRecordFilter { readonly runIds?: ReadonlyArray; readonly turnItemTypes?: ReadonlyArray; readonly turnItemStatuses?: ReadonlyArray; + // Skips finished tool call, assistant message and reasoning nodes of finished + // runs. Command decisions only read root, subagent and active-run nodes. + readonly omitSettledLeafNodes?: boolean; } + +type ProjectionNode = OrchestrationV2ThreadProjection["nodes"][number]; +type ProjectionRun = OrchestrationV2ThreadProjection["runs"][number]; +const settledLeafNodeKinds: ReadonlySet = new Set([ + "tool_call", + "assistant_message", + "reasoning", +]); +const settledNodeStatuses: ReadonlySet = new Set([ + "completed", + "interrupted", + "failed", + "cancelled", + "rolled_back", +]); +const activeRunStatuses: ReadonlySet = new Set([ + "preparing", + "queued", + "starting", + "running", + "waiting", +]); + +// In-memory twin of the omitSettledLeafNodes SQL clause in readCanonicalProjection. +const isSettledLeafNode = (node: ProjectionNode, runs: ReadonlyArray) => + settledLeafNodeKinds.has(node.kind) && + settledNodeStatuses.has(node.status) && + node.runId !== null && + !runs.some((run) => run.id === node.runId && activeRunStatuses.has(run.status)); export type ProjectionRecordField = Exclude< keyof OrchestrationV2ThreadProjection, "thread" | "updatedAt" | "visibleTurnItems" @@ -3069,9 +3101,23 @@ export const layer: Layer.Layer = : window === undefined ? sql` SELECT payload_json - FROM orchestration_v2_projection_nodes - WHERE thread_id = ${threadId} - ORDER BY COALESCE(started_at, ''), node_id ASC + FROM orchestration_v2_projection_nodes AS node + WHERE node.thread_id = ${threadId} + ${ + filter?.omitSettledLeafNodes === true + ? sql`AND NOT ( + node.kind IN ('tool_call', 'assistant_message', 'reasoning') + AND node.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') + AND node.run_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS run + WHERE run.run_id = node.run_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + )` + : sql`` + } + ORDER BY COALESCE(node.started_at, ''), node.node_id ASC ` : sql` -- Keep retained IDs outermost so SQLite looks up each node by its @@ -6423,6 +6469,10 @@ export const layerMemory: Layer.Layer = Layer.effect( filter.turnItemStatuses.includes(row.status)) && (filter?.turnItemRunId === undefined || filter.turnItemRunId === row.runId), ), + nodes: + filter?.omitSettledLeafNodes === true + ? projection.nodes.filter((node) => !isSettledLeafNode(node, projection.runs)) + : projection.nodes, }; return Object.fromEntries([ ["thread", projection.thread],