From d36d09892c7482ba271251b7faf948e0ad6013c3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Wed, 14 Oct 2026 05:25:37 -0700 Subject: [PATCH] perf(server): unchanged node updates no longer write to SQLite Adapters send the same running reasoning node again with each reasoning delta, and each copy became its own event row and commit. In the #15707 trace these copies were 91% of all events. ProviderEventIngestor now skips a provider node.updated when the encoded node equals the stored projection row. It compares with the stored row, not with the last node the run sent, because the server also writes nodes (for example when a user answers an approval). Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ProjectionStore.ts | 24 ++++++++ .../ProviderEventIngestor.test.ts | 60 +++++++++++++++++++ .../orchestration-v2/ProviderEventIngestor.ts | 3 + .../ProviderTurnControlService.test.ts | 1 + 4 files changed, 88 insertions(+) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 03ddc771a595..11f228f40418 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -73,6 +73,7 @@ import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2Pend import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Equal from "effect/Equal"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -460,6 +461,10 @@ export interface ProjectionStoreV2Shape { requestId: TurnItemId, resultId: TurnItemId, ) => Effect.Effect; + /** True when the projection already stores this exact node, so a write would change nothing. */ + readonly isNodeStored: ( + node: OrchestrationV2ExecutionNode, + ) => Effect.Effect; readonly getRuntimeRequest: ( threadId: ThreadId, requestId: RuntimeRequestId, @@ -4404,6 +4409,15 @@ export const layer: Layer.Layer = return rows[0]?.pending === 1; }).pipe(Effect.mapError(controlReadError(threadId))); + const isNodeStored: ProjectionStoreV2Shape["isNodeStored"] = (node) => + Effect.gen(function* () { + const payloadJson = yield* encodeNodePayload(node); + const rows = yield* sql` + SELECT payload_json FROM orchestration_v2_projection_nodes WHERE node_id = ${node.id} + `; + return rows[0]?.payload_json === payloadJson; + }).pipe(Effect.mapError(controlReadError(node.threadId))); + const getRuntimeRequest: ProjectionStoreV2Shape["getRuntimeRequest"] = (threadId, requestId) => sql .withTransaction( @@ -5992,6 +6006,7 @@ export const layer: Layer.Layer = canStartQueuedRun, getPendingNativeUserInputs, hasUnpairedRunInterruptRequest, + isNodeStored, getMessageCount, getNextTurnItemOrdinal, getTurnItem, @@ -6255,6 +6270,15 @@ export const layerMemory: Layer.Layer = Layer.effect( !projection.turnItems.some((item) => item.id === resultId) ); }), + isNodeStored: (node) => + Ref.get(replayState).pipe( + Effect.map((state) => { + const stored = state.projections + .get(node.threadId) + ?.nodes.find((candidate) => candidate.id === node.id); + return stored !== undefined && Equal.equals(stored, node); + }), + ), getMessageCount: (threadId) => Ref.get(replayState).pipe( Effect.map((state) => state.projections.get(threadId)?.messages.length ?? 0), diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index aed7ec03c739..f2a8ed7ff0e1 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -325,6 +325,66 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + it.effect("skips a node update only when it matches the stored node", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + const threadId = threadEvent.threadId; + yield* eventSink.write({ events: [threadEvent] }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const nodeId = NodeId.make(`${threadId}:approval`); + const node: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId: nodeId, + kind: "approval_request", + status: "waiting", + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }; + const ingestNode = (sent: OrchestrationV2ExecutionNode) => + ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event: { type: "node.updated", driver: CODEX_DRIVER, node: sent }, + }); + + assert.lengthOf(yield* ingestNode(node), 1); + assert.isEmpty(yield* ingestNode({ ...node })); + // The server resolves the request, then the provider asks again with the same node. + yield* eventSink.write({ + events: [ + { + id: yield* idAllocator.allocate.event({ threadId }), + type: "node.updated", + threadId, + occurredAt: now, + payload: { ...node, status: "completed", completedAt: now }, + }, + ], + }); + assert.lengthOf(yield* ingestNode(node), 1); + const projection = yield* projectionStore.getThreadProjection(threadId); + assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, "waiting"); + }), + ); + it.effect("rolls back runs a provider rewound off its active branch", () => Effect.gen(function* () { const now = yield* DateTime.now; diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 37361e7198d5..ddf2efbcff8b 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -611,6 +611,9 @@ export const layer: Layer.Layer< }), ]; case "node.updated": + // Adapters send an unchanged node again with each text delta. Compare with the + // stored node, not the last one sent, because the server also writes nodes. + if (yield* projections.isNodeStored(input.event.node)) return []; return [ yield* makeDomainEvent(input, { type: "node.updated", diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 2ac00838018b..cafd31852643 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -255,6 +255,7 @@ it.effect( getCheckpointContext: () => Effect.die("not used"), getCheckpointCaptureContext: () => Effect.die("not used"), getRunMessage: () => Effect.die("not used"), + isNodeStored: () => Effect.die("not used"), canStartQueuedRun: () => Effect.die("not used"), getRecoveryThreadIds: () => Effect.die("unused getRecoveryThreadIds"), getUnreadableThreadIds: () => Effect.die("unused getUnreadableThreadIds"),