| author |  Theo Browne <me@t3.gg> 2026-10-05 13:51:15-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-05 13:51:15-07:00 |
| log | f32c23cf1aa9ff177224e93c239165adc67798c0 |
| tree | d2f55da23a732a6c38f23453a585b6140006bbe7 |
| parent | 2bffcc52f77bd6ff8be8e91be7a09f455eecbe07 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): Stop also stops delegated tasks and pull request watches (#16002)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>18 files changed, 1271 insertions(+), 183 deletions(-)
apps
server
src
mcp
- OrchestratorMcpService.test.ts
- OrchestratorMcpService.ts
- OrchestratorMcpToolkit.integration.test.ts
toolkits
orchestrator
pullRequests
docs
orchestration-v2
packages
contracts
apps/server/src/mcp/OrchestratorMcpService.test.ts+6-5
| ... | ... | @@ -316,7 +316,7 @@ describe("OrchestratorMcpService", () => { |
| 316 | 316 | }), |
| 317 | 317 | ); |
| 318 | 318 | |
| 319 | it.effect("does not dispose delivery when the child interrupt fails", () => | |
| 319 | it.effect("does not dispose delivery when the child stop fails", () => | |
| 320 | 320 | Effect.gen(function* () { |
| 321 | 321 | const parentThreadId = ThreadId.make("thread:mcp-cancel-failed-parent"); |
| 322 | 322 | const childThreadId = ThreadId.make("thread:mcp-cancel-failed-child"); |
| ... | ... | @@ -355,7 +355,7 @@ describe("OrchestratorMcpService", () => { |
| 355 | 355 | Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection), |
| 356 | 356 | dispatch: (command) => |
| 357 | 357 | Ref.update(dispatched, (commands) => [...commands, command]).pipe( |
| 358 | Effect.andThen(Effect.fail(new Error("simulated interrupt failure") as never)), | |
| 358 | Effect.andThen(Effect.fail(new Error("simulated stop failure") as never)), | |
| 359 | 359 | ), |
| 360 | 360 | }), |
| 361 | 361 | Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), |
| ... | ... | @@ -386,13 +386,13 @@ describe("OrchestratorMcpService", () => { |
| 386 | 386 | assert.equal(error.code, "task_not_cancellable"); |
| 387 | 387 | assert.deepEqual( |
| 388 | 388 | (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type), |
| 389 | ["run.interrupt"], | |
| 389 | ["thread.stop"], | |
| 390 | 390 | ); |
| 391 | 391 | }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(dependencies)))); |
| 392 | 392 | }), |
| 393 | 393 | ); |
| 394 | 394 | |
| 395 | it.effect("returns cancel requested when post-interrupt disposal fails", () => | |
| 395 | it.effect("returns cancel requested when post-stop disposal fails", () => | |
| 396 | 396 | Effect.gen(function* () { |
| 397 | 397 | const parentThreadId = ThreadId.make("thread:mcp-cancel-dispose-failed-parent"); |
| 398 | 398 | const childThreadId = ThreadId.make("thread:mcp-cancel-dispose-failed-child"); |
| ... | ... | @@ -437,6 +437,7 @@ describe("OrchestratorMcpService", () => { |
| 437 | 437 | : Effect.succeed({} as never), |
| 438 | 438 | ), |
| 439 | 439 | ), |
| 440 | stopDelegatedTasks: () => Effect.void, | |
| 440 | 441 | }), |
| 441 | 442 | Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }), |
| 442 | 443 | Layer.mock(ProviderAdapterRegistry.ProviderAdapterRegistryV2)({ |
| ... | ... | @@ -467,7 +468,7 @@ describe("OrchestratorMcpService", () => { |
| 467 | 468 | assert.equal(result.status, "cancel_requested"); |
| 468 | 469 | assert.deepEqual( |
| 469 | 470 | (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type), |
| 470 | ["run.interrupt", "delegated_task.completion-delivery.dispose"], | |
| 471 | ["thread.stop", "delegated_task.completion-delivery.dispose"], | |
| 471 | 472 | ); |
| 472 | 473 | }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(dependencies)))); |
| 473 | 474 | }), |
apps/server/src/mcp/OrchestratorMcpService.ts+41-24
| ... | ... | @@ -1662,43 +1662,60 @@ const make = Effect.gen(function* () { |
| 1662 | 1662 | ), |
| 1663 | 1663 | ), |
| 1664 | 1664 | ); |
| 1665 | // Published task results stay terminal. Later child-thread messages do not | |
| 1666 | // reopen the task, so cancelling it must not interrupt those separate runs. | |
| 1665 | // Cancelling stops the child thread the way Stop does, including work that came | |
| 1666 | // after a published result: follow-up runs, pull request watch wakes, and the | |
| 1667 | // tasks it delegated. The published result stays terminal. | |
| 1668 | const stopChild = Effect.gen(function* () { | |
| 1669 | const commandId = stableCommandId({ scope, requestKey: key, operation: "cancel-task" }); | |
| 1670 | const reason = input.reason; | |
| 1671 | yield* threadManagement | |
| 1672 | .dispatch({ | |
| 1673 | type: "thread.stop", | |
| 1674 | commandId, | |
| 1675 | threadId: current.childThreadId, | |
| 1676 | ...(reason === undefined ? {} : { reason }), | |
| 1677 | }) | |
| 1678 | .pipe( | |
| 1679 | Effect.mapError((error) => | |
| 1680 | failure( | |
| 1681 | "task_not_cancellable", | |
| 1682 | `Unable to stop delegated task ${input.taskId}: ${errorMessage(error)}`, | |
| 1683 | ), | |
| 1684 | ), | |
| 1685 | ); | |
| 1686 | // A retry with the same clientRequestId repeats only the stops that failed. | |
| 1687 | yield* threadManagement | |
| 1688 | .stopDelegatedTasks({ threadId: current.childThreadId, commandId, reason }) | |
| 1689 | .pipe( | |
| 1690 | Effect.mapError((error) => | |
| 1691 | failure( | |
| 1692 | "task_not_cancellable", | |
| 1693 | `Stopped delegated task ${input.taskId}, but not every task it delegated: ${errorMessage(error)}`, | |
| 1694 | ), | |
| 1695 | ), | |
| 1696 | ); | |
| 1697 | }); | |
| 1667 | 1698 | if (isTerminalTaskStatus(current.status)) { |
| 1699 | yield* stopChild; | |
| 1668 | 1700 | yield* disposeCompletionDelivery; |
| 1669 | 1701 | return { |
| 1670 | 1702 | taskId: input.taskId, |
| 1671 | 1703 | status: current.status, |
| 1672 | 1704 | } satisfies OrchestratorMcpTaskCancelResult; |
| 1673 | 1705 | } |
| 1706 | // A task waiting on its own delegates has work to stop below it. One with neither | |
| 1707 | // a running turn nor delegates (such as one awaiting a restart) keeps its delivery. | |
| 1674 | 1708 | const child = yield* loadProjection(current.childThreadId); |
| 1675 | const activeRun = ThreadManagementService.latestActiveRun(child); | |
| 1676 | if (activeRun === undefined) { | |
| 1709 | if ( | |
| 1710 | current.workState !== "waiting_for_children" && | |
| 1711 | ThreadManagementService.latestActiveRun(child) === undefined | |
| 1712 | ) { | |
| 1677 | 1713 | return yield* failure( |
| 1678 | 1714 | "task_not_cancellable", |
| 1679 | 1715 | `Delegated task ${input.taskId} has no interruptible child run.`, |
| 1680 | 1716 | ); |
| 1681 | 1717 | } |
| 1682 | yield* threadManagement | |
| 1683 | .dispatch({ | |
| 1684 | type: "run.interrupt", | |
| 1685 | commandId: stableCommandId({ | |
| 1686 | scope, | |
| 1687 | requestKey: key, | |
| 1688 | operation: "cancel-task", | |
| 1689 | }), | |
| 1690 | threadId: current.childThreadId, | |
| 1691 | runId: activeRun.id, | |
| 1692 | ...(input.reason === undefined ? {} : { reason: input.reason }), | |
| 1693 | }) | |
| 1694 | .pipe( | |
| 1695 | Effect.mapError((error) => | |
| 1696 | failure( | |
| 1697 | "task_not_cancellable", | |
| 1698 | `Unable to interrupt delegated task ${input.taskId}: ${errorMessage(error)}`, | |
| 1699 | ), | |
| 1700 | ), | |
| 1701 | ); | |
| 1718 | yield* stopChild; | |
| 1702 | 1719 | yield* disposeCompletionDelivery.pipe( |
| 1703 | 1720 | Effect.catchCause((cause) => |
| 1704 | 1721 | Effect.logWarning("orchestrator-mcp.cancel-task.delivery-dispose-failed", { |
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+6-23
| ... | ... | @@ -1663,7 +1663,7 @@ describe("orchestrator MCP toolkit", () => { |
| 1663 | 1663 | ).toBe(true); |
| 1664 | 1664 | const completedTaskCancelCall = yield* invoke("task_cancel", { |
| 1665 | 1665 | taskId: delegated.taskId, |
| 1666 | reason: "Must not interrupt a later unrelated child run.", | |
| 1666 | reason: "Stop the child's later work too.", | |
| 1667 | 1667 | clientRequestId: "cancel-completed-delegated-task-1", |
| 1668 | 1668 | }); |
| 1669 | 1669 | const completedTaskCancel = yield* decodeTaskCancelResult( |
| ... | ... | @@ -1681,36 +1681,19 @@ describe("orchestrator MCP toolkit", () => { |
| 1681 | 1681 | result: delegatedResult, |
| 1682 | 1682 | completionDelivery: { state: "disposed" }, |
| 1683 | 1683 | }); |
| 1684 | expect( | |
| 1685 | (yield* orchestrator.getThreadProjection(delegated.childThreadId)).runs.find( | |
| 1686 | (run) => run.id === activeChildFollowup.runId, | |
| 1687 | )?.status, | |
| 1688 | ).toBe("running"); | |
| 1689 | const activeChildCleanupCall = yield* invoke("t3_thread_interrupt", { | |
| 1690 | threadId: delegated.childThreadId, | |
| 1691 | runId: activeChildFollowup.runId, | |
| 1692 | reason: "Clean up the active follow-up after verifying task cancellation isolation.", | |
| 1693 | clientRequestId: "interrupt-delegated-child-followup-1", | |
| 1694 | }); | |
| 1695 | const activeChildCleanup = yield* decodeThreadInterruptResult( | |
| 1696 | activeChildCleanupCall.structuredContent, | |
| 1697 | ).pipe(Effect.orDie); | |
| 1698 | expect(activeChildCleanup).toMatchObject({ | |
| 1699 | runId: activeChildFollowup.runId, | |
| 1700 | status: "interrupt_requested", | |
| 1701 | }); | |
| 1684 | // Cancelling a finished task still stops the child thread's later work. | |
| 1702 | 1685 | yield* waitForProjection(orchestrator, delegated.childThreadId, (projection) => |
| 1703 | 1686 | projection.runs.some( |
| 1704 | 1687 | (run) => run.id === activeChildFollowup.runId && run.status === "interrupted", |
| 1705 | 1688 | ), |
| 1706 | 1689 | ); |
| 1707 | const delegatedStatusAfterCleanupCall = yield* invoke("task_status", { | |
| 1690 | const delegatedStatusAfterCancelCall = yield* invoke("task_status", { | |
| 1708 | 1691 | taskId: delegated.taskId, |
| 1709 | 1692 | }); |
| 1710 | const delegatedStatusAfterCleanup = yield* decodeDelegateTaskResult( | |
| 1711 | delegatedStatusAfterCleanupCall.structuredContent, | |
| 1693 | const delegatedStatusAfterCancel = yield* decodeDelegateTaskResult( | |
| 1694 | delegatedStatusAfterCancelCall.structuredContent, | |
| 1712 | 1695 | ).pipe(Effect.orDie); |
| 1713 | expect(delegatedStatusAfterCleanup).toMatchObject({ | |
| 1696 | expect(delegatedStatusAfterCancel).toMatchObject({ | |
| 1714 | 1697 | childRunId: delegated.childRunId, |
| 1715 | 1698 | status: "completed", |
| 1716 | 1699 | summary: delegatedResult, |
apps/server/src/mcp/toolkits/orchestrator/tools.test.ts+1-1
| ... | ... | @@ -26,7 +26,7 @@ describe("orchestrator MCP tool guidance", () => { |
| 26 | 26 | ); |
| 27 | 27 | assert.include( |
| 28 | 28 | OrchestratorToolkit.tools.task_cancel.description ?? "", |
| 29 | "without interrupting later child-thread runs", | |
| 29 | "This includes later child-thread runs, even after the task is terminal", | |
| 30 | 30 | ); |
| 31 | 31 | }); |
| 32 | 32 |
apps/server/src/mcp/toolkits/orchestrator/tools.ts+2-2
| ... | ... | @@ -71,7 +71,7 @@ export const DelegateTaskTool = Tool.make("delegate_task", { |
| 71 | 71 | |
| 72 | 72 | const TaskStatusTool = Tool.make("task_status", { |
| 73 | 73 | description: |
| 74 | "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task or extend task_cancel to those turns. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.", | |
| 74 | "Needs an agent running inside a T3 thread. Read a T3-owned delegated task created by this parent thread. childRunId identifies the original delegated run. workState distinguishes working, waiting_for_children, and result_available; a completed turn with live nested work is not a completed task. summary is the final task result, including provider errors on failure, and remains stable after publication. hasPendingChildRuns reports later queued or executing turns in the backing child thread, even after the task is terminal; it does not reopen the task, and task_cancel stops those turns too. latestTerminal* provides later non-monitor turn results. Reading a terminal result acknowledges its automatic parent delivery.", | |
| 75 | 75 | parameters: OrchestratorMcpTaskStatusInput, |
| 76 | 76 | success: OrchestratorMcpDelegateTaskResult, |
| 77 | 77 | failure: OrchestratorMcpFailure, |
| ... | ... | @@ -85,7 +85,7 @@ const TaskStatusTool = Tool.make("task_status", { |
| 85 | 85 | |
| 86 | 86 | const TaskCancelTool = Tool.make("task_cancel", { |
| 87 | 87 | description: |
| 88 | "Needs an agent running inside a T3 thread. Request interruption of an active T3-owned delegated task and dispose its automatic parent delivery. For a terminal task, return its existing status and dispose delivery without interrupting later child-thread runs, even when task_status reports hasPendingChildRuns=true. Published task results remain available. Use t3_thread_interrupt for a later active run.", | |
| 88 | "Needs an agent running inside a T3 thread. Stop a T3-owned delegated task and dispose its automatic parent delivery. Its child thread stops like a user Stop: the running turn is interrupted, queued turns are held, pull request watches end, and the tasks it delegated stop too. This includes later child-thread runs, even after the task is terminal. A terminal task returns its existing status, and published task results remain available.", | |
| 89 | 89 | parameters: OrchestratorMcpTaskCancelInput, |
| 90 | 90 | success: OrchestratorMcpTaskCancelResult, |
| 91 | 91 | failure: OrchestratorMcpFailure, |
apps/server/src/mcp/toolkits/pullRequests/tools.ts+1-1
| ... | ... | @@ -289,7 +289,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", { |
| 289 | 289 | |
| 290 | 290 | const WatchPullRequestTool = Tool.make("watch_pull_request", { |
| 291 | 291 | description: |
| 292 | "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every minute and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. Watching ends when the pull request merges or closes, when its thread settles, when T3 Code cannot read it for 15 minutes, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch.", | |
| 292 | "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every minute and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. Watching ends when the pull request merges or closes, when its thread settles, when T3 Code cannot read it for 15 minutes, when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch.", | |
| 293 | 293 | parameters: PullRequestTargetInput, |
| 294 | 294 | success: WatchPullRequestResult, |
| 295 | 295 | failure: PullRequestToolError, |
apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts+134-20
| ... | ... | @@ -39,9 +39,17 @@ const instanceId = ProviderInstanceId.make("codex"); |
| 39 | 39 | const modelSelection = { instanceId, model: "test-model" }; |
| 40 | 40 | |
| 41 | 41 | // Codex turns leave commands running, then the thread moves to another |
| 42 | // provider thread (a provider switch). Stop on the newer, settled run must | |
| 43 | // reach both provider threads and end all of the Codex work. | |
| 44 | const stopEarlierBackgroundWork = (failedStart: boolean) => | |
| 42 | // provider thread (a provider switch). Stop must end all of the Codex work, | |
| 43 | // including when it selects an older resumed run on the other provider thread. | |
| 44 | const stopEarlierBackgroundWork = ({ | |
| 45 | failedStart = false, | |
| 46 | stopWithQueue, | |
| 47 | olderStart = false, | |
| 48 | }: { | |
| 49 | readonly failedStart?: boolean; | |
| 50 | readonly stopWithQueue?: "thread.stop" | "run.interrupt"; | |
| 51 | readonly olderStart?: boolean; | |
| 52 | }) => | |
| 45 | 53 | Effect.scoped( |
| 46 | 54 | Effect.gen(function* () { |
| 47 | 55 | const cwd = yield* checkpointWorkspace("background-work-stop"); |
| ... | ... | @@ -377,10 +385,11 @@ const stopEarlierBackgroundWork = (failedStart: boolean) => |
| 377 | 385 | // the thread moves on to another provider thread, which also has a |
| 378 | 386 | // live session. |
| 379 | 387 | const watcherId = TurnItemId.make("turn-item:watcher"); |
| 388 | const otherProviderThreadId = ProviderThreadId.make("provider-thread:other"); | |
| 380 | 389 | const watcherRun = settledRun({ |
| 381 | 390 | ordinal: 2, |
| 382 | providerThreadId: codexProviderThread.id, | |
| 383 | runningItem: { id: watcherId, kind: "command" }, | |
| 391 | providerThreadId: olderStart ? otherProviderThreadId : codexProviderThread.id, | |
| 392 | ...(olderStart ? {} : { runningItem: { id: watcherId, kind: "command" as const } }), | |
| 384 | 393 | }); |
| 385 | 394 | const reviewerId = TurnItemId.make("turn-item:reviewer"); |
| 386 | 395 | const reviewerRun = settledRun({ |
| ... | ... | @@ -388,7 +397,6 @@ const stopEarlierBackgroundWork = (failedStart: boolean) => |
| 388 | 397 | providerThreadId: codexProviderThread.id, |
| 389 | 398 | runningItem: { id: reviewerId, kind: "subagent" }, |
| 390 | 399 | }); |
| 391 | const otherProviderThreadId = ProviderThreadId.make("provider-thread:other"); | |
| 392 | 400 | const latestRun = settledRun({ ordinal: 4, providerThreadId: otherProviderThreadId }); |
| 393 | 401 | yield* sink.write({ |
| 394 | 402 | events: [ |
| ... | ... | @@ -440,30 +448,126 @@ const stopEarlierBackgroundWork = (failedStart: boolean) => |
| 440 | 448 | }); |
| 441 | 449 | } |
| 442 | 450 | |
| 443 | yield* orchestrator.dispatch({ | |
| 444 | type: "run.interrupt", | |
| 445 | commandId: CommandId.make("stop-background-work"), | |
| 446 | threadId, | |
| 447 | runId: failedStart ? failedRun.runId : latestRun.runId, | |
| 448 | }); | |
| 451 | if (stopWithQueue !== undefined) { | |
| 452 | const owner = (yield* orchestrator.getThreadProjection(threadId)).runs.find( | |
| 453 | (run) => run.id === latestRun.runId, | |
| 454 | )!; | |
| 455 | // A message queues while checkpointing finishes, then the user holds the queue. | |
| 456 | yield* sink.write({ | |
| 457 | events: [ | |
| 458 | { | |
| 459 | id: EventId.make("latest-run-waiting"), | |
| 460 | type: "run.updated", | |
| 461 | threadId, | |
| 462 | runId: owner.id, | |
| 463 | occurredAt: now, | |
| 464 | payload: { ...owner, status: "waiting", completedAt: null }, | |
| 465 | }, | |
| 466 | ], | |
| 467 | }); | |
| 468 | yield* orchestrator.dispatch({ | |
| 469 | type: "message.dispatch", | |
| 470 | commandId: CommandId.make("queue-follow-up"), | |
| 471 | threadId, | |
| 472 | messageId: MessageId.make("message:queue-follow-up"), | |
| 473 | text: "Follow up after the background work", | |
| 474 | attachments: [], | |
| 475 | dispatchMode: { type: "queue_after_active" }, | |
| 476 | createdBy: "user", | |
| 477 | creationSource: "web", | |
| 478 | }); | |
| 479 | const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!; | |
| 480 | assert.equal(queued.status, "queued"); | |
| 481 | yield* sink.write({ | |
| 482 | events: [ | |
| 483 | { | |
| 484 | id: EventId.make("queue-held"), | |
| 485 | type: "run.updated", | |
| 486 | threadId, | |
| 487 | runId: queued.id, | |
| 488 | occurredAt: now, | |
| 489 | payload: { ...queued, queueHeld: true }, | |
| 490 | }, | |
| 491 | { | |
| 492 | id: EventId.make("latest-run-settled"), | |
| 493 | type: "run.updated", | |
| 494 | threadId, | |
| 495 | runId: owner.id, | |
| 496 | occurredAt: now, | |
| 497 | payload: owner, | |
| 498 | }, | |
| 499 | ], | |
| 500 | }); | |
| 501 | } | |
| 502 | ||
| 503 | if (olderStart) { | |
| 504 | const older = (yield* orchestrator.getThreadProjection(threadId)).runs.find( | |
| 505 | (run) => run.id === watcherRun.runId, | |
| 506 | )!; | |
| 507 | // A resumed queue can start a lower-ordinal run after later runs have ended. | |
| 508 | yield* sink.write({ | |
| 509 | events: [ | |
| 510 | { | |
| 511 | id: EventId.make("older-run-starting"), | |
| 512 | type: "run.updated", | |
| 513 | threadId, | |
| 514 | runId: older.id, | |
| 515 | occurredAt: now, | |
| 516 | payload: { ...older, status: "starting", startedAt: null, completedAt: null }, | |
| 517 | }, | |
| 518 | ], | |
| 519 | }); | |
| 520 | } | |
| 521 | ||
| 522 | yield* orchestrator.dispatch( | |
| 523 | stopWithQueue === "thread.stop" | |
| 524 | ? { | |
| 525 | type: "thread.stop", | |
| 526 | commandId: CommandId.make("stop-background-work"), | |
| 527 | threadId, | |
| 528 | } | |
| 529 | : { | |
| 530 | type: "run.interrupt", | |
| 531 | commandId: CommandId.make("stop-background-work"), | |
| 532 | threadId, | |
| 533 | runId: olderStart | |
| 534 | ? watcherRun.runId | |
| 535 | : failedStart | |
| 536 | ? failedRun.runId | |
| 537 | : latestRun.runId, | |
| 538 | ...(stopWithQueue === undefined ? {} : { holdQueue: true }), | |
| 539 | }, | |
| 540 | ); | |
| 449 | 541 | yield* worker.drain(); |
| 450 | 542 | |
| 451 | // Stop reaches both provider threads. The Codex one is interrupted at | |
| 452 | // its latest pending work, the subagent's parent turn, so its settle | |
| 453 | // covers all three Codex runs. | |
| 543 | // The Codex interrupt targets its latest pending work, the subagent's parent turn, | |
| 544 | // so its settlement covers the Codex background work. | |
| 454 | 545 | assert.sameDeepMembers( |
| 455 | 546 | interrupts.map((interrupt) => [interrupt.providerThread.id, interrupt.providerTurnId]), |
| 456 | 547 | [ |
| 457 | [otherProviderThreadId, latestRun.providerTurnId], | |
| 548 | ...(olderStart ? [] : [[otherProviderThreadId, latestRun.providerTurnId]]), | |
| 458 | 549 | [codexProviderThread.id, reviewerRun.providerTurnId], |
| 459 | 550 | ], |
| 460 | 551 | ); |
| 461 | 552 | const after = yield* orchestrator.getThreadProjection(threadId); |
| 553 | if (olderStart) { | |
| 554 | assert.equal( | |
| 555 | after.runs.find((run) => run.id === watcherRun.runId)?.status, | |
| 556 | "interrupted", | |
| 557 | ); | |
| 558 | } | |
| 559 | if (stopWithQueue !== undefined) { | |
| 560 | assert.equal(after.runs.at(-1)?.status, "queued"); | |
| 561 | assert.equal(after.runs.at(-1)?.queueHeld, true); | |
| 562 | assert.lengthOf(started, 1); | |
| 563 | } | |
| 462 | 564 | assert.deepEqual( |
| 463 | [devServerId, watcherId, reviewerId].map( | |
| 565 | [devServerId, ...(olderStart ? [] : [watcherId]), reviewerId].map( | |
| 464 | 566 | (id) => after.turnItems.find((item) => item.id === id)?.status, |
| 465 | 567 | ), |
| 466 | ["interrupted", "interrupted", "interrupted"], | |
| 568 | olderStart | |
| 569 | ? ["interrupted", "interrupted"] | |
| 570 | : ["interrupted", "interrupted", "interrupted"], | |
| 467 | 571 | ); |
| 468 | 572 | }).pipe( |
| 469 | 573 | Effect.provide( |
| ... | ... | @@ -478,10 +582,20 @@ const stopEarlierBackgroundWork = (failedStart: boolean) => |
| 478 | 582 | ); |
| 479 | 583 | |
| 480 | 584 | it.effect("Stop reaches background work an earlier provider thread still runs", () => |
| 481 | stopEarlierBackgroundWork(false), | |
| 585 | stopEarlierBackgroundWork({}), | |
| 482 | 586 | ); |
| 483 | 587 | |
| 484 | 588 | it.effect( |
| 485 | 589 | "Stop reaches earlier background work after the newest run fails before provider start", |
| 486 | () => stopEarlierBackgroundWork(true), | |
| 590 | () => stopEarlierBackgroundWork({ failedStart: true }), | |
| 591 | ); | |
| 592 | ||
| 593 | it.effect.each(["thread.stop", "run.interrupt"] as const)( | |
| 594 | "%s stops background work when a later message is queued", | |
| 595 | (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType }), | |
| 596 | ); | |
| 597 | ||
| 598 | it.effect.each(["thread.stop", "run.interrupt"] as const)( | |
| 599 | "%s reaches later background work when an older run is starting", | |
| 600 | (stopType) => stopEarlierBackgroundWork({ stopWithQueue: stopType, olderStart: true }), | |
| 487 | 601 | ); |
apps/server/src/orchestration-v2/EffectOutbox.ts+6
| ... | ... | @@ -102,6 +102,11 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ |
| 102 | 102 | Schema.Struct({ type: Schema.Literal("regenerate") }), |
| 103 | 103 | ]), |
| 104 | 104 | }), |
| 105 | /** Follows a Stop: sends `thread.stop` to every delegated task under the stopped thread. */ | |
| 106 | Schema.Struct({ | |
| 107 | type: Schema.Literal("delegated-tasks.stop"), | |
| 108 | reason: Schema.optional(Schema.String), | |
| 109 | }), | |
| 105 | 110 | ]); |
| 106 | 111 | export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.Type; |
| 107 | 112 | |
| ... | ... | @@ -113,6 +118,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ |
| 113 | 118 | "terminal.cleanup", |
| 114 | 119 | "attachment.cleanup", |
| 115 | 120 | "thread-title.generate", |
| 121 | "delegated-tasks.stop", | |
| 116 | 122 | ] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>; |
| 117 | 123 | |
| 118 | 124 | export const PROCESS_BOUND_EFFECT_TYPES = [ |
apps/server/src/orchestration-v2/EffectWorker.ts+17
| ... | ... | @@ -459,6 +459,23 @@ export const executorLayer: Layer.Layer< |
| 459 | 459 | }), |
| 460 | 460 | ), |
| 461 | 461 | ); |
| 462 | case "delegated-tasks.stop": | |
| 463 | return threads | |
| 464 | .stopDelegatedTasks({ | |
| 465 | threadId: effect.threadId, | |
| 466 | commandId: effect.commandId, | |
| 467 | reason: effect.request.reason, | |
| 468 | }) | |
| 469 | .pipe( | |
| 470 | Effect.mapError( | |
| 471 | (cause) => | |
| 472 | new OrchestrationEffectExecutionError({ | |
| 473 | effectId: effect.id, | |
| 474 | effectType: effect.request.type, | |
| 475 | cause, | |
| 476 | }), | |
| 477 | ), | |
| 478 | ); | |
| 462 | 479 | } |
| 463 | 480 | }, |
| 464 | 481 | }); |
apps/server/src/orchestration-v2/Orchestrator.ts+378-96
| ... | ... | @@ -59,6 +59,7 @@ import { |
| 59 | 59 | import * as Context from "effect/Context"; |
| 60 | 60 | import * as DateTime from "effect/DateTime"; |
| 61 | 61 | import * as Effect from "effect/Effect"; |
| 62 | import * as Exit from "effect/Exit"; | |
| 62 | 63 | import * as FileSystem from "effect/FileSystem"; |
| 63 | 64 | import * as Path from "effect/Path"; |
| 64 | 65 | import * as Layer from "effect/Layer"; |
| ... | ... | @@ -418,6 +419,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { |
| 418 | 419 | case "checkpoint.rollback": |
| 419 | 420 | case "checkpoint.rollback.fail": |
| 420 | 421 | case "thread.background-work.settle": |
| 422 | case "thread.stop": | |
| 421 | 423 | case "provider.switch": |
| 422 | 424 | return command.threadId; |
| 423 | 425 | case "delegated_task.request": |
| ... | ... | @@ -1038,7 +1040,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 1038 | 1040 | }); |
| 1039 | 1041 | |
| 1040 | 1042 | const emitQueuedRunCancellation = (input: { |
| 1041 | readonly command: OrchestrationV2Command; | |
| 1043 | readonly command: OrchestrationV2ServerCommand; | |
| 1042 | 1044 | readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>; |
| 1043 | 1045 | readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "attempts">; |
| 1044 | 1046 | readonly run: OrchestrationV2Run; |
| ... | ... | @@ -2012,7 +2014,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 2012 | 2014 | }); |
| 2013 | 2015 | |
| 2014 | 2016 | const disposeDelegatedCompletionCohort = (input: { |
| 2015 | readonly command: OrchestrationV2Command; | |
| 2017 | readonly command: OrchestrationV2ServerCommand; | |
| 2016 | 2018 | readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>; |
| 2017 | 2019 | readonly projection: Pick< |
| 2018 | 2020 | OrchestrationV2ThreadProjection, |
| ... | ... | @@ -2379,6 +2381,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 2379 | 2381 | cause: `Thread ${command.threadId} is settled and cannot watch pull requests.`, |
| 2380 | 2382 | }); |
| 2381 | 2383 | } |
| 2384 | // Only the agent's watch_pull_request links as "agent". A call that raced a Stop may land | |
| 2385 | // after its run ended, so the latest run that executed decides. A user can still watch. | |
| 2386 | if ( | |
| 2387 | command.type === "thread.pull-request.watch" && | |
| 2388 | command.watching && | |
| 2389 | command.link?.source === "agent" | |
| 2390 | ) { | |
| 2391 | const { runs } = yield* projectionStore | |
| 2392 | .getThreadRecords(command.threadId, ["runs"]) | |
| 2393 | .pipe(mapDispatchError(command)); | |
| 2394 | const latest = latestExecutedRun(runs); | |
| 2395 | if (latest !== null && (yield* stopReachedRun(command, command.threadId, latest.id))) { | |
| 2396 | return yield* new OrchestratorDispatchError({ | |
| 2397 | commandId: command.commandId, | |
| 2398 | commandType: command.type, | |
| 2399 | cause: `Thread ${command.threadId} was stopped.`, | |
| 2400 | }); | |
| 2401 | } | |
| 2402 | } | |
| 2382 | 2403 | if ( |
| 2383 | 2404 | command.type === "thread.metadata.update" && |
| 2384 | 2405 | command.expectedWorktreePath !== undefined && |
| ... | ... | @@ -4402,7 +4423,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 4402 | 4423 | // Held queued runs never started; they wait behind the continuation. |
| 4403 | 4424 | projection.runs.some( |
| 4404 | 4425 | (run) => run.id !== source.id && run.status !== "queued" && runRanAfter(run, source), |
| 4405 | ) | |
| 4426 | ) || | |
| 4427 | (yield* stopReachedRun(command, command.threadId, source.id)) | |
| 4406 | 4428 | ) { |
| 4407 | 4429 | // Preserve the current row so stale automatic deliveries receive an |
| 4408 | 4430 | // accepted receipt without changing work or repeatedly retrying. |
| ... | ... | @@ -6362,6 +6384,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 6362 | 6384 | cause: `Parent run ${command.parentRunId} is not active.`, |
| 6363 | 6385 | }); |
| 6364 | 6386 | } |
| 6387 | if (yield* stopReachedRun(command, command.parentThreadId, parentRun.id)) { | |
| 6388 | return yield* new OrchestratorDispatchError({ | |
| 6389 | commandId: command.commandId, | |
| 6390 | commandType: command.type, | |
| 6391 | cause: `Parent run ${command.parentRunId} is stopping.`, | |
| 6392 | }); | |
| 6393 | } | |
| 6365 | 6394 | const parentNode = parentProjection.nodes.find( |
| 6366 | 6395 | (candidate) => candidate.id === command.parentNodeId, |
| 6367 | 6396 | ); |
| ... | ... | @@ -7960,6 +7989,159 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 7960 | 7989 | } |
| 7961 | 7990 | }); |
| 7962 | 7991 | |
| 7992 | /** | |
| 7993 | * Whether Stop reached a run, as its interrupt request records. Its agent can call tools | |
| 7994 | * until the provider stops it, and a slow call can land after that, but nothing it starts | |
| 7995 | * may outlive the Stop. A restart continuation of that run must not start either. | |
| 7996 | */ | |
| 7997 | const stopReachedRun = ( | |
| 7998 | command: OrchestrationV2ServerCommand, | |
| 7999 | threadId: ThreadId, | |
| 8000 | runId: RunId, | |
| 8001 | ) => | |
| 8002 | projectionStore | |
| 8003 | .getThreadRecords(threadId, ["turnItems"], { | |
| 8004 | turnItemTypes: ["run_interrupt_request"], | |
| 8005 | turnItemRunIds: [runId], | |
| 8006 | }) | |
| 8007 | .pipe( | |
| 8008 | Effect.map(({ turnItems }) => turnItems.length > 0), | |
| 8009 | mapDispatchError(command), | |
| 8010 | ); | |
| 8011 | ||
| 8012 | /** Records Stop for a run without replacing a marker already written by its interrupt. */ | |
| 8013 | const markStoppedRun = Effect.fnUntraced(function* (input: { | |
| 8014 | readonly command: Extract< | |
| 8015 | OrchestrationV2ServerCommand, | |
| 8016 | { readonly type: "run.interrupt" | "thread.stop" } | |
| 8017 | >; | |
| 8018 | readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>; | |
| 8019 | readonly thread: OrchestrationV2AppThread; | |
| 8020 | readonly run: OrchestrationV2Run; | |
| 8021 | readonly now: DateTime.Utc; | |
| 8022 | }) { | |
| 8023 | const { command, events, thread, run, now } = input; | |
| 8024 | if (run.rootNodeId === null || run.providerThreadId === null) return; | |
| 8025 | const pendingItems = (yield* Ref.get(events)).flatMap((event) => | |
| 8026 | event.type === "turn-item.updated" ? [event.payload] : [], | |
| 8027 | ); | |
| 8028 | if ( | |
| 8029 | pendingItems.some((item) => item.runId === run.id && item.type === "run_interrupt_request") || | |
| 8030 | (yield* stopReachedRun(command, thread.id, run.id)) | |
| 8031 | ) { | |
| 8032 | return; | |
| 8033 | } | |
| 8034 | yield* emit( | |
| 8035 | events, | |
| 8036 | command, | |
| 8037 | )({ | |
| 8038 | type: "turn-item.updated", | |
| 8039 | threadId: thread.id, | |
| 8040 | runId: run.id, | |
| 8041 | nodeId: run.rootNodeId, | |
| 8042 | providerInstanceId: run.providerInstanceId, | |
| 8043 | occurredAt: now, | |
| 8044 | payload: { | |
| 8045 | id: idAllocator.derive.runSignalTurnItem({ runId: run.id, signal: "interrupt-request" }), | |
| 8046 | threadId: thread.id, | |
| 8047 | runId: run.id, | |
| 8048 | nodeId: run.rootNodeId, | |
| 8049 | providerThreadId: run.providerThreadId, | |
| 8050 | providerTurnId: null, | |
| 8051 | nativeItemRef: null, | |
| 8052 | parentItemId: null, | |
| 8053 | ordinal: yield* nextTurnItemOrdinal({ thread, turnItems: pendingItems }), | |
| 8054 | status: "completed", | |
| 8055 | title: "Interrupt requested", | |
| 8056 | startedAt: now, | |
| 8057 | completedAt: now, | |
| 8058 | updatedAt: now, | |
| 8059 | type: "run_interrupt_request", | |
| 8060 | message: command.reason ?? "Interrupt requested", | |
| 8061 | }, | |
| 8062 | }); | |
| 8063 | }); | |
| 8064 | ||
| 8065 | /** | |
| 8066 | * What Stop holds besides the run it interrupts: queued runs wait for the user, the | |
| 8067 | * thread's pull request watches end, and every wake its delegated tasks still owe is | |
| 8068 | * dropped, since Stop stops those tasks too. Nothing automatic starts the thread again. | |
| 8069 | * `cohortRunIds` are dropped even when they owe nothing, like a plain interrupt's. | |
| 8070 | */ | |
| 8071 | const holdStoppedThread = (input: { | |
| 8072 | readonly command: Extract< | |
| 8073 | OrchestrationV2ServerCommand, | |
| 8074 | { readonly type: "run.interrupt" | "thread.stop" } | |
| 8075 | >; | |
| 8076 | readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>; | |
| 8077 | readonly projection: Pick<OrchestrationV2ThreadProjection, "thread" | "runs" | "subagents">; | |
| 8078 | readonly cohortRunIds: ReadonlyArray<RunId>; | |
| 8079 | readonly now: DateTime.Utc; | |
| 8080 | }) => | |
| 8081 | Effect.gen(function* () { | |
| 8082 | const emitEvent = emit(input.events, input.command); | |
| 8083 | const { thread, runs, subagents } = input.projection; | |
| 8084 | for (const run of runs) { | |
| 8085 | // A resumed queue can have a lower ordinal than a pending continuation source. | |
| 8086 | // Stop must reach those sources even when interrupting the resumed run succeeds. | |
| 8087 | if ( | |
| 8088 | run.status === "cancelled" && | |
| 8089 | (yield* awaitsRestartContinuation(run).pipe(mapDispatchError(input.command))) | |
| 8090 | ) { | |
| 8091 | yield* markStoppedRun({ ...input, thread, run }); | |
| 8092 | } | |
| 8093 | if (run.status !== "queued" || run.queueHeld === true) continue; | |
| 8094 | yield* emitEvent({ | |
| 8095 | type: "run.updated", | |
| 8096 | threadId: thread.id, | |
| 8097 | runId: run.id, | |
| 8098 | providerInstanceId: run.providerInstanceId, | |
| 8099 | occurredAt: input.now, | |
| 8100 | payload: { ...run, queueHeld: true }, | |
| 8101 | }); | |
| 8102 | } | |
| 8103 | const pullRequests = thread.pullRequests ?? []; | |
| 8104 | if (pullRequests.some((link) => link.watch !== undefined)) { | |
| 8105 | yield* emitEvent({ | |
| 8106 | type: "thread.pull-request-synced", | |
| 8107 | threadId: thread.id, | |
| 8108 | providerInstanceId: thread.providerInstanceId, | |
| 8109 | occurredAt: input.now, | |
| 8110 | payload: { | |
| 8111 | ...thread, | |
| 8112 | pullRequests: pullRequests.map((link) => withPullRequestWatch(link, undefined)), | |
| 8113 | }, | |
| 8114 | }); | |
| 8115 | } | |
| 8116 | const cohortRunIds = new Set(input.cohortRunIds); | |
| 8117 | for (const run of runs) { | |
| 8118 | if (run.delegatedCompletion?.delivery != null) cohortRunIds.add(run.id); | |
| 8119 | } | |
| 8120 | for (const task of subagents) { | |
| 8121 | const delivery = task.completionDelivery?.state; | |
| 8122 | if ( | |
| 8123 | task.origin === "app_owned" && | |
| 8124 | task.runId !== null && | |
| 8125 | delivery !== "acknowledged" && | |
| 8126 | delivery !== "delivered" && | |
| 8127 | delivery !== "disposed" | |
| 8128 | ) { | |
| 8129 | cohortRunIds.add(task.runId); | |
| 8130 | } | |
| 8131 | } | |
| 8132 | // Each disposal reads the events the ones before it wrote. | |
| 8133 | for (const parentRunId of cohortRunIds) { | |
| 8134 | yield* disposeDelegatedCompletionCohort({ | |
| 8135 | command: input.command, | |
| 8136 | events: input.events, | |
| 8137 | projection: yield* getProjectionWithPendingEvents(thread.id, input.events), | |
| 8138 | parentRunId, | |
| 8139 | disposition: "stopped", | |
| 8140 | now: input.now, | |
| 8141 | }); | |
| 8142 | } | |
| 8143 | }); | |
| 8144 | ||
| 7963 | 8145 | const dispatchBackgroundWorkSettle = ( |
| 7964 | 8146 | command: Extract< |
| 7965 | 8147 | OrchestrationV2InternalCommand, |
| ... | ... | @@ -8045,7 +8227,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8045 | 8227 | ? undefined |
| 8046 | 8228 | : projection.providerThreads.find((candidate) => candidate.id === run?.providerThreadId); |
| 8047 | 8229 | const hasBackgroundWork = |
| 8048 | run?.id === projection.runs.at(-1)?.id && | |
| 8230 | run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id && | |
| 8049 | 8231 | derivePendingBackgroundWork({ |
| 8050 | 8232 | latestRun: run, |
| 8051 | 8233 | providerThreads: projection.providerThreads, |
| ... | ... | @@ -8078,34 +8260,29 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8078 | 8260 | (candidate) => candidate.id === run.userMessageId, |
| 8079 | 8261 | ); |
| 8080 | 8262 | const completionCohortRunId = completionMessage?.delegatedCompletion?.parentRunId ?? run.id; |
| 8081 | const stopCompletionCohort = () => | |
| 8082 | Effect.gen(function* () { | |
| 8083 | yield* disposeDelegatedCompletionCohort({ | |
| 8084 | command, | |
| 8085 | events, | |
| 8086 | projection: yield* getProjectionWithPendingEvents(command.threadId, events), | |
| 8087 | parentRunId: completionCohortRunId, | |
| 8088 | disposition: "stopped", | |
| 8089 | now, | |
| 8090 | }); | |
| 8091 | }); | |
| 8263 | // Stop holds the rest of the thread too. A plain interrupt only drops the wakes owed by | |
| 8264 | // this run's delegated tasks, or on a wake run, by the cohort that woke it. | |
| 8265 | const stopRemainingWork = () => | |
| 8266 | command.holdQueue === true | |
| 8267 | ? holdStoppedThread({ | |
| 8268 | command, | |
| 8269 | events, | |
| 8270 | projection, | |
| 8271 | cohortRunIds: [completionCohortRunId], | |
| 8272 | now, | |
| 8273 | }) | |
| 8274 | : Effect.gen(function* () { | |
| 8275 | yield* disposeDelegatedCompletionCohort({ | |
| 8276 | command, | |
| 8277 | events, | |
| 8278 | projection: yield* getProjectionWithPendingEvents(command.threadId, events), | |
| 8279 | parentRunId: completionCohortRunId, | |
| 8280 | disposition: "stopped", | |
| 8281 | now, | |
| 8282 | }); | |
| 8283 | }); | |
| 8092 | 8284 | |
| 8093 | 8285 | const emitEvent = emit(events, command); |
| 8094 | const holdQueuedRuns = Effect.forEach( | |
| 8095 | projection.runs.filter( | |
| 8096 | (candidate) => candidate.status === "queued" && !candidate.queueHeld, | |
| 8097 | ), | |
| 8098 | (queuedRun) => | |
| 8099 | emitEvent({ | |
| 8100 | type: "run.updated", | |
| 8101 | threadId: command.threadId, | |
| 8102 | runId: queuedRun.id, | |
| 8103 | providerInstanceId: queuedRun.providerInstanceId, | |
| 8104 | occurredAt: now, | |
| 8105 | payload: { ...queuedRun, queueHeld: true }, | |
| 8106 | }), | |
| 8107 | { discard: true }, | |
| 8108 | ); | |
| 8109 | 8286 | const interruptRequestItem: OrchestrationV2TurnItem = { |
| 8110 | 8287 | id: idAllocator.derive.runSignalTurnItem({ |
| 8111 | 8288 | runId: run.id, |
| ... | ... | @@ -8128,6 +8305,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8128 | 8305 | message: command.reason ?? "Interrupt requested", |
| 8129 | 8306 | }; |
| 8130 | 8307 | |
| 8308 | // Background work can outlive a provider switch, such as a Codex dev | |
| 8309 | // server left running when the thread moved to Claude. Stop also reaches | |
| 8310 | // each other live provider thread that owns pending work; that | |
| 8311 | // interrupt's settle follow-up ends what its provider leaves behind. | |
| 8312 | // Work on a dead session is settled with this run's below. | |
| 8313 | const otherProviderInterrupts: Array<PendingOrchestrationEffectV2> = []; | |
| 8314 | if (hasBackgroundWork || command.holdQueue === true) { | |
| 8315 | const runOrdinals = new Map( | |
| 8316 | projection.runs.map((candidate) => [candidate.id, candidate.ordinal]), | |
| 8317 | ); | |
| 8318 | const runOrdinalOf = (item: (typeof projection.turnItems)[number]) => | |
| 8319 | item.runId === null ? -1 : (runOrdinals.get(item.runId) ?? -1); | |
| 8320 | // Each provider thread is interrupted at its latest pending work: that | |
| 8321 | // turn's run bounds the settle follow-up, which must cover all of the | |
| 8322 | // thread's work. The target comes from the item's provider turn, since | |
| 8323 | // a native subagent item names its own provider thread but its | |
| 8324 | // parent's turn; interrupting the parent's turn reaches the subagent. | |
| 8325 | const latestTurnByProviderThread = new Map< | |
| 8326 | OrchestrationV2ProviderThread["id"], | |
| 8327 | { readonly turn: (typeof projection.providerTurns)[number]; readonly runOrdinal: number } | |
| 8328 | >(); | |
| 8329 | for (const item of pendingBackgroundTurnItems({ | |
| 8330 | turnItems: projection.turnItems, | |
| 8331 | runs: projection.runs, | |
| 8332 | })) { | |
| 8333 | const turn = projection.providerTurns.find( | |
| 8334 | (candidate) => candidate.id === item.providerTurnId, | |
| 8335 | ); | |
| 8336 | if (turn === undefined || turn.providerThreadId === providerThread.id) continue; | |
| 8337 | const runOrdinal = runOrdinalOf(item); | |
| 8338 | const latest = latestTurnByProviderThread.get(turn.providerThreadId); | |
| 8339 | if (latest === undefined || runOrdinal > latest.runOrdinal) { | |
| 8340 | latestTurnByProviderThread.set(turn.providerThreadId, { turn, runOrdinal }); | |
| 8341 | } | |
| 8342 | } | |
| 8343 | for (const { turn } of latestTurnByProviderThread.values()) { | |
| 8344 | const owner = projection.providerThreads.find( | |
| 8345 | (candidate) => candidate.id === turn.providerThreadId, | |
| 8346 | ); | |
| 8347 | if (owner === undefined || owner.providerSessionId === null) continue; | |
| 8348 | const ownerSession = yield* providerSessions | |
| 8349 | .get(owner.providerSessionId) | |
| 8350 | .pipe(Effect.orElseSucceed(() => Option.none())); | |
| 8351 | if (Option.isNone(ownerSession)) continue; | |
| 8352 | otherProviderInterrupts.push({ | |
| 8353 | id: `effect:${command.commandId}:provider-turn.interrupt:${turn.id}`, | |
| 8354 | commandId: command.commandId, | |
| 8355 | threadId: command.threadId, | |
| 8356 | request: { | |
| 8357 | type: "provider-turn.interrupt", | |
| 8358 | providerSessionId: owner.providerSessionId, | |
| 8359 | providerThreadId: owner.id, | |
| 8360 | providerTurnId: turn.id, | |
| 8361 | }, | |
| 8362 | }); | |
| 8363 | } | |
| 8364 | } | |
| 8365 | ||
| 8131 | 8366 | if ( |
| 8132 | 8367 | providerTurn === undefined && |
| 8133 | 8368 | (run.status === "preparing" || run.status === "starting" || run.status === "running") |
| ... | ... | @@ -8238,8 +8473,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8238 | 8473 | occurredAt: now, |
| 8239 | 8474 | payload: { ...run, status: "interrupted", completedAt: now }, |
| 8240 | 8475 | }); |
| 8241 | if (command.holdQueue === true) yield* holdQueuedRuns; | |
| 8242 | yield* stopCompletionCohort(); | |
| 8476 | yield* stopRemainingWork(); | |
| 8477 | yield* Ref.update(effects, (existing) => [...existing, ...otherProviderInterrupts]); | |
| 8243 | 8478 | return { |
| 8244 | 8479 | effectTypes: ["provider-turn.start", "provider-turn.restart"], |
| 8245 | 8480 | reason: `Run ${run.id} was interrupted before its provider turn started.`, |
| ... | ... | @@ -8256,63 +8491,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8256 | 8491 | cause: `Run ${command.runId} is not interruptible.`, |
| 8257 | 8492 | }); |
| 8258 | 8493 | } |
| 8259 | // Background work can outlive a provider switch, such as a Codex dev | |
| 8260 | // server left running when the thread moved to Claude. Stop also reaches | |
| 8261 | // each other live provider thread that owns pending work; that | |
| 8262 | // interrupt's settle follow-up ends what its provider leaves behind. | |
| 8263 | // Work on a dead session is settled with this run's below. | |
| 8264 | const otherProviderInterrupts: Array<PendingOrchestrationEffectV2> = []; | |
| 8265 | if (hasBackgroundWork) { | |
| 8266 | const runOrdinals = new Map( | |
| 8267 | projection.runs.map((candidate) => [candidate.id, candidate.ordinal]), | |
| 8268 | ); | |
| 8269 | const runOrdinalOf = (item: (typeof projection.turnItems)[number]) => | |
| 8270 | item.runId === null ? -1 : (runOrdinals.get(item.runId) ?? -1); | |
| 8271 | // Each provider thread is interrupted at its latest pending work: that | |
| 8272 | // turn's run bounds the settle follow-up, which must cover all of the | |
| 8273 | // thread's work. The target comes from the item's provider turn, since | |
| 8274 | // a native subagent item names its own provider thread but its | |
| 8275 | // parent's turn; interrupting the parent's turn reaches the subagent. | |
| 8276 | const latestTurnByProviderThread = new Map< | |
| 8277 | OrchestrationV2ProviderThread["id"], | |
| 8278 | { readonly turn: (typeof projection.providerTurns)[number]; readonly runOrdinal: number } | |
| 8279 | >(); | |
| 8280 | for (const item of pendingBackgroundTurnItems({ | |
| 8281 | turnItems: projection.turnItems, | |
| 8282 | runs: projection.runs, | |
| 8283 | })) { | |
| 8284 | const turn = projection.providerTurns.find( | |
| 8285 | (candidate) => candidate.id === item.providerTurnId, | |
| 8286 | ); | |
| 8287 | if (turn === undefined || turn.providerThreadId === providerThread.id) continue; | |
| 8288 | const runOrdinal = runOrdinalOf(item); | |
| 8289 | const latest = latestTurnByProviderThread.get(turn.providerThreadId); | |
| 8290 | if (latest === undefined || runOrdinal > latest.runOrdinal) { | |
| 8291 | latestTurnByProviderThread.set(turn.providerThreadId, { turn, runOrdinal }); | |
| 8292 | } | |
| 8293 | } | |
| 8294 | for (const { turn } of latestTurnByProviderThread.values()) { | |
| 8295 | const owner = projection.providerThreads.find( | |
| 8296 | (candidate) => candidate.id === turn.providerThreadId, | |
| 8297 | ); | |
| 8298 | if (owner === undefined || owner.providerSessionId === null) continue; | |
| 8299 | const ownerSession = yield* providerSessions | |
| 8300 | .get(owner.providerSessionId) | |
| 8301 | .pipe(Effect.orElseSucceed(() => Option.none())); | |
| 8302 | if (Option.isNone(ownerSession)) continue; | |
| 8303 | otherProviderInterrupts.push({ | |
| 8304 | id: `effect:${command.commandId}:provider-turn.interrupt:${turn.id}`, | |
| 8305 | commandId: command.commandId, | |
| 8306 | threadId: command.threadId, | |
| 8307 | request: { | |
| 8308 | type: "provider-turn.interrupt", | |
| 8309 | providerSessionId: owner.providerSessionId, | |
| 8310 | providerThreadId: owner.id, | |
| 8311 | providerTurnId: turn.id, | |
| 8312 | }, | |
| 8313 | }); | |
| 8314 | } | |
| 8315 | } | |
| 8316 | 8494 | // Stop on a settled thread's background work. Its process may be gone |
| 8317 | 8495 | // (released, restarted) and only the projection still shows the work; |
| 8318 | 8496 | // the settle follow-up ends whatever no provider reports ending. |
| ... | ... | @@ -8334,8 +8512,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8334 | 8512 | occurredAt: now, |
| 8335 | 8513 | payload: interruptRequestItem, |
| 8336 | 8514 | }); |
| 8337 | if (command.holdQueue === true) yield* holdQueuedRuns; | |
| 8338 | yield* stopCompletionCohort(); | |
| 8515 | yield* stopRemainingWork(); | |
| 8339 | 8516 | yield* settleBackgroundWork({ |
| 8340 | 8517 | command, |
| 8341 | 8518 | events, |
| ... | ... | @@ -8391,8 +8568,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8391 | 8568 | occurredAt: now, |
| 8392 | 8569 | payload: interruptRequestItem, |
| 8393 | 8570 | }); |
| 8394 | if (command.holdQueue === true) yield* holdQueuedRuns; | |
| 8395 | yield* stopCompletionCohort(); | |
| 8571 | yield* stopRemainingWork(); | |
| 8396 | 8572 | yield* Ref.update(effects, (existing) => [ |
| 8397 | 8573 | ...existing, |
| 8398 | 8574 | { |
| ... | ... | @@ -8411,6 +8587,91 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8411 | 8587 | return undefined; |
| 8412 | 8588 | }); |
| 8413 | 8589 | |
| 8590 | /** | |
| 8591 | * Stop for a thread whatever it is doing, the way the Stop button stops the run it | |
| 8592 | * shows: the running turn (or one whose background work is still pending) is interrupted, | |
| 8593 | * and the rest of the thread is held. A turn that cannot be interrupted still leaves the | |
| 8594 | * thread held. Delegated tasks under the thread are stopped by the sender. | |
| 8595 | */ | |
| 8596 | const dispatchThreadStop = ( | |
| 8597 | command: Extract<OrchestrationV2ServerCommand, { readonly type: "thread.stop" }>, | |
| 8598 | events: Ref.Ref<Array<OrchestrationV2DomainEvent>>, | |
| 8599 | effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>, | |
| 8600 | ) => | |
| 8601 | Effect.gen(function* () { | |
| 8602 | const projection = yield* projectionStore | |
| 8603 | .getThreadRecords(command.threadId, ["runs", "providerThreads", "turnItems", "subagents"], { | |
| 8604 | turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], | |
| 8605 | turnItemStatuses: ["pending", "running", "waiting"], | |
| 8606 | }) | |
| 8607 | .pipe( | |
| 8608 | Effect.mapError( | |
| 8609 | (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), | |
| 8610 | ), | |
| 8611 | ); | |
| 8612 | if (projection.thread.deletedAt !== null) return undefined; | |
| 8613 | // Queued messages never started and cannot own background work. | |
| 8614 | const latestRun = projection.runs.findLast((run) => run.status !== "queued"); | |
| 8615 | const target = | |
| 8616 | projection.runs.findLast( | |
| 8617 | (run) => | |
| 8618 | run.status === "preparing" || run.status === "starting" || run.status === "running", | |
| 8619 | ) ?? | |
| 8620 | (derivePendingBackgroundWork({ | |
| 8621 | latestRun, | |
| 8622 | providerThreads: projection.providerThreads, | |
| 8623 | turnItems: projection.turnItems, | |
| 8624 | activeProviderThreadId: projection.thread.activeProviderThreadId, | |
| 8625 | runs: projection.runs, | |
| 8626 | }).length > 0 | |
| 8627 | ? latestRun | |
| 8628 | : undefined); | |
| 8629 | if (target !== undefined) { | |
| 8630 | const interruptEvents = yield* Ref.make<Array<OrchestrationV2DomainEvent>>([]); | |
| 8631 | const interruptEffects = yield* Ref.make<Array<PendingOrchestrationEffectV2>>([]); | |
| 8632 | const interrupted = yield* Effect.exit( | |
| 8633 | dispatchRunInterrupt( | |
| 8634 | { | |
| 8635 | type: "run.interrupt", | |
| 8636 | commandId: command.commandId, | |
| 8637 | threadId: command.threadId, | |
| 8638 | runId: target.id, | |
| 8639 | holdQueue: true, | |
| 8640 | ...(command.reason === undefined ? {} : { reason: command.reason }), | |
| 8641 | }, | |
| 8642 | interruptEvents, | |
| 8643 | interruptEffects, | |
| 8644 | ), | |
| 8645 | ); | |
| 8646 | // A failed interrupt may have written part of its events, so they are kept apart. | |
| 8647 | if (Exit.isSuccess(interrupted)) { | |
| 8648 | const written = yield* Ref.get(interruptEvents); | |
| 8649 | const enqueued = yield* Ref.get(interruptEffects); | |
| 8650 | yield* Ref.update(events, (existing) => [...existing, ...written]); | |
| 8651 | yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); | |
| 8652 | return interrupted.value; | |
| 8653 | } | |
| 8654 | yield* Effect.logWarning("thread.stop could not interrupt the running turn", { | |
| 8655 | threadId: command.threadId, | |
| 8656 | runId: target.id, | |
| 8657 | cause: interrupted.cause, | |
| 8658 | }); | |
| 8659 | } | |
| 8660 | const now = yield* DateTime.now; | |
| 8661 | // Record that Stop reached the turn it could not interrupt and the thread's last executed | |
| 8662 | // run, so a late tool call from their agent starts nothing. | |
| 8663 | const executed = latestExecutedRun(projection.runs); | |
| 8664 | const reached = new Map<RunId, OrchestrationV2Run>(); | |
| 8665 | if (target !== undefined) reached.set(target.id, target); | |
| 8666 | if (executed !== null && !hasLiveRun({ runs: [executed] })) | |
| 8667 | reached.set(executed.id, executed); | |
| 8668 | for (const run of reached.values()) { | |
| 8669 | yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); | |
| 8670 | } | |
| 8671 | yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); | |
| 8672 | return undefined; | |
| 8673 | }); | |
| 8674 | ||
| 8414 | 8675 | const dispatchCheckpointRollback = ( |
| 8415 | 8676 | command: Extract<OrchestrationV2Command, { readonly type: "checkpoint.rollback" }>, |
| 8416 | 8677 | events: Ref.Ref<Array<OrchestrationV2DomainEvent>>, |
| ... | ... | @@ -9591,6 +9852,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 9591 | 9852 | break; |
| 9592 | 9853 | case "run.interrupt": |
| 9593 | 9854 | cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects); |
| 9855 | // Stop also stops every delegated task under the thread once it commits. | |
| 9856 | if (command.holdQueue === true) { | |
| 9857 | yield* Ref.update(effects, (existing) => [ | |
| 9858 | ...existing, | |
| 9859 | { | |
| 9860 | id: `effect:${command.commandId}:delegated-tasks.stop`, | |
| 9861 | commandId: command.commandId, | |
| 9862 | threadId: command.threadId, | |
| 9863 | request: { | |
| 9864 | type: "delegated-tasks.stop", | |
| 9865 | ...(command.reason === undefined ? {} : { reason: command.reason }), | |
| 9866 | }, | |
| 9867 | } satisfies PendingOrchestrationEffectV2, | |
| 9868 | ]); | |
| 9869 | } | |
| 9870 | break; | |
| 9871 | case "thread.stop": | |
| 9872 | cancelUnsettledEffects = yield* dispatchThreadStop(command, events, effects); | |
| 9594 | 9873 | break; |
| 9595 | 9874 | case "queued-message.promote-to-steer": |
| 9596 | 9875 | yield* dispatchQueuedMessagePromoteToSteer(command, events, effects); |
| ... | ... | @@ -9763,9 +10042,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 9763 | 10042 | |
| 9764 | 10043 | const plan = yield* dispatchOnce(command).pipe( |
| 9765 | 10044 | Effect.flatMap((planned) => |
| 9766 | // A settle that finds the provider already ended everything has | |
| 9767 | // nothing to record, which is its expected outcome, not a failure. | |
| 9768 | planned.events.length > 0 || command.type === "thread.background-work.settle" | |
| 10045 | // A settle that finds the provider already ended everything, or a | |
| 10046 | // stop that finds nothing running, has nothing to record. That is | |
| 10047 | // its expected outcome, not a failure. | |
| 10048 | planned.events.length > 0 || | |
| 10049 | command.type === "thread.background-work.settle" || | |
| 10050 | command.type === "thread.stop" | |
| 9769 | 10051 | ? Effect.succeed(planned) |
| 9770 | 10052 | : Effect.fail( |
| 9771 | 10053 | new OrchestratorDispatchError({ |
apps/server/src/orchestration-v2/ThreadManagementService.ts+38-1
| ... | ... | @@ -5,7 +5,7 @@ import type { |
| 5 | 5 | } from "./ProjectionStore.ts"; |
| 6 | 6 | import { |
| 7 | 7 | type ChatAttachment, |
| 8 | type CommandId, | |
| 8 | CommandId, | |
| 9 | 9 | MessageId, |
| 10 | 10 | type ModelSelection, |
| 11 | 11 | type OrchestrationV2Actor, |
| ... | ... | @@ -323,6 +323,16 @@ export interface ThreadManagementServiceShape { |
| 323 | 323 | readonly interruptThread: ( |
| 324 | 324 | input: ThreadManagementInterruptInput, |
| 325 | 325 | ) => Effect.Effect<ThreadManagementInterruptResult, ThreadManagementFailure>; |
| 326 | /** | |
| 327 | * Sends `thread.stop` to every delegated task under a thread, depth first. Their command IDs | |
| 328 | * derive from `commandId`, so a retry repeats nothing that already stopped. A task that | |
| 329 | * cannot be stopped does not keep the others running: all are tried, then it fails. | |
| 330 | */ | |
| 331 | readonly stopDelegatedTasks: (input: { | |
| 332 | readonly threadId: ThreadId; | |
| 333 | readonly commandId: CommandId; | |
| 334 | readonly reason?: string | undefined; | |
| 335 | }) => Effect.Effect<void, Orchestrator.OrchestratorV2Error>; | |
| 326 | 336 | readonly getThreadEventSequence: Orchestrator.OrchestratorV2["Service"]["getThreadEventSequence"]; |
| 327 | 337 | readonly recoverDelegatedTask: Orchestrator.OrchestratorV2["Service"]["recoverDelegatedTask"]; |
| 328 | 338 | readonly delegatedTaskResultPending: Orchestrator.OrchestratorV2["Service"]["delegatedTaskResultPending"]; |
| ... | ... | @@ -721,6 +731,32 @@ const make = Effect.gen(function* () { |
| 721 | 731 | return { type: "interrupt_requested", run: interruptibleRun, dispatch } as const; |
| 722 | 732 | }); |
| 723 | 733 | |
| 734 | const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = (input) => | |
| 735 | Effect.gen(function* () { | |
| 736 | const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]); | |
| 737 | const failures: Array<Orchestrator.OrchestratorV2Error> = []; | |
| 738 | for (const task of subagents) { | |
| 739 | if (task.origin !== "app_owned" || task.childThreadId === null) continue; | |
| 740 | const threadId = task.childThreadId; | |
| 741 | yield* dispatch({ | |
| 742 | type: "thread.stop", | |
| 743 | commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), | |
| 744 | threadId, | |
| 745 | ...(input.reason === undefined ? {} : { reason: input.reason }), | |
| 746 | }).pipe( | |
| 747 | Effect.andThen(stopDelegatedTasks({ ...input, threadId })), | |
| 748 | Effect.catch((error) => | |
| 749 | Effect.logWarning("Unable to stop a delegated task", { | |
| 750 | parentThreadId: input.threadId, | |
| 751 | threadId, | |
| 752 | error, | |
| 753 | }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))), | |
| 754 | ), | |
| 755 | ); | |
| 756 | } | |
| 757 | if (failures[0] !== undefined) return yield* Effect.fail(failures[0]); | |
| 758 | }); | |
| 759 | ||
| 724 | 760 | return ThreadManagementService.of({ |
| 725 | 761 | ensureLegacyTranscript, |
| 726 | 762 | dispatch, |
| ... | ... | @@ -753,6 +789,7 @@ const make = Effect.gen(function* () { |
| 753 | 789 | sendToThread, |
| 754 | 790 | waitForThread, |
| 755 | 791 | interruptThread, |
| 792 | stopDelegatedTasks, | |
| 756 | 793 | getThreadEventSequence: orchestrator.getThreadEventSequence, |
| 757 | 794 | recoverDelegatedTask: orchestrator.recoverDelegatedTask, |
| 758 | 795 | delegatedTaskResultPending: orchestrator.delegatedTaskResultPending, |
apps/server/src/orchestration-v2/ThreadStop.test.ts created+605
| ... | ... | @@ -0,0 +1,605 @@ |
| 1 | import { assert, it } from "@effect/vitest"; | |
| 2 | import { | |
| 3 | CommandId, | |
| 4 | EventId, | |
| 5 | MessageId, | |
| 6 | NodeId, | |
| 7 | ProjectId, | |
| 8 | ProviderDriverKind, | |
| 9 | ProviderInstanceId, | |
| 10 | ProviderTurnId, | |
| 11 | ThreadId, | |
| 12 | TurnItemId, | |
| 13 | } from "@t3tools/contracts"; | |
| 14 | import * as DateTime from "effect/DateTime"; | |
| 15 | import * as Effect from "effect/Effect"; | |
| 16 | import * as Exit from "effect/Exit"; | |
| 17 | import * as Layer from "effect/Layer"; | |
| 18 | import * as Schema from "effect/Schema"; | |
| 19 | import * as SqlClient from "effect/unstable/sql/SqlClient"; | |
| 20 | import * as TestClock from "effect/testing/TestClock"; | |
| 21 | ||
| 22 | import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; | |
| 23 | import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; | |
| 24 | import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; | |
| 25 | import * as Orchestrator from "./Orchestrator.ts"; | |
| 26 | import * as ProjectionStore from "./ProjectionStore.ts"; | |
| 27 | import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; | |
| 28 | import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; | |
| 29 | import * as ThreadManagementService from "./ThreadManagementService.ts"; | |
| 30 | import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; | |
| 31 | ||
| 32 | const instanceId = ProviderInstanceId.make("codex"); | |
| 33 | const modelSelection = { instanceId, model: "gpt-5.1-codex" }; | |
| 34 | const adapter = { | |
| 35 | instanceId, | |
| 36 | driver: ProviderDriverKind.make("codex"), | |
| 37 | getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), | |
| 38 | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), | |
| 39 | openSession: () => Effect.die("Runs here never reach a provider"), | |
| 40 | } as ProviderAdapterV2Shape; | |
| 41 | const database = SqlitePersistenceMemory; | |
| 42 | // No effect worker: runs stay unstarted, so Stop ends them without a provider. | |
| 43 | const testLayer = ThreadManagementService.layer.pipe( | |
| 44 | Layer.provideMerge( | |
| 45 | Layer.mergeAll( | |
| 46 | database, | |
| 47 | ProjectionStore.layer.pipe(Layer.provide(database)), | |
| 48 | makeOrchestratorV2ReplayLayerWithRegistry( | |
| 49 | { name: "thread-stop" }, | |
| 50 | ProviderAdapterRegistry.makeLayer([adapter]), | |
| 51 | { databaseLayer: database, runEffectWorker: false }, | |
| 52 | ), | |
| 53 | ), | |
| 54 | ), | |
| 55 | ); | |
| 56 | ||
| 57 | const encodeEffectRequest = Schema.encodeSync(Schema.fromJsonString(OrchestrationEffectRequestV2)); | |
| 58 | ||
| 59 | const pullRequest = (number: number) => ({ | |
| 60 | host: "github.com", | |
| 61 | repository: "pingdotgg/t3code", | |
| 62 | number, | |
| 63 | }); | |
| 64 | ||
| 65 | const createWatchingThread = (threadId: ThreadId, number: number) => | |
| 66 | Effect.gen(function* () { | |
| 67 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 68 | yield* orchestrator.dispatch({ | |
| 69 | type: "thread.create", | |
| 70 | commandId: CommandId.make(`create:${threadId}`), | |
| 71 | threadId, | |
| 72 | projectId: ProjectId.make("project:thread-stop"), | |
| 73 | title: threadId, | |
| 74 | modelSelection, | |
| 75 | runtimeMode: "full-access", | |
| 76 | interactionMode: "default", | |
| 77 | branch: null, | |
| 78 | worktreePath: null, | |
| 79 | createdBy: "user", | |
| 80 | creationSource: "web", | |
| 81 | }); | |
| 82 | yield* watch(threadId, number); | |
| 83 | }); | |
| 84 | ||
| 85 | const watch = (threadId: ThreadId, number: number) => | |
| 86 | Effect.gen(function* () { | |
| 87 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 88 | yield* orchestrator.dispatch({ | |
| 89 | type: "thread.pull-request.watch", | |
| 90 | commandId: CommandId.make(`watch:${threadId}:${number}`), | |
| 91 | threadId, | |
| 92 | ...pullRequest(number), | |
| 93 | watching: true, | |
| 94 | link: { url: `https://github.com/pingdotgg/t3code/pull/${number}`, source: "agent" }, | |
| 95 | }); | |
| 96 | }); | |
| 97 | ||
| 98 | const send = (threadId: ThreadId, text: string, type: "start_immediately" | "queue_after_active") => | |
| 99 | Effect.gen(function* () { | |
| 100 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 101 | yield* orchestrator.dispatch({ | |
| 102 | type: "message.dispatch", | |
| 103 | commandId: CommandId.make(`send:${threadId}:${text}`), | |
| 104 | threadId, | |
| 105 | messageId: MessageId.make(`message:${threadId}:${text}`), | |
| 106 | text, | |
| 107 | attachments: [], | |
| 108 | dispatchMode: { type }, | |
| 109 | createdBy: "user", | |
| 110 | creationSource: "web", | |
| 111 | }); | |
| 112 | }); | |
| 113 | ||
| 114 | /** Delegates `task` from the parent's latest run and returns the child thread. */ | |
| 115 | const delegate = (parentThreadId: ThreadId, task: string) => | |
| 116 | Effect.gen(function* () { | |
| 117 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 118 | const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!; | |
| 119 | yield* orchestrator.dispatch({ | |
| 120 | type: "delegated_task.request", | |
| 121 | commandId: CommandId.make(`delegate:${task}`), | |
| 122 | parentThreadId, | |
| 123 | parentRunId: parentRun.id, | |
| 124 | parentNodeId: parentRun.rootNodeId!, | |
| 125 | task, | |
| 126 | modelSelection, | |
| 127 | runtimeMode: "full-access", | |
| 128 | interactionMode: "default", | |
| 129 | completionWake: "always", | |
| 130 | createdBy: "agent", | |
| 131 | creationSource: "mcp", | |
| 132 | }); | |
| 133 | const projection = yield* orchestrator.getThreadProjection(parentThreadId); | |
| 134 | return projection.subagents.find((candidate) => candidate.prompt === task)!.childThreadId!; | |
| 135 | }); | |
| 136 | ||
| 137 | const threadState = (threadId: ThreadId) => | |
| 138 | Effect.gen(function* () { | |
| 139 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 140 | const projection = yield* orchestrator.getThreadProjection(threadId); | |
| 141 | return { | |
| 142 | runs: projection.runs.map((run) => `${run.status}${run.queueHeld === true ? ":held" : ""}`), | |
| 143 | watched: (projection.thread.pullRequests ?? []) | |
| 144 | .filter((link) => link.watch !== undefined) | |
| 145 | .map((link) => link.number), | |
| 146 | }; | |
| 147 | }); | |
| 148 | ||
| 149 | it.effect("Stop ends watches, holds queues, and stops the delegated tasks under the thread", () => | |
| 150 | Effect.gen(function* () { | |
| 151 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 152 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 153 | const threads = yield* ThreadManagementService.ThreadManagementService; | |
| 154 | const sql = yield* SqlClient.SqlClient; | |
| 155 | const parentThreadId = ThreadId.make("thread:stop-parent"); | |
| 156 | yield* createWatchingThread(parentThreadId, 1); | |
| 157 | yield* send(parentThreadId, "first", "start_immediately"); | |
| 158 | const childThreadId = yield* delegate(parentThreadId, "child task"); | |
| 159 | yield* watch(childThreadId, 2); | |
| 160 | const grandchildThreadId = yield* delegate(childThreadId, "grandchild task"); | |
| 161 | yield* send(childThreadId, "child follow-up", "queue_after_active"); | |
| 162 | ||
| 163 | // The run that delegated the child ends before the run the user stops, so | |
| 164 | // only a Stop that covers earlier runs keeps the stopped child from waking it. | |
| 165 | const now = yield* DateTime.now; | |
| 166 | const firstRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; | |
| 167 | yield* projections.apply({ | |
| 168 | id: EventId.make("event:stop-parent:first-run-completed"), | |
| 169 | type: "run.updated", | |
| 170 | threadId: parentThreadId, | |
| 171 | runId: firstRun.id, | |
| 172 | occurredAt: now, | |
| 173 | payload: { ...firstRun, status: "completed", completedAt: now }, | |
| 174 | }); | |
| 175 | yield* send(parentThreadId, "second", "start_immediately"); | |
| 176 | const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!; | |
| 177 | const childWatch = (yield* orchestrator.getThreadProjection(childThreadId)).thread | |
| 178 | .pullRequests?.[0]?.watch; | |
| 179 | assert.isDefined(childWatch); | |
| 180 | ||
| 181 | const stopCommandId = CommandId.make("stop-parent"); | |
| 182 | yield* orchestrator.dispatch({ | |
| 183 | type: "run.interrupt", | |
| 184 | commandId: stopCommandId, | |
| 185 | threadId: parentThreadId, | |
| 186 | runId: secondRun.id, | |
| 187 | holdQueue: true, | |
| 188 | }); | |
| 189 | ||
| 190 | assert.deepEqual(yield* threadState(parentThreadId), { | |
| 191 | runs: ["completed", "interrupted"], | |
| 192 | watched: [], | |
| 193 | }); | |
| 194 | const parent = yield* orchestrator.getThreadProjection(parentThreadId); | |
| 195 | assert.equal(parent.subagents[0]?.completionDelivery?.state, "disposed"); | |
| 196 | const effects = yield* sql<{ readonly effect_type: string }>` | |
| 197 | SELECT effect_type FROM orchestration_v2_effect_outbox WHERE command_id = ${stopCommandId} | |
| 198 | `; | |
| 199 | assert.include( | |
| 200 | effects.map((row) => row.effect_type), | |
| 201 | "delegated-tasks.stop", | |
| 202 | ); | |
| 203 | ||
| 204 | // What the delegated-tasks.stop effect runs once the Stop commits. | |
| 205 | yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); | |
| 206 | assert.deepEqual(yield* threadState(childThreadId), { | |
| 207 | runs: ["interrupted", "queued:held"], | |
| 208 | watched: [], | |
| 209 | }); | |
| 210 | assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]); | |
| 211 | ||
| 212 | // A watch read that raced the Stop cannot wake the stopped child. | |
| 213 | const lateWake = yield* Effect.exit( | |
| 214 | orchestrator.dispatch({ | |
| 215 | type: "thread.pull-request-watch.sync", | |
| 216 | commandId: CommandId.make("late-watch-wake"), | |
| 217 | threadId: childThreadId, | |
| 218 | ...pullRequest(2), | |
| 219 | startedAt: childWatch!.startedAt, | |
| 220 | watch: null, | |
| 221 | wake: { | |
| 222 | messageId: MessageId.make("message:late-watch-wake"), | |
| 223 | text: "Checks passed.", | |
| 224 | notification: { source: { kind: "monitor" }, outcome: "completed", summary: "#2" }, | |
| 225 | }, | |
| 226 | }), | |
| 227 | ); | |
| 228 | assert.isTrue(Exit.isFailure(lateWake)); | |
| 229 | assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); | |
| 230 | ||
| 231 | // A retried effect stops nothing twice. | |
| 232 | yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); | |
| 233 | assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); | |
| 234 | }).pipe(Effect.provide(testLayer)), | |
| 235 | ); | |
| 236 | ||
| 237 | it.effect( | |
| 238 | "thread.stop ends an idle thread's watches and accepts a thread with nothing to stop", | |
| 239 | () => | |
| 240 | Effect.gen(function* () { | |
| 241 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 242 | const threadId = ThreadId.make("thread:stop-idle"); | |
| 243 | yield* createWatchingThread(threadId, 3); | |
| 244 | ||
| 245 | yield* orchestrator.dispatch({ | |
| 246 | type: "thread.stop", | |
| 247 | commandId: CommandId.make("stop-idle"), | |
| 248 | threadId, | |
| 249 | }); | |
| 250 | assert.deepEqual(yield* threadState(threadId), { runs: [], watched: [] }); | |
| 251 | ||
| 252 | const again = yield* orchestrator.dispatch({ | |
| 253 | type: "thread.stop", | |
| 254 | commandId: CommandId.make("stop-idle-again"), | |
| 255 | threadId, | |
| 256 | }); | |
| 257 | assert.lengthOf(again.storedEvents, 0); | |
| 258 | }).pipe(Effect.provide(testLayer)), | |
| 259 | ); | |
| 260 | ||
| 261 | it.effect("a run Stop reached cannot delegate or start a watch, even after it ends", () => | |
| 262 | Effect.gen(function* () { | |
| 263 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 264 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 265 | const threadId = ThreadId.make("thread:stop-barrier"); | |
| 266 | yield* createWatchingThread(threadId, 4); | |
| 267 | yield* send(threadId, "work", "start_immediately"); | |
| 268 | const now = yield* DateTime.now; | |
| 269 | const run = { ...(yield* orchestrator.getThreadProjection(threadId)).runs[0]!, startedAt: now }; | |
| 270 | yield* projections.apply({ | |
| 271 | id: EventId.make("event:stop-barrier:running"), | |
| 272 | type: "run.updated", | |
| 273 | threadId, | |
| 274 | runId: run.id, | |
| 275 | occurredAt: now, | |
| 276 | payload: { ...run, status: "running" }, | |
| 277 | }); | |
| 278 | // A queued message the user cancelled does not count as the latest run. | |
| 279 | yield* send(threadId, "queued", "queue_after_active"); | |
| 280 | const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!; | |
| 281 | yield* orchestrator.dispatch({ | |
| 282 | type: "queued-run.cancel", | |
| 283 | commandId: CommandId.make("cancel-queued"), | |
| 284 | threadId, | |
| 285 | runId: queued.id, | |
| 286 | }); | |
| 287 | // Stop reached the run, but its provider has not stopped it yet. | |
| 288 | yield* projections.apply({ | |
| 289 | id: EventId.make("event:stop-barrier:interrupt-request"), | |
| 290 | type: "turn-item.updated", | |
| 291 | threadId, | |
| 292 | runId: run.id, | |
| 293 | occurredAt: now, | |
| 294 | payload: { | |
| 295 | id: TurnItemId.make("turn-item:stop-barrier:interrupt-request"), | |
| 296 | threadId, | |
| 297 | runId: run.id, | |
| 298 | nodeId: run.rootNodeId!, | |
| 299 | providerThreadId: run.providerThreadId!, | |
| 300 | providerTurnId: null, | |
| 301 | nativeItemRef: null, | |
| 302 | parentItemId: null, | |
| 303 | ordinal: 1_000, | |
| 304 | status: "completed", | |
| 305 | title: "Interrupt requested", | |
| 306 | startedAt: now, | |
| 307 | completedAt: now, | |
| 308 | updatedAt: now, | |
| 309 | type: "run_interrupt_request", | |
| 310 | message: "Stop", | |
| 311 | }, | |
| 312 | }); | |
| 313 | ||
| 314 | assert.isTrue(Exit.isFailure(yield* Effect.exit(delegate(threadId, "late task")))); | |
| 315 | assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 5)))); | |
| 316 | ||
| 317 | // A slow watch_pull_request call can land after the stopped run ended. | |
| 318 | yield* projections.apply({ | |
| 319 | id: EventId.make("event:stop-barrier:interrupted"), | |
| 320 | type: "run.updated", | |
| 321 | threadId, | |
| 322 | runId: run.id, | |
| 323 | occurredAt: now, | |
| 324 | payload: { ...run, status: "interrupted", completedAt: now }, | |
| 325 | }); | |
| 326 | assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 5)))); | |
| 327 | assert.deepEqual(yield* threadState(threadId), { | |
| 328 | runs: ["interrupted", "cancelled"], | |
| 329 | watched: [4], | |
| 330 | }); | |
| 331 | ||
| 332 | // The user can still stop and restart a watch by hand. | |
| 333 | for (const watching of [false, true]) { | |
| 334 | yield* orchestrator.dispatch({ | |
| 335 | type: "thread.pull-request.watch", | |
| 336 | commandId: CommandId.make(`manual-watch:${watching}`), | |
| 337 | threadId, | |
| 338 | ...pullRequest(4), | |
| 339 | watching, | |
| 340 | }); | |
| 341 | } | |
| 342 | assert.deepEqual((yield* threadState(threadId)).watched, [4]); | |
| 343 | }).pipe(Effect.provide(testLayer)), | |
| 344 | ); | |
| 345 | ||
| 346 | it.effect("thread.stop keeps a restart continuation of the stopped run from starting", () => | |
| 347 | Effect.gen(function* () { | |
| 348 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 349 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 350 | const sql = yield* SqlClient.SqlClient; | |
| 351 | const threadId = ThreadId.make("thread:stop-restart"); | |
| 352 | yield* createWatchingThread(threadId, 8); | |
| 353 | yield* send(threadId, "work", "start_immediately"); | |
| 354 | const now = yield* DateTime.now; | |
| 355 | const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; | |
| 356 | // A server restart cut the run before its provider started; its continuation is pending. | |
| 357 | yield* projections.apply({ | |
| 358 | id: EventId.make("event:stop-restart:cancelled"), | |
| 359 | type: "run.updated", | |
| 360 | threadId, | |
| 361 | runId: run.id, | |
| 362 | occurredAt: now, | |
| 363 | payload: { ...run, status: "cancelled", startedAt: null, completedAt: now }, | |
| 364 | }); | |
| 365 | const at = DateTime.formatIso(now); | |
| 366 | yield* sql` | |
| 367 | INSERT INTO orchestration_v2_effect_outbox ( | |
| 368 | effect_id, command_id, thread_id, effect_type, payload_json, status, | |
| 369 | attempt_count, available_at, created_at, updated_at | |
| 370 | ) VALUES ( | |
| 371 | ${`effect:restart-continuation:${run.id}`}, 'command:restart', ${threadId}, | |
| 372 | 'provider-runtime.continue', | |
| 373 | ${encodeEffectRequest({ type: "provider-runtime.continue", sourceRunId: run.id })}, | |
| 374 | 'pending', 0, ${at}, ${at}, ${at} | |
| 375 | ) | |
| 376 | `; | |
| 377 | ||
| 378 | yield* orchestrator.dispatch({ | |
| 379 | type: "thread.stop", | |
| 380 | commandId: CommandId.make("stop-restart"), | |
| 381 | threadId, | |
| 382 | }); | |
| 383 | yield* orchestrator.dispatch({ | |
| 384 | type: "message.dispatch", | |
| 385 | commandId: CommandId.make("restart-continuation"), | |
| 386 | threadId, | |
| 387 | messageId: MessageId.make("message:restart-continuation"), | |
| 388 | text: "Continue.", | |
| 389 | attachments: [], | |
| 390 | dispatchMode: { type: "start_immediately" }, | |
| 391 | createdBy: "agent", | |
| 392 | creationSource: "server", | |
| 393 | restartContinuationOfRunId: run.id, | |
| 394 | }); | |
| 395 | assert.deepEqual(yield* threadState(threadId), { runs: ["cancelled"], watched: [] }); | |
| 396 | }).pipe(Effect.provide(testLayer)), | |
| 397 | ); | |
| 398 | ||
| 399 | it.effect.each(["thread.stop", "run.interrupt"] as const)( | |
| 400 | "a successful %s blocks a newer restart continuation when completion timestamps tie", | |
| 401 | (stopType) => | |
| 402 | Effect.gen(function* () { | |
| 403 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 404 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 405 | const sql = yield* SqlClient.SqlClient; | |
| 406 | const threadId = ThreadId.make("thread:stop-resumed-queue"); | |
| 407 | yield* createWatchingThread(threadId, 12); | |
| 408 | yield* send(threadId, "first", "start_immediately"); | |
| 409 | yield* send(threadId, "queued", "queue_after_active"); | |
| 410 | const now = yield* DateTime.now; | |
| 411 | const [first, queued] = (yield* orchestrator.getThreadProjection(threadId)).runs; | |
| 412 | assert.isDefined(first); | |
| 413 | assert.isDefined(queued); | |
| 414 | for (const run of [first!, queued!]) { | |
| 415 | yield* projections.apply({ | |
| 416 | id: EventId.make(`event:stop-resumed-queue:hold:${run.id}`), | |
| 417 | type: "run.updated", | |
| 418 | threadId, | |
| 419 | runId: run.id, | |
| 420 | occurredAt: now, | |
| 421 | payload: | |
| 422 | run.id === first!.id | |
| 423 | ? { ...run, status: "completed", completedAt: now } | |
| 424 | : { ...run, queueHeld: true }, | |
| 425 | }); | |
| 426 | } | |
| 427 | // A later turn runs ahead of the held queue, then recovery cancels it. | |
| 428 | yield* send(threadId, "restart source", "start_immediately"); | |
| 429 | const source = (yield* orchestrator.getThreadProjection(threadId)).runs.at(-1)!; | |
| 430 | yield* projections.apply({ | |
| 431 | id: EventId.make("event:stop-resumed-queue:restart-cancelled"), | |
| 432 | type: "run.updated", | |
| 433 | threadId, | |
| 434 | runId: source.id, | |
| 435 | occurredAt: now, | |
| 436 | payload: { ...source, status: "cancelled", startedAt: now, completedAt: now }, | |
| 437 | }); | |
| 438 | const at = DateTime.formatIso(now); | |
| 439 | yield* sql` | |
| 440 | INSERT INTO orchestration_v2_effect_outbox ( | |
| 441 | effect_id, command_id, thread_id, effect_type, payload_json, status, | |
| 442 | attempt_count, available_at, created_at, updated_at | |
| 443 | ) VALUES ( | |
| 444 | ${`effect:restart-continuation:${source.id}`}, 'command:restart', ${threadId}, | |
| 445 | 'provider-runtime.continue', | |
| 446 | ${encodeEffectRequest({ type: "provider-runtime.continue", sourceRunId: source.id })}, | |
| 447 | 'pending', 0, ${at}, ${at}, ${at} | |
| 448 | ) | |
| 449 | `; | |
| 450 | yield* orchestrator.dispatch({ | |
| 451 | type: "queue.resume", | |
| 452 | commandId: CommandId.make("resume-older-queue"), | |
| 453 | threadId, | |
| 454 | }); | |
| 455 | assert.equal( | |
| 456 | (yield* orchestrator.getThreadProjection(threadId)).runs.find( | |
| 457 | (run) => run.id === queued!.id, | |
| 458 | )?.status, | |
| 459 | "starting", | |
| 460 | ); | |
| 461 | yield* orchestrator.dispatch( | |
| 462 | stopType === "thread.stop" | |
| 463 | ? { type: "thread.stop", commandId: CommandId.make("stop-resumed-queue"), threadId } | |
| 464 | : { | |
| 465 | type: "run.interrupt", | |
| 466 | commandId: CommandId.make("stop-resumed-queue"), | |
| 467 | threadId, | |
| 468 | runId: queued!.id, | |
| 469 | holdQueue: true, | |
| 470 | }, | |
| 471 | ); | |
| 472 | const stopped = yield* orchestrator.getThreadProjection(threadId); | |
| 473 | const interrupted = stopped.runs.find((run) => run.id === queued!.id)!; | |
| 474 | assert.equal(interrupted.status, "interrupted"); | |
| 475 | assert.equal(DateTime.toEpochMillis(interrupted.completedAt!), DateTime.toEpochMillis(now)); | |
| 476 | yield* orchestrator.dispatch({ | |
| 477 | type: "message.dispatch", | |
| 478 | commandId: CommandId.make("restart-after-stop"), | |
| 479 | threadId, | |
| 480 | messageId: MessageId.make("message:restart-after-stop"), | |
| 481 | text: "Continue.", | |
| 482 | attachments: [], | |
| 483 | dispatchMode: { type: "start_immediately" }, | |
| 484 | createdBy: "agent", | |
| 485 | creationSource: "server", | |
| 486 | restartContinuationOfRunId: source.id, | |
| 487 | }); | |
| 488 | assert.deepEqual(yield* threadState(threadId), { | |
| 489 | runs: ["completed", "interrupted", "cancelled"], | |
| 490 | watched: [], | |
| 491 | }); | |
| 492 | }).pipe(Effect.provide(testLayer.pipe(Layer.provideMerge(TestClock.layer())))), | |
| 493 | ); | |
| 494 | ||
| 495 | it.effect("a delegated task that cannot be stopped fails the walk after its siblings stop", () => | |
| 496 | Effect.gen(function* () { | |
| 497 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 498 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 499 | const threads = yield* ThreadManagementService.ThreadManagementService; | |
| 500 | const parentThreadId = ThreadId.make("thread:stop-partial"); | |
| 501 | yield* createWatchingThread(parentThreadId, 6); | |
| 502 | yield* send(parentThreadId, "work", "start_immediately"); | |
| 503 | const childThreadId = yield* delegate(parentThreadId, "stoppable task"); | |
| 504 | const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!; | |
| 505 | // A task whose thread is gone, listed before the one that can stop. | |
| 506 | yield* projections.apply({ | |
| 507 | id: EventId.make("event:stop-partial:missing-task"), | |
| 508 | type: "subagent.updated", | |
| 509 | threadId: parentThreadId, | |
| 510 | runId: task.runId!, | |
| 511 | nodeId: NodeId.make("node:stop-partial:missing"), | |
| 512 | providerInstanceId: instanceId, | |
| 513 | occurredAt: yield* DateTime.now, | |
| 514 | payload: { | |
| 515 | ...task, | |
| 516 | id: NodeId.make("node:stop-partial:missing"), | |
| 517 | childThreadId: ThreadId.make("thread:stop-partial:missing"), | |
| 518 | startedAt: DateTime.subtract(task.startedAt!, { hours: 1 }), | |
| 519 | }, | |
| 520 | }); | |
| 521 | ||
| 522 | const walked = yield* Effect.exit( | |
| 523 | threads.stopDelegatedTasks({ | |
| 524 | threadId: parentThreadId, | |
| 525 | commandId: CommandId.make("stop-partial"), | |
| 526 | }), | |
| 527 | ); | |
| 528 | assert.isTrue(Exit.isFailure(walked)); | |
| 529 | assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted"]); | |
| 530 | }).pipe(Effect.provide(testLayer)), | |
| 531 | ); | |
| 532 | ||
| 533 | it.effect("thread.stop on a finished thread refuses a late agent watch", () => | |
| 534 | Effect.gen(function* () { | |
| 535 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 536 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 537 | const threadId = ThreadId.make("thread:stop-finished"); | |
| 538 | yield* createWatchingThread(threadId, 9); | |
| 539 | yield* send(threadId, "work", "start_immediately"); | |
| 540 | const now = yield* DateTime.now; | |
| 541 | const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; | |
| 542 | yield* projections.apply({ | |
| 543 | id: EventId.make("event:stop-finished:completed"), | |
| 544 | type: "run.updated", | |
| 545 | threadId, | |
| 546 | runId: run.id, | |
| 547 | occurredAt: now, | |
| 548 | payload: { ...run, status: "completed", startedAt: now, completedAt: now }, | |
| 549 | }); | |
| 550 | ||
| 551 | yield* orchestrator.dispatch({ | |
| 552 | type: "thread.stop", | |
| 553 | commandId: CommandId.make("stop-finished"), | |
| 554 | threadId, | |
| 555 | }); | |
| 556 | assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 10)))); | |
| 557 | assert.deepEqual(yield* threadState(threadId), { runs: ["completed"], watched: [] }); | |
| 558 | }).pipe(Effect.provide(testLayer)), | |
| 559 | ); | |
| 560 | ||
| 561 | it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is refused", () => | |
| 562 | Effect.gen(function* () { | |
| 563 | const orchestrator = yield* Orchestrator.OrchestratorV2; | |
| 564 | const projections = yield* ProjectionStore.ProjectionStoreV2; | |
| 565 | const threadId = ThreadId.make("thread:stop-lost-session"); | |
| 566 | yield* createWatchingThread(threadId, 11); | |
| 567 | yield* send(threadId, "work", "start_immediately"); | |
| 568 | const now = yield* DateTime.now; | |
| 569 | const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; | |
| 570 | // The turn is running, but its provider session is gone, so the interrupt fails. | |
| 571 | yield* projections.apply({ | |
| 572 | id: EventId.make("event:stop-lost-session:running"), | |
| 573 | type: "run.updated", | |
| 574 | threadId, | |
| 575 | runId: run.id, | |
| 576 | occurredAt: now, | |
| 577 | payload: { ...run, status: "running", startedAt: now }, | |
| 578 | }); | |
| 579 | yield* projections.apply({ | |
| 580 | id: EventId.make("event:stop-lost-session:turn"), | |
| 581 | type: "provider-turn.updated", | |
| 582 | threadId, | |
| 583 | occurredAt: now, | |
| 584 | payload: { | |
| 585 | id: ProviderTurnId.make("provider-turn:stop-lost-session"), | |
| 586 | providerThreadId: run.providerThreadId!, | |
| 587 | nodeId: run.rootNodeId!, | |
| 588 | runAttemptId: run.activeAttemptId!, | |
| 589 | nativeTurnRef: null, | |
| 590 | ordinal: 1, | |
| 591 | status: "running", | |
| 592 | startedAt: now, | |
| 593 | completedAt: null, | |
| 594 | }, | |
| 595 | }); | |
| 596 | ||
| 597 | yield* orchestrator.dispatch({ | |
| 598 | type: "thread.stop", | |
| 599 | commandId: CommandId.make("stop-lost-session"), | |
| 600 | threadId, | |
| 601 | }); | |
| 602 | assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(threadId, 12)))); | |
| 603 | assert.deepEqual(yield* threadState(threadId), { runs: ["running"], watched: [] }); | |
| 604 | }).pipe(Effect.provide(testLayer)), | |
| 605 | ); |
apps/server/src/relay/AgentAwarenessRelay.test.ts+1
| ... | ... | @@ -206,6 +206,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( |
| 206 | 206 | sendToThread: unused, |
| 207 | 207 | waitForThread: unused, |
| 208 | 208 | interruptThread: unused, |
| 209 | stopDelegatedTasks: unused, | |
| 209 | 210 | getThreadEventSequence: unused, |
| 210 | 211 | recoverDelegatedTask: unused, |
| 211 | 212 | delegatedTaskResultPending: unused, |
docs/orchestration-v2/feature-lifecycles.md+8
| ... | ... | @@ -251,6 +251,14 @@ If the provider does not support interruption: |
| 251 | 251 | - best case: stop provider session and mark run interrupted or cancelled by policy. |
| 252 | 252 | - worst case: mark interrupt unsupported and leave run active until provider exits. |
| 253 | 253 | |
| 254 | The Stop button sends `run.interrupt` with `holdQueue`. A Stop also holds the queue, | |
| 255 | ends the thread's pull request watches, and drops wakes its delegated tasks still | |
| 256 | owe, so nothing automatic restarts the thread. It then stops every delegated task | |
| 257 | under the thread: a `delegated-tasks.stop` effect sends the internal `thread.stop` | |
| 258 | command to each child thread, depth first. This runs after commit, not inside the | |
| 259 | parent's command, because a command holds only its own thread's lock. `task_cancel` | |
| 260 | sends the same `thread.stop` to its child. | |
| 261 | ||
| 254 | 262 | ## Resumption |
| 255 | 263 | |
| 256 | 264 | Resumption has three related meanings: |
docs/orchestration-v2/orchestrator-mcp-server.md+7-7
| ... | ... | @@ -279,13 +279,13 @@ the published task result. |
| 279 | 279 | |
| 280 | 280 | ### `task_cancel` |
| 281 | 281 | |
| 282 | Interrupts the currently active task run through the normal V2 `run.interrupt` | |
| 283 | command and disposes automatic parent delivery. Native background work between | |
| 284 | turns currently has no interruptible run. For a terminal task, it returns the | |
| 285 | existing status and disposes delivery without interrupting later child-thread runs, | |
| 286 | even when `task_status` reports `hasPendingChildRuns: true`. Published task results | |
| 287 | remain available. It accepts an optional cancellation reason. Use | |
| 288 | `t3_thread_interrupt` to stop a later active run. | |
| 282 | Stops the child thread with the internal `thread.stop` command, then stops every | |
| 283 | task the child delegated, and disposes automatic parent delivery. Like a user Stop, | |
| 284 | `thread.stop` interrupts the running turn, holds queued turns, and ends pull request | |
| 285 | watches. A nonterminal task with no interruptible run is rejected. A terminal task | |
| 286 | returns its existing status, and its child thread still stops, including later | |
| 287 | runs and watch wakes. Published task results remain available. It accepts an | |
| 288 | optional cancellation reason. | |
| 289 | 289 | |
| 290 | 290 | ### `create_threads` |
| 291 | 291 |
docs/user/source-control.md+4-3
| ... | ... | @@ -189,9 +189,10 @@ Ask the agent to watch, monitor, or babysit a pull request and it calls `watch_p |
| 189 | 189 | the thread is active, the server checks the pull request every minute and wakes the agent when a check |
| 190 | 190 | fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict. |
| 191 | 191 | Comments from your own account do not wake it. Watching ends when the pull request merges or closes, |
| 192 | after 10 wakes in a row that bring only comments, or when the server cannot read the pull request for | |
| 193 | 15 minutes. Settling a thread also ends all its watches. Unsettle the thread before starting a new | |
| 194 | watch. To start or stop it yourself, use the row menu in the **Linked pull requests** panel. | |
| 192 | after 10 wakes in a row that bring only comments, when the server cannot read the pull request for | |
| 193 | 15 minutes, or when you press Stop on the thread. Settling a thread also ends all its watches. | |
| 194 | Unsettle the thread before starting a new watch. To start or stop it yourself, use the row menu in | |
| 195 | the **Linked pull requests** panel. | |
| 195 | 196 | |
| 196 | 197 | Cross-repository links use a project on the same host. Azure DevOps reviews require a project checked |
| 197 | 198 | out from the matching organization and repository. |
docs/user/thread-sidebar.md+1
| ... | ... | @@ -208,6 +208,7 @@ snooze limit stops by default. Providers without a reset time offer manual |
| 208 | 208 | retry and the normal snooze choices. |
| 209 | 209 | |
| 210 | 210 | On web and desktop, use **Agents** to follow work delegated to subagents. |
| 211 | Stop on a thread also stops the subagents it delegated to. | |
| 211 | 212 | |
| 212 | 213 | Subagent threads started by the agent can't take messages; message the parent |
| 213 | 214 | thread instead. When such a subagent needs an approval or an answer, the parent |
packages/contracts/src/orchestrationV2.ts+15
| ... | ... | @@ -2830,6 +2830,10 @@ export const OrchestrationV2Command = Schema.Union([ |
| 2830 | 2830 | threadId: ThreadId, |
| 2831 | 2831 | runId: RunId, |
| 2832 | 2832 | reason: Schema.optional(Schema.String), |
| 2833 | /** | |
| 2834 | * Set by the Stop button. Stop also holds the queue, ends the thread's pull request | |
| 2835 | * watches, and stops every delegated task under the thread. | |
| 2836 | */ | |
| 2833 | 2837 | holdQueue: Schema.optional(Schema.Boolean), |
| 2834 | 2838 | }), |
| 2835 | 2839 | Schema.Struct({ |
| ... | ... | @@ -3013,6 +3017,17 @@ const OrchestrationV2InternalCommand = Schema.Union([ |
| 3013 | 3017 | providerThreadId: ProviderThreadId, |
| 3014 | 3018 | providerTurnId: ProviderTurnId, |
| 3015 | 3019 | }), |
| 3020 | /** | |
| 3021 | * Stop for one thread, whatever it is doing: interrupts its running turn, holds its queue, | |
| 3022 | * ends its pull request watches, and drops pending delegated-task wakes. Nothing to stop is | |
| 3023 | * an accepted no-op. Delegated tasks under the thread get their own `thread.stop`. | |
| 3024 | */ | |
| 3025 | Schema.Struct({ | |
| 3026 | type: Schema.Literal("thread.stop"), | |
| 3027 | commandId: CommandId, | |
| 3028 | threadId: ThreadId, | |
| 3029 | reason: Schema.optional(Schema.String), | |
| 3030 | }), | |
| 3016 | 3031 | ]); |
| 3017 | 3032 | export type OrchestrationV2InternalCommand = typeof OrchestrationV2InternalCommand.Type; |
| 3018 | 3033 |