From 091c086985c0ac770903a5312a4266a9387042e9 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 09:13:44 +0000 Subject: [PATCH 1/2] test(server): deferred-steer check and queued cancel read a long thread's whole history Failing tests for #36095: a running provider-turn update reads every run of the thread, and queued-run.cancel reads every node and orders every user message, both under the thread's dispatch lock. Co-Authored-By: Claude Opus 5.5 --- .../Orchestrator.control-reads.test.ts | 122 +++++++++++++++++- 1 file changed, 121 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 21afe0b0574c8..f36ce5bdc564b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -2468,6 +2468,19 @@ it.effect("queued-message commands read a long thread's history without walking }), ); walks.push(...(yield* historyWalks("queued-message.promote-to-steer"))); + yield* dispatch("cancelled"); + const cancelledRun = (yield* projections.getThreadProjection(threadId)).runs.at(-1)!; + assert.equal(cancelledRun.status, "queued"); + statements.splice(0); + yield* observed( + orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make("cancel:long-history-reads"), + threadId, + runId: cancelledRun.id, + }), + ); + walks.push(...(yield* historyWalks("queued-run.cancel"))); yield* observed( orchestrator.dispatch({ type: "run.interrupt", @@ -2481,7 +2494,14 @@ it.effect("queued-message commands read a long thread's history without walking const after = yield* projections.getThreadProjection(threadId); assert.deepEqual( after.messages.filter((message) => message.role === "user").map((message) => message.text), - ["Request 1", "Request 2", "Request 3", "/compact", "Request queued, edited"], + [ + "Request 1", + "Request 2", + "Request 3", + "/compact", + "Request cancelled", + "Request queued, edited", + ], ); assert.deepEqual( after.providerThreads.map((providerThread) => providerThread.id), @@ -2490,3 +2510,103 @@ it.effect("queued-message commands read a long thread's history without walking assert.deepEqual(walks, [], walks.join("\n")); }).pipe(Effect.provide(layerTest)), ); + +it.effect("a running provider-turn update reads only the thread's queued and running runs", () => + Effect.gen(function* () { + const runsReads = yield* Deferred.make["compile"]>>(); + let armed = false; + // The deferred-steer reactor runs in a fiber forked by the layer, so observe at layer level. + const observe: Statement.Transformer = (statement) => { + const compiled = statement.compile(); + return armed && /FROM orchestration_v2_projection_runs\b/.test(compiled[0]) + ? Deferred.succeed(runsReads, compiled).pipe(Effect.as(statement)) + : Effect.succeed(statement); + }; + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const eventSink = yield* EventSink.EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:deferred-steer-reads"); + const driver = ProviderDriverKind.make("codex"); + const now = yield* DateTime.now; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-deferred-steer-reads"), + threadId, + projectId: ProjectId.make("project:deferred-steer-reads"), + title: "Deferred steer reads", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("dispatch:deferred-steer-reads"), + threadId, + messageId: MessageId.make("message:deferred-steer-reads"), + text: "Request", + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const run = (yield* projections.getThreadProjection(threadId)).runs.at(-1)!; + const providerThreadId = ProviderThreadId.make("provider-thread:deferred-steer-reads"); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:deferred-steer-reads:run"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, providerThreadId, status: "running", queuePosition: null }, + }, + ], + }); + armed = true; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:deferred-steer-reads:turn"), + type: "provider-turn.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:deferred-steer-reads"), + providerThreadId, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId!, + nativeTurnRef: { + driver, + nativeId: "native:deferred-steer-reads", + strength: "strong", + }, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ], + }); + const [query, params] = yield* Deferred.await(runsReads); + const plan = yield* sql.unsafe<{ detail: string }>(`EXPLAIN QUERY PLAN ${query}`, params); + // Every run the thread ever had, when only queued steers and the running run matter. + assert.isFalse( + plan.some(({ detail }) => /^SEARCH .*runs.* \(thread_id=\?\)$/.test(detail)), + `${plan.map(({ detail }) => detail).join("\n")}\n in: ${query.replace(/\s+/g, " ").trim()}`, + ); + }).pipe( + Effect.provide( + layerTest.pipe(Layer.provide(Layer.succeed(Statement.CurrentTransformer, observe))), + ), + ); + }), +); From 3eca0df05fb8caf634d8a8cdb7efb2156ec1fde5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 09:19:15 +0000 Subject: [PATCH 2/2] fix(server): deferred-steer check and queued cancel no longer read a thread's whole history The deferred-steer reactor read every run of the thread on each running provider-turn update, and queued-run.cancel read every node and ordered every user message, both under the thread's dispatch lock. The steer check now reads only queued and running runs, and the cancel reads only the queued run's own message and skips settled leaf nodes. Fixes #36095 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 45edf2c74282e..469fa20838d2c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11169,7 +11169,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const projection = yield* loadProjectionForCommand( command, ["runs", "messages", "nodes", "attempts", "subagents", "turnItems"], - { turnItemTypes: [], messageRoles: ["user"] }, + // Cancelling reads the queued run's own message and root node, never a settled leaf. + { + turnItemTypes: [], + messageRoles: ["user"], + messageRunIds: [command.runId], + omitSettledLeafNodes: true, + }, ); const queuedRun = projection.runs.find((candidate) => candidate.id === command.runId); if (queuedRun === undefined || queuedRun.status !== "queued") { @@ -15281,7 +15287,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (stored.event.type !== "provider-turn.updated") return; const threadId = stored.event.threadId; const turn = stored.event.payload; - const projection = yield* projectionStore.getThreadRecords(threadId, ["runs"]); + const projection = yield* projectionStore.getThreadRecords(threadId, ["runs"], { + runStatuses: ["queued", "running"], + }); const run = projection.runs.find( (candidate) => candidate.activeAttemptId === turn.runAttemptId && candidate.status === "running",