Skip to content

Commit c814e51

Browse files
flambohmaria-rcks
andauthored
fix(server): a cancelled delegated task no longer shows its child's question (#18459)
Co-authored-by: maria-rcks <maria@kuuro.net>
1 parent fdacb3a commit c814e51

2 files changed

Lines changed: 131 additions & 0 deletions

File tree

‎apps/server/src/orchestration-v2/Orchestrator.ts‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10476,6 +10476,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1047610476
...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }),
1047710477
});
1047810478
}
10479+
10480+
const { resultRun } = progress;
10481+
if (progress.state !== "result_available" || resultRun === undefined) return;
10482+
const status = delegatedTaskTerminalStatus(runOutcome(resultRun));
10483+
if (status !== "cancelled" && status !== "interrupted") return;
10484+
if (
10485+
yield* childAwaitsRestartContinuation(
10486+
childControls.runs,
10487+
resultRun,
10488+
options?.settledContinuationOf,
10489+
)
10490+
)
10491+
return;
10492+
// Cancelled or interrupted tasks must not leave async questions that resume the child.
10493+
// Callers hold the parent lock; dismiss takes the child lock, as follow-up intake does.
10494+
const { runtimeRequests } = yield* projectionStore.getThreadRecords(childThreadId, [
10495+
"runtimeRequests",
10496+
]);
10497+
const pendingQuestions = runtimeRequests.filter(
10498+
(request) =>
10499+
request.status === "pending" &&
10500+
request.kind === "user_input" &&
10501+
request.responseCapability.type === "message",
10502+
);
10503+
if (pendingQuestions.length === 0) return;
10504+
const { nodes } = yield* projectionStore.getThreadRecords(childThreadId, ["nodes"]);
10505+
const resultRunIds = new Set<RunId>();
10506+
let run: OrchestrationV2Run | undefined = resultRun;
10507+
while (run !== undefined && !resultRunIds.has(run.id)) {
10508+
resultRunIds.add(run.id);
10509+
const previousRunId: RunId | undefined = run.restartContinuationOfRunId;
10510+
run =
10511+
previousRunId === undefined
10512+
? undefined
10513+
: childControls.runs.find((candidate) => candidate.id === previousRunId);
10514+
}
10515+
for (const request of pendingQuestions) {
10516+
if (
10517+
nodes.some(
10518+
(node) =>
10519+
node.id === request.nodeId && node.runId !== null && resultRunIds.has(node.runId),
10520+
)
10521+
) {
10522+
yield* dispatchWithReceipt({
10523+
type: "thread.user-input.dismiss",
10524+
commandId: CommandId.make(`command:system:delegated-task-question:${request.id}`),
10525+
threadId: childThreadId,
10526+
requestId: request.id,
10527+
}).pipe(
10528+
Effect.catchCause((cause) =>
10529+
Effect.logWarning("Failed to dismiss task question", {
10530+
requestId: request.id,
10531+
cause,
10532+
}),
10533+
),
10534+
);
10535+
}
10536+
}
1047910537
});
1048010538

1048110539
const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) =>

‎apps/server/src/orchestration-v2/ThreadStop.test.ts‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
ProviderInstanceId,
1010
ProviderSessionId,
1111
ProviderTurnId,
12+
RuntimeRequestId,
1213
ThreadId,
1314
TurnItemId,
1415
} from "@t3tools/contracts";
@@ -539,6 +540,78 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)(
539540
}).pipe(Effect.provide(layerTest.pipe(Layer.provideMerge(TestClock.layer())))),
540541
);
541542

543+
it.effect.each([
544+
{ ending: "cancelled", question: "resolved" },
545+
{ ending: "interrupted", question: "resolved" },
546+
{ ending: "completed", question: "pending" },
547+
{ ending: "failed", question: "pending" },
548+
] as const)(
549+
"a $ending delegated task leaves its async question $question",
550+
({ ending, question }) =>
551+
Effect.gen(function* () {
552+
const orchestrator = yield* Orchestrator.OrchestratorV2;
553+
const projections = yield* ProjectionStore.ProjectionStoreV2;
554+
const parentThreadId = ThreadId.make(`thread:stop-question-parent:${ending}`);
555+
yield* createWatchingThread(parentThreadId, 20);
556+
yield* send(parentThreadId, "first", "start_immediately");
557+
const childThreadId = yield* delegate(parentThreadId, `question task ${ending}`);
558+
const childRun = (yield* orchestrator.getThreadProjection(childThreadId)).runs[0]!;
559+
assert.equal(
560+
(yield* orchestrator.getThreadProjection(childThreadId)).nodes.find(
561+
(node) => node.id === childRun.rootNodeId,
562+
)?.runId,
563+
childRun.id,
564+
);
565+
const now = yield* DateTime.now;
566+
yield* projections.apply({
567+
id: EventId.make("event:stop-question:request"),
568+
type: "runtime-request.updated",
569+
threadId: childThreadId,
570+
runId: childRun.id,
571+
occurredAt: now,
572+
payload: {
573+
id: RuntimeRequestId.make("request:stop-question"),
574+
nodeId: childRun.rootNodeId!,
575+
providerTurnId: null,
576+
nativeRequestRef: null,
577+
kind: "user_input",
578+
status: "pending",
579+
responseCapability: { type: "message" },
580+
createdAt: now,
581+
resolvedAt: null,
582+
},
583+
});
584+
const request = Effect.map(
585+
orchestrator.getThreadProjection(childThreadId),
586+
(projection) => projection.runtimeRequests[0],
587+
);
588+
589+
// While the task runs, its question stays open.
590+
yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id);
591+
assert.equal((yield* request)?.status, "pending");
592+
if (ending === "interrupted") {
593+
// What task_cancel dispatches.
594+
yield* orchestrator.dispatch({
595+
type: "thread.stop",
596+
commandId: CommandId.make("stop-question-child"),
597+
threadId: childThreadId,
598+
});
599+
} else {
600+
yield* projections.apply({
601+
id: EventId.make(`event:stop-question:${ending}`),
602+
type: "run.updated",
603+
threadId: childThreadId,
604+
runId: childRun.id,
605+
occurredAt: now,
606+
payload: { ...childRun, status: ending, startedAt: now, completedAt: now },
607+
});
608+
}
609+
yield* orchestrator.recoverDelegatedTask(childThreadId, childRun.id);
610+
assert.equal((yield* request)?.status, question);
611+
assert.equal((yield* request)?.decision, question === "resolved" ? "cancel" : undefined);
612+
}).pipe(Effect.provide(layerTest)),
613+
);
614+
542615
it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () =>
543616
Effect.gen(function* () {
544617
const orchestrator = yield* Orchestrator.OrchestratorV2;

0 commit comments

Comments
 (0)