| author |  Theo Browne <me@t3.gg> 2026-10-14 17:05:21-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-14 17:05:21-07:00 |
| log | ab3a3cd162d90fd9f01ac62efdb9d4ad55c33639 |
| tree | 96fad38975a7f8775bd473bdc82ce024464702dd |
| parent | 742c229957b69a0cc5a1379dff6c66b595854057 |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(server): commands on busy threads stop decoding thousands of finished tool nodes (#19675)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>3 files changed, 147 insertions(+), 4 deletions(-)
apps
server
src
orchestration-v2
apps/server/src/orchestration-v2/Orchestrator.ts+7-1
| ... | @@ -940,6 +940,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -940,6 +940,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 940 | 940 | ||
| 941 | // Command decisions need control records, not historical assistant/tool output. | 941 | // Command decisions need control records, not historical assistant/tool output. |
| 942 | // Handoff preparation explicitly reads its history after choosing a strategy. | 942 | // Handoff preparation explicitly reads its history after choosing a strategy. |
| 943 | // This runs inside the write transaction for every command, so finished leaf | ||
| 944 | // nodes of finished runs stay unread: busy threads hold thousands of them. | ||
| 943 | const readCommandProjection = (threadId: ThreadId) => | 945 | const readCommandProjection = (threadId: ThreadId) => |
| 944 | projectionStore | 946 | projectionStore |
| 945 | .getThreadRecords( | 947 | .getThreadRecords( |
| ... | @@ -958,7 +960,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -958,7 +960,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 958 | "checkpointScopes", | 960 | "checkpointScopes", |
| 959 | "contextTransfers", | 961 | "contextTransfers", |
| 960 | ], | 962 | ], |
| 961 | { turnItemTypes: ["user_message", "error"], messageRoles: ["user"] }, | 963 | { |
| 964 | turnItemTypes: ["user_message", "error"], | ||
| 965 | messageRoles: ["user"], | ||
| 966 | omitSettledLeafNodes: true, | ||
| 967 | }, | ||
| 962 | ) | 968 | ) |
| 963 | .pipe( | 969 | .pipe( |
| 964 | Effect.map((records): OrchestrationV2ThreadProjection => ({ | 970 | Effect.map((records): OrchestrationV2ThreadProjection => ({ |
apps/server/src/orchestration-v2/ProjectionStore.test.ts+87
| ... | @@ -382,6 +382,89 @@ const threadWorktreePathsReadActiveThreads = Effect.gen(function* () { | ... | @@ -382,6 +382,89 @@ const threadWorktreePathsReadActiveThreads = Effect.gen(function* () { |
| 382 | ]); | 382 | ]); |
| 383 | }); | 383 | }); |
| 384 | 384 | ||
| 385 | const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { | ||
| 386 | const store = yield* ProjectionStore.ProjectionStoreV2; | ||
| 387 | const now = yield* DateTime.now; | ||
| 388 | const threadId = yield* addRolledBackRecoveryCandidate("settled-leaf-nodes"); | ||
| 389 | const settledRun = (yield* store.getThreadProjection(threadId)).runs[0]!; | ||
| 390 | const activeRun = { | ||
| 391 | ...settledRun, | ||
| 392 | id: RunId.make("run:settled-leaf-nodes:active"), | ||
| 393 | ordinal: 2, | ||
| 394 | userMessageId: MessageId.make("message:settled-leaf-nodes:active"), | ||
| 395 | rootNodeId: NodeId.make("node:settled-leaf-nodes:active-root"), | ||
| 396 | status: "running" as const, | ||
| 397 | completedAt: null, | ||
| 398 | }; | ||
| 399 | yield* store.apply({ | ||
| 400 | id: EventId.make("event:settled-leaf-nodes:active-run"), | ||
| 401 | type: "run.created", | ||
| 402 | threadId, | ||
| 403 | runId: activeRun.id, | ||
| 404 | nodeId: activeRun.rootNodeId, | ||
| 405 | driver, | ||
| 406 | providerInstanceId, | ||
| 407 | occurredAt: now, | ||
| 408 | payload: activeRun, | ||
| 409 | }); | ||
| 410 | const nodes = [ | ||
| 411 | ["settled-root", settledRun, "root_turn", "completed"], | ||
| 412 | ["settled-tool", settledRun, "tool_call", "completed"], | ||
| 413 | ["settled-message", settledRun, "assistant_message", "failed"], | ||
| 414 | ["settled-reasoning", settledRun, "reasoning", "cancelled"], | ||
| 415 | ["settled-subagent", settledRun, "subagent", "completed"], | ||
| 416 | ["settled-run-live-tool", settledRun, "tool_call", "running"], | ||
| 417 | ["active-run-done-tool", activeRun, "tool_call", "completed"], | ||
| 418 | ] as const; | ||
| 419 | for (const [name, run, kind, status] of nodes) { | ||
| 420 | const id = NodeId.make(`node:settled-leaf-nodes:${name}`); | ||
| 421 | yield* store.apply({ | ||
| 422 | id: EventId.make(`event:settled-leaf-nodes:${name}`), | ||
| 423 | type: "node.updated", | ||
| 424 | threadId, | ||
| 425 | runId: run.id, | ||
| 426 | nodeId: id, | ||
| 427 | driver, | ||
| 428 | occurredAt: now, | ||
| 429 | payload: { | ||
| 430 | id, | ||
| 431 | threadId, | ||
| 432 | runId: run.id, | ||
| 433 | rootNodeId: kind === "root_turn" ? id : run.rootNodeId!, | ||
| 434 | parentNodeId: kind === "root_turn" ? null : run.rootNodeId, | ||
| 435 | kind, | ||
| 436 | status, | ||
| 437 | countsForRun: kind === "root_turn", | ||
| 438 | providerThreadId: null, | ||
| 439 | providerTurnId: null, | ||
| 440 | nativeItemRef: null, | ||
| 441 | runtimeRequestId: null, | ||
| 442 | checkpointScopeId: null, | ||
| 443 | startedAt: now, | ||
| 444 | completedAt: status === "running" ? null : now, | ||
| 445 | }, | ||
| 446 | }); | ||
| 447 | } | ||
| 448 | const nodeNames = (records: { readonly nodes: ReadonlyArray<{ readonly id: string }> }) => | ||
| 449 | records.nodes | ||
| 450 | .map((node) => node.id.replace("node:settled-leaf-nodes:", "")) | ||
| 451 | .filter((name) => nodes.some(([candidate]) => candidate === name)) | ||
| 452 | .toSorted(); | ||
| 453 | |||
| 454 | assert.deepEqual( | ||
| 455 | nodeNames(yield* store.getThreadRecords(threadId, ["nodes"], { omitSettledLeafNodes: true })), | ||
| 456 | ["active-run-done-tool", "settled-root", "settled-run-live-tool", "settled-subagent"], | ||
| 457 | ); | ||
| 458 | assert.deepEqual( | ||
| 459 | nodeNames(yield* store.getThreadRecords(threadId, ["nodes"])), | ||
| 460 | nodes.map(([name]) => name).toSorted(), | ||
| 461 | ); | ||
| 462 | }); | ||
| 463 | |||
| 464 | it.effect("memory command reads skip finished leaf nodes of finished runs", () => | ||
| 465 | commandReadsSkipSettledLeafNodes.pipe(Effect.provide(ProjectionStore.layerMemory)), | ||
| 466 | ); | ||
| 467 | |||
| 385 | it.effect("memory worktree path lookup reads only active threads", () => | 468 | it.effect("memory worktree path lookup reads only active threads", () => |
| 386 | threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), | 469 | threadWorktreePathsReadActiveThreads.pipe(Effect.provide(ProjectionStore.layerMemory)), |
| 387 | ); | 470 | ); |
| ... | @@ -657,6 +740,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { | ... | @@ -657,6 +740,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { |
| 657 | assert.deepEqual(recovery?.restartCancelledBackgroundWork, work); | 740 | assert.deepEqual(recovery?.restartCancelledBackgroundWork, work); |
| 658 | }), | 741 | }), |
| 659 | ); | 742 | ); |
| 743 | it.effect( | ||
| 744 | "command reads skip finished leaf nodes of finished runs", | ||
| 745 | () => commandReadsSkipSettledLeafNodes, | ||
| 746 | ); | ||
| 660 | it.effect("limits turn-start history to the requested runs, including an empty selection", () => | 747 | it.effect("limits turn-start history to the requested runs, including an empty selection", () => |
| 661 | Effect.gen(function* () { | 748 | Effect.gen(function* () { |
| 662 | const store = yield* ProjectionStore.ProjectionStoreV2; | 749 | const store = yield* ProjectionStore.ProjectionStoreV2; |
apps/server/src/orchestration-v2/ProjectionStore.ts+53-3
| ... | @@ -304,7 +304,39 @@ export interface ProjectionRecordFilter { | ... | @@ -304,7 +304,39 @@ export interface ProjectionRecordFilter { |
| 304 | readonly runIds?: ReadonlyArray<RunId>; | 304 | readonly runIds?: ReadonlyArray<RunId>; |
| 305 | readonly turnItemTypes?: ReadonlyArray<OrchestrationV2TurnItem["type"]>; | 305 | readonly turnItemTypes?: ReadonlyArray<OrchestrationV2TurnItem["type"]>; |
| 306 | readonly turnItemStatuses?: ReadonlyArray<OrchestrationV2TurnItem["status"]>; | 306 | readonly turnItemStatuses?: ReadonlyArray<OrchestrationV2TurnItem["status"]>; |
| 307 | // Skips finished tool call, assistant message and reasoning nodes of finished | ||
| 308 | // runs. Command decisions only read root, subagent and active-run nodes. | ||
| 309 | readonly omitSettledLeafNodes?: boolean; | ||
| 307 | } | 310 | } |
| 311 | |||
| 312 | type ProjectionNode = OrchestrationV2ThreadProjection["nodes"][number]; | ||
| 313 | type ProjectionRun = OrchestrationV2ThreadProjection["runs"][number]; | ||
| 314 | const settledLeafNodeKinds: ReadonlySet<ProjectionNode["kind"]> = new Set([ | ||
| 315 | "tool_call", | ||
| 316 | "assistant_message", | ||
| 317 | "reasoning", | ||
| 318 | ]); | ||
| 319 | const settledNodeStatuses: ReadonlySet<ProjectionNode["status"]> = new Set([ | ||
| 320 | "completed", | ||
| 321 | "interrupted", | ||
| 322 | "failed", | ||
| 323 | "cancelled", | ||
| 324 | "rolled_back", | ||
| 325 | ]); | ||
| 326 | const activeRunStatuses: ReadonlySet<ProjectionRun["status"]> = new Set([ | ||
| 327 | "preparing", | ||
| 328 | "queued", | ||
| 329 | "starting", | ||
| 330 | "running", | ||
| 331 | "waiting", | ||
| 332 | ]); | ||
| 333 | |||
| 334 | // In-memory twin of the omitSettledLeafNodes SQL clause in readCanonicalProjection. | ||
| 335 | const isSettledLeafNode = (node: ProjectionNode, runs: ReadonlyArray<ProjectionRun>) => | ||
| 336 | settledLeafNodeKinds.has(node.kind) && | ||
| 337 | settledNodeStatuses.has(node.status) && | ||
| 338 | node.runId !== null && | ||
| 339 | !runs.some((run) => run.id === node.runId && activeRunStatuses.has(run.status)); | ||
| 308 | export type ProjectionRecordField = Exclude< | 340 | export type ProjectionRecordField = Exclude< |
| 309 | keyof OrchestrationV2ThreadProjection, | 341 | keyof OrchestrationV2ThreadProjection, |
| 310 | "thread" | "updatedAt" | "visibleTurnItems" | 342 | "thread" | "updatedAt" | "visibleTurnItems" |
| ... | @@ -3069,9 +3101,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = | ... | @@ -3069,9 +3101,23 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = |
| 3069 | : window === undefined | 3101 | : window === undefined |
| 3070 | ? sql<PayloadRow>` | 3102 | ? sql<PayloadRow>` |
| 3071 | SELECT payload_json | 3103 | SELECT payload_json |
| 3072 | FROM orchestration_v2_projection_nodes | 3104 | FROM orchestration_v2_projection_nodes AS node |
| 3073 | WHERE thread_id = ${threadId} | 3105 | WHERE node.thread_id = ${threadId} |
| 3074 | ORDER BY COALESCE(started_at, ''), node_id ASC | 3106 | ${ |
| 3107 | filter?.omitSettledLeafNodes === true | ||
| 3108 | ? sql`AND NOT ( | ||
| 3109 | node.kind IN ('tool_call', 'assistant_message', 'reasoning') | ||
| 3110 | AND node.status IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back') | ||
| 3111 | AND node.run_id IS NOT NULL | ||
| 3112 | AND NOT EXISTS ( | ||
| 3113 | SELECT 1 FROM orchestration_v2_projection_runs AS run | ||
| 3114 | WHERE run.run_id = node.run_id | ||
| 3115 | AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') | ||
| 3116 | ) | ||
| 3117 | )` | ||
| 3118 | : sql`` | ||
| 3119 | } | ||
| 3120 | ORDER BY COALESCE(node.started_at, ''), node.node_id ASC | ||
| 3075 | ` | 3121 | ` |
| 3076 | : sql<PayloadRow & { readonly position: number }>` | 3122 | : sql<PayloadRow & { readonly position: number }>` |
| 3077 | -- Keep retained IDs outermost so SQLite looks up each node by its | 3123 | -- Keep retained IDs outermost so SQLite looks up each node by its |
| ... | @@ -6423,6 +6469,10 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect( | ... | @@ -6423,6 +6469,10 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect( |
| 6423 | filter.turnItemStatuses.includes(row.status)) && | 6469 | filter.turnItemStatuses.includes(row.status)) && |
| 6424 | (filter?.turnItemRunId === undefined || filter.turnItemRunId === row.runId), | 6470 | (filter?.turnItemRunId === undefined || filter.turnItemRunId === row.runId), |
| 6425 | ), | 6471 | ), |
| 6472 | nodes: | ||
| 6473 | filter?.omitSettledLeafNodes === true | ||
| 6474 | ? projection.nodes.filter((node) => !isSettledLeafNode(node, projection.runs)) | ||
| 6475 | : projection.nodes, | ||
| 6426 | }; | 6476 | }; |
| 6427 | return Object.fromEntries([ | 6477 | return Object.fromEntries([ |
| 6428 | ["thread", projection.thread], | 6478 | ["thread", projection.thread], |