diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 003844efa7206f55289db15ee5fc89e36097d32e..f31d1bb5d36b43d487d91a249f590b9bf6151671 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import * as Tracer from "effect/Tracer"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../config.ts"; @@ -749,6 +750,24 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { // Runtime reconciliation writes restart cancellations under this command // prefix, which the live terminal-run listener skips. +// Collects the SQL text of every statement run under the returned tracer. +const traceSqlStatements = () => { + const statements: Array = []; + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const query = span.attributes.get("db.query.text"); + if (typeof query === "string") statements.push(query); + }; + return span; + }, + }); + return { statements, tracer }; +}; + const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`); const parentProviderThreadId = (threadId: ThreadId) => @@ -1578,6 +1597,111 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => { }), ); + it.effect("settles only the current delivery of a parent with long delivery history", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:delivery-history"); + const runId = RunId.make("run:delivery-history"); + const taskId = NodeId.make("node:delivery-history-task"); + const currentMessageId = MessageId.make(`message:delegated-delivery:${threadId}`); + const currentRunId = RunId.make("run:delivery-history:current"); + const HISTORY = 40; + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:delivery-history"), + runId, + rootNodeId: NodeId.make("node:delivery-history-root"), + taskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [taskId], + now, + }); + const deliveryMessage = (id: MessageId, deliveryRunId: RunId) => ({ + id: EventId.make(`event:${deliveryRunId}:message`), + type: "message.updated" as const, + threadId, + runId: deliveryRunId, + occurredAt: now, + payload: { + id, + threadId, + runId: deliveryRunId, + nodeId: null, + role: "user" as const, + text: `Delegated task ${taskId} reached a terminal state.`, + attachments: [], + streaming: false, + createdBy: "agent" as const, + creationSource: "server" as const, + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }); + // Earlier deliveries that already settled. Each one still owns a + // delegated message and a terminal run. + const history = Array.from({ length: HISTORY }, (_, index) => { + const historyRunId = RunId.make(`run:delivery-history:${index}`); + const messageId = MessageId.make(`message:delivery-history:${index}`); + return [ + deliveryMessage(messageId, historyRunId), + runEvent({ + threadId, + runId: historyRunId, + ordinal: index + 2, + status: "completed", + now, + providerThreadId: parentProviderThreadId(threadId), + userMessageId: messageId, + }), + ]; + }).flat(); + yield* eventSink.write({ + commandId: reconcileCommandId("delivery-history"), + events: [ + ...history, + deliveryMessage(currentMessageId, currentRunId), + runEvent({ + threadId, + runId: currentRunId, + ordinal: HISTORY + 2, + status: "completed", + now, + providerThreadId: parentProviderThreadId(threadId), + userMessageId: currentMessageId, + }), + ], + }); + + const { statements, tracer } = traceSqlStatements(); + yield* orchestrator.recoverDelegatedTasks.pipe(Effect.withTracer(tracer)); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + "delivered", + ); + assert.isNull(projection.runs.find((row) => row.id === runId)?.delegatedCompletion?.delivery); + // Recovery reads the delivery message once, for the current delivery + // run only. History must not cost one thread reload per past delivery. + assert.lengthOf( + statements.filter((statement) => + statement.includes("FROM orchestration_v2_projection_messages AS message"), + ), + 1, + ); + + // A second recovery finds nothing open on this thread and changes nothing. + yield* orchestrator.recoverDelegatedTasks; + const again = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(again.runs, projection.runs); + assert.deepEqual(again.subagents, projection.subagents); + }), + ); + it.effect("settles a cancelled child whose held wakes wait behind it", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 35a2ecf02c69185dea806950c30678eef5c34688..e793548ea50d954d363bcf24d639f86edd3664d9 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11333,6 +11333,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // queue.resume command arrives. const recoverDelegatedTasks = Effect.gen(function* () { yield* projectionStore.getRecoveryThreadIds("subagent-results").pipe( + // Candidate counts let a startup trace show whether recovery scales + // with pending work or with history. + Effect.tap((threadIds) => + Effect.annotateCurrentSpan("recovery.subagentResultThreads", threadIds.length), + ), Effect.flatMap((threadIds) => Effect.forEach( threadIds, @@ -11360,6 +11365,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); yield* projectionStore.getRecoveryThreadIds("delegated-completions").pipe( + Effect.tap((threadIds) => + Effect.annotateCurrentSpan("recovery.delegatedCompletionThreads", threadIds.length), + ), Effect.flatMap((threadIds) => Effect.forEach( threadIds, @@ -11368,18 +11376,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .withLock( threadId, Effect.gen(function* () { - const projection = yield* projectionStore.getThreadRecords(threadId, [ - "runs", - "messages", - ]); + const projection = yield* projectionStore.getThreadRecords(threadId, ["runs"]); + // Only the run that consumed a cohort's current delivery + // message can settle it. A long-lived parent keeps every + // past delivery run, and finalizing each of them would + // reload the thread once per historical delivery. + const openDeliveryMessageIds = new Set( + projection.runs.flatMap((run) => + run.delegatedCompletion?.delivery == null + ? [] + : [run.delegatedCompletion.delivery.messageId], + ), + ); const terminalDeliveryRunIds = projection.runs - .filter((run) => delegatedTaskTerminalStatus(run.status) !== null) - .filter((run) => - projection.messages.some( - (message) => - message.id === run.userMessageId && - message.delegatedCompletion !== undefined, - ), + .filter( + (run) => + delegatedTaskTerminalStatus(run.status) !== null && + openDeliveryMessageIds.has(run.userMessageId), ) .map((run) => run.id); for (const runId of terminalDeliveryRunIds) { @@ -11388,9 +11401,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const refreshed = terminalDeliveryRunIds.length === 0 ? projection - : yield* projectionStore.getThreadRecords(threadId, ["runs", "messages"], { - messageRoles: ["user"], - }); + : yield* projectionStore.getThreadRecords(threadId, ["runs"]); for (const run of refreshed.runs) { if ( run.delegatedCompletion?.delivery !== null &&