diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 03ddc771a59525c20596e0ec0a6a04481c7051e0..11f228f4041803affdddd09e6299d913b84eb26c 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 aed7ec03c739afb13cc78bd08b55ea58b14c99b8..f2a8ed7ff0e18c6915d756710af7cae8ba400312 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 37361e7198d57faa019cf10417ffb90674b059cc..ddf2efbcff8bf1f2b2c65fd35750790f84e05972 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 2ac00838018b4844a38c00bd795f62d750557673..cafd318526434a9f35ed520d4008def832b0bf75 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"),