diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6f4e99814696572efcb51112b1dab9ab3d3b7d15..c7b322f2c49203403feceb024ab9fd9304323951 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10476,6 +10476,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }), }); } + + const { resultRun } = progress; + if (progress.state !== "result_available" || resultRun === undefined) return; + const status = delegatedTaskTerminalStatus(runOutcome(resultRun)); + if (status !== "cancelled" && status !== "interrupted") return; + if ( + yield* childAwaitsRestartContinuation( + childControls.runs, + resultRun, + options?.settledContinuationOf, + ) + ) + return; + // Cancelled or interrupted tasks must not leave async questions that resume the child. + // Callers hold the parent lock; dismiss takes the child lock, as follow-up intake does. + const { runtimeRequests } = yield* projectionStore.getThreadRecords(childThreadId, [ + "runtimeRequests", + ]); + const pendingQuestions = runtimeRequests.filter( + (request) => + request.status === "pending" && + request.kind === "user_input" && + request.responseCapability.type === "message", + ); + if (pendingQuestions.length === 0) return; + const { nodes } = yield* projectionStore.getThreadRecords(childThreadId, ["nodes"]); + const resultRunIds = new Set(); + let run: OrchestrationV2Run | undefined = resultRun; + while (run !== undefined && !resultRunIds.has(run.id)) { + resultRunIds.add(run.id); + const previousRunId: RunId | undefined = run.restartContinuationOfRunId; + run = + previousRunId === undefined + ? undefined + : childControls.runs.find((candidate) => candidate.id === previousRunId); + } + for (const request of pendingQuestions) { + if ( + nodes.some( + (node) => + node.id === request.nodeId && node.runId !== null && resultRunIds.has(node.runId), + ) + ) { + yield* dispatchWithReceipt({ + type: "thread.user-input.dismiss", + commandId: CommandId.make(`command:system:delegated-task-question:${request.id}`), + threadId: childThreadId, + requestId: request.id, + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to dismiss task question", { + requestId: request.id, + cause, + }), + ), + ); + } + } }); const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) => diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 2357c149090129720fb7035cc325380ae2773071..1e141cea588e2ce35049997e9af408ed20957177 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -9,6 +9,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderTurnId, + RuntimeRequestId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -539,6 +540,78 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)( }).pipe(Effect.provide(layerTest.pipe(Layer.provideMerge(TestClock.layer())))), ); +it.effect.each([ + { ending: "cancelled", question: "resolved" }, + { ending: "interrupted", question: "resolved" }, + { ending: "completed", question: "pending" }, + { ending: "failed", question: "pending" }, +] as const)( + "a $ending delegated task leaves its async question $question", + ({ ending, question }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const parentThreadId = ThreadId.make(`thread:stop-question-parent:${ending}`); + yield* createWatchingThread(parentThreadId, 20); + yield* send(parentThreadId, "first", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, `question task ${ending}`); + const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!; + assert.equal( + (yield* orchestrator.getThreadProjection(childThreadId)).nodes.find( + (node) => node.id === childRun.rootNodeId, + )?.runId, + childRun.id, + ); + const now = yield* DateTime.now; + yield* projections.apply({ + id: EventId.make("event:stop-question:request"), + type: "runtime-request.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:stop-question"), + nodeId: childRun.rootNodeId!, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }, + }); + const request = Effect.map( + orchestrator.getThreadProjection(childThreadId), + (projection) => projection.runtimeRequests[0], + ); + + // While the task runs, its question stays open. + yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id); + assert.equal((yield* request)?.status, "pending"); + if (ending === "interrupted") { + // What task_cancel dispatches. + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("stop-question-child"), + threadId: childThreadId, + }); + } else { + yield* projections.apply({ + id: EventId.make(`event:stop-question:${ending}`), + type: "run.updated", + threadId: childThreadId, + runId: childRun.id, + occurredAt: now, + payload: { ...childRun, status: ending, startedAt: now, completedAt: now }, + }); + } + yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id); + assert.equal((yield* request)?.status, question); + assert.equal((yield* request)?.decision, question === "resolved" ? "cancel" : undefined); + }).pipe(Effect.provide(layerTest)), +); + it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;