diff --git a/.agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md b/.agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md index 125ef0946c..ef5bb5847a 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md +++ b/.agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md @@ -12,7 +12,7 @@ Three seams: the queue-aware cancel, the `AgentHandle` disposer, and the bash ow ### 1. Queue-aware `Agent.cancel(reason?)` -A new `cancel()` verb on the `Agent` interface — the single public stop primitive. (It originally shipped alongside a narrower step-only `abort()`; that verb was later removed as unused, leaving `cancel()` the only public way to stop work.) It clears the inbox's queued + steering FIFOs, aborts the in-flight step if any, and drives a **turn-scoped cancellation marker** the driver loop checks at every turn-decision point — so a prompt that is queued-but-not-yet-started never runs, a cancel landing in the pre-step / continuation window drops the about-to-run turn (ending it `aborted`), and a later prompt cannot be batched into the cancelled turn. `whenIdle()` reaches post-cancel quiescence. ACP `session/cancel` maps to `cancel()`. The marker is armed ONLY when there is something to cancel, so an idle no-op cancel cannot strand the next prompt. +A new `cancel()` verb on the `Agent` interface — the single public stop primitive. (It originally shipped alongside a narrower step-only `abort()`; that verb was later removed as unused, leaving `cancel()` the only public way to stop work.) It clears the inbox's queued + steering FIFOs, aborts the in-flight step if any, and drives a **turn-scoped cancellation marker** the driver loop checks at every turn-decision point — so a prompt that is queued-but-not-yet-started never runs, a cancel landing in the pre-step / continuation window drops the about-to-run turn (ending it `aborted`), and a later accepted prompt remains an independent queued turn. `whenIdle()` reaches post-cancel quiescence. ACP `session/cancel` maps to `cancel()`. The marker is armed ONLY when there is something to cancel, so an idle no-op cancel cannot strand the next prompt. ### 2. `AgentHandle` async disposer @@ -29,7 +29,7 @@ Background-task ownership moved from a `tool-bash` plugin-local `Map-session-`; `sessionId` resumes or creates; `resumeSessionId` requires history. Active failures emit `agent-loop/config-start-failed(sessionId, error)`, so front doors reject work; teardown stays silent. @@ -69,13 +69,13 @@ choose declarative identity and fresh/resume path -> enter session + agent -> session/created -> agent/created -> enable driving -> agent/session-start(source) -> start driver forever: - wait for queued messages + wait for a queued message emit agent/status(running) TURN: 'turn/start' - each queued message -> agent/prompt-submit + claimed message -> agent/prompt-submit allowed prompt -> 'user/message' plus injected context - every prompt blocked -> 'turn/end'(rejected) + blocked prompt -> 'prompt/blocked' -> 'turn/end'(rejected) STEP loop: drain steering assemble system prompt and tool schemas diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 357623f1b6..ae17230a7a 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -33,7 +33,7 @@ A fully configured agent and live session were published. Setup is composition-o Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:143`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:150`](../../packages/core/agent/src/types.ts) ### `agent/disposed` — emit @@ -53,7 +53,7 @@ An agent left the registry; AgentLoop emits this after driver quiescence but bef Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:152`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:159`](../../packages/core/agent/src/types.ts) ### `agent/error` — emit @@ -75,7 +75,7 @@ A step or turn errored. The loop reports a failure here (plus the logger) even w Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:307`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:314`](../../packages/core/agent/src/types.ts) ### `agent/post-step` — serial @@ -98,7 +98,7 @@ Awaited serial checkpoint after the response, real or synthetic tool results, in Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:260`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:267`](../../packages/core/agent/src/types.ts) ### `agent/pre-step` — serial @@ -121,18 +121,18 @@ Awaited serial checkpoint before `step/start`; appends land outside the pending Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:200`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:207`](../../packages/core/agent/src/types.ts) ### `agent/prompt-submit` — waterfall -Allow, rewrite, or block one drained prompt before it becomes a user message. Call `next()` for the unchanged default. +Allow, rewrite, or block one claimed prompt before it becomes a user message. Call `next()` for the unchanged default. ```ts cordis-catalog /** - * Allow, rewrite, or block one drained prompt before it becomes a user + * Allow, rewrite, or block one claimed prompt before it becomes a user * message. Call `next()` for the unchanged default. - * @param agent - the agent draining its inbox. - * @param content - the drained message's blocks, as queued. + * @param agent - the agent whose turn claimed the message. + * @param content - the claimed message's blocks, as queued. * @param source - the message's resolved source. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode waterfall @@ -142,7 +142,7 @@ Allow, rewrite, or block one drained prompt before it becomes a user message. Ca Types: [Agent](../core-data-structures/core.md) · [ContentBlock](../core-data-structures/core.md) · [MessageSource](../core-data-structures/core.md) · [PromptDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:210`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:217`](../../packages/core/agent/src/types.ts) ### `agent/queued` — emit @@ -163,7 +163,7 @@ Detached, frozen content entered the agent's inbox. Source defaults have already Types: [Agent](../core-data-structures/core.md) · [ContentBlock](../core-data-structures/core.md) · [MessageSource](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:171`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:178`](../../packages/core/agent/src/types.ts) ### `agent/request` — waterfall @@ -186,7 +186,7 @@ Replace the frozen call configuration. Model-visible content must use logged cha Types: [Agent](../core-data-structures/core.md) · [LlmCallConfig](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:222`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:229`](../../packages/core/agent/src/types.ts) ### `agent/request-error` — waterfall @@ -211,7 +211,7 @@ Recover a model-request failure after its failed step has closed. `retry` opens Types: [Agent](../core-data-structures/core.md) · [RequestError](../core-data-structures/core.md) · [RequestErrorDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:274`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:281`](../../packages/core/agent/src/types.ts) ### `agent/session-prefix` — waterfall @@ -237,7 +237,7 @@ Compose request-only messages placed before derived history. The frozen result i Types: [Agent](../core-data-structures/core.md) · [Message](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:237`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:244`](../../packages/core/agent/src/types.ts) ### `agent/session-start` — emit @@ -259,7 +259,7 @@ The session lifecycle began, once before the first turn. Use `agent.inject()` to Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [SessionStartSource](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:184`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:191`](../../packages/core/agent/src/types.ts) ### `agent/status` — emit @@ -279,7 +279,7 @@ Agent status changed (`idle` ⇄ `running`, or → `disposed`). `send()` does no Types: [Agent](../core-data-structures/core.md) · [AgentStatus](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:161`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:168`](../../packages/core/agent/src/types.ts) ### `agent/step-result` — waterfall @@ -301,7 +301,7 @@ Waterfall: post-process the assembled assistant Message before tool dispatch (va Types: [Agent](../core-data-structures/core.md) · [Message](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:248`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:255`](../../packages/core/agent/src/types.ts) ### `agent/turn-continuation` — waterfall @@ -322,7 +322,7 @@ Override whether the turn continues. The default continues after tool calls or s Types: [Agent](../core-data-structures/core.md) · [ContinuationDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:284`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:291`](../../packages/core/agent/src/types.ts) ### `agent/turn-stop` — serial @@ -343,7 +343,7 @@ Monotonic terminal-stop checkpoint after continuation and steering are folded; a Types: [Agent](../core-data-structures/core.md) · [ContinuationStop](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:294`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:301`](../../packages/core/agent/src/types.ts) ## `agent-loop/*` diff --git a/docs/core-data-structures/core.md b/docs/core-data-structures/core.md index 5b3d5135b2..6adee2e281 100644 --- a/docs/core-data-structures/core.md +++ b/docs/core-data-structures/core.md @@ -360,15 +360,20 @@ interface Agent { readonly ctx: Context /** - * Queue detached, frozen lossless-JSON input; starts a turn when idle. + * Queue one detached, frozen lossless-JSON item. If claimed, it is the sole + * ordinary message in its FIFO-ordered turn; the next claimed item waits for + * that turn's checkpoint. * Invalid input throws synchronously before notification or enqueue. */ send(content: ContentBlock[], options?: SendOptions): void /** - * Steer a running turn: content is injected between steps of the current - * turn. Uses the same owned-value and synchronous-validation boundary as - * {@link send}; when idle, behaves exactly like that method. + * Submit steering while the agent is `running`. An open turn records it at + * the next steering checkpoint before a request or continuation decision; + * policy may stop before another step. After turn close and its checkpoint, + * any remainder is queued for a later turn; terminal `agent/turn-stop`, + * cancellation, or disposal may discard it. Uses the same synchronous + * snapshot-and-validation boundary as {@link send}; when idle, delegates to it. */ steer(content: ContentBlock[], options?: SendOptions): void @@ -382,10 +387,11 @@ interface Agent { inject(content: ContentBlock[], options?: InjectOptions): void /** - * Clear queued and steering work, including work waiting to start, and abort - * the active step. The supplied reason is preserved across pre-step and active - * cancellation windows, and `whenIdle()` resolves after cancellation reaches - * quiescence. Idle cancellation is a no-op and does not arm a later cancel. + * Clear all queued and steering work, including items waiting to start, and + * abort the active step. The supplied reason is preserved across pre-step + * and active cancellation windows, and `whenIdle()` resolves after + * cancellation reaches quiescence. Idle cancellation is a no-op and does not + * arm a later cancel. */ cancel(reason?: string): void @@ -395,7 +401,7 @@ interface Agent { } ``` -`AgentStatus` is `'idle' | 'running' | 'disposed'`, and `SessionId` is branded. `AgentOptions` is merge-extensible and currently includes `provider?` and `model?`; dispatch requires both after `agent/request`. Persona belongs to `dsh-system-prompt`: an agent-scoped `deployment:persona` may shadow the global default. +`AgentStatus` is `'idle' | 'running' | 'disposed'`, and `SessionId` is branded. `running` describes the driver-wide drain interval, which can span turn close, its durability checkpoint, and consecutive queued turns; it does not prove a turn is still open. `AgentOptions` is merge-extensible and currently includes `provider?` and `model?`; dispatch requires both after `agent/request`. Persona belongs to `dsh-system-prompt`: an agent-scoped `deployment:persona` may shadow the global default. The [event taxonomy](../architecture.md#event) owns the `agent/*` lifecycle, checkpoint, and waterfall contracts. Turn and step boundaries are durable session events rather than agent emits. @@ -419,13 +425,14 @@ interface HookContext { } ``` -`agent/prompt-submit` returns a `PromptDecision` (allow a drained queued message — optionally rewriting its `content` or attaching `additionalContexts` — or block it; a batch whose every prompt is blocked opens a zero-step turn that ends `rejected`): +`agent/prompt-submit` returns a `PromptDecision` (allow the turn's claimed queued message — optionally rewriting its `content` or attaching `additionalContexts` — or record `prompt/blocked` and end that zero-step turn as `rejected`): ```ts type-equiv /** * Prompt interception result. `allow.content` replaces the prompt and each - * `additionalContexts` entry becomes a separate context message. `block` records a - * durable `prompt/blocked`; an all-blocked batch ends a zero-step rejected turn. + * `additionalContexts` entry becomes a separate context message. `block` + * records a durable `prompt/blocked` and ends the claimed prompt's zero-step + * turn as rejected. */ type PromptDecision = | { kind: 'allow'; content?: ContentBlock[]; additionalContexts?: HookContext[] } diff --git a/docs/core-data-structures/persistence.md b/docs/core-data-structures/persistence.md index a80b9bc896..18ef172ef8 100644 --- a/docs/core-data-structures/persistence.md +++ b/docs/core-data-structures/persistence.md @@ -6,7 +6,7 @@ The seam is a textbook [capability seam](../../.agents/notes/implemented/archite ## The flush checkpoint -`session/event` is a *synchronous* notification; persistence plugins buffer it (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. Flush is `ctx.parallel` (awaited): a turn's events are durably committed before the next turn starts, and the turn boundary is the commit boundary. A rejecting flush is reported via `agent/error` and the logger — never as a session event (it would land past the commit boundary), so the backend keeps its buffered events for the next flush. +`session/event` is a *synchronous* notification; persistence plugins buffer it (write-behind) until `session/flush`. The loop awaits an ordinary turn's checkpoint before claiming the next queue item; synchronous idle `inject()` schedules its checkpoint without blocking `send()`, and disposal still drains it. A successful flush durably commits the closed turn as one unit; a rejecting flush is reported through `agent/error` and the logger — never as a session event past the closed turn — while the backend keeps its buffered events for the next flush. ## Crash recovery preserves an interrupted turn diff --git a/docs/core-data-structures/session.md b/docs/core-data-structures/session.md index 502da9d00d..d1c97b1fae 100644 --- a/docs/core-data-structures/session.md +++ b/docs/core-data-structures/session.md @@ -17,27 +17,28 @@ The append-only event types. Merge-extensible: a plugin declares extra event typ */ interface SessionEventMap { /** - * Opens turn `turn`. `trigger` records what started it — a drained message - * batch or an idle-time injection. The turn is the durability/replay + * Opens turn `turn`. `trigger` records what started it — one claimed queued + * message or an idle-time injection. The turn is the durability/replay * boundary: every event sits between a `turn/start` and its matching * `turn/end` (the turn-enclosure invariant). */ 'turn/start': { turn: number; trigger: TurnTrigger } /** * Closes turn `turn` with the {@link TurnEndReason} that ended it. The loop - * fires the awaited `session/flush` checkpoint at every turn end, so the turn - * boundary is also the durable-commit boundary. + * awaits `session/flush` after an ordinary turn ends before claiming the next + * queued item. Success commits the turn; rejection is reported live and does + * not prevent later work. */ 'turn/end': { turn: number; reason: TurnEndReason } /** Opens step `step` of turn `turn` — one model call plus the tool executions it requested. */ 'step/start': { turn: number; step: number } /** Closes step `step` of turn `turn`. */ 'step/end': { turn: number; step: number } - /** A user-visible prompt (queued message drained at turn start). */ + /** A user-visible prompt (the queued message claimed for this turn). */ 'user/message': { content: ContentBlock[]; source: MessageSource } /** * Durable record of a prompt veto and its reason. It is log-only: the blocked - * prompt never enters the model-visible surface, including in a mixed batch. + * prompt never enters the model-visible surface, and its turn runs zero steps. */ 'prompt/blocked': { content: ContentBlock[]; source: MessageSource; reason: string } /** @@ -480,8 +481,8 @@ interface TurnEndReasonMap { /** At least one step reached its output-token ceiling, even if a plugin continued the turn. */ 'max-tokens': { kind: 'max-tokens' } /** - * Policy blocked every prompt before the first step. The zero-step turn still - * records a balanced durable boundary and the veto reason. + * Policy blocked the turn's claimed prompt before the first step. The + * zero-step turn still records a balanced durable boundary and veto reason. */ rejected: { kind: 'rejected'; reason: string } /** @@ -492,7 +493,7 @@ interface TurnEndReasonMap { } ``` -`max-tokens` mirrors the model-call `FinishReason` of the same name: any `max-tokens` step in a turn makes the whole turn end `max-tokens` rather than `completed` (the cut-short fact wins over a later continuation), so a consumer can tell a clean stop from a truncated one — but only over `completed`: the `disposed`/`aborted`/`error` outcomes take precedence. `rejected` is a zero-step turn whose whole prompt batch an `agent/prompt-submit` hook blocked (the ACP bridge maps it to `cancelled`). `interrupted` is the one reason no loop emits — it is synthesized by crash recovery (see [persistence.md](persistence.md)). Both maps are merge-extensible. +`max-tokens` mirrors the model-call `FinishReason` of the same name: any `max-tokens` step in a turn makes the whole turn end `max-tokens` rather than `completed` (the cut-short fact wins over a later continuation), so a consumer can tell a clean stop from a truncated one — but only over `completed`: the `disposed`/`aborted`/`error` outcomes take precedence. `rejected` is a zero-step turn whose claimed prompt an `agent/prompt-submit` hook blocked (the ACP bridge maps it to `cancelled`). `interrupted` is the one reason no loop emits — it is synthesized by crash recovery (see [persistence.md](persistence.md)). Both maps are merge-extensible. ## The turn-enclosure invariant diff --git a/docs/defensive-patterns.md b/docs/defensive-patterns.md index cf30072094..fe74a9d19f 100644 --- a/docs/defensive-patterns.md +++ b/docs/defensive-patterns.md @@ -12,7 +12,7 @@ When an interface documents two valid ways to signal something — an adapter ma ## Async state is not synchronous state -`agent.send()` does not flip status before returning; a background task's completion races turn boundaries; `reader.close()` fires for both EOF and disposal. Never gate control flow on a status you only just requested — drive lifecycle off the events/promises that actually fire (`agent/status`, `task.done`), and observe the transition (saw `running` THEN `idle`) rather than counting actions you assume map 1:1 to turns (the loop batches queued messages). The guard cuts both ways: if the awaited transition can never occur (EOF with no work submitted → never `running`), the wait hangs — handle the "nothing to wait for" branch explicitly. +`agent.send()` does not flip status before returning; a background task's completion races turn boundaries; `reader.close()` fires for both EOF and disposal. Never gate control flow on a status you only just requested — drive lifecycle off the events/promises that actually fire (`agent/status`, `task.done`), and observe the transition (saw `running` THEN `idle`) instead of treating status as a per-send result: several queued sends run as consecutive turns under one `running` interval, while cancellation or disposal can discard unstarted items. The guard cuts both ways: if the awaited transition can never occur (EOF with no work submitted → never `running`), the wait hangs — handle the "nothing to wait for" branch explicitly. ## Dispose must reach quiescence, not just request it diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index ed028010d9..b678c4b7e2 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -8,21 +8,21 @@ This matrix shows which packages dispatch each harness-owned event and which pac | Event | Mode | Declared in | Dispatchers | Listeners | | --- | --- | --- | --- | --- | | `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:362`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | -| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:143`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | -| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:152`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | -| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:307`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`tui`](../packages/ui/tui) | -| `agent/post-step` | `serial` | [`packages/core/agent/src/types.ts:260`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic) | -| `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:200`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`time-context`](../packages/context/time-context), [`user-approval`](../packages/ui/user-approval) | -| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:210`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard) | -| `agent/queued` | `emit` | [`packages/core/agent/src/types.ts:171`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | -| `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:222`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp) | -| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:274`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic) | -| `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:237`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`tool-skill`](../packages/skill/tool-skill), [`workspace-context`](../packages/context/workspace-context) | -| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:184`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`stdio`](../packages/ui/stdio) | -| `agent/status` | `emit` | [`packages/core/agent/src/types.ts:161`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`invariants`](../packages/support/invariants), [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | -| `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:248`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | -| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:284`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | -| `agent/turn-stop` | `serial` | [`packages/core/agent/src/types.ts:294`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`subagent-inprocess`](../packages/subagent/subagent-inprocess) | +| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:150`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | +| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:159`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | +| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:314`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`tui`](../packages/ui/tui) | +| `agent/post-step` | `serial` | [`packages/core/agent/src/types.ts:267`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic) | +| `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:207`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`time-context`](../packages/context/time-context), [`user-approval`](../packages/ui/user-approval) | +| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:217`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard) | +| `agent/queued` | `emit` | [`packages/core/agent/src/types.ts:178`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | +| `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:229`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp) | +| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:281`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic) | +| `agent/session-prefix` | `waterfall` | [`packages/core/agent/src/types.ts:244`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`tool-skill`](../packages/skill/tool-skill), [`workspace-context`](../packages/context/workspace-context) | +| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:191`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`stdio`](../packages/ui/stdio) | +| `agent/status` | `emit` | [`packages/core/agent/src/types.ts:168`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`invariants`](../packages/support/invariants), [`stdio`](../packages/ui/stdio), [`tui`](../packages/ui/tui) | +| `agent/step-result` | `waterfall` | [`packages/core/agent/src/types.ts:255`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | - | +| `agent/turn-continuation` | `waterfall` | [`packages/core/agent/src/types.ts:291`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | +| `agent/turn-stop` | `serial` | [`packages/core/agent/src/types.ts:301`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`subagent-inprocess`](../packages/subagent/subagent-inprocess) | | `approval/request` | `waterfall` | [`packages/ui/user-approval/src/index.ts:31`](../packages/ui/user-approval/src/index.ts) | [`user-approval`](../packages/ui/user-approval) (`waterfall`) | [`acp`](../packages/ui/acp) | | `fs/edit-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:62`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:71`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) | diff --git a/docs/i18n/style-samples.md b/docs/i18n/style-samples.md index 2d9de63bc6..dd970b7c12 100644 --- a/docs/i18n/style-samples.md +++ b/docs/i18n/style-samples.md @@ -28,9 +28,9 @@ **dispose(资源释放)必须等待所有任务完全停稳,不能仅下发终止指令就返回**:如果清理过程只发出终止或中断信号,却不等任务停止就返回,就会留下孤儿进程。清理应采用异步方式,等待所有子任务彻底退出(先发出终止信号,再等待退出);发出信号前应先关闭监听器与通知注册表,使延迟到达的完成事件不再触发通知。测试要证明 dispose 的确等到清理完成:执行完 `await fiber.dispose()` 后进程 PID 立即消失,不能只检查进程最终会自行消亡。 -> **Async state is not synchronous state** — `agent.send()` does not flip status before returning; a background task's completion races turn boundaries; `reader.close()` fires for both EOF and disposal. Never gate control flow on a status you only just requested — drive lifecycle off the events/promises that actually fire (`agent/status`, `task.done`), and observe the transition (saw `running` THEN `idle`) rather than counting actions you assume map 1:1 to turns. +> **Async state is not synchronous state** — `agent.send()` does not flip status before returning; a background task's completion races turn boundaries; `reader.close()` fires for both EOF and disposal. Never gate control flow on a status you only just requested — drive lifecycle off the events/promises that actually fire (`agent/status`, `task.done`), and observe the transition (saw `running` THEN `idle`) instead of treating status as a per-send result: several queued sends run as consecutive turns under one `running` interval, while cancellation or disposal can discard unstarted items. -**异步状态不等同于同步瞬时状态**:调用 `agent.send()` 不会在返回前同步更新状态;后台任务的完成时间与轮次边界存在竞态;`reader.close()` 既会在读到文件末尾时触发,也会在资源释放时触发。切勿把刚刚发起的状态变更当成已经生效,据此控制流程;生命周期逻辑应以实际触发的事件和已完成的 promise(`agent/status`、`task.done`)为准,并观察完整的状态变化(先 `running`,再 `idle`),不要根据操作次数推断操作与轮次一一对应。 +**异步状态不等同于同步瞬时状态**:调用 `agent.send()` 不会在返回前同步更新状态;后台任务的完成时间与轮次边界存在竞态;`reader.close()` 既会在读到文件末尾时触发,也会在资源释放时触发。切勿把刚刚发起的状态变更当成已经生效,据此控制流程;生命周期逻辑应以实际触发的事件和已完成的 promise(`agent/status`、`task.done`)为准,并观察完整的状态变化(先 `running`,再 `idle`),不要把状态当作逐次 `send()` 的结果:多次排队的 `send()` 会作为连续轮次运行,但可能共用一个 `running` 区间;取消或资源释放还可能丢弃尚未启动的队列项。 ## ③ 测试政策清单 diff --git a/docs/persistence-catalog.md b/docs/persistence-catalog.md index c4d01702d1..6ab6f5f1d4 100644 --- a/docs/persistence-catalog.md +++ b/docs/persistence-catalog.md @@ -79,7 +79,7 @@ export type SessionEvent = { }[T] ``` -Sources: [`packages/core/session/src/types.ts:255`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:262`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:292`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:324`](../packages/core/session/src/types.ts) +Sources: [`packages/core/session/src/types.ts:256`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:263`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:293`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:325`](../packages/core/session/src/types.ts) ## Events @@ -151,7 +151,7 @@ Source: [`packages/ui/user-approval/src/index.ts:68`](../packages/ui/user-approv Types: [StreamChunk](core-data-structures/llm-streaming.md) -Source: [`packages/core/session/src/types.ts:219`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:220`](../packages/core/session/src/types.ts) #### `assistant/message` — surface @@ -167,7 +167,7 @@ Source: [`packages/core/session/src/types.ts:219`](../packages/core/session/src/ Types: [ContentBlock](core-data-structures/core.md) · [TokenUsage](core-data-structures/llm-streaming.md) -Source: [`packages/core/session/src/types.ts:226`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:227`](../packages/core/session/src/types.ts) ### `compact/*` @@ -246,7 +246,7 @@ Source: [`packages/compact/compact/src/types.ts:22`](../packages/compact/compact Types: [ContentBlock](core-data-structures/core.md) · [MessageSource](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:213`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:214`](../packages/core/session/src/types.ts) ### `hook/*` @@ -317,14 +317,14 @@ Source: [`packages/ui/permission/src/index.ts:36`](../packages/ui/permission/src ```ts persistence-catalog /** * Durable record of a prompt veto and its reason. It is log-only: the blocked - * prompt never enters the model-visible surface, including in a mixed batch. + * prompt never enters the model-visible surface, and its turn runs zero steps. */ 'prompt/blocked': { content: ContentBlock[]; source: MessageSource; reason: string } ``` Types: [ContentBlock](core-data-structures/core.md) · [MessageSource](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:201`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:202`](../packages/core/session/src/types.ts) ### `request/*` @@ -338,7 +338,7 @@ Source: [`packages/core/session/src/types.ts:201`](../packages/core/session/src/ 'request/header': { header: EpochHeader; reason: RequestHeaderReason } ``` -Source: [`packages/core/session/src/types.ts:251`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:252`](../packages/core/session/src/types.ts) ### `sandbox/*` @@ -369,7 +369,7 @@ Source: [`packages/sandbox/sandbox-policy/src/session-mode.ts:34`](../packages/s Types: [ContentBlock](core-data-structures/core.md) · [MessageSource](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:244`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:245`](../packages/core/session/src/types.ts) ### `step/*` @@ -380,7 +380,7 @@ Source: [`packages/core/session/src/types.ts:244`](../packages/core/session/src/ 'step/end': { turn: number; step: number } ``` -Source: [`packages/core/session/src/types.ts:194`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:195`](../packages/core/session/src/types.ts) #### `step/start` — log-only @@ -389,7 +389,7 @@ Source: [`packages/core/session/src/types.ts:194`](../packages/core/session/src/ 'step/start': { turn: number; step: number } ``` -Source: [`packages/core/session/src/types.ts:192`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:193`](../packages/core/session/src/types.ts) ### `todo/*` @@ -402,7 +402,7 @@ Source: [`packages/core/session/src/types.ts:192`](../packages/core/session/src/ Types: [TodoItem](core-data-structures/session.md) -Source: [`packages/core/session/src/types.ts:246`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:247`](../packages/core/session/src/types.ts) ### `tool/*` @@ -419,7 +419,7 @@ Source: [`packages/core/session/src/types.ts:246`](../packages/core/session/src/ Types: [CallId](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:232`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:233`](../packages/core/session/src/types.ts) #### `tool/code-dispatch` — log-only @@ -463,7 +463,7 @@ Source: [`packages/core/tools/src/code-mode.ts:34`](../packages/core/tools/src/c Types: [CallId](core-data-structures/core.md) · [ContentBlock](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:242`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:243`](../packages/core/session/src/types.ts) ### `turn/*` @@ -472,22 +472,23 @@ Source: [`packages/core/session/src/types.ts:242`](../packages/core/session/src/ ```ts persistence-catalog /** * Closes turn `turn` with the {@link TurnEndReason} that ended it. The loop - * fires the awaited `session/flush` checkpoint at every turn end, so the turn - * boundary is also the durable-commit boundary. + * awaits `session/flush` after an ordinary turn ends before claiming the next + * queued item. Success commits the turn; rejection is reported live and does + * not prevent later work. */ 'turn/end': { turn: number; reason: TurnEndReason } ``` Types: [TurnEndReason](core-data-structures/session.md) -Source: [`packages/core/session/src/types.ts:190`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:191`](../packages/core/session/src/types.ts) #### `turn/start` — log-only ```ts persistence-catalog /** - * Opens turn `turn`. `trigger` records what started it — a drained message - * batch or an idle-time injection. The turn is the durability/replay + * Opens turn `turn`. `trigger` records what started it — one claimed queued + * message or an idle-time injection. The turn is the durability/replay * boundary: every event sits between a `turn/start` and its matching * `turn/end` (the turn-enclosure invariant). */ @@ -503,10 +504,10 @@ Source: [`packages/core/session/src/types.ts:184`](../packages/core/session/src/ #### `user/message` — surface ```ts persistence-catalog -/** A user-visible prompt (queued message drained at turn start). */ +/** A user-visible prompt (the queued message claimed for this turn). */ 'user/message': { content: ContentBlock[]; source: MessageSource } ``` Types: [ContentBlock](core-data-structures/core.md) · [MessageSource](core-data-structures/core.md) -Source: [`packages/core/session/src/types.ts:196`](../packages/core/session/src/types.ts) +Source: [`packages/core/session/src/types.ts:197`](../packages/core/session/src/types.ts) diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 26ce375479..efb29e547d 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -675,8 +675,8 @@ export const EVENT_API: readonly EventApiEntry[] = [ name: 'agent/prompt-submit', mode: 'waterfall', signature: '\'agent/prompt-submit\'(this: Scoped, agent: Agent, content: ContentBlock[], source: MessageSource, next: () => Promise): Promise', - jsDoc: '/**\n * Allow, rewrite, or block one drained prompt before it becomes a user\n * message. Call `next()` for the unchanged default.\n * @param agent - the agent draining its inbox.\n * @param content - the drained message\'s blocks, as queued.\n * @param source - the message\'s resolved source.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode waterfall\n */', - summary: 'Allow, rewrite, or block one drained prompt before it becomes a user message.', + jsDoc: '/**\n * Allow, rewrite, or block one claimed prompt before it becomes a user\n * message. Call `next()` for the unchanged default.\n * @param agent - the agent whose turn claimed the message.\n * @param content - the claimed message\'s blocks, as queued.\n * @param source - the message\'s resolved source.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode waterfall\n */', + summary: 'Allow, rewrite, or block one claimed prompt before it becomes a user message.', }, { name: 'agent/queued', diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index 9c5c869168..acb49b1ba5 100644 --- a/packages/core/agent-loop/README.md +++ b/packages/core/agent-loop/README.md @@ -46,7 +46,9 @@ Configured agents start automatically. A model call requires both `provider` and ### Internal concrete driver -The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. The concrete `send()`, running `steer()`, and open-turn `inject()` materialize content plus resolved source once as detached, deeply frozen lossless JSON; malformed data throws before enqueue or append. An injection that arrives while the current step executes assistant tool calls stays in a FIFO until the batch settles; successful batches place it after the complete result batch, and interrupted batches drain it before the turn closes. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy. +The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy. + +Each concrete `send()` materializes content plus resolved source once as a detached, deeply frozen lossless-JSON FIFO item. If claimed, it is the sole ordinary message in its turn; a successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, or a pre-start failure may drop it without a turn. Running `steer()` enters the steering FIFO: an open turn records it at the next steering checkpoint before a request or continuation decision, but policy can still stop before another step; steering left after turn close and its checkpoint becomes later queued input unless terminal turn policy, cancellation, or disposal discards it. Open-turn `inject()` uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls; successful batches place it after all results, and interrupted batches drain it before turn close. Malformed data throws before enqueue or append. ### Loop lifecycle (`loop.ts`) diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index a9c7f3b766..d569146d64 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -397,7 +397,7 @@ export class ReactLoopAgent implements Agent { cancelReason: () => this.cancelReason, clearCancel: () => { this.cancelRequested = false }, withToolBatch: run => this.withToolBatch(run), - // Pre-step cancellation re-parks without emitting a status transition. + // Pre-start cancellation settles queued-work waiters before publishing idle. settleIdle: () => { this.settleIdleWaiters() }, })) } diff --git a/packages/core/agent-loop/src/inbox.ts b/packages/core/agent-loop/src/inbox.ts index abb588b919..b26a79a1ef 100644 --- a/packages/core/agent-loop/src/inbox.ts +++ b/packages/core/agent-loop/src/inbox.ts @@ -15,7 +15,7 @@ export interface InboxMessage { } /** - * Per-agent inbox: a queued FIFO (drained at turn start) and a steering FIFO + * Per-agent inbox: a queued FIFO (dequeued once per turn start) and a steering FIFO * (drained between steps of a running turn). Purely an in-memory mechanism of * the loop — the public surface is `Agent.send()` / `Agent.steer()`. */ @@ -54,11 +54,11 @@ export class Inbox { } /** - * Drain all queued messages (turn start). - * @returns the drained messages in arrival order; the queued FIFO is left empty. + * Remove the oldest queued message for one turn start. + * @returns the oldest message, or `undefined` when the queued FIFO is empty. */ - drainQueued(): InboxMessage[] { - return this.queuedMessages.splice(0) + dequeueQueued(): InboxMessage | undefined { + return this.queuedMessages.shift() } /** @@ -72,7 +72,7 @@ export class Inbox { /** * Discard all pending messages (queued + steering) without delivering them — * used by `cancel()`, which drops un-started work rather than draining it into - * a turn. Unlike `drainQueued`/`drainSteering`, the messages are thrown away. + * a turn. Unlike `dequeueQueued`/`drainSteering`, the messages are thrown away. */ clear(): void { this.queuedMessages.length = 0 diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index bc18ee52d3..e8008996a9 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -91,16 +91,16 @@ export interface LoopHandle { cancelReason(): string /** Clear the cancel marker (called once per iteration after the turn returns). */ clearCancel(): void - /** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */ + /** Settle idle waiters before pre-running cancellation publishes idle. */ settleIdle(): void /** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */ readonly withToolBatch: (run: (acceptContext: (context: HookContext) => void) => Promise) => Promise } /** - * Drive queued batches as durable turns until disposal. Plugin failures end the - * current turn without terminating the driver. The caller establishes the - * `ctx.agents.withInitiator()` boundary before entry; package-private + * Drive queued messages as independent durable turns until disposal. Plugin + * failures end the current turn without terminating the driver. The caller + * establishes the `ctx.agents.withInitiator()` boundary before entry; package-private * orchestration recovers that exact Agent and captures its Session locally. * @param ctx - the plugin context the loop reaches its initiating Agent, * events (agent/…, session/flush), and services (systemPrompt, llm, tools) @@ -118,20 +118,35 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise { const events = agentEvents(ctx, agent) while (!handle.isDisposed()) { - await handle.inbox.waitForQueued(handle.disposed) - if (handle.isDisposed()) break - - // Cancellation between wake and `running` skips only the cancelled work; - // a replacement prompt still runs and owns the eventual idle transition. + // An idle listener can enqueue and cancel replacement work before the next + // wait is installed. Consume that empty marker before parking the driver. if (handle.isCancelled()) { handle.clearCancel() if (!handle.inbox.hasQueued) { handle.settleIdle() + handle.setStatus('idle') + continue + } + } + + await handle.inbox.waitForQueued(handle.disposed) + if (handle.isDisposed()) break + + // Cancellation between wake and `running` skips only the cancelled work; + // a replacement prompt still runs before the eventual idle transition. + if (handle.isCancelled()) { + handle.clearCancel() + if (!handle.inbox.hasQueued) { + // Settle before publishing idle: the already-idle path has no status + // transition, while an idle listener can register waiters for new work. + handle.settleIdle() + handle.setStatus('idle') continue } } handle.setStatus('running') + if (handle.isDisposed()) break // A synchronous `running` listener can cancel before `runTurn`; balance the // status only when no replacement prompt was queued by that listener. @@ -182,12 +197,11 @@ async function runTurn( return messages.length > 0 } - // Drain before opening the turn, but append only after `turn/start`. - const queued = handle.inbox.drainQueued() - const first = queued[0] + // Claim one queued message before opening its turn, but append it only after `turn/start`. + const message = handle.inbox.dequeueQueued() /* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */ - if (!first) throw new Error('runTurn invariant violated: no queued message at turn start') - const trigger: TurnTrigger = { kind: 'message', source: first.source } + if (!message) throw new Error('runTurn invariant violated: no queued message at turn start') + const trigger: TurnTrigger = { kind: 'message', source: message.source } let reason: TurnEndReason = { kind: 'completed' } let step = 0 @@ -226,42 +240,26 @@ async function runTurn( // matter what throws below; the catch + closeTurn guarantee it. A pre-commit // veto leaves no turn/start in the log and therefore owes no turn/end. session.append('turn/start', { turn, trigger }) - // Each drained queued message runs the `agent/prompt-submit` waterfall before - // it becomes a `user/message` — a hook can rewrite the prompt or block it. + // The claimed message runs the `agent/prompt-submit` waterfall before it + // becomes a `user/message` — a hook can rewrite the prompt or block it. // Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed; // turn/end is now owed, so a throwing prompt-submit listener (the waterfall // throws) is caught below and the turn still closes. - let anyAllowed = false - // Seeded with a floor (only observable if the batch were empty, which - // runTurn never allows — it is called with ≥1 queued message); each `block` - // decision carries a required `reason` and overwrites it, so a fully-blocked - // batch always reports the last vetoing reason. - let lastBlockReason = 'prompt blocked by hook' - for (const message of queued) { - const decision = await events.waterfall( - 'agent/prompt-submit', message.content, message.source, - () => Promise.resolve({ kind: 'allow' }), - ) - if (decision.kind === 'block') { - lastBlockReason = decision.reason - // Record the veto durably: `PromptDecision.reason` is the durable record - // of why a prompt was blocked, but a fully-blocked batch's `rejected` - // turn/end only preserves the LAST reason, and a MIXED batch (this prompt - // blocked, another allowed) does not end `rejected` at all — so without - // this append a blocked prompt would vanish from the log whenever any - // sibling prompt is allowed. `prompt/blocked` sits in the open turn in - // place of the `user/message` this prompt would have become. - session.append('prompt/blocked', { content: message.content, source: message.source, reason: decision.reason }) - continue - } - anyAllowed = true + const promptDecision = await events.waterfall( + 'agent/prompt-submit', message.content, message.source, + () => Promise.resolve({ kind: 'allow' }), + ) + if (promptDecision.kind === 'block') { + session.append('prompt/blocked', { content: message.content, source: message.source, reason: promptDecision.reason }) + reason = { kind: 'rejected', reason: promptDecision.reason } + } else { // `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them. - const content = decision.content ?? message.content + const content = promptDecision.content ?? message.content session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' }) // Every `allow.additionalContexts` entry is a separate context/message the // next request also sees. The turn is open, so inject() appends each one // into THIS turn without flattening provenance or metadata. - for (const context of decision.additionalContexts ?? []) { + for (const context of promptDecision.additionalContexts ?? []) { agent.inject(context.content, { source: context.source, ...context.meta !== undefined ? { meta: context.meta } : {}, @@ -270,11 +268,8 @@ async function runTurn( } while (true) { - // A fully blocked batch closes its zero-step turn as rejected. - if (!anyAllowed) { - reason = { kind: 'rejected', reason: lastBlockReason } - break - } + // A blocked prompt closes its zero-step turn as rejected. + if (promptDecision.kind === 'block') break step += 1 // Steering from the previous round's continuation listeners joins before diff --git a/packages/core/agent-loop/tests/cancel.spec.ts b/packages/core/agent-loop/tests/cancel.spec.ts index 14e7cf8976..39289779be 100644 --- a/packages/core/agent-loop/tests/cancel.spec.ts +++ b/packages/core/agent-loop/tests/cancel.spec.ts @@ -79,7 +79,8 @@ describe('Agent.cancel()', () => { // send() queues synchronously (status still idle, loop microtask not yet // resumed). Cancel in that pre-step window: the queued turn must not run. - send(agent, 'drop me') + send(agent, 'drop me first') + send(agent, 'drop me second') agent.cancel('pre-step') // Give the loop a chance to wake and process the cancel. @@ -91,6 +92,35 @@ describe('Agent.cancel()', () => { expect(agent.status).toBe('idle') }) + it('disposal from the running notification drops queued work before turn start', async () => { + const adapter = new MockAdapter([textResponse('should not run')]) + const ctx = await harness(adapter) + const handle = await ctx.agents.create({ + sessionId: SessionId('dispose-running-session'), + agentOptions: { provider: 'mock', model: 'mock' }, + }) + const agent = handle.agent + + const running = Promise.withResolvers() + let disposalDone: Promise | undefined + ctx.on('agent/status', (subject, status) => { + if (subject !== agent || status !== 'running') return + disposalDone = handle.dispose() + running.resolve(undefined) + }) + + send(agent, 'drop before claim') + await running.promise + if (disposalDone === undefined) throw new Error('running listener did not start disposal') + await disposalDone + await driverDone(agent) + + expect(agent.status).toBe('disposed') + expect(agent.session.events.some(event => event.type === 'turn/start')).toBe(false) + expect(userTexts(agent)).toEqual([]) + expect(adapter.requests).toHaveLength(0) + }) + it('a whenIdle() waiter registered BEFORE a pre-step cancel resolves (F1 hang guard)', async () => { const adapter = new MockAdapter([textResponse('x')]) const ctx = await harness(adapter) @@ -110,7 +140,162 @@ describe('Agent.cancel()', () => { expect(agent.status).toBe('idle') }) - it('cancel() mid-step aborts the in-flight model call; the turn ends aborted', async () => { + it('cancel() between consecutive turns restores idle and leaves idle steer usable', async () => { + const adapter = new MockAdapter([textResponse('first reply'), textResponse('steer reply')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('between-turn-cancel'), { provider: 'mock', model: 'mock' }) + + let rejectFirstFlush = true + ctx.on('session/flush', (session) => { + if (session !== agent.session || !rejectFirstFlush) return + rejectFirstFlush = false + throw new Error('first flush failed') + }) + + const cancelled = Promise.withResolvers() + ctx.on('agent/error', (subject, _turn, _step, error) => { + if (subject !== agent || error.message !== 'first flush failed') return + // The first hop runs before runLoop resumes from runTurn; the second lands + // before its resolved waitForQueued continuation checks cancellation. + queueMicrotask(() => { + queueMicrotask(() => { + agent.cancel('between turns') + cancelled.resolve(undefined) + }) + }) + }) + + const statuses: string[] = [] + ctx.on('agent/status', (subject, status) => { + if (subject === agent) statuses.push(status) + }) + + send(agent, 'first') + send(agent, 'queued tail') + await cancelled.promise + + expect(agent.status).toBe('idle') + expect(statuses).toEqual(['running', 'idle']) + expect(adapter.requests).toHaveLength(1) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1) + expect(userTexts(agent)).toEqual(['first']) + + let idleResolved = false + void agent.whenIdle().then(() => { idleResolved = true }) + await Promise.resolve() + expect(idleResolved).toBe(true) + + const idle = waitForIdle(ctx, agent) + agent.steer([{ type: 'text', text: 'idle steer' }]) + await idle + + expect(statuses).toEqual(['running', 'idle', 'running', 'idle']) + expect(adapter.requests).toHaveLength(2) + expect(userTexts(agent)).toEqual(['first', 'idle steer']) + }) + + it('an idle-listener replacement keeps whenIdle pending until the replacement turn finishes', async () => { + const adapter = new MockAdapter([textResponse('first reply'), textResponse('replacement reply')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('between-turn-idle-listener'), { provider: 'mock', model: 'mock' }) + + let rejectFirstFlush = true + ctx.on('session/flush', (session) => { + if (session !== agent.session || !rejectFirstFlush) return + rejectFirstFlush = false + throw new Error('first flush failed') + }) + + ctx.on('agent/error', (subject, _turn, _step, error) => { + if (subject !== agent || error.message !== 'first flush failed') return + queueMicrotask(() => { + queueMicrotask(() => { agent.cancel('between turns') }) + }) + }) + + const replacementRegistered = Promise.withResolvers() + let replacementObservation: Promise<{ status: string; requests: number; turns: number }> | undefined + ctx.on('agent/status', (subject, status) => { + if (subject !== agent || status !== 'idle' || replacementObservation !== undefined) return + send(agent, 'replacement') + replacementObservation = agent.whenIdle().then(() => ({ + status: agent.status, + requests: adapter.requests.length, + turns: agent.session.events.filter(event => event.type === 'turn/start').length, + })) + replacementRegistered.resolve(undefined) + }) + + send(agent, 'first') + send(agent, 'cancelled tail') + await replacementRegistered.promise + if (replacementObservation === undefined) throw new Error('idle listener did not register replacement work') + + await expect(replacementObservation).resolves.toEqual({ status: 'idle', requests: 2, turns: 2 }) + expect(userTexts(agent)).toEqual(['first', 'replacement']) + }) + + it('idle-listener cancellation settles its waiter without cancelling later work', async () => { + const adapter = new MockAdapter([textResponse('first reply'), textResponse('later reply')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('idle-listener-cancel'), { provider: 'mock', model: 'mock' }) + + const replacementRegistered = Promise.withResolvers() + let replacementObservation: Promise<{ status: string; requests: number; turns: number }> | undefined + ctx.on('agent/status', (subject, status) => { + if (subject !== agent || status !== 'idle' || replacementObservation !== undefined) return + send(agent, 'cancelled replacement') + replacementObservation = agent.whenIdle().then(() => ({ + status: agent.status, + requests: adapter.requests.length, + turns: agent.session.events.filter(event => event.type === 'turn/start').length, + })) + agent.cancel('idle listener') + replacementRegistered.resolve(undefined) + }) + + send(agent, 'first') + await replacementRegistered.promise + if (replacementObservation === undefined) throw new Error('idle listener did not register replacement work') + + await expect(Promise.race([ + replacementObservation, + new Promise((_resolve, reject) => setTimeout(() => { reject(new Error('whenIdle hung after idle-listener cancel')) }, 1000)), + ])).resolves.toEqual({ status: 'idle', requests: 1, turns: 1 }) + + const idle = waitForIdle(ctx, agent) + send(agent, 'later') + await idle + expect(adapter.requests).toHaveLength(2) + expect(userTexts(agent)).toEqual(['first', 'later']) + }) + + it('replacement work queued after idle-listener cancellation still runs', async () => { + const adapter = new MockAdapter([textResponse('first reply'), textResponse('replacement reply')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('idle-listener-post-cancel-send'), { provider: 'mock', model: 'mock' }) + + const replacementRegistered = Promise.withResolvers() + let replacementIdle: Promise | undefined + ctx.on('agent/status', (subject, status) => { + if (subject !== agent || status !== 'idle' || replacementIdle !== undefined) return + send(agent, 'cancelled replacement') + agent.cancel('idle listener') + send(agent, 'surviving replacement') + replacementIdle = agent.whenIdle() + replacementRegistered.resolve(undefined) + }) + + send(agent, 'first') + await replacementRegistered.promise + if (replacementIdle === undefined) throw new Error('idle listener did not register replacement work') + await replacementIdle + + expect(adapter.requests).toHaveLength(2) + expect(userTexts(agent)).toEqual(['first', 'surviving replacement']) + }) + + it('cancel() mid-step aborts the active turn and drops every queued tail item', async () => { const adapter = new MockAdapter(['hang']) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -121,10 +306,14 @@ describe('Agent.cancel()', () => { send(agent, 'go') await new Promise(r => setTimeout(r, 30)) expect(agent.status).toBe('running') + send(agent, 'queued tail') agent.cancel('mid-step') await waitForIdle(ctx, agent) expect(reasons).toEqual([{ kind: 'aborted', reason: 'mid-step' }]) + expect(userTexts(agent)).toEqual(['go']) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1) + expect(adapter.requests).toHaveLength(1) }) it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => { diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index 80d085c32b..b436c4468b 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -632,15 +632,20 @@ describe('plugin exceptions are contained', () => { expect(agent.status).toBe('idle') }) - it('a rejecting session/flush listener is reported but does not kill the agent', async () => { + it('a rejecting first-turn flush settles before the queued tail starts', async () => { const adapter = new MockAdapter([textResponse('one'), textResponse('two')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - let rejectedOnce = false - ctx.on('session/flush', async () => { - if (!rejectedOnce) { - rejectedOnce = true + const firstFlush = Promise.withResolvers() + const releaseFirstFlush = Promise.withResolvers() + let flushes = 0 + ctx.on('session/flush', async (session) => { + if (session !== agent.session) return + flushes += 1 + if (flushes === 1) { + firstFlush.resolve(undefined) + await releaseFirstFlush.promise throw new Error('disk full') } }) @@ -648,18 +653,25 @@ describe('plugin exceptions are contained', () => { const errors: Error[] = [] ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error)) + const idle = waitForIdle(ctx, agent) send(agent, 'first') - await waitForIdle(ctx, agent) - expect(errors.map(e => e.message)).toEqual(['disk full']) - send(agent, 'second') - await waitForIdle(ctx, agent) + + await firstFlush.promise + expect(adapter.requests).toHaveLength(1) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1) + + releaseFirstFlush.resolve(undefined) + await idle + + expect(errors.map(e => e.message)).toEqual(['disk full']) expect(adapter.requests).toHaveLength(2) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2) }) }) describe('disposed status is part of the agent/status contract', () => { - it('disposing the fiber emits agent/status(disposed) and ends the turn with reason disposed', async () => { + it('disposing the fiber ends the active turn and never starts its queued tail', async () => { const adapter = new MockAdapter(['hang']) const ctx = await harness(adapter) @@ -675,11 +687,19 @@ describe('disposed status is part of the agent/status contract', () => { send(agent, 'go') await new Promise(r => setTimeout(r, 30)) + send(agent, 'queued tail') await fiber.dispose() await driverDone(agent) expect(statuses).toEqual(['running', 'disposed']) expect(reasons).toEqual([{ kind: 'disposed' }]) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1) + const messages = agent.session.events + .filter(event => event.type === 'user/message') + .flatMap(event => event.data.content) + .flatMap(block => block.type === 'text' ? [block.text] : []) + expect(messages).toEqual(['go']) + expect(adapter.requests).toHaveLength(1) }) it('a throwing agent/status listener cannot break disposal or leak the registry entry', async () => { diff --git a/packages/core/agent-loop/tests/coverage-edges.spec.ts b/packages/core/agent-loop/tests/coverage-edges.spec.ts index a1d449d273..85f99e8a66 100644 --- a/packages/core/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts @@ -146,12 +146,22 @@ describe('toError normalization', () => { const errors: Error[] = [] ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error)) - send(agent, 'go') + send(agent, 'fails before turn start') + send(agent, 'survives as the next item') await waitForIdle(ctx, agent) expect(errors).toHaveLength(1) expect(errors[0]).toMatchObject({ message: 'naked string error', code: 'UNKNOWN' }) - expect(adapter.requests).toEqual([]) - expect(agent.session.events.some(event => event.type === 'turn/start' || event.type === 'turn/end')).toBe(false) + expect(adapter.requests).toHaveLength(1) + const starts = agent.session.events.filter(event => event.type === 'turn/start') + const ends = agent.session.events.filter(event => event.type === 'turn/end') + const messages = agent.session.events.filter(event => event.type === 'user/message') + expect(starts).toHaveLength(1) + expect(starts[0]?.type === 'turn/start' && starts[0].data.turn).toBe(1) + expect(ends).toHaveLength(1) + expect(messages).toHaveLength(1) + expect(messages[0]?.type === 'user/message' && messages[0].data.content).toEqual([ + { type: 'text', text: 'survives as the next item' }, + ]) }) it('normalizes non-Error throws from agent/request waterfall via inline toError in runStep catch', async () => { diff --git a/packages/core/agent-loop/tests/inbox.spec.ts b/packages/core/agent-loop/tests/inbox.spec.ts index 4bea62abe2..f4eea9fdd0 100644 --- a/packages/core/agent-loop/tests/inbox.spec.ts +++ b/packages/core/agent-loop/tests/inbox.spec.ts @@ -8,17 +8,17 @@ function resolverPair() { } describe('Inbox', () => { - it('enqueues and drains queued messages in FIFO order', () => { + it('dequeues one queued message at a time in FIFO order', () => { const inbox = new Inbox() inbox.enqueue({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } }) inbox.enqueue({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } }) expect(inbox.hasQueued).toBe(true) - const drained = inbox.drainQueued() - expect(drained).toHaveLength(2) - expect(drained[0]!.content[0]).toMatchObject({ text: 'first' }) - expect(drained[1]!.content[0]).toMatchObject({ text: 'second' }) + expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'first' }) + expect(inbox.hasQueued).toBe(true) + expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'second' }) expect(inbox.hasQueued).toBe(false) + expect(inbox.dequeueQueued()).toBeUndefined() }) it('pushes and drains steering messages separately from queued', () => { diff --git a/packages/core/agent-loop/tests/interception.spec.ts b/packages/core/agent-loop/tests/interception.spec.ts index 92281cde94..83b156b0ef 100644 --- a/packages/core/agent-loop/tests/interception.spec.ts +++ b/packages/core/agent-loop/tests/interception.spec.ts @@ -177,9 +177,7 @@ describe('agent/prompt-submit', () => { expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'rejected', reason: 'blocked by policy' }) }) - it('a mixed batch records a prompt/blocked for the vetoed prompt while the allowed one runs', async () => { - // Blocking one prompt in a mixed batch must persist its reason even though - // the allowed prompt keeps the turn from ending rejected. + it('adjacent blocked and allowed prompts keep independent turn outcomes', async () => { const adapter = new MockAdapter([textResponse('ran once')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -192,13 +190,13 @@ describe('agent/prompt-submit', () => { const reasons: TurnEndReason[] = [] ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) }) - // both sends land before the loop drains → one batched turn + // Both sends land before the driver wakes, but each remains its own turn. send(agent, 'secret') send(agent, 'safe') await waitForIdle(ctx, agent) const log = events(agent) - // the allowed prompt became a user/message and drove exactly one model call + // The allowed prompt became a user/message and drove exactly one model call. const userMsgs = log.filter(e => e.type === 'user/message') expect(userMsgs).toHaveLength(1) expect(userMsgs[0]?.type === 'user/message' && userMsgs[0].data.content).toEqual([{ type: 'text', text: 'safe' }]) @@ -210,12 +208,14 @@ describe('agent/prompt-submit', () => { content: [{ type: 'text', text: 'secret' }], reason: 'policy: no secrets', }) - // the turn did NOT reject — a sibling was allowed — so the boundary reason - // alone would not have preserved the block - expect(reasons.some(r => r.kind === 'rejected')).toBe(false) + expect(log.filter(e => e.type === 'turn/start')).toHaveLength(2) + expect(reasons).toEqual([ + { kind: 'rejected', reason: 'policy: no secrets' }, + { kind: 'completed' }, + ]) }) - it('a throwing prompt-submit listener ends the turn balanced (error), loop survives', async () => { + it('a throwing prompt-submit listener ends its turn balanced while an adjacent message survives', async () => { const adapter = new MockAdapter([textResponse('after')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -226,20 +226,31 @@ describe('agent/prompt-submit', () => { return { kind: 'allow' as const } }) const errors: Error[] = [] + const reasons: TurnEndReason[] = [] + const statuses: string[] = [] ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + ctx.on('agent/status', (subject, status) => { if (subject === agent) statuses.push(status) }) + ctx.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'turn/end') reasons.push(event.data.reason) + }) + const idle = waitForIdle(ctx, agent) send(agent, 'first') - await waitForIdle(ctx, agent) - expect(errors.map(e => e.message)).toEqual(['prompt hook broke']) - // turn balanced - const log = events(agent) - expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1) - expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1) - - // loop survives: a second prompt runs normally send(agent, 'second') - await waitForIdle(ctx, agent) - expect(adapter.requests.length).toBeGreaterThanOrEqual(1) + await idle + expect(errors.map(e => e.message)).toEqual(['prompt hook broke']) + // The failed prompt forms one balanced error turn; the adjacent prompt forms + // the following normal turn without an intermediate idle transition. + const log = events(agent) + expect(log.filter(e => e.type === 'turn/start')).toHaveLength(2) + expect(log.filter(e => e.type === 'turn/end')).toHaveLength(2) + expect(reasons).toEqual([ + { kind: 'error', step: 0, message: 'prompt hook broke' }, + { kind: 'completed' }, + ]) + expect(statuses).toEqual(['running', 'idle']) + expect(adapter.requests).toHaveLength(1) + expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('second') }) }) diff --git a/packages/core/agent-loop/tests/loop.spec.ts b/packages/core/agent-loop/tests/loop.spec.ts index 5652f0715e..dd686edfcb 100644 --- a/packages/core/agent-loop/tests/loop.spec.ts +++ b/packages/core/agent-loop/tests/loop.spec.ts @@ -354,14 +354,24 @@ describe('agent loop', () => { expect(flat).toContain('change of plans') }) - it('steering while idle behaves like send (starts a turn)', async () => { - const adapter = new MockAdapter([textResponse('ok')]) + it('same-tick idle steering inherits one-send-one-turn FIFO behavior', async () => { + const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - agent.steer([{ type: 'text', text: 'hello' }]) - await waitForIdle(ctx, agent) - expect(agent.session.events.some(e => e.type === 'user/message')).toBe(true) + const idle = waitForIdle(ctx, agent) + agent.steer([{ type: 'text', text: 'first idle steer' }]) + agent.steer([{ type: 'text', text: 'second idle steer' }]) + await idle + + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2) + expect(agent.session.events + .filter(event => event.type === 'user/message') + .map(event => event.data.content)).toEqual([ + [{ type: 'text', text: 'first idle steer' }], + [{ type: 'text', text: 'second idle steer' }], + ]) + expect(adapter.requests).toHaveLength(2) }) it('inject() while idle wraps context in a one-shot turn, visible to the next request', async () => { @@ -922,7 +932,149 @@ describe('agent loop', () => { expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('completed') }) - it('chains queued messages into consecutive turns', async () => { + it('keeps same-tick sends in separate turns and checkpoints before the next starts', async () => { + const adapter = new MockAdapter([textResponse('first answer'), textResponse('second answer')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) + + const firstFlush = Promise.withResolvers() + const releaseFirstFlush = Promise.withResolvers() + let flushes = 0 + ctx.on('session/flush', async (session) => { + if (session !== agent.session) return + flushes += 1 + if (flushes === 1) { + firstFlush.resolve(undefined) + await releaseFirstFlush.promise + } + }) + + const turns: number[] = [] + ctx.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'turn/start') turns.push(event.data.turn) + }) + + const idle = waitForIdle(ctx, agent) + send(agent, 'first message') + send(agent, 'second message') + + await firstFlush.promise + expect(turns).toEqual([1]) + expect(adapter.requests).toHaveLength(1) + + releaseFirstFlush.resolve(undefined) + await idle + + expect(turns).toEqual([1, 2]) + expect(flushes).toBe(2) + expect(adapter.requests).toHaveLength(2) + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first answer') + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('second message') + }) + + it('holds a turn-end listener send behind the closing turn checkpoint', async () => { + const adapter = new MockAdapter([textResponse('first answer'), textResponse('second answer')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) + + const firstFlush = Promise.withResolvers() + const releaseFirstFlush = Promise.withResolvers() + let flushes = 0 + ctx.on('session/flush', async (session) => { + if (session !== agent.session) return + flushes += 1 + if (flushes === 1) { + firstFlush.resolve(undefined) + await releaseFirstFlush.promise + } + }) + + const turns: number[] = [] + const statuses: string[] = [] + ctx.on('agent/status', (subject, status) => { + if (subject === agent) statuses.push(status) + }) + ctx.on('session/event', (session, event) => { + if (session !== agent.session) return + if (event.type === 'turn/start') turns.push(event.data.turn) + if (event.type === 'turn/end' && event.data.turn === 1) send(agent, 'turn-end listener message') + }) + + const idle = waitForIdle(ctx, agent) + send(agent, 'first message') + await firstFlush.promise + + expect(turns).toEqual([1]) + expect(adapter.requests).toHaveLength(1) + + releaseFirstFlush.resolve(undefined) + await idle + + expect(turns).toEqual([1, 2]) + expect(statuses).toEqual(['running', 'idle']) + expect(adapter.requests).toHaveLength(2) + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first answer') + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('turn-end listener message') + }) + + it('keeps a reentrant agent/queued send as the next independent turn', async () => { + const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) + + let nested = false + ctx.on('agent/queued', (subject) => { + if (subject !== agent || nested) return + nested = true + send(agent, 'queued listener message') + }) + + const idle = waitForIdle(ctx, agent) + send(agent, 'outer message') + await idle + + const turns = agent.session.events.filter(event => event.type === 'turn/start') + const messages = agent.session.events + .filter(event => event.type === 'user/message') + .map(event => event.data.content) + expect(turns).toHaveLength(2) + expect(messages).toEqual([ + [{ type: 'text', text: 'outer message' }], + [{ type: 'text', text: 'queued listener message' }], + ]) + }) + + it('preserves independent turn sources across an adjacent microtask send', async () => { + const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) + + const idle = waitForIdle(ctx, agent) + agent.send([{ type: 'text', text: 'user message' }]) + await Promise.resolve() + agent.send( + [{ type: 'text', text: 'plugin message' }], + { source: { kind: 'plugin', plugin: 'test' } }, + ) + await idle + + const triggers = agent.session.events + .filter(event => event.type === 'turn/start') + .map(event => event.data.trigger) + const sources = agent.session.events + .filter(event => event.type === 'user/message') + .map(event => event.data.source) + expect(triggers).toEqual([ + { kind: 'message', source: { kind: 'user' } }, + { kind: 'message', source: { kind: 'plugin', plugin: 'test' } }, + ]) + expect(sources).toEqual([ + { kind: 'user' }, + { kind: 'plugin', plugin: 'test' }, + ]) + }) + + it('keeps a session-listener send after dequeue in the following turn', async () => { const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -945,6 +1097,37 @@ describe('agent loop', () => { expect(turns).toEqual([1, 2]) expect(adapter.requests).toHaveLength(2) + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first') + expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('second message') + }) + + it('keeps a model-adapter callback send in the following turn', async () => { + const agentRef: { current?: Agent } = {} + const adapter = new MockAdapter([ + () => { + const agent = agentRef.current + if (agent === undefined) throw new Error('model callback ran before agent setup') + send(agent, 'model callback message') + return textResponse('first') + }, + textResponse('second'), + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) + agentRef.current = agent + + const idle = waitForIdle(ctx, agent) + send(agent, 'outer message') + await idle + + const messages = agent.session.events + .filter(event => event.type === 'user/message') + .map(event => event.data.content) + expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2) + expect(messages).toEqual([ + [{ type: 'text', text: 'outer message' }], + [{ type: 'text', text: 'model callback message' }], + ]) }) it('awaits session/flush at turn end (persistence checkpoint)', async () => { diff --git a/packages/core/agent-loop/tests/properties.spec.ts b/packages/core/agent-loop/tests/properties.spec.ts index ff7a6337ce..85efda0e4e 100644 --- a/packages/core/agent-loop/tests/properties.spec.ts +++ b/packages/core/agent-loop/tests/properties.spec.ts @@ -81,6 +81,21 @@ function turnNumbers(agent: Agent): number[] { .map(e => (e.data as { turn: number }).turn) } +function turnEndNumbers(agent: Agent): number[] { + return agent.session.events + .filter(e => e.type === 'turn/end') + .map(e => (e.data as { turn: number }).turn) +} + +function userMessageCountsByTurn(agent: Agent): number[] { + const counts: number[] = [] + for (const event of agent.session.events) { + if (event.type === 'turn/start') counts.push(0) + if (event.type === 'user/message') counts[counts.length - 1]! += 1 + } + return counts +} + /** Assert a status trace is a legal run: idle/running alternating, ending idle. */ function assertLegalStatusTrace(trace: string[]): void { for (let i = 1; i < trace.length; i++) { @@ -90,7 +105,7 @@ function assertLegalStatusTrace(trace: string[]): void { } describe('agent loop scheduling properties', () => { - it('a synchronous burst loses no message and uses strictly increasing turns', async () => { + it('a synchronous burst gives every message its own strictly increasing turn', async () => { await fc.assert(fc.asyncProperty( fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 6 }), async (texts) => { @@ -105,8 +120,11 @@ describe('agent loop scheduling properties', () => { // No message lost: every send appears as a user/message, in order. expect(userMessageTexts(agent)).toEqual(texts) - // A synchronous burst batches into exactly one turn. - expect(turnNumbers(agent)).toEqual([1]) + // This failure-free fixture maps every item to an independent turn. + expect(turnNumbers(agent)).toEqual(texts.map((_, i) => i + 1)) + expect(turnEndNumbers(agent)).toEqual(texts.map((_, i) => i + 1)) + expect(userMessageCountsByTurn(agent)).toEqual(texts.map(() => 1)) + expect(trace).toEqual(['running', 'idle']) assertLegalStatusTrace(trace) } finally { await ctx.fiber.dispose() @@ -137,9 +155,9 @@ describe('agent loop scheduling properties', () => { ), { numRuns: 20, timeout: 2000 }) }) - it('mixed schedule (send, optionally settle) loses no message and orders turns', async () => { - // Each step is a (text, settle?) pair: settle=true awaits idle before the - // next send (own turn); settle=false sends in the same tick (batches). + it('mixed settled and same-tick sends preserve one turn per message', async () => { + // Each step optionally waits for idle before the next send; that scheduling + // choice must not change the ordinary message-to-turn mapping. const stepArb = fc.record({ text: fc.string({ minLength: 1 }), settle: fc.boolean() }) await fc.assert(fc.asyncProperty( fc.array(stepArb, { minLength: 1, maxLength: 6 }), @@ -158,14 +176,13 @@ describe('agent loop scheduling properties', () => { } await lastIdle - // No message lost or reordered, regardless of batching. + // No message is lost or reordered, regardless of driver timing. expect(userMessageTexts(agent)).toEqual(steps.map(s => s.text)) - // Turn numbers are a strictly increasing 1..N prefix (N = turn count). + // Every item forms one FIFO-ordered turn containing only that message. const turns = turnNumbers(agent) - expect(turns).toEqual(turns.map((_, i) => i + 1)) - // Every message landed in some turn; turns never exceed messages. - expect(turns.length).toBeLessThanOrEqual(steps.length) - expect(turns.length).toBeGreaterThanOrEqual(1) + expect(turns).toEqual(steps.map((_, i) => i + 1)) + expect(turnEndNumbers(agent)).toEqual(turns) + expect(userMessageCountsByTurn(agent)).toEqual(steps.map(() => 1)) } finally { await ctx.fiber.dispose() } diff --git a/packages/core/agent/README.md b/packages/core/agent/README.md index 08e2fa1112..23a7bfddf0 100644 --- a/packages/core/agent/README.md +++ b/packages/core/agent/README.md @@ -54,13 +54,15 @@ Turn and step boundaries and the model token stream are durable `session/event` The handle every plugin programs against: -- `agent.send(content, options?)` — queue a message; starts a turn when idle. Content and resolved source become one detached, deeply frozen lossless-JSON record before `agent/queued` and enqueue; invalid data throws synchronously, and caller or notification-listener in-place mutation cannot change the log or model input (`agent/prompt-submit` still rewrites by returning replacement content). -- `agent.steer(content, options?)` — steer a running turn (inject between steps); uses the same owned acceptance boundary and behaves like `send` when idle +- `agent.send(content, options?)` — queue one independent FIFO item. If claimed, that item becomes the sole ordinary message in its turn; a claimed FIFO successor waits for that turn's checkpoint to settle. Broad cancellation, disposal, or a pre-start failure may instead drop it without a turn. Content and resolved source become one detached, deeply frozen lossless-JSON record before `agent/queued` and enqueue; invalid data throws synchronously, and caller or notification-listener in-place mutation cannot change the log or model input (`agent/prompt-submit` still rewrites by returning replacement content). The [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md) owns the rationale. +- `agent.steer(content, options?)` — submit steering while the agent is `running`. An open turn records it at the next steering checkpoint before a request or continuation decision; policy can still stop before another step. After turn close and its checkpoint, remaining steering becomes later queued input unless terminal turn policy, cancellation, or disposal discards it. The method uses the same synchronous snapshot-and-validation boundary as `send` and delegates to `send` when idle - `agent.inject(content, options?)` — accept detached in-session context without running the model; the next request sees its `context/message` with `content` rendered verbatim as a user-role message. `options.meta` persists opaque JSON state without rendering it. While a turn is open it joins that turn, deferring FIFO while the current tool batch executes and draining before turn close if execution is interrupted; while idle it is wrapped in a one-shot `injection` turn and durability checkpoint ([the turn-enclosure invariant](../../../.agents/notes/implemented/architecture/2026-06-15-turn-enclosure-invariant.md)). - `agent.cancel(reason?)` — cancel ALL pending work: clears the queued + steering FIFOs, aborts the in-flight step, and drops a turn about to start (the pre-step window) so a queued-but-not-started prompt never runs. A UI/ACP `session/cancel` maps to this. The single public stop primitive. Idle with nothing pending → a safe no-op. - `agent.whenIdle()` — resolve once the agent reaches quiescence after settling out of `running` (idle → immediately; disposed → awaits the loop exit). A non-owner's quiescence-observation hook: it observes the work settling WITHOUT tearing the agent down. Teardown is separate — a lifecycle owner stops and unregisters via `AgentHandle.dispose()`, which awaits the loop exit directly. - `agent.session`, `agent.status`, `agent.options`, `agent.id` +`running` describes a driver-wide drain interval, not proof that a turn is still open; it can cover turn close, the durability checkpoint, and consecutive queued turns. + ### Extension points - Agent creation: `AgentLoop.create()` is the concrete config-path implementation (in `dsh-agent-loop`), while programmatic consumers create/resume owned agents through `ctx.agents.create()` / `ctx.agents.resume()`. Replace the loop by implementing `Agent` and registering via `ctx.agents.register()`. diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index ebe1e503af..3af6e76371 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -38,9 +38,9 @@ export interface InjectOptions extends SendOptions { /** * An agent's lifecycle state, emitted on every transition as `agent/status`: - * `idle` (parked, waiting for queued work), `running` (a turn is in progress), - * `disposed` (terminal — no transition leaves it, and `send`/`steer`/`inject` - * throw). + * `idle` (parked, waiting for queued work), `running` (the driver is draining + * work and may be closing or checkpointing a turn), `disposed` (terminal — no + * transition leaves it, and `send`/`steer`/`inject` throw). */ export type AgentStatus = 'idle' | 'running' | 'disposed' @@ -54,8 +54,9 @@ export interface HookContext { /** * Prompt interception result. `allow.content` replaces the prompt and each - * `additionalContexts` entry becomes a separate context message. `block` records a - * durable `prompt/blocked`; an all-blocked batch ends a zero-step rejected turn. + * `additionalContexts` entry becomes a separate context message. `block` + * records a durable `prompt/blocked` and ends the claimed prompt's zero-step + * turn as rejected. */ export type PromptDecision = | { kind: 'allow'; content?: ContentBlock[]; additionalContexts?: HookContext[] } @@ -93,15 +94,20 @@ export interface Agent { readonly ctx: Context /** - * Queue detached, frozen lossless-JSON input; starts a turn when idle. + * Queue one detached, frozen lossless-JSON item. If claimed, it is the sole + * ordinary message in its FIFO-ordered turn; the next claimed item waits for + * that turn's checkpoint. * Invalid input throws synchronously before notification or enqueue. */ send(content: ContentBlock[], options?: SendOptions): void /** - * Steer a running turn: content is injected between steps of the current - * turn. Uses the same owned-value and synchronous-validation boundary as - * {@link send}; when idle, behaves exactly like that method. + * Submit steering while the agent is `running`. An open turn records it at + * the next steering checkpoint before a request or continuation decision; + * policy may stop before another step. After turn close and its checkpoint, + * any remainder is queued for a later turn; terminal `agent/turn-stop`, + * cancellation, or disposal may discard it. Uses the same synchronous + * snapshot-and-validation boundary as {@link send}; when idle, delegates to it. */ steer(content: ContentBlock[], options?: SendOptions): void @@ -115,10 +121,11 @@ export interface Agent { inject(content: ContentBlock[], options?: InjectOptions): void /** - * Clear queued and steering work, including work waiting to start, and abort - * the active step. The supplied reason is preserved across pre-step and active - * cancellation windows, and `whenIdle()` resolves after cancellation reaches - * quiescence. Idle cancellation is a no-op and does not arm a later cancel. + * Clear all queued and steering work, including items waiting to start, and + * abort the active step. The supplied reason is preserved across pre-step + * and active cancellation windows, and `whenIdle()` resolves after + * cancellation reaches quiescence. Idle cancellation is a no-op and does not + * arm a later cancel. */ cancel(reason?: string): void @@ -199,10 +206,10 @@ declare module 'cordis' { */ 'agent/pre-step'(this: Scoped, agent: Agent, turn: number, step: number, signal: AbortSignal): Promise | void /** - * Allow, rewrite, or block one drained prompt before it becomes a user + * Allow, rewrite, or block one claimed prompt before it becomes a user * message. Call `next()` for the unchanged default. - * @param agent - the agent draining its inbox. - * @param content - the drained message's blocks, as queued. + * @param agent - the agent whose turn claimed the message. + * @param content - the claimed message's blocks, as queued. * @param source - the message's resolved source. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode waterfall diff --git a/packages/core/session/src/types.ts b/packages/core/session/src/types.ts index 50950e3f6b..7188c58b97 100644 --- a/packages/core/session/src/types.ts +++ b/packages/core/session/src/types.ts @@ -106,8 +106,8 @@ export interface TurnEndReasonMap { /** At least one step reached its output-token ceiling, even if a plugin continued the turn. */ 'max-tokens': { kind: 'max-tokens' } /** - * Policy blocked every prompt before the first step. The zero-step turn still - * records a balanced durable boundary and the veto reason. + * Policy blocked the turn's claimed prompt before the first step. The + * zero-step turn still records a balanced durable boundary and veto reason. */ rejected: { kind: 'rejected'; reason: string } /** @@ -176,27 +176,28 @@ export type RequestHeaderReason = 'initial' | 'resume' | 'change' */ export interface SessionEventMap { /** - * Opens turn `turn`. `trigger` records what started it — a drained message - * batch or an idle-time injection. The turn is the durability/replay + * Opens turn `turn`. `trigger` records what started it — one claimed queued + * message or an idle-time injection. The turn is the durability/replay * boundary: every event sits between a `turn/start` and its matching * `turn/end` (the turn-enclosure invariant). */ 'turn/start': { turn: number; trigger: TurnTrigger } /** * Closes turn `turn` with the {@link TurnEndReason} that ended it. The loop - * fires the awaited `session/flush` checkpoint at every turn end, so the turn - * boundary is also the durable-commit boundary. + * awaits `session/flush` after an ordinary turn ends before claiming the next + * queued item. Success commits the turn; rejection is reported live and does + * not prevent later work. */ 'turn/end': { turn: number; reason: TurnEndReason } /** Opens step `step` of turn `turn` — one model call plus the tool executions it requested. */ 'step/start': { turn: number; step: number } /** Closes step `step` of turn `turn`. */ 'step/end': { turn: number; step: number } - /** A user-visible prompt (queued message drained at turn start). */ + /** A user-visible prompt (the queued message claimed for this turn). */ 'user/message': { content: ContentBlock[]; source: MessageSource } /** * Durable record of a prompt veto and its reason. It is log-only: the blocked - * prompt never enters the model-visible surface, including in a mixed batch. + * prompt never enters the model-visible surface, and its turn runs zero steps. */ 'prompt/blocked': { content: ContentBlock[]; source: MessageSource; reason: string } /** diff --git a/packages/examples/stdio-demo/tests/built-bin.e2e.ts b/packages/examples/stdio-demo/tests/built-bin.e2e.ts index d3e0a32d09..c2bb459cc9 100644 --- a/packages/examples/stdio-demo/tests/built-bin.e2e.ts +++ b/packages/examples/stdio-demo/tests/built-bin.e2e.ts @@ -104,8 +104,8 @@ async function makeConsumer( return dir } -/** Run the built bin in `cwd` against `configArg` with one stdin line; resolve with stdout/stderr + exit code. */ -function runBuiltBin(cwd: string, configArg: string, line: string): Promise<{ stdout: string; code: number; stderr: string }> { +/** Run the built bin in `cwd` against `configArg` with piped stdin; resolve with stdout/stderr + exit code. */ +function runBuiltBin(cwd: string, configArg: string, input: string): Promise<{ stdout: string; code: number; stderr: string }> { return new Promise((resolve, reject) => { // --expose-internals: the cordis Loader resolves bare plugin specifiers via // its internal module loader (active only under this flag); demo:echo passes @@ -128,7 +128,7 @@ function runBuiltBin(cwd: string, configArg: string, line: string): Promise<{ st }, 25_000) child.on('exit', (code) => { clearTimeout(timer); resolve({ stdout, code: code ?? -1, stderr }) }) child.on('error', (err) => { clearTimeout(timer); reject(err) }) - child.stdin.write(`${line}\n`) + child.stdin.write(`${input}\n`) child.stdin.end() }) } @@ -167,6 +167,17 @@ describe.skipIf(!existsSync(stdioBin))('dsh-stdio-demo BUILT bin (node lib/bin.j expect(code).toBe(0) }, 30_000) + it('runs two synchronously piped lines as two ordinary turns', async () => { + consumer = await makeConsumer('TWO-TURNS ready.') + const { stdout, code, stderr } = await runBuiltBin(consumer, './cordis.yml', 'first\nsecond') + expect(stderr).not.toContain('UNHANDLED') + expect(stdout).toContain('[main turn 1]') + expect(stdout).toContain('You said: "first"') + expect(stdout).toContain('[main turn 2]') + expect(stdout).toContain('You said: "second"') + expect(code).toBe(0) + }, 30_000) + it('boots when optional spill plugins are loaded from a built consumer install', async () => { consumer = await makeConsumer( 'SPILL-OK ready.', diff --git a/packages/ui/acp/src/index.ts b/packages/ui/acp/src/index.ts index 540d95b203..b190aeb47d 100644 --- a/packages/ui/acp/src/index.ts +++ b/packages/ui/acp/src/index.ts @@ -827,10 +827,11 @@ export function apply(ctx: Context, config: AcpConfig): void { // session/cancel maps to the queue-aware agent.cancel(reason): it aborts // a RUNNING step, clears the queued + steering FIFOs, and drops a // turn that is about to start (the pre-step window) — so a queued-but- - // not-yet-started prompt never runs, and a prompt accepted right after - // cannot be batched into the cancelled turn. Scoped to THIS session's + // not-yet-started prompt never runs, while a prompt accepted afterward + // remains a separate queued turn. Scoped to THIS session's // agent — a cancel in one session never touches another's stream or - // pending prompt (RFC 011 isolation). We ALSO settle the in-flight prompt + // pending prompt (multi-session isolation). + // We ALSO settle the in-flight prompt // as cancelled directly here: do NOT rely on the resulting turn/end to // settle it, because cancel() may drop the turn before any turn/end is // emitted, and removing this direct settle would move the RPC's diff --git a/packages/ui/acp/tests/dispose.spec.ts b/packages/ui/acp/tests/dispose.spec.ts index 5cbdb6859b..8baf076d58 100644 --- a/packages/ui/acp/tests/dispose.spec.ts +++ b/packages/ui/acp/tests/dispose.spec.ts @@ -223,7 +223,8 @@ describe('acp bridge — disposal & HMR safety', () => { it('per-session AgentHandle dispose leaves sibling agents untouched', async () => { // The factory returns a per-agent AgentHandle whose dispose() tears down - // EXACTLY that agent + its session — RFC 011 isolation. Create two agents + // EXACTLY that agent + its session — the registry's per-handle isolation + // contract. Create two agents // directly through the registry factory (the same path the ACP bridge uses), // dispose one handle, and assert the other survives, registered and // queryable, with its session still in the store. diff --git a/packages/ui/acp/tests/multi-session.spec.ts b/packages/ui/acp/tests/multi-session.spec.ts index 0881fe4199..efeb00f9ad 100644 --- a/packages/ui/acp/tests/multi-session.spec.ts +++ b/packages/ui/acp/tests/multi-session.spec.ts @@ -14,7 +14,7 @@ function messageTextFor(updates: { sessionId?: string; update: CapturedUpdate }[ .join('') } -describe('acp bridge — RFC 011 multi-session isolation', () => { +describe('acp bridge — multi-session isolation', () => { let storageDir: string let harness: BridgeHarness | undefined diff --git a/packages/ui/stdio/src/index.ts b/packages/ui/stdio/src/index.ts index 497f609c93..ac22cf6730 100644 --- a/packages/ui/stdio/src/index.ts +++ b/packages/ui/stdio/src/index.ts @@ -168,9 +168,9 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt // immediately — no turn will ever start, so there is nothing to wait // for. (Gating on an observed 'running' here would hang forever.) // - If work WAS submitted, exit the next time the agent settles to idle - // AFTER having run. Two subtleties this handles: the loop batches - // several queued messages into ONE turn (one idle), so we don't count - // sends; and agent.send() does NOT synchronously flip status to + // AFTER having run. Later lines may steer the active turn, and consecutive + // queued turns can share one running interval, so we don't count inputs; + // agent.send() also does NOT synchronously flip status to // 'running', so requiring an observed 'running' first (`sawRunning`) // avoids exiting in the gap before the turn starts and dropping work. let stdinClosed = false