| author |  oliver <97427849+flamboh@users.noreply.github.com> 2026-10-14 20:18:17-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-15 03:18:17+00:00 |
| log | c814e51af2ba962c9291c9459ee64b4f57e0b5ce |
| tree | 0ad2de6403527804077f7efa805890e0177e875d |
| parent | fdacb3a3804bb535e8f5c7b733c4445e8567ea62 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): a cancelled delegated task no longer shows its child's question (#18459)
Co-authored-by: maria-rcks <maria@kuuro.net>2 files changed, 131 insertions(+), 0 deletions(-)
apps
server
src
orchestration-v2
apps/server/src/orchestration-v2/Orchestrator.ts+58
| ... | @@ -10476,6 +10476,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -10476,6 +10476,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 10476 | ...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }), | 10476 | ...(progress.resultRun === undefined ? {} : { resultRun: progress.resultRun }), |
| 10477 | }); | 10477 | }); |
| 10478 | } | 10478 | } |
| 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 | } | ||
| 10479 | }); | 10537 | }); |
| 10480 | 10538 | ||
| 10481 | const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) => | 10539 | const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) => |
apps/server/src/orchestration-v2/ThreadStop.test.ts+73
| ... | @@ -9,6 +9,7 @@ import { | ... | @@ -9,6 +9,7 @@ import { |
| 9 | ProviderInstanceId, | 9 | ProviderInstanceId, |
| 10 | ProviderSessionId, | 10 | ProviderSessionId, |
| 11 | ProviderTurnId, | 11 | ProviderTurnId, |
| 12 | RuntimeRequestId, | ||
| 12 | ThreadId, | 13 | ThreadId, |
| 13 | TurnItemId, | 14 | TurnItemId, |
| 14 | } from "@t3tools/contracts"; | 15 | } from "@t3tools/contracts"; |
| ... | @@ -539,6 +540,78 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)( | ... | @@ -539,6 +540,78 @@ it.effect.each(["thread.stop", "run.interrupt"] as const)( |
| 539 | }).pipe(Effect.provide(layerTest.pipe(Layer.provideMerge(TestClock.layer())))), | 540 | }).pipe(Effect.provide(layerTest.pipe(Layer.provideMerge(TestClock.layer())))), |
| 540 | ); | 541 | ); |
| 541 | 542 | ||
| 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 | |||
| 542 | it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () => | 615 | it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () => |
| 543 | Effect.gen(function* () { | 616 | Effect.gen(function* () { |
| 544 | const orchestrator = yield* Orchestrator.OrchestratorV2; | 617 | const orchestrator = yield* Orchestrator.OrchestratorV2; |