diff --git a/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.i18n.yaml b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.i18n.yaml new file mode 100644 index 0000000000..fe6e8870f5 --- /dev/null +++ b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.i18n.yaml @@ -0,0 +1,6 @@ +# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each +# side as of the last confirmed-consistent state. Both languages carry equal authority; +# after editing either side, bring the other along and re-record with: +# pnpm run verify-translation-pairing --write .agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.md +2026-08-06-continuable-subagent-interrupt.md: 729f1eb8259aa28aa771ed71872ebce9cadd4ed8 +2026-08-06-continuable-subagent-interrupt.zh.md: 15fe34d8b5e1d67623aabfe08710f341739e92e1 diff --git a/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.md b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.md new file mode 100644 index 0000000000..729f1eb825 --- /dev/null +++ b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.md @@ -0,0 +1,42 @@ +# Agent Note: Continuable subagent current-turn interrupt + +Status: implemented + +English | [中文](2026-08-06-continuable-subagent-interrupt.zh.md) + +## Problem + +A running continuable subagent could not be stopped without destroying it. The continuation manager cancels child Agents only inside whole-Activation teardown (settlement, drain, scoped drain), `send_message`/`subagent.prompt` only add work, and the Web composer's Stop button was deliberately limited to ordinary sessions. A human watching a continuable child burn tokens on a wrong path had no lever short of killing the parent tree, and when the direct parent Agent was offline the child was entirely untouchable even though its Activation stayed live. One-shot runs have holder-owned disposal and task-kill; continuable children had no analogous current-turn control. + +## Decision + +`ctx.subagents.interrupt(targetSessionId, authority)` stops only the live target's current turn. The manager primitive authorizes synchronously, calls the existing `Agent.cancel(cause, { keepInbox: true })`, and returns `void` — fire-and-return: the cancel signal is guaranteed issued, target quiescence is not awaited. Nothing else changes: no Activation disposal, no handle release, no descendant cascade, no inbox clearing, and no `AgentLoop` or `CancelOptions` change. Because `keepInbox` parks the pending queue at idle, an interrupt never auto-starts the next queued follow-up; only a later explicit waking send resumes the preserved FIFO order. + +Authority is a closed two-variant union, deliberately wider than delivery authority because stopping a turn is idempotent and delivers no content: + +- `{ kind: 'user', parentSessionId }` — a human presents the durable direct-parent address. The live target's `session.header.parentSession` must match; no live parent Agent, catalog read, or persistence access is involved, which is exactly what keeps a live child stoppable while its parent Agent is offline. Cancel cause `user`. +- `{ kind: 'ancestor', agent }` — an exact live ancestor Agent (direct parent or deeper). The caller must be the registry's current entry for its id (stale callers are rejected even for absent targets), must not be the target itself, and must appear in the Activation's materialization-time `ancestry` WeakSet. Cancel cause `parent`. + +Targets are resolved only in the manager's process-local Activation map. An absent id — unknown, one-shot, or naturally settled — is an accepted no-op, which uniformly covers completion races and repeat requests without leaking durable-catalog information; a target whose disposal transaction is already open is likewise an accepted no-op after authorization. One-shot lifecycle (holder `dispose()`, task-kill) is untouched. `SubagentService.interrupt()` treats a manager-less composition as an accepted no-op rather than `CONTINUATION_UNAVAILABLE`, because without a manager no manager-owned live Activation can exist. + +The Host RPC `subagent.interrupt` takes the continuable `SubagentAddress` and returns `{ accepted: true }`. Its implementation calls only the core primitive with `user` authority — deliberately no `catalogChild()`, `listChildren()`, `sessionQuery`, or parent-registry lookup. A live target with a mismatched parent address maps to `subagent-unauthorized`; unexpected failures map to `internal` without leaking error text onto the wire. + +## Alternatives considered + +**Route human interrupts through `session.cancel`.** The generic session cancel requires an attached ordinary session and rejects subagent-owned sessions; widening it would entangle subagent authority rules with ordinary session routing. A subagent-domain RPC keeps the address-based authorization and the parent-offline guarantee explicit. + +**Await target quiescence and return the turn outcome.** Cancellation is cooperative, so quiescence is unbounded; holding the RPC (and a `ChildLock` slot) open invites timeouts and convoying against delivery and disposal. Acceptance-of-signal is the only fact the caller needs, and races (natural completion, disposal) already settle idempotently. + +**Reuse whole-Activation disposal for interrupt.** Disposal cancels without `keepInbox`, flushes, captures, and releases the handle — it destroys queued work and the child's residency. Interrupt is a control operation on one turn, not a lifecycle operation on the Activation. + +**Extend `send_message`/`followup` authority to ancestors while at it.** Delivery injects content into a conversation and is not idempotent; its exact-direct-parent authority stays unchanged. Only interrupt gets the wider ancestor and address-based user authority. + +**Auto-resume the parked queue after an interrupt.** Immediately starting queued follow-up B after aborting A would make the interrupt look ignored and steal the human's window to redirect the child. Parking until an explicit waking send keeps the stop observable and the FIFO order intact. + +## Consequences + +A human or ancestor can now stop a runaway continuable turn without losing the child, its queued work, or its running descendants; the cost is a deliberately weak postcondition (`accepted` means "signal issued", so a target may remain visibly `running` until it observes the signal) that clients must render honestly. The parked-queue rule means an interrupted child sits idle with retained work until someone sends a waking message — an intentional human-in-the-loop pause, not a scheduler defect. The Web Stop action and the model-facing `interrupt_agent` tool build on this primitive in the stacked follow-up PRs for issue #1535. + +## Testing + +Core coverage in `packages/subagent/subagent/tests/continuation.spec.ts` proves the durable `turn/end` abort, parked-then-FIFO-resumed queue, untouched descendant, both authority kinds with their cancel causes, self/sibling/stale/non-ancestor rejection, absent/one-shot/disposal-race no-ops, and the unchanged `keepInbox` loop behavior. Host coverage in `packages/host/apiproxy/tests` proves the RPC calls only the core primitive (no agents/catalog/history reads), the `subagent-unauthorized`/`internal` mappings, the wire schema's continuable-mode fence, and carrier round-trips. diff --git a/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.zh.md b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.zh.md new file mode 100644 index 0000000000..15fe34d8b5 --- /dev/null +++ b/.agents/notes/implemented/feature/2026-08-06-continuable-subagent-interrupt.zh.md @@ -0,0 +1,42 @@ +# Agent Note: Continuable subagent 当前轮次中断 + +Status: implemented + +[English](2026-08-06-continuable-subagent-interrupt.md) | 中文 + +## Problem + +一个正在运行的 continuable subagent 无法在不销毁它的前提下被停止。继续执行管理器只在整个 Activation 拆除(结算、drain、scoped drain)内部取消子 Agent,`send_message`/`subagent.prompt` 只能增加工作,而 Web composer 的 Stop 按钮被刻意限制在普通会话。人类眼看着一个 continuable child 在错误路径上烧 token,除了干掉整个 parent 树没有任何手段;当直接 parent Agent 离线时,即使 child 的 Activation 仍然在线,它也完全不可触及。一次性运行有持有方拥有的 disposal 和 task-kill;continuable child 没有对应的当前轮次控制。 + +## Decision + +`ctx.subagents.interrupt(targetSessionId, authority)` 只停止在线目标的当前轮次。管理器原语同步完成鉴权,调用现有的 `Agent.cancel(cause, { keepInbox: true })`,然后返回 `void`——fire-and-return:保证取消信号已发出,但不等待目标静止。其余一切不变:不 dispose Activation、不释放 handle、不级联后代、不清空 inbox,也不改动 `AgentLoop` 或 `CancelOptions`。由于 `keepInbox` 让待处理队列停在 idle,中断绝不会自动启动下一个排队的 follow-up;只有之后一次显式唤醒发送才按保留的 FIFO 顺序恢复。 + +授权是一个封闭的双变体 union,刻意比投递权限更宽,因为停止一个轮次是幂等的且不投递任何内容: + +- `{ kind: 'user', parentSessionId }`——人类出示持久化直接 parent 地址。在线目标的 `session.header.parentSession` 必须匹配;不涉及在线 parent Agent、目录读取或持久化访问,这正是 parent Agent 离线时在线 child 仍可被停止的原因。取消 cause 为 `user`。 +- `{ kind: 'ancestor', agent }`——一个确切在线的 ancestor Agent(直接 parent 或更深)。调用方必须是注册表中其 id 的当前条目(过期调用方即使目标不存在也被拒绝),不得是目标本身,并且必须出现在 Activation 物化时记录的 `ancestry` WeakSet 中。取消 cause 为 `parent`。 + +目标只在管理器进程本地的 Activation map 中解析。不存在的 id——未知、一次性或已自然结算——是被接受的 no-op,统一覆盖完成竞态和重复请求而不泄露持久化目录信息;disposal 事务已打开的目标在鉴权后同样是被接受的 no-op。一次性生命周期(持有方 `dispose()`、task-kill)不受影响。`SubagentService.interrupt()` 把未绑定管理器的组合视为被接受的 no-op 而不是 `CONTINUATION_UNAVAILABLE`,因为没有管理器就不可能存在管理器拥有的在线 Activation。 + +Host RPC `subagent.interrupt` 接收 continuable 的 `SubagentAddress` 并返回 `{ accepted: true }`。它的实现只以 `user` 授权调用核心原语——刻意不调用 `catalogChild()`、`listChildren()`、`sessionQuery` 或 parent 注册表查找。parent 地址不匹配的在线目标映射为 `subagent-unauthorized`;意外失败映射为 `internal`,不把错误文本泄漏到 wire。 + +## Alternatives considered + +**让人类中断走 `session.cancel`。** 通用会话取消要求附着的普通会话并拒绝 subagent 拥有的会话;放宽它会把 subagent 权限规则缠进普通会话路由。subagent 域的 RPC 让基于地址的鉴权和 parent 离线保证保持显式。 + +**等待目标静止并返回轮次结果。** 取消是协作式的,静止时间无上界;让 RPC(以及一个 `ChildLock` 槽位)保持打开会招致超时并与投递、disposal 形成排队。调用方需要的唯一事实是信号已被接受,而竞态(自然完成、disposal)本就幂等收敛。 + +**复用整个 Activation 的 disposal 来做中断。** disposal 的取消不带 `keepInbox`,还会 flush、capture 并释放 handle——它销毁排队工作和 child 的驻留。中断是针对一个轮次的控制操作,不是针对 Activation 的生命周期操作。 + +**顺手把 `send_message`/`followup` 权限扩展到 ancestor。** 投递向对话注入内容且不幂等;其确切直接 parent 权限保持不变。只有中断获得更宽的 ancestor 与基于地址的用户授权。 + +**中断后自动恢复被暂停的队列。** 在中止 A 后立即启动排队的 follow-up B 会让中断看起来被忽略,并夺走人类重新引导 child 的窗口。暂停到显式唤醒发送为止,让停止可观察且 FIFO 顺序完整。 + +## Consequences + +人类或 ancestor 现在可以停止一个失控的 continuable 轮次,而不丢失 child、其排队工作或正在运行的后代;代价是一个刻意保持弱的后置条件(`accepted` 表示"信号已发出",目标在观察到信号前可能仍显示 `running`),客户端必须如实呈现。暂停队列规则意味着被中断的 child 会带着保留的工作停在 idle,直到有人发送唤醒消息——这是有意的 human-in-the-loop 暂停,不是调度器缺陷。Web 的 Stop 操作和面向模型的 `interrupt_agent` 工具在 issue #1535 的后续 stacked PR 中基于此原语构建。 + +## Testing + +`packages/subagent/subagent/tests/continuation.spec.ts` 中的核心覆盖证明了持久化 `turn/end` 中止、队列先暂停后按 FIFO 恢复、后代不受影响、两种授权及其取消 cause、self/sibling/stale/非 ancestor 拒绝、absent/一次性/disposal 竞态 no-op,以及 `keepInbox` 循环行为不变。`packages/host/apiproxy/tests` 中的 Host 覆盖证明 RPC 只调用核心原语(不读 agents/目录/历史)、`subagent-unauthorized`/`internal` 映射、wire schema 的 continuable 模式围栏以及 carrier 往返。 diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index ae16d10f07..edfd046430 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -699,7 +699,7 @@ A published child settled. Scope-filtered dispatch uses the same delegating pare Types: [Scoped](../core-data-structures/scope.md) · [SubagentService](../core-data-structures/subagent.md) -Source: [`packages/subagent/subagent/src/index.ts:160`](../../packages/subagent/subagent/src/index.ts) +Source: [`packages/subagent/subagent/src/index.ts:162`](../../packages/subagent/subagent/src/index.ts) ### `subagent/provider-added` — emit @@ -716,7 +716,7 @@ A provider became resolvable in the registry. Types: [SubagentProvider](../core-data-structures/subagent.md) -Source: [`packages/subagent/subagent/src/index.ts:134`](../../packages/subagent/subagent/src/index.ts) +Source: [`packages/subagent/subagent/src/index.ts:136`](../../packages/subagent/subagent/src/index.ts) ### `subagent/provider-removed` — emit @@ -731,7 +731,7 @@ A provider left the registry. Accepted runs remain holder-owned. 'subagent/provider-removed'(name: string): void ``` -Source: [`packages/subagent/subagent/src/index.ts:140`](../../packages/subagent/subagent/src/index.ts) +Source: [`packages/subagent/subagent/src/index.ts:142`](../../packages/subagent/subagent/src/index.ts) ### `subagent/start` — emit @@ -753,7 +753,7 @@ A provider established a published child. For in-process providers, `ctx.agents. Types: [Scoped](../core-data-structures/scope.md) · [SubagentService](../core-data-structures/subagent.md) -Source: [`packages/subagent/subagent/src/index.ts:151`](../../packages/subagent/subagent/src/index.ts) +Source: [`packages/subagent/subagent/src/index.ts:153`](../../packages/subagent/subagent/src/index.ts) ## `system-prompt/*` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index a22c3464eb..caf081b5a6 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -2071,6 +2071,22 @@ async startContinuable(spec: ContinuableStartSpec): Promise */ async followup( parent: Agent, childId: SessionId, content: ContentBlock[], options: SubagentFollowupOptions, ): Promise +/** + * Interrupt one live continuable child's current turn under a human parent + * address or an exact live ancestor Agent. Fire-and-return: the cancel + * signal is issued before this returns, but the target may keep running + * until it observes the signal. Pending inbox work, the Activation, and + * published descendants are preserved; only a later waking send resumes the + * parked FIFO queue. An absent target — including a one-shot or unknown id — + * is an accepted no-op, as is a manager-less composition, which cannot own a + * live Activation. + * @param targetSessionId - the durable child session id to interrupt. + * @param authority - the human parent address or exact live ancestor Agent. + * @throws {SubagentError} `UNAUTHORIZED` when the authority does not own the + * live target. + */ +interrupt(targetSessionId: SessionId, authority: SubagentInterruptAuthority): void + /** * Deliver selected content from one live continuable child to its durable * direct parent. The child is the authority credential; callers cannot name a @@ -2171,9 +2187,9 @@ list(): string[] async start(name: string, request: SubagentStartRequest): Promise ``` -Types: [Agent](../core-data-structures/core.md) · [ContentBlock](../core-data-structures/core.md) · [ContinuableSetupContribution](../core-data-structures/subagent.md) · [ContinuableStart](../core-data-structures/subagent.md) · [ContinuableStartSpec](../core-data-structures/subagent.md) · [MessageId](../core-data-structures/core.md) · [SessionId](../core-data-structures/core.md) · [SubagentFollowupOptions](../core-data-structures/subagent.md) · [SubagentListEntry](../core-data-structures/subagent.md) · [SubagentProvider](../core-data-structures/subagent.md) · [SubagentReportOptions](../core-data-structures/subagent.md) · [SubagentRun](../core-data-structures/subagent.md) · [SubagentStartRequest](../core-data-structures/subagent.md) +Types: [Agent](../core-data-structures/core.md) · [ContentBlock](../core-data-structures/core.md) · [ContinuableSetupContribution](../core-data-structures/subagent.md) · [ContinuableStart](../core-data-structures/subagent.md) · [ContinuableStartSpec](../core-data-structures/subagent.md) · [MessageId](../core-data-structures/core.md) · [SessionId](../core-data-structures/core.md) · [SubagentFollowupOptions](../core-data-structures/subagent.md) · [SubagentInterruptAuthority](../core-data-structures/subagent.md) · [SubagentListEntry](../core-data-structures/subagent.md) · [SubagentProvider](../core-data-structures/subagent.md) · [SubagentReportOptions](../core-data-structures/subagent.md) · [SubagentRun](../core-data-structures/subagent.md) · [SubagentStartRequest](../core-data-structures/subagent.md) -Source: [`packages/subagent/subagent/src/index.ts:165`](../../packages/subagent/subagent/src/index.ts) +Source: [`packages/subagent/subagent/src/index.ts:167`](../../packages/subagent/subagent/src/index.ts) ## `ctx.subprocess` — `SubprocessService` (abstract seam) diff --git a/packages/client/connection/src/client/fixture.ts b/packages/client/connection/src/client/fixture.ts index ec3199ee02..2a081284b2 100644 --- a/packages/client/connection/src/client/fixture.ts +++ b/packages/client/connection/src/client/fixture.ts @@ -2203,6 +2203,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld { prompt: request => Promise.resolve(ok(request, { messageId: `fixture-message-${request.payload.childSessionId}` as never, })), + interrupt: request => Promise.resolve(ok(request, { accepted: true as const })), }, host: { describe: request => ok(request, { version: '0.0.0-fixture', cwd: '/tmp/fixture', attachedSessions }), @@ -2748,6 +2749,7 @@ export class FixtureApiClient extends AbstractApiClient { case 'subagent.list': return this.api.subagents.list(request) case 'subagent.history': return this.api.subagents.history(request) case 'subagent.prompt': return this.api.subagents.prompt(request, signal) + case 'subagent.interrupt': return this.api.subagents.interrupt(request) case 'host.describe': return this.api.host.describe(request) case 'host.pickDirectory': return this.api.host.pickDirectory(request, new AbortController().signal) case 'host.listDirectory': return this.api.host.listDirectory(request, new AbortController().signal) diff --git a/packages/client/connection/tests/fake-api.ts b/packages/client/connection/tests/fake-api.ts index cc4504e538..ef5071316e 100644 --- a/packages/client/connection/tests/fake-api.ts +++ b/packages/client/connection/tests/fake-api.ts @@ -126,6 +126,9 @@ export class FakeApiClient implements IApiClient { prompt: (payload: unknown) => this.record('subagent.prompt', payload, Promise.resolve(ok({ messageId: 'fake-message' as never, }))), + interrupt: (payload: unknown) => this.record('subagent.interrupt', payload, Promise.resolve(ok({ + accepted: true as const, + }))), } readonly host: IApiClient['host'] = { diff --git a/packages/client/runtime/tests/fake-api.ts b/packages/client/runtime/tests/fake-api.ts index 2f4299ce6c..5e510c0012 100644 --- a/packages/client/runtime/tests/fake-api.ts +++ b/packages/client/runtime/tests/fake-api.ts @@ -140,10 +140,14 @@ export class FakeApiClient implements IApiClient { onSubagentPrompt: (payload: unknown) => Promise> = () => Promise.resolve(ok({ messageId: 'fake-message' as never })) + onSubagentInterrupt: (payload: unknown) => Promise> + = () => Promise.resolve(ok({ accepted: true as const })) + readonly subagents: IApiClient['subagents'] = { list: (payload: unknown) => this.record('subagent.list', payload, this.onSubagentList(payload)), history: (payload: unknown) => this.record('subagent.history', payload, this.onSubagentHistory(payload)), prompt: (payload: unknown) => this.record('subagent.prompt', payload, this.onSubagentPrompt(payload)), + interrupt: (payload: unknown) => this.record('subagent.interrupt', payload, this.onSubagentInterrupt(payload)), } readonly host: IApiClient['host'] = { diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index dec95f3b5f..acaefa5d5f 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -928,6 +928,10 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ signature: 'async followup( parent: Agent, childId: SessionId, content: ContentBlock[], options: SubagentFollowupOptions, ): Promise', jsDoc: '/**\n * Deliver one later message to a continuable child as its next FIFO turn. A\n * resident child\'s Agent inbox accepts it directly (waking a `waiting`\n * Activation), while an absent one is cold-resumed from its persisted\n * Session. The Agent inbox is the only queue, so every accepted message has\n * one observable order.\n * @param parent - the exact live direct parent authorizing this delivery.\n * @param childId - durable child session id.\n * @param content - user-role content to deliver.\n * @param options - durable provenance and caller cancellation, which stops the\n * operation only before inbox acceptance.\n * @returns the accepted message\'s inbox id.\n * @throws when continuation services are unavailable, parent authority is\n * rejected, or the message was not admitted.\n */', }, + { + signature: 'interrupt(targetSessionId: SessionId, authority: SubagentInterruptAuthority): void', + jsDoc: '/**\n * Interrupt one live continuable child\'s current turn under a human parent\n * address or an exact live ancestor Agent. Fire-and-return: the cancel\n * signal is issued before this returns, but the target may keep running\n * until it observes the signal. Pending inbox work, the Activation, and\n * published descendants are preserved; only a later waking send resumes the\n * parked FIFO queue. An absent target — including a one-shot or unknown id —\n * is an accepted no-op, as is a manager-less composition, which cannot own a\n * live Activation.\n * @param targetSessionId - the durable child session id to interrupt.\n * @param authority - the human parent address or exact live ancestor Agent.\n * @throws {SubagentError} `UNAUTHORIZED` when the authority does not own the\n * live target.\n */', + }, { signature: 'async reportFrom( child: Agent, content: ContentBlock[], options: SubagentReportOptions, ): Promise', jsDoc: '/**\n * Deliver selected content from one live continuable child to its durable\n * direct parent. The child is the authority credential; callers cannot name a\n * recipient. Reporting does not conclude the child\'s turn or Activation.\n * @param child - exact live reporting child.\n * @param content - selected model-facing content.\n * @param options - parent scheduling and pre-acceptance cancellation.\n * @returns the stable identity of the parent-accepted message.\n * @throws when continuation services are unavailable, sender authorization\n * fails, or the direct parent is not live.\n */', @@ -2787,6 +2791,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [ name: 'SubagentFollowupOptions', declaration: 'export interface SubagentFollowupOptions {\n readonly source: MessageSource;\n readonly signal: AbortSignal;\n}', }, + { + name: 'SubagentInterruptAuthority', + declaration: 'export type SubagentInterruptAuthority = {\n readonly kind: \'user\';\n readonly parentSessionId: SessionId;\n} | {\n readonly kind: \'ancestor\';\n readonly agent: Agent;\n};', + }, { name: 'SubagentListEntry', declaration: 'export type SubagentListEntry = {\n readonly kind: \'child\';\n readonly id: SessionId;\n readonly activity: \'running\' | \'inactive\';\n readonly hasChildren: boolean;\n} & ({\n readonly mode: \'one-shot\';\n readonly label?: string;\n} | {\n readonly mode: \'continuable\';\n readonly label: string;\n}) | {\n readonly kind: \'diagnostic\';\n readonly id: SessionId;\n readonly reason: \'corrupt\' | \'unsupported\' | \'unavailable\';\n};', diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index cfcae423bc..52236b0f29 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -2013,6 +2013,31 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro return subagentPromptError(request, error, signal) } }, + + // Deliberately no catalog, history, persistence, or parent Agent lookup: + // the core primitive alone authorizes the durable address against the + // live Activation, which is what keeps a live child interruptible while + // its parent Agent is offline. Absent targets are accepted no-ops there. + interrupt(request) { + const { parentSessionId, childSessionId } = request.payload + try { + ctx.subagents.interrupt(childSessionId, { kind: 'user', parentSessionId }) + } catch (error: unknown) { + if (error instanceof SubagentError && error.code === 'UNAUTHORIZED') { + return Promise.resolve(err(request, { + code: 'subagent-unauthorized', + message: 'subagent does not belong to this parent', + details: { childSessionId }, + })) + } + return Promise.resolve(err(request, { + code: 'internal', + message: 'subagent interrupt failed', + details: {}, + })) + } + return Promise.resolve(ok(request, { accepted: true as const })) + }, }, workspace: { diff --git a/packages/host/apiproxy/src/api/index.ts b/packages/host/apiproxy/src/api/index.ts index cb83c5328d..5537a0e382 100644 --- a/packages/host/apiproxy/src/api/index.ts +++ b/packages/host/apiproxy/src/api/index.ts @@ -42,7 +42,8 @@ export type { } from './sessions.ts' export type { DirectoryEntry, DirectoryListing, HostApi } from './host.ts' export type { - SubagentAddress, SubagentCatalog, SubagentListEntry, SubagentPromptReceipt, SubagentsApi, + SubagentAddress, SubagentCatalog, SubagentInterruptReceipt, SubagentListEntry, + SubagentPromptReceipt, SubagentsApi, } from './subagents.ts' export type { WorkspaceApi, WorkspaceId, WorkspaceView } from './workspace.ts' export type { CommandsApi, CommandDescriptor } from './commands.ts' diff --git a/packages/host/apiproxy/src/api/rpc-map.ts b/packages/host/apiproxy/src/api/rpc-map.ts index 9a8750c722..deb963db07 100644 --- a/packages/host/apiproxy/src/api/rpc-map.ts +++ b/packages/host/apiproxy/src/api/rpc-map.ts @@ -36,6 +36,7 @@ export interface RpcMethodMap { 'subagent.list': SubagentsApi['list'] 'subagent.history': SubagentsApi['history'] 'subagent.prompt': SubagentsApi['prompt'] + 'subagent.interrupt': SubagentsApi['interrupt'] 'host.describe': HostApi['describe'] 'host.pickDirectory': HostApi['pickDirectory'] 'host.listDirectory': HostApi['listDirectory'] diff --git a/packages/host/apiproxy/src/api/subagents.schema.ts b/packages/host/apiproxy/src/api/subagents.schema.ts index 1987d568d3..6ed8bd3263 100644 --- a/packages/host/apiproxy/src/api/subagents.schema.ts +++ b/packages/host/apiproxy/src/api/subagents.schema.ts @@ -69,6 +69,18 @@ export const subagentPromptRequestSchema = z.object({ content: z.array(contentBlockSchema), }) as unknown as z.ZodType> +/** subagent.interrupt request payload. */ +export const subagentInterruptRequestSchema = z.object({ + parentSessionId: sessionIdSchema, + childSessionId: sessionIdSchema, + mode: z.literal('continuable'), +}) satisfies z.ZodType>> + +/** subagent.interrupt response value. */ +export const subagentInterruptValueSchema = z.object({ + accepted: z.literal(true), +}) satisfies z.ZodType>> + const messageIdSchema = z.string() as unknown as z.ZodType /** subagent.prompt response value. */ diff --git a/packages/host/apiproxy/src/api/subagents.ts b/packages/host/apiproxy/src/api/subagents.ts index 4c251dca7c..8efb452c2e 100644 --- a/packages/host/apiproxy/src/api/subagents.ts +++ b/packages/host/apiproxy/src/api/subagents.ts @@ -40,6 +40,11 @@ export interface SubagentPromptReceipt { messageId: MessageId } +/** Uniform acknowledgement that one interrupt request was admitted. */ +export interface SubagentInterruptReceipt { + accepted: true +} + /** Durable parent/child address that selects subagent transport in the client. */ export type SubagentAddress = & { @@ -94,4 +99,17 @@ export interface SubagentsApi { >, signal: AbortSignal, ): Promise> + + /** + * Interrupts a live continuable child's current turn under the address's + * durable direct-parent authority, without requiring a live parent Agent, + * consulting the catalog, or resuming anything. Fire-and-return: `accepted` + * acknowledges the admitted cancel signal, not target quiescence, so the + * child may remain visibly running briefly. Queued follow-ups are kept and + * parked; an absent, idle, or already-completed target is likewise + * `accepted`. + */ + interrupt( + request: RpcRequest>, + ): Promise> } diff --git a/packages/host/apiproxy/src/fetch/client.ts b/packages/host/apiproxy/src/fetch/client.ts index 0f54d76dbc..0ce935809f 100644 --- a/packages/host/apiproxy/src/fetch/client.ts +++ b/packages/host/apiproxy/src/fetch/client.ts @@ -58,6 +58,7 @@ import { import { llmDiscoverModelsValueSchema, llmModelsValueSchema, llmProvidersValueSchema } from '../api/llm.schema.ts' import { subagentHistoryValueSchema, + subagentInterruptValueSchema, subagentListValueSchema, subagentPromptValueSchema, } from '../api/subagents.schema.ts' @@ -96,6 +97,7 @@ export interface IApiClient { list(payload: RequestPayload<'subagent.list'>, signal?: AbortSignal): Promise>> history(payload: RequestPayload<'subagent.history'>, signal?: AbortSignal): Promise>> prompt(payload: RequestPayload<'subagent.prompt'>, signal?: AbortSignal): Promise>> + interrupt(payload: RequestPayload<'subagent.interrupt'>, signal?: AbortSignal): Promise>> } host: { describe(payload: RequestPayload<'host.describe'>, signal?: AbortSignal): Promise>> @@ -171,6 +173,7 @@ const UNARY_VALUE_SCHEMAS: { [K in keyof RpcMethodMap]: z.ZodType this.callUnary('subagent.list', payload, signal), history: (payload, signal) => this.callUnary('subagent.history', payload, signal), prompt: (payload, signal) => this.callUnary('subagent.prompt', payload, signal), + interrupt: (payload, signal) => this.callUnary('subagent.interrupt', payload, signal), } readonly host: IApiClient['host'] = { diff --git a/packages/host/apiproxy/src/fetch/handler.ts b/packages/host/apiproxy/src/fetch/handler.ts index d41b51ad6d..bd3bc3827a 100644 --- a/packages/host/apiproxy/src/fetch/handler.ts +++ b/packages/host/apiproxy/src/fetch/handler.ts @@ -60,6 +60,7 @@ import { import { llmDiscoverModelsRequestSchema, llmModelsRequestSchema, llmProvidersRequestSchema } from '../api/llm.schema.ts' import { subagentHistoryRequestSchema, + subagentInterruptRequestSchema, subagentListRequestSchema, subagentPromptRequestSchema, } from '../api/subagents.schema.ts' @@ -95,6 +96,7 @@ const UNARY_ROUTES: UnaryRoutes = { 'subagent.list': { schema: subagentListRequestSchema, invoke: (api, r, signal) => api.subagents.list(r, signal) }, 'subagent.history': { schema: subagentHistoryRequestSchema, invoke: (api, r, signal) => api.subagents.history(r, signal) }, 'subagent.prompt': { schema: subagentPromptRequestSchema, invoke: (api, r, signal) => api.subagents.prompt(r, signal) }, + 'subagent.interrupt': { schema: subagentInterruptRequestSchema, invoke: (api, r) => api.subagents.interrupt(r) }, 'host.describe': { schema: hostDescribeRequestSchema, invoke: (api, r) => api.host.describe(r) }, 'host.pickDirectory': { schema: hostPickDirectoryRequestSchema, invoke: (api, r, signal) => api.host.pickDirectory(r, signal) }, 'host.listDirectory': { schema: hostListDirectoryRequestSchema, invoke: (api, r, signal) => api.host.listDirectory(r, signal) }, diff --git a/packages/host/apiproxy/tests/api-proxy-subagents.spec.ts b/packages/host/apiproxy/tests/api-proxy-subagents.spec.ts index feb9ecb073..62c7e653e0 100644 --- a/packages/host/apiproxy/tests/api-proxy-subagents.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-subagents.spec.ts @@ -19,6 +19,7 @@ function bench(options: { childStatus?: 'idle' | 'running' entries?: object[] followupError?: Error + interruptError?: Error listError?: Error /** Persistence forgets the child entirely (the vanished-mid-read race). */ storedChild?: false @@ -53,6 +54,12 @@ function bench(options: { ) => options.followupError === undefined ? Promise.resolve('message-1') : Promise.reject(options.followupError)) + const interrupt = vi.fn(( + _targetSessionId: SessionId, + _authority: { kind: 'user'; parentSessionId: SessionId }, + ) => { + if (options.interruptError !== undefined) throw options.interruptError + }) const childHeader = { version: 0, id: CHILD, createdAt: 1, cwd: '/proj', parentSession: options.historyParent ?? PARENT, } satisfies SessionHeader @@ -72,7 +79,7 @@ function bench(options: { }) const ctx = new Context() ctx.provide('agents', { get: getAgent }) - ctx.provide('subagents', { listChildren, followup }) + ctx.provide('subagents', { listChildren, followup, interrupt }) ctx.provide('sessions', { get: (id: SessionId) => options.liveChild === true && id === CHILD ? { id: CHILD, header: childHeader, events: childEvents } @@ -90,7 +97,7 @@ function bench(options: { const api = createApiProxy(ctx, { defaultTarget: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp', workspaceRoot: '/tmp', }) - return { api, getAgent, listChildren, inspect, snapshot, restore, followup, parent } + return { api, getAgent, listChildren, inspect, snapshot, restore, followup, interrupt, parent } } describe('subagent gateway', () => { @@ -309,4 +316,48 @@ describe('subagent gateway', () => { error: { code: 'internal', message: 'subagent prompt failed' }, }) }) + + it('interrupts through the core primitive alone while the parent Agent is offline', async () => { + const { api, interrupt, getAgent, listChildren, inspect } = bench({ parentLive: false }) + const response = await api.subagents.interrupt(request({ + parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const, + })) + expect(response.rpcId).toBe('subagent-rpc') + expect(response.result).toEqual({ ok: true, value: { accepted: true } }) + expect(interrupt).toHaveBeenCalledExactlyOnceWith(CHILD, { kind: 'user', parentSessionId: PARENT }) + // No parent-registry, catalog, or history dependency: this is what keeps a + // live child interruptible after its parent Agent went offline. + expect(getAgent).not.toHaveBeenCalled() + expect(listChildren).not.toHaveBeenCalled() + expect(inspect).not.toHaveBeenCalled() + }) + + it('maps interrupt authorization rejection without touching other services', async () => { + const { api, listChildren } = bench({ + interruptError: new SubagentError('secret lineage', 'UNAUTHORIZED'), + }) + const response = await api.subagents.interrupt(request({ + parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const, + })) + expect(response.result).toEqual({ + ok: false, + error: { + code: 'subagent-unauthorized', + message: 'subagent does not belong to this parent', + details: { childSessionId: CHILD }, + }, + }) + expect(listChildren).not.toHaveBeenCalled() + }) + + it('hides unexpected interrupt failures behind the internal code', async () => { + const { api } = bench({ interruptError: new Error('secret activation state') }) + const response = await api.subagents.interrupt(request({ + parentSessionId: PARENT, childSessionId: CHILD, mode: 'continuable' as const, + })) + expect(response.result).toEqual({ + ok: false, + error: { code: 'internal', message: 'subagent interrupt failed', details: {} }, + }) + }) }) diff --git a/packages/host/apiproxy/tests/client-handler.spec.ts b/packages/host/apiproxy/tests/client-handler.spec.ts index ebd56ee551..476ebf344f 100644 --- a/packages/host/apiproxy/tests/client-handler.spec.ts +++ b/packages/host/apiproxy/tests/client-handler.spec.ts @@ -63,6 +63,7 @@ function scriptedApi(overrides: { list: r => ok(r, { entries: [], parentAvailable: false }), history: r => ok(r, { events: [], hasMore: false }), prompt: r => ok(r, { messageId: 'message-1' as never }), + interrupt: r => ok(r, { accepted: true as const }), ...overrides.subagents, }, host: { @@ -248,6 +249,32 @@ describe('unary round trip', () => { } }) + it('round-trips subagent.interrupt and rejects a one-shot or incomplete address', async () => { + const interrupt = vi.fn((r: RpcRequest) => ok(r, { accepted: true as const })) + const api = scriptedApi({ subagents: { interrupt } }) + const c = client(api) + + const accepted = await c.subagents.interrupt({ + parentSessionId: sid('parent'), childSessionId: sid('child'), mode: 'continuable', + }) + expect(accepted.result).toEqual({ ok: true, value: { accepted: true } }) + expect(interrupt).toHaveBeenCalledTimes(1) + + // The wire schema owns the mode fence: a one-shot address never reaches the impl. + const oneShot = await c.subagents.interrupt({ + parentSessionId: sid('parent'), childSessionId: sid('child'), mode: 'one-shot', + } as never) + expect(oneShot.result.ok).toBe(false) + if (!oneShot.result.ok) expect(oneShot.result.error.code).toBe('bad-request') + + const incomplete = await c.subagents.interrupt({ + parentSessionId: sid('parent'), mode: 'continuable', + } as never) + expect(incomplete.result.ok).toBe(false) + if (!incomplete.result.ok) expect(incomplete.result.error.code).toBe('bad-request') + expect(interrupt).toHaveBeenCalledTimes(1) + }) + it('rejects a method/path mismatch as bad-request', async () => { const handler = toFetchHandler(scriptedApi()) const body = { type: 'client-request', rpcId: 'r1', method: 'session.create', payload: {} } diff --git a/packages/host/apiproxy/tests/fetch-carrier.spec.ts b/packages/host/apiproxy/tests/fetch-carrier.spec.ts index 6481d75837..83ada22644 100644 --- a/packages/host/apiproxy/tests/fetch-carrier.spec.ts +++ b/packages/host/apiproxy/tests/fetch-carrier.spec.ts @@ -128,6 +128,9 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra result: { ok: true, value: { messageId: 'message-1' as never } }, } }, + async interrupt(request) { + return { rpcId: request.rpcId, result: { ok: true, value: { accepted: true as const } } } + }, }, host: { async describe(request) { @@ -433,6 +436,11 @@ describe('unary round trip (handler ⇄ client, no network)', () => { mode: 'continuable', content: [], })).result).toEqual({ ok: true, value: { messageId: 'message-1' } }) + expect((await c.subagents.interrupt({ + parentSessionId: 'parent' as never, + childSessionId: 'child' as never, + mode: 'continuable', + })).result).toEqual({ ok: true, value: { accepted: true } }) }) it('keeps caller and connection aborts on command.execute', async () => { diff --git a/packages/subagent/subagent/README.i18n.yaml b/packages/subagent/subagent/README.i18n.yaml index f5ceedc743..250fd1a078 100644 --- a/packages/subagent/subagent/README.i18n.yaml +++ b/packages/subagent/subagent/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/subagent/subagent/README.md -README.md: 9d2e38c8730f7b7f26e690aa878a4466fa7c2829 -README.zh.md: 341c18617af4d040ec44814fac1ec4502d9b8902 +README.md: 7de3a5563b274e925fba931a6d5de17e68cc397c +README.zh.md: 6067555544ec0c32729beb2b4e3f773e31b747a1 diff --git a/packages/subagent/subagent/README.md b/packages/subagent/subagent/README.md index 9d2e38c873..7de3a5563b 100644 --- a/packages/subagent/subagent/README.md +++ b/packages/subagent/subagent/README.md @@ -18,6 +18,7 @@ The [subagent family overview](../README.md) maps implementations and model-faci | `start(name, request)` | Validate an ordinary caller request, resolve its detached `one-shot` descriptor, then await the provider until a real one-shot child is published. Fulfillment returns a holder-owned `SubagentRun`; rejection means the provider has already cleaned every unpublished startup resource, while post-publication turn or infrastructure faults settle through the run. Continuable children never enter through this operation. | | `startContinuable(spec)` | Establish one durable continuable child and deliver its initial prompt. Resolves with `{ childId, messageId }` when the child's inbox accepts that prompt, without waiting for the turn to start or for the message to reach the Session log; any earlier failure rejects with no ids and rolls the child back entirely. Requires `ctx.agents`, session persistence, and a provider with the `prepareContinuable` capability. | | `followup(parent, childId, content, { source, signal })` | Deliver one later message from the exact live direct parent as the child's next FIFO turn, matching `Agent.followup()` terminology, and return the accepted `MessageId`. A resident child's inbox accepts it directly (waking a waiting Activation); an absent one cold-resumes from its persisted Session. Requires `ctx.agents`; cold resume also requires session persistence. | +| `interrupt(targetSessionId, authority)` | Interrupt one live continuable child's current turn under a human durable parent address (`{ kind: 'user', parentSessionId }`) or an exact live ancestor Agent (`{ kind: 'ancestor', agent }`). Admission is synchronous and the effect asynchronous: it issues `Agent.cancel(cause, { keepInbox: true })` and returns without waiting for the target to observe the signal. Pending inbox work, the Activation, and published descendants are preserved; only a later waking send resumes the parked FIFO queue. An absent target — unknown, one-shot, or already-settled id — and a manager-less composition are accepted no-ops; a wrong parent address or a stale, self-targeting, or non-ancestor caller rejects with `UNAUTHORIZED`. | | `reportFrom(child, content, { delivery, signal })` | Deliver one selected message from the exact live continuable child to its exact live direct parent and return the accepted stable `MessageId`. Quiet delivery injects context; waking delivery submits one later parent turn. | | `registerContinuableSetup(contribution)` | Compose an optional deployment capability into each continuable child's unpublished scope, with immediate revocation from resident children. | | `drainContinuableDescendants(parents)` | Close admission below exact live host-owned parent Agents, stop only their visible continuable descendants, await materializations admitted below those roots through publication or rollback, then release the selected forests child-first. The cutoff lasts until each exact parent leaves the registry; unrelated parent forests and manager-wide admission remain live. | @@ -76,7 +77,7 @@ Run events are scoped to the delegating parent. Every listener is independently Provider additions and removals also emit `subagent/provider-added` and `subagent/provider-removed`. Consumers such as the model-facing tool use those events because Cordis may load sibling plugins concurrently; configuration order does not prove registration order. -Continuable children do not create `SubagentRun` or Tasks. The continuation manager directly owns one process-local Activation and retained `AgentHandle` per resident child Session, uses the Agent inbox as the only FIFO, and cold-resumes from the durable descriptor. Exact live direct-parent identity authorizes parent-to-child delivery. Exact live child identity authorizes reports; the manager derives the recipient from durable `parentSession`, and `MessageSource` remains provenance rather than authority. +Continuable children do not create `SubagentRun` or Tasks. The continuation manager directly owns one process-local Activation and retained `AgentHandle` per resident child Session, uses the Agent inbox as the only FIFO, and cold-resumes from the durable descriptor. Exact live direct-parent identity authorizes parent-to-child delivery. Exact live child identity authorizes reports; the manager derives the recipient from durable `parentSession`, and `MessageSource` remains provenance rather than authority. Interrupt authority is deliberately wider than delivery authority: a human presents the durable direct-parent address so a live child stays stoppable while its parent Agent is offline, and any exact live ancestor recorded in the Activation's materialization lineage may stop its descendant, because stopping a turn is idempotent and delivers no content. When `ctx.sessionProjections` is available, the service registers two projection units. `subagentTiming` resets at each descriptor so a fork seed's ancestor work cannot enter the child's total, then accumulates `turn/start` → `turn/end` active time and retains same-cut `active.since` and `active.through` bounds for an open turn; while that turn remains open, `active.through` follows the latest folded event, giving an inactive consumer a conservative crash bound without mixing in newer session metadata. `subagent` folds the durable identity — mode plus creation label — from `subagent/descriptor` events with the same last-wins reset discipline, so a fork seed's ancestor descriptor stands only until the child's own overrides it; a malformed or unrecognized-version payload folds to the serializable `null` sentinel — indistinguishable from a log with no descriptor, and surviving every JSON push frame so a consumer replaces a stale identity instead of keeping it — and never throws. @@ -84,7 +85,7 @@ When `ctx.sessionProjections` is available, the service registers two projection ## Collection model -The model-facing tool collects synchronously by default: it awaits the child result and disposes the run before returning. One-shot background delegation registers a plain Task in the tool, whose generic status, collection, and cancellation tools own later interaction, and persists its model-supplied `description` as the optional display label. Continuable background delegation calls `ctx.subagents.startContinuable()` and returns only the durable child id; the child owns its own turns from inbox acceptance, so there is no Task, no result promise, and no public subagent cancellation — a caller sends later work with the `send_message` follow-up tool, and the durable child Session remains the source of the child's detailed output. The continuation manager exists only while `ctx.agents` is available, and session persistence is resolved per continuation operation. Independently, `listChildren()` enumerates the live-preferred merge of the live session store and optional session persistence — live-only when persistence is absent, since a cold child cannot be resumed then either — and serves each child's durable mode/label from the registered `subagent` projection unit: the registry's watermark snapshot for a live child; for a cold one, a durable projection-cache row when it serves an own-suffix identity — its `seq` gate proves the value postdates the fork seed, where a child's own descriptor is immutable once appended — else one bounded-concurrency persistence inspection folded through the registry, whose result must still name the enumerated lifecycle (a re-published id degrades to a `corrupt` diagnostic). A throwing cache read renders no verdict — the cache is derived data — and silently falls through to that authoritative re-fold. The projection fold is the single classification authority; listing parses no descriptor itself. A served identity produces a child row; a settled candidate whose fold served no identity is a `corrupt` diagnostic, a failed inspection is a transient `unavailable` retried on the next listing, and a running candidate without an identity yet is omitted (the creation window before its descriptor is appended). It never consults the continuation manager, Agent registrations, Activations, or providers. Each child row derives its read-time `hasChildren` hint from merged headers carrying durable `origin: 'subagent'`; it does not read descendant event logs, and the descriptor-backed child catalog remains authoritative when expanded. Service consumers such as a UI can retain both modes and choose a fallback for an unlabeled one-shot child; the model-facing `list_agents` tool projects only `continuable` entries and maps service activity to its existing `running`/`complete` vocabulary. The listing forwards the caller's signal to every persistence read, checks cancellation around each of those awaits, and reports every observed abort as `SubagentError` code `CANCELLED`; an unmounted projection registry fails loud with `SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE`, and a missing session store with `SUBAGENT_CONTROL_SESSION_STORE_UNAVAILABLE`. See the [background subagent tasks Agent Note](../../../.agents/notes/implemented/feature/2026-07-08-background-subagent-tasks.md), the [continuable background subagents Agent Note](../../../.agents/notes/implemented/feature/2026-07-21-continuable-background-subagents.md), the [durable catalog Agent Note](../../../.agents/notes/implemented/feature/2026-07-22-durable-subagent-catalog-and-list-agents.md), the [merged-service Agent Note](../../../.agents/notes/implemented/simplification/2026-07-26-merge-subagent-control-service.md), the [capability-seam Agent Note](../../../.agents/notes/implemented/feature/2026-06-21-subagent-capability-seam.md), and `src/types.ts` for the complete contracts. +The model-facing tool collects synchronously by default: it awaits the child result and disposes the run before returning. One-shot background delegation registers a plain Task in the tool, whose generic status, collection, and cancellation tools own later interaction, and persists its model-supplied `description` as the optional display label. Continuable background delegation calls `ctx.subagents.startContinuable()` and returns only the durable child id; the child owns its own turns from inbox acceptance, so there is no Task or result promise — a caller sends later work with the `send_message` follow-up tool, while `interrupt()` stops only the current turn without disposing the child. The durable child Session remains the source of the child's detailed output. The continuation manager exists only while `ctx.agents` is available, and session persistence is resolved per continuation operation. Independently, `listChildren()` enumerates the live-preferred merge of the live session store and optional session persistence — live-only when persistence is absent, since a cold child cannot be resumed then either — and serves each child's durable mode/label from the registered `subagent` projection unit: the registry's watermark snapshot for a live child; for a cold one, a durable projection-cache row when it serves an own-suffix identity — its `seq` gate proves the value postdates the fork seed, where a child's own descriptor is immutable once appended — else one bounded-concurrency persistence inspection folded through the registry, whose result must still name the enumerated lifecycle (a re-published id degrades to a `corrupt` diagnostic). A throwing cache read renders no verdict — the cache is derived data — and silently falls through to that authoritative re-fold. The projection fold is the single classification authority; listing parses no descriptor itself. A served identity produces a child row; a settled candidate whose fold served no identity is a `corrupt` diagnostic, a failed inspection is a transient `unavailable` retried on the next listing, and a running candidate without an identity yet is omitted (the creation window before its descriptor is appended). It never consults the continuation manager, Agent registrations, Activations, or providers. Each child row derives its read-time `hasChildren` hint from merged headers carrying durable `origin: 'subagent'`; it does not read descendant event logs, and the descriptor-backed child catalog remains authoritative when expanded. Service consumers such as a UI can retain both modes and choose a fallback for an unlabeled one-shot child; the model-facing `list_agents` tool projects only `continuable` entries and maps service activity to its existing `running`/`complete` vocabulary. The listing forwards the caller's signal to every persistence read, checks cancellation around each of those awaits, and reports every observed abort as `SubagentError` code `CANCELLED`; an unmounted projection registry fails loud with `SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE`, and a missing session store with `SUBAGENT_CONTROL_SESSION_STORE_UNAVAILABLE`. See the [background subagent tasks Agent Note](../../../.agents/notes/implemented/feature/2026-07-08-background-subagent-tasks.md), the [continuable background subagents Agent Note](../../../.agents/notes/implemented/feature/2026-07-21-continuable-background-subagents.md), the [durable catalog Agent Note](../../../.agents/notes/implemented/feature/2026-07-22-durable-subagent-catalog-and-list-agents.md), the [merged-service Agent Note](../../../.agents/notes/implemented/simplification/2026-07-26-merge-subagent-control-service.md), the [capability-seam Agent Note](../../../.agents/notes/implemented/feature/2026-06-21-subagent-capability-seam.md), and `src/types.ts` for the complete contracts. Continuable Activations await a best-effort final session flush without treating listener participation as durability confirmation. One-shot runs retain best-effort session checkpointing, so a completed one-shot child is discoverable after disposal only when its session actually reached persistence; the service does not invent a catalog entry from Task history when that checkpoint is absent. @@ -99,7 +100,7 @@ No direct invalidation; the named consumers own any request-prefix changes. ## Known Limitations and Deferred Work - **ACP children remain one-shot and are not trace-enumerable** — an ACP run has no local child session in the parent's session corpus. An ACP `prepareContinuable` requires persisting the remote session id in provider-specific descriptor data and a per-child continuation advertisement, since ACP `loadSession` support is negotiated per child rather than established by the method's presence. Remote providers also require a separate Activation ownership contract with equivalent authenticated control and child-first quiescence before they support continuable children. -- **No host-user continuation** — `followup()` requires the exact live direct parent. A future host adapter needs a concrete authenticated interaction before the seam gains a separate user capability. +- **No host-user continuation** — `followup()` requires the exact live direct parent. Only `interrupt()` accepts a durable parent-address user authority, because stopping a turn is idempotent and delivers no content; a future host adapter needs a concrete authenticated interaction before the seam gains a user delivery capability. - **No current-turn steering** — continuable messages and waking reports enqueue later turns; neither redirects an open turn. - **Process-local residency** — the Activation inbox and ownership graph do not coordinate two harness processes; concurrent access to one persistence store still requires a durable mailbox and cross-process lease protocol. - **No replay of accepted-but-unlogged messages** — only messages written to the child Session log are reconstructable with their admitted provenance. A crash may lose an accepted initial prompt or follow-up that never reached the log; a later authorized message can cold-resume the child, but the lost message is not replayed automatically. diff --git a/packages/subagent/subagent/README.zh.md b/packages/subagent/subagent/README.zh.md index 341c18617a..6067555544 100644 --- a/packages/subagent/subagent/README.zh.md +++ b/packages/subagent/subagent/README.zh.md @@ -18,6 +18,7 @@ subagent seam 允许一个 agent(智能体)通过具名提供方把工作委 | `start(name, request)` | 校验普通调用方请求,解析其分离的 `one-shot` 描述符,然后等待提供方,直到真实的一次性子 agent 发布。兑现时返回由持有方拥有的 `SubagentRun`;拒绝表示提供方已清理所有未发布的启动资源,而发布后的轮次或基础设施故障会通过该 run 结算。可继续子 agent 绝不通过此操作进入。 | | `startContinuable(spec)` | 建立一个持久化可继续子 agent,并投递其初始提示词。子 agent 的 inbox 接受该提示词时,兑现为 `{ childId, messageId }`,无需等待轮次开始或消息写入 Session 日志;此前任何失败都会以无 id 拒绝,并完全回滚该子 agent。要求 `ctx.agents`、会话持久化以及具备 `prepareContinuable` 能力的提供方。 | | `followup(parent, childId, content, { source, signal })` | 将来自确切在线直接父级的一条后续消息作为子 agent 的下一个 FIFO 轮次投递,术语与 `Agent.followup()` 一致,并返回被接受的 `MessageId`。驻留中的子 agent 由其 inbox 直接接受(唤醒处于 waiting 的 Activation);不驻留的则从其持久化 Session 冷恢复。要求 `ctx.agents`;冷恢复还要求会话持久化。 | +| `interrupt(targetSessionId, authority)` | 以人类持久化 parent 地址(`{ kind: 'user', parentSessionId }`)或确切在线 ancestor Agent(`{ kind: 'ancestor', agent }`)为授权,中断一个在线可继续 child 的当前轮次。准入是同步的、生效是异步的:它发出 `Agent.cancel(cause, { keepInbox: true })` 后立即返回,不等待目标观察到信号。待处理的 inbox 工作、Activation 与已发布的后代均保持不变;只有之后的一次唤醒发送才会恢复被暂停的 FIFO 队列。目标不存在——未知、一次性或已结算的 id——以及未绑定管理器的组合都是被接受的 no-op;错误的 parent 地址,或过期、指向自身、非 ancestor 的调用方会以 `UNAUTHORIZED` 拒绝。 | | `reportFrom(child, content, { delivery, signal })` | 从确切在线可继续 child 向其确切在线直接 parent 投递一条选中消息,并返回已接受的稳定 `MessageId`。静默投递会注入上下文;唤醒投递会提交一个后续 parent 轮次。 | | `registerContinuableSetup(contribution)` | 把一项可选部署能力组合到每个可继续 child 尚未发布的作用域中,并支持从驻留 child 立即撤销。 | | `drainContinuableDescendants(parents)` | 在由 host 确切拥有的在线 parent Agent 之下关闭准入,只停止其可见的可继续后代,等待在这些根之下已获准的物化过程完成发布或回滚,再按 child-first 顺序释放所选森林。该截止状态会持续到每个确切 parent 离开注册表;无关的 parent 森林和管理器全局准入保持在线。 | @@ -76,7 +77,7 @@ subagent seam 允许一个 agent(智能体)通过具名提供方把工作委 提供方新增和移除还会发出 `subagent/provider-added` 与 `subagent/provider-removed`。面向模型的工具等消费方使用这些事件,因为 Cordis 可能并发加载同级插件;配置顺序不能证明注册顺序。 -可继续子级不会创建 `SubagentRun` 或 Task。继续执行管理器为每个驻留子 Session 直接拥有一个仅存在于当前进程的 Activation 和一个留存的 `AgentHandle`,使用 Agent inbox 作为唯一 FIFO,并从持久化描述符冷恢复。父到子投递由确切在线的直接父级身份授权。上报则由确切在线的子级身份授权;管理器根据持久化的 `parentSession` 推导接收方,`MessageSource` 仍只表示来源,不表示权限。 +可继续子级不会创建 `SubagentRun` 或 Task。继续执行管理器为每个驻留子 Session 直接拥有一个仅存在于当前进程的 Activation 和一个留存的 `AgentHandle`,使用 Agent inbox 作为唯一 FIFO,并从持久化描述符冷恢复。父到子投递由确切在线的直接父级身份授权。上报则由确切在线的子级身份授权;管理器根据持久化的 `parentSession` 推导接收方,`MessageSource` 仍只表示来源,不表示权限。中断权限被刻意设计得比投递权限更宽:人类出示持久化直接 parent 地址,因此即使 parent Agent 离线,在线 child 仍可被停止;Activation 物化时记录的任何确切在线 ancestor 也可以停止其后代——因为停止一个轮次是幂等的,且不投递任何内容。 当 `ctx.sessionProjections` 可用时,服务会注册两个投影单元。`subagentTiming` 会在每个描述符处重置,使 fork 种子中的祖先工作不会计入 child 总量,随后累加 `turn/start` → `turn/end` 活跃时间,并为未结束的轮次保留同一切面的 `active.since` 和 `active.through` 边界;在该轮次保持未结束期间,`active.through` 会跟随最近折叠的事件,从而为 inactive 消费方提供保守的崩溃上界,又不会混入更新的会话元数据。`subagent` 以同样的 last-wins 重置纪律从 `subagent/descriptor` 事件折叠持久化身份——模式与创建标签——因此 fork 种子中的祖先描述符只在 child 自身的描述符覆盖之前有效;畸形或版本不识别的载荷折叠为可序列化的 `null` 哨兵——与没有描述符的日志不可区分,且能完好通过每个 JSON 推送帧,让消费方以之替换掉手中过时的身份而非永久滞留——绝不抛错。 @@ -84,7 +85,7 @@ subagent seam 允许一个 agent(智能体)通过具名提供方把工作委 ## 收集模型 -面向模型的工具默认同步收集:先等待子 agent 结果,再 dispose 运行,然后才返回。一次性后台委派会在工具中注册普通 Task,其通用状态、收集和取消工具负责后续交互,并将模型提供的 `description` 持久化为可选显示标签。可继续后台委派会调用 `ctx.subagents.startContinuable()`,只返回持久化子 agent id;子 agent 自 inbox 接受起就拥有自己的轮次,因此没有 Task、没有结果 promise,也没有公开的子 agent 取消操作——调用方通过 `send_message` 后续操作工具发送后续工作,而持久化子 agent Session 仍是子 agent 详细输出的来源。只有 `ctx.agents` 可用时,继续执行管理器才会存在,而会话持久化按每项继续执行操作解析。与此独立,`listChildren()` 枚举在线会话存储与可选会话持久化的在线优先合并——持久化缺席时仅枚举在线 child,因为那时冷 child 本就无法恢复——并由已注册的 `subagent` 投影单元供给每个 child 的持久化模式与标签:在线 child 取注册表的水位快照;冷 child 先取可选投影缓存的持久化行,且仅当其 `seq` 门证明该值折叠自 child 自身后缀(fork 种子之后——自有描述符一经追加即不可变)才直接采用,否则经一次有界并发的持久化 inspect 再经注册表折叠,且 inspect 结果必须仍指向枚举时的生命周期(同 id 被重新发布的会话降级为 `corrupt` diagnostic)。缓存读取抛错不产生判决——缓存是派生数据——静默落到该权威重折。投影折叠是唯一的分类权威;列表自身不解析任何描述符。取得身份值即产出 child 行;已定局而折叠未产出身份的候选是 `corrupt` diagnostic,inspect 失败是瞬时的 `unavailable`(下次列表重试),运行中而暂无身份值的候选整行省略(描述符尚未追加的创建窗口)。它不查询继续执行管理器、Agent 注册信息、Activation 或提供方。每个 child 行都会根据合并结果中携带持久化 `origin: 'subagent'` 的 header 派生读取时的 `hasChildren` 提示;它不会读取后代事件日志,展开后仍以描述符支撑的 child 目录为权威依据。UI 等服务消费方可以保留两种模式,并为无标签的一次性 child 选择回退展示;面向模型的 `list_agents` 工具只投影 `continuable` 条目,并将服务活动状态映射到现有的 `running`/`complete` 词汇。列表操作会把调用方的取消信号转发到每次持久化读取,在这些 await 前后检查取消,并将每次检测到的中止报告为 `SubagentError` 错误码 `CANCELLED`;投影注册表未挂载则以 `SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE` 响亮失败,会话存储缺失则以 `SUBAGENT_CONTROL_SESSION_STORE_UNAVAILABLE` 响亮失败。完整契约见[后台 subagent 任务 Agent Note](../../../.agents/notes/implemented/feature/2026-07-08-background-subagent-tasks.md)、[可继续后台 subagent Agent Note](../../../.agents/notes/implemented/feature/2026-07-21-continuable-background-subagents.md)、[持久化目录 Agent Note](../../../.agents/notes/implemented/feature/2026-07-22-durable-subagent-catalog-and-list-agents.md)、[服务合并 Agent Note](../../../.agents/notes/implemented/simplification/2026-07-26-merge-subagent-control-service.md)、[能力 seam Agent Note](../../../.agents/notes/implemented/feature/2026-06-21-subagent-capability-seam.md)和 `src/types.ts`。 +面向模型的工具默认同步收集:先等待子 agent 结果,再 dispose 运行,然后才返回。一次性后台委派会在工具中注册普通 Task,其通用状态、收集和取消工具负责后续交互,并将模型提供的 `description` 持久化为可选显示标签。可继续后台委派会调用 `ctx.subagents.startContinuable()`,只返回持久化子 agent id;子 agent 自 inbox 接受起就拥有自己的轮次,因此没有 Task 或结果 promise——调用方通过 `send_message` 后续操作工具发送后续工作,而 `interrupt()` 只停止当前轮次,不 dispose 子 agent。持久化子 agent Session 仍是子 agent 详细输出的来源。只有 `ctx.agents` 可用时,继续执行管理器才会存在,而会话持久化按每项继续执行操作解析。与此独立,`listChildren()` 枚举在线会话存储与可选会话持久化的在线优先合并——持久化缺席时仅枚举在线 child,因为那时冷 child 本就无法恢复——并由已注册的 `subagent` 投影单元供给每个 child 的持久化模式与标签:在线 child 取注册表的水位快照;冷 child 先取可选投影缓存的持久化行,且仅当其 `seq` 门证明该值折叠自 child 自身后缀(fork 种子之后——自有描述符一经追加即不可变)才直接采用,否则经一次有界并发的持久化 inspect 再经注册表折叠,且 inspect 结果必须仍指向枚举时的生命周期(同 id 被重新发布的会话降级为 `corrupt` diagnostic)。缓存读取抛错不产生判决——缓存是派生数据——静默落到该权威重折。投影折叠是唯一的分类权威;列表自身不解析任何描述符。取得身份值即产出 child 行;已定局而折叠未产出身份的候选是 `corrupt` diagnostic,inspect 失败是瞬时的 `unavailable`(下次列表重试),运行中而暂无身份值的候选整行省略(描述符尚未追加的创建窗口)。它不查询继续执行管理器、Agent 注册信息、Activation 或提供方。每个 child 行都会根据合并结果中携带持久化 `origin: 'subagent'` 的 header 派生读取时的 `hasChildren` 提示;它不会读取后代事件日志,展开后仍以描述符支撑的 child 目录为权威依据。UI 等服务消费方可以保留两种模式,并为无标签的一次性 child 选择回退展示;面向模型的 `list_agents` 工具只投影 `continuable` 条目,并将服务活动状态映射到现有的 `running`/`complete` 词汇。列表操作会把调用方的取消信号转发到每次持久化读取,在这些 await 前后检查取消,并将每次检测到的中止报告为 `SubagentError` 错误码 `CANCELLED`;投影注册表未挂载则以 `SUBAGENT_CONTROL_PROJECTIONS_UNAVAILABLE` 响亮失败,会话存储缺失则以 `SUBAGENT_CONTROL_SESSION_STORE_UNAVAILABLE` 响亮失败。完整契约见[后台 subagent 任务 Agent Note](../../../.agents/notes/implemented/feature/2026-07-08-background-subagent-tasks.md)、[可继续后台 subagent Agent Note](../../../.agents/notes/implemented/feature/2026-07-21-continuable-background-subagents.md)、[持久化目录 Agent Note](../../../.agents/notes/implemented/feature/2026-07-22-durable-subagent-catalog-and-list-agents.md)、[服务合并 Agent Note](../../../.agents/notes/implemented/simplification/2026-07-26-merge-subagent-control-service.md)、[能力 seam Agent Note](../../../.agents/notes/implemented/feature/2026-06-21-subagent-capability-seam.md)和 `src/types.ts`。 可继续 Activation 会等待 best-effort 的最终会话 flush,但不会把 listener 参与视为持久性确认。一次性运行保留尽力执行的会话检查点,因此已完成的一次性 child 只有在其会话确实进入持久化存储时,才可在 dispose 后继续被发现;如果该检查点缺失,服务不会根据 Task 历史虚构目录条目。 @@ -99,7 +100,7 @@ subagent seam 允许一个 agent(智能体)通过具名提供方把工作委 ## 已知限制与暂缓事项 - **ACP 子 agent 仍为一次性,且无法通过追踪枚举**:ACP 运行在 parent 会话语料中没有本地 child 会话。ACP 的 `prepareContinuable` 需要在提供方专用描述符数据中持久化远端会话 id,以及逐子 agent 的继续执行能力声明,因为 ACP 的 `loadSession` 支持按子 agent 协商,而不是通过方法是否存在来确定。远程提供方还需要一份独立的 Activation 所有权契约,具备等效的经认证控制和子先于父的停稳保证,才能支持可继续子 agent。 -- **无 host-user 继续执行**:`followup()` 要求确切在线直接父级。未来 host 适配器需要具体的经认证交互,才能让该 seam 获得单独的用户能力。 +- **无 host-user 继续执行**:`followup()` 要求确切在线直接父级。只有 `interrupt()` 接受持久化 parent 地址形式的用户授权,因为停止一个轮次是幂等的且不投递任何内容;未来 host 适配器需要具体的经认证交互,才能让该 seam 获得用户投递能力。 - **不对当前轮次进行 steering**:可继续消息和唤醒式 report 会排入后续轮次,均不会重定向正在进行的轮次。 - **驻留仅限进程内**:Activation inbox 与所有权图不会在两个 harness 进程之间协调;对单个持久化存储的并发访问仍然需要持久化邮箱和跨进程租约协议。 - **不重放已接受但未记录的消息**:只有写入子 agent Session 日志的消息才能连同其被接受时的来源一起重建。崩溃可能丢失从未写入日志、已被接受的初始提示词或后续消息;此后一条经授权的消息可以冷恢复该子 agent,但丢失的消息不会自动重放。 diff --git a/packages/subagent/subagent/src/continuation.ts b/packages/subagent/subagent/src/continuation.ts index 212536e713..ebac0a1296 100644 --- a/packages/subagent/subagent/src/continuation.ts +++ b/packages/subagent/subagent/src/continuation.ts @@ -103,6 +103,15 @@ export interface ContinuableStart { readonly messageId: MessageId } +/** + * Authority under which one interrupt request is admitted. `user` carries the + * durable direct-parent address a human client presented; `ancestor` carries + * the exact live Agent object whose recorded lineage must contain the caller. + */ +export type SubagentInterruptAuthority = + | { readonly kind: 'user'; readonly parentSessionId: SessionId } + | { readonly kind: 'ancestor'; readonly agent: Agent } + /** Options for following up with one continuable child. */ export interface SubagentFollowupOptions { /** Durable attribution retained on the delivered message; it grants no authority. */ @@ -412,6 +421,68 @@ export class SubagentContinuationManager { } } + /** + * Interrupt one live continuable child's current turn. Admission is + * synchronous and the effect is asynchronous: this authorizes the caller, + * requests `Agent.cancel(cause, { keepInbox: true })` on the target, and + * returns without waiting for the target to observe the signal or reach + * quiescence. The Activation, its handle, accepted pending inbox work, and + * already-published descendants are untouched; the parked queue resumes only + * on a later waking send. + * + * An absent target is an accepted no-op, which uniformly covers natural + * completion races, repeated requests, one-shot ids, and unknown ids without + * consulting the durable catalog. A target whose disposal transaction is + * already open is likewise an accepted no-op after authorization. + * @param targetSessionId - the durable child session id to interrupt. + * @param authority - the human parent address or exact live ancestor Agent. + * @throws {SubagentError} `UNAUTHORIZED` when the presented authority does + * not own the live target: a stale or self-targeting ancestor caller, a + * parent address that is not the live target's durable direct parent, or + * an ancestor outside the target's recorded live lineage. + */ + interrupt(targetSessionId: SessionId, authority: SubagentInterruptAuthority): void { + if (authority.kind === 'ancestor') { + const caller = authority.agent + // A stale caller is rejected even when the target is absent, so a + // replaced same-id Agent can never probe this manager's state. + if (this.ctx.agents.get(caller.id) !== caller) { + throw new SubagentError( + `interrupting "${targetSessionId}" requires the exact live ancestor agent`, + 'UNAUTHORIZED', + ) + } + if (caller.id === targetSessionId) { + throw new SubagentError( + `agent "${caller.id}" cannot interrupt itself`, + 'UNAUTHORIZED', + ) + } + } + const activation = this.activations.get(targetSessionId) + if (activation === undefined) return + if (authority.kind === 'user') { + if (activation.handle.agent.session.header.parentSession !== authority.parentSessionId) { + throw new SubagentError( + `subagent "${targetSessionId}" belongs to another parent session`, + 'UNAUTHORIZED', + ) + } + } else if (!activation.ancestry.has(authority.agent)) { + throw new SubagentError( + `subagent "${targetSessionId}" is not a live descendant of agent "${authority.agent.id}"`, + 'UNAUTHORIZED', + ) + } + // Disposal already stopped the target with a whole-Activation teardown; + // a second cancel would be a redundant signal on a closing handle. + if (activation.disposal !== undefined) return + activation.handle.agent.cancel( + authority.kind === 'user' ? { kind: 'user' } : { kind: 'parent' }, + { keepInbox: true }, + ) + } + /** * Deliver explicitly selected content from one resident continuable child to * its durable direct parent. Sender authorization, parent resolution, and diff --git a/packages/subagent/subagent/src/index.ts b/packages/subagent/subagent/src/index.ts index 5bcd53d6a5..e179bfd8b4 100644 --- a/packages/subagent/subagent/src/index.ts +++ b/packages/subagent/subagent/src/index.ts @@ -58,6 +58,7 @@ import type { ContinuableStart, ContinuableStartSpec, SubagentFollowupOptions, + SubagentInterruptAuthority, SubagentReportOptions, } from './continuation.ts' import SubagentActivationSetupRegistry from './activation-setup-registry.ts' @@ -111,6 +112,7 @@ export type { ContinuableStartSpec, CoordinatorMessageSource, SubagentFollowupOptions, + SubagentInterruptAuthority, SubagentReportDelivery, SubagentReportMessageSource, SubagentReportOptions, @@ -231,6 +233,24 @@ export class SubagentService extends Service { return this.requireContinuations().followup(parent, childId, content, options) } + /** + * Interrupt one live continuable child's current turn under a human parent + * address or an exact live ancestor Agent. Fire-and-return: the cancel + * signal is issued before this returns, but the target may keep running + * until it observes the signal. Pending inbox work, the Activation, and + * published descendants are preserved; only a later waking send resumes the + * parked FIFO queue. An absent target — including a one-shot or unknown id — + * is an accepted no-op, as is a manager-less composition, which cannot own a + * live Activation. + * @param targetSessionId - the durable child session id to interrupt. + * @param authority - the human parent address or exact live ancestor Agent. + * @throws {SubagentError} `UNAUTHORIZED` when the authority does not own the + * live target. + */ + interrupt(targetSessionId: SessionId, authority: SubagentInterruptAuthority): void { + this.continuations?.interrupt(targetSessionId, authority) + } + /** * Deliver selected content from one live continuable child to its durable * direct parent. The child is the authority credential; callers cannot name a diff --git a/packages/subagent/subagent/tests/continuation.spec.ts b/packages/subagent/subagent/tests/continuation.spec.ts index dca06add38..9370676f76 100644 --- a/packages/subagent/subagent/tests/continuation.spec.ts +++ b/packages/subagent/subagent/tests/continuation.spec.ts @@ -1717,3 +1717,223 @@ describe('continuable errors', () => { expect(ctx.agents.get(started.childId)).toBeUndefined() }) }) + +describe('SubagentService.interrupt', () => { + it('aborts the current turn durably, parks accepted follow-ups, and resumes them only on a waking send', async () => { + const releaseFirst = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('first'), gate: releaseFirst.promise }, + { chunks: textResponse('second') }, + { chunks: textResponse('third') }, + { chunks: textResponse('fourth') }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId)! + await followup(ctx, parent, started.childId, message('parked B')) + await followup(ctx, parent, started.childId, message('parked C')) + const cancelSpy = vi.spyOn(child, 'cancel') + + ctx.subagents.interrupt(started.childId, { kind: 'user', parentSessionId: parent.id }) + + expect(cancelSpy).toHaveBeenCalledTimes(1) + expect(cancelSpy).toHaveBeenCalledWith({ kind: 'user' }, { keepInbox: true }) + // Cancellation is cooperative: the held model call observes it on release. + releaseFirst.resolve(undefined) + await child.whenIdle() + // Parked, not resumed: no second model request follows the abort, the + // accepted follow-ups stay pending, and the same Activation stays resident. + expect(adapter.requests).toHaveLength(1) + expect(child.inbox.nextTurn).toHaveLength(2) + expect(child.status).toBe('idle') + expect(ctx.agents.get(started.childId)).toBe(child) + + // Only an explicit waking send restores the driver; the parked items then + // run before it in the existing FIFO order. + await followup(ctx, parent, started.childId, message('waking D')) + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(started.childId) + expect(userTexts(loaded.events)).toEqual(['child task', 'parked B', 'parked C', 'waking D']) + const turnEnds = loaded.events + .filter(event => event.type === 'turn/end') + .map(event => (event).data.reason.kind) + expect(turnEnds).toEqual(['aborted', 'completed', 'completed', 'completed']) + }) + + it('interrupts only the target while its resident descendant keeps running', async () => { + const releaseChild = Promise.withResolvers() + const releaseGrandchild = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('child'), gate: releaseChild.promise }, + { chunks: textResponse('grandchild'), gate: releaseGrandchild.promise }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId)! + const grandchild = await ctx.subagents.startContinuable(startSpec(child)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) }) + const grandchildAgent = ctx.agents.get(grandchild.childId)! + const childCancel = vi.spyOn(child, 'cancel') + const grandchildCancel = vi.spyOn(grandchildAgent, 'cancel') + + ctx.subagents.interrupt(started.childId, { kind: 'user', parentSessionId: parent.id }) + + expect(childCancel).toHaveBeenCalledTimes(1) + releaseChild.resolve(undefined) + await child.whenIdle() + // The target parks as a waiting owner; the published descendant was never + // signalled and keeps its own turn open. + expect(grandchildCancel).not.toHaveBeenCalled() + expect(ctx.agents.get(started.childId)).toBe(child) + expect(ctx.agents.get(grandchild.childId)).toBe(grandchildAgent) + + releaseGrandchild.resolve(undefined) + await waitNoActivation(ctx, grandchild.childId) + await waitNoActivation(ctx, started.childId) + const loaded = await ctx.sessionPersistence.load(grandchild.childId) + const turnEnds = loaded.events + .filter(event => event.type === 'turn/end') + .map(event => (event).data.reason.kind) + expect(turnEnds).toEqual(['completed']) + }) + + it('authorizes the human address against the live target\'s durable direct parent', async () => { + const hold = Promise.withResolvers() + const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: hold.promise }]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId)! + const cancelSpy = vi.spyOn(child, 'cancel') + + expect(() => { ctx.subagents.interrupt(started.childId, { + kind: 'user', + parentSessionId: SessionId('stranger'), + }) }).toThrow(/belongs to another parent session/) + expect(cancelSpy).not.toHaveBeenCalled() + + ctx.subagents.interrupt(started.childId, { kind: 'user', parentSessionId: parent.id }) + expect(cancelSpy).toHaveBeenCalledWith({ kind: 'user' }, { keepInbox: true }) + hold.resolve(undefined) + await waitNoActivation(ctx, started.childId) + }) + + it('lets a deep exact live ancestor interrupt its descendant with the parent cause', async () => { + const releaseChild = Promise.withResolvers() + const releaseGrandchild = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('child'), gate: releaseChild.promise }, + { chunks: textResponse('grandchild'), gate: releaseGrandchild.promise }, + ]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId)! + const grandchild = await ctx.subagents.startContinuable(startSpec(child)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) }) + const grandchildAgent = ctx.agents.get(grandchild.childId)! + const childCancel = vi.spyOn(child, 'cancel') + const grandchildCancel = vi.spyOn(grandchildAgent, 'cancel') + + // Deep ancestor: the top-level parent interrupts the grandchild. + ctx.subagents.interrupt(grandchild.childId, { kind: 'ancestor', agent: parent }) + expect(grandchildCancel).toHaveBeenCalledWith({ kind: 'parent' }, { keepInbox: true }) + // Direct ancestor: the same authority kind covers the immediate parent. + ctx.subagents.interrupt(started.childId, { kind: 'ancestor', agent: parent }) + expect(childCancel).toHaveBeenCalledWith({ kind: 'parent' }, { keepInbox: true }) + + releaseChild.resolve(undefined) + releaseGrandchild.resolve(undefined) + await waitNoActivation(ctx, grandchild.childId) + await waitNoActivation(ctx, started.childId) + }) + + it('rejects self, sibling, stale, and unrelated ancestor callers without touching the target', async () => { + const releaseA = Promise.withResolvers() + const releaseB = Promise.withResolvers() + const adapter = new GatedAdapter([ + { chunks: textResponse('a'), gate: releaseA.promise }, + { chunks: textResponse('b'), gate: releaseB.promise }, + ]) + const { ctx, parent } = await setupWith(adapter) + const targetStart = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const target = ctx.agents.get(targetStart.childId)! + const siblingStart = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(2) }) + const sibling = ctx.agents.get(siblingStart.childId)! + const stranger = ctx.agentLoop.create(SessionId('stranger'), { provider: 'mock', model: 'mock' }) + const stale = { ...parent, id: parent.id } as unknown as Agent + const cancelSpy = vi.spyOn(target, 'cancel') + + expect(() => { ctx.subagents.interrupt(targetStart.childId, { kind: 'ancestor', agent: target }) }) + .toThrow(/cannot interrupt itself/) + expect(() => { ctx.subagents.interrupt(targetStart.childId, { kind: 'ancestor', agent: sibling }) }) + .toThrow(/not a live descendant/) + expect(() => { ctx.subagents.interrupt(targetStart.childId, { kind: 'ancestor', agent: stranger }) }) + .toThrow(/not a live descendant/) + expect(() => { ctx.subagents.interrupt(targetStart.childId, { kind: 'ancestor', agent: stale }) }) + .toThrow(/exact live ancestor/) + // A stale caller is rejected before target lookup, even for an absent id. + expect(() => { ctx.subagents.interrupt(SessionId('missing'), { kind: 'ancestor', agent: stale }) }) + .toThrow(/exact live ancestor/) + expect(cancelSpy).not.toHaveBeenCalled() + + releaseA.resolve(undefined) + releaseB.resolve(undefined) + await waitNoActivation(ctx, targetStart.childId) + await waitNoActivation(ctx, siblingStart.childId) + }) + + it('accepts absent and one-shot ids as no-ops without touching the one-shot Agent', async () => { + const { ctx, parent } = await setup([textResponse('one shot')]) + ctx.subagents.interrupt(SessionId('missing'), { kind: 'user', parentSessionId: parent.id }) + ctx.subagents.interrupt(SessionId('missing'), { kind: 'ancestor', agent: parent }) + + const run = await ctx.subagents.start('spawn', { + label: 'one-shot work', + prompt: message('one-shot work'), + parent, + signal: testSignal, + }) + const oneShot = run.localAgent! + const cancelSpy = vi.spyOn(oneShot, 'cancel') + ctx.subagents.interrupt(run.id, { kind: 'user', parentSessionId: parent.id }) + ctx.subagents.interrupt(run.id, { kind: 'ancestor', agent: parent }) + expect(cancelSpy).not.toHaveBeenCalled() + await run.result + await run.dispose() + }) + + it('accepts an interrupt after natural completion', async () => { + const { ctx, parent } = await setup([textResponse('done')]) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await waitNoActivation(ctx, started.childId) + ctx.subagents.interrupt(started.childId, { kind: 'user', parentSessionId: parent.id }) + ctx.subagents.interrupt(started.childId, { kind: 'ancestor', agent: parent }) + }) + + it('accepts an interrupt that lost the race with disposal without signalling twice', async () => { + const hold = Promise.withResolvers() + const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: hold.promise }]) + const { ctx, parent } = await setupWith(adapter) + const started = await ctx.subagents.startContinuable(startSpec(parent)) + await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) }) + const child = ctx.agents.get(started.childId)! + const cancelSpy = vi.spyOn(child, 'cancel') + + // Scoped teardown opens the disposal transaction synchronously and issues + // its own whole-Activation cancel before this call returns. + const drained = ctx.subagents.drainContinuableDescendants([parent]) + expect(cancelSpy).toHaveBeenCalledTimes(1) + + // Interrupt after the cutoff: accepted no-op, no second signal, no waiting. + ctx.subagents.interrupt(started.childId, { kind: 'user', parentSessionId: parent.id }) + expect(cancelSpy).toHaveBeenCalledTimes(1) + + hold.resolve(undefined) + await drained + }) +}) diff --git a/packages/subagent/subagent/tests/service.spec.ts b/packages/subagent/subagent/tests/service.spec.ts index 8b90719227..a50696cf2a 100644 --- a/packages/subagent/subagent/tests/service.spec.ts +++ b/packages/subagent/subagent/tests/service.spec.ts @@ -133,6 +133,16 @@ describe('SubagentService', () => { await expect(subagents.drainContinuableDescendants([])).resolves.toBeUndefined() }) + it('treats interrupt as an accepted no-op when no manager was bound', async () => { + const { subagents } = await service() + // Without a continuation manager no live Activation can exist, so there is + // nothing to stop and nothing to authorize against. + expect(() => { subagents.interrupt(SessionId('child'), { + kind: 'user', + parentSessionId: SessionId('parent-1'), + }) }).not.toThrow() + }) + it('rejects continuable operations when their runtime services are absent', async () => { const { subagents } = await service() await expect(subagents.startContinuable({ diff --git a/scripts/gen-cordis-catalog.ts b/scripts/gen-cordis-catalog.ts index 54581a8605..c8ddb487ef 100644 --- a/scripts/gen-cordis-catalog.ts +++ b/scripts/gen-cordis-catalog.ts @@ -173,6 +173,7 @@ export const LINK_MAP: Readonly> = { ContinuableStartSpec: 'subagent.md', CoordinatorMessageSource: 'subagent.md', SubagentFollowupOptions: 'subagent.md', + SubagentInterruptAuthority: 'subagent.md', SubagentListEntry: 'subagent.md', SubagentProvider: 'subagent.md', SubagentReportDelivery: 'subagent.md',