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))), + ), + ); + }), +); 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",