Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into codex/project-instruction-files

Browse Source 
# Conflicts:
#	AGENTS.md
#	docs/config-catalog.md
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/core-data-structures/core.md
#	docs/event-producer-consumer.md
#	docs/persistence-catalog.md
#	docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md
#	docs/rfc/implemented/feature/2026-06-15-code-mode.md
#	docs/rfc/implemented/feature/2026-06-30-hook-bridges.md
#	docs/rfc/implemented/feature/2026-06-30-interception-seams.md
#	docs/rfc/implemented/feature/2026-07-08-repeat-tool-guard.md
#	docs/rfc/proposed/simplification/2026-07-04-prune-dead-core-spine-surface.md
#	examples/AGENTS.md
#	examples/acp-agent/cordis.yml
#	examples/acp-agent/tests/acp.snapshot.ts
#	examples/echo-agent/cordis.yml
#	examples/sandbox-acp-agent/cordis.yml
#	packages/cordis/tool-cordis/src/api-catalog.ts
#	packages/core/agent-core/README.md
#	packages/core/agent-core/src/index.ts
#	packages/core/agent-loop/README.md
#	packages/core/agent-loop/src/loop.ts
#	packages/core/agent-loop/tests/interception.spec.ts
#	packages/core/agent/src/types.ts
#	packages/core/tools/README.md
#	packages/core/tools/src/code-mode.ts
#	packages/core/tools/src/index.ts
#	packages/fs/fs-local/src/index.ts
#	packages/fs/fs/README.md
#	packages/fs/fs/src/index.ts
#	packages/guard/repeat-tool-guard/README.md
#	packages/guard/repeat-tool-guard/src/index.ts
#	packages/hooks/hooks-claude/src/index.ts
#	packages/hooks/hooks-codex/src/index.ts
#	packages/ui/acp-agent/src/index.ts
This commit is contained in:
Yichen Jiang
2026-07-14 19:50:25 +08:00
parent c2f2740a3e fb126e164a
commit ee111fd978
720 changed files with 21199 additions and 14129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
packages/core/agent-loop/README.md
View File

@@ -1,6 +1,6 @@
# dsh-agent-loop
THE concrete agent plugin: `ReactLoopAgent` and the loop driver. Implements the `Agent` interface and drives the session/turn/step lifecycle.
Concrete `ReactLoopAgent` implementation and loop driver.
This is the only package in the harness that contains concrete loop logic. Everything else is an abstract service or a plugin against extension seams — new behavior goes into plugins, not here.
@@ -8,18 +8,16 @@ This is the only package in the harness that contains concrete loop logic. Every
### Public API
Creation and resume are one rollback-covered transaction: construct a private session, concrete agent, and scoped context; await optional setup; enter both registries; announce `session/created` then `agent/created`; emit `agent/session-start`; and only then start the driver. Setup receives the full scoped `Context` as trusted same-process composition code and must not drive the unpublished agent. Ordinary typed identity and option inputs are borrowed under their readonly contract, while seed events and session metadata are validated and snapshotted because they cross the durable session boundary. An optional `AbortSignal` cancels only load/setup/publication and is detached before the returned handle becomes visible.
Creation and resume use one caller-owned transaction: compose while unpublished, enter both registries, announce lifecycle edges, then start the driver. Failure rolls back private resources; caller, handle, and provider teardown share one quiescence boundary. The interface contract and ownership order live in [`dsh-agent`](../agent/README.md) and the [agent-scope runtime RFC](../../../docs/rfc/implemented/architecture/2026-07-12-agent-scope-runtime-design.md).
The caller fiber and the AgentLoop provider are co-owners. `AgentFactory.createAgent(ownerCtx, options)` and `resume(ownerCtx, options)` receive caller ownership explicitly, while the factory keeps its own dependency context for `sessions`/`llm`/`tools`/`systemPrompt`; this lets a caller inject only `agents` without shrinking the new agent's service surface. Caller unload, handle disposal, or provider unload converge on one memoized quiescence boundary. Provider shutdown waits both resource teardown and the public create/resume wrapper that observed deactivation, so no continuation can publish after dependencies disappear.
Caller-chosen ids arbitrate only at final registry entry, so concurrent contenders may prepare but every loser rolls back. Entry-bound detach capabilities cannot remove a later same-id replacement. Teardown stops and drains—including idle-injection flushes—before detaching agent, session, and scope; ids become reusable at detach.
IDs are caller-chosen and assumed globally unique; accidental UUID collisions are outside the supported model. Two concurrent operations with the same agent or session id may both prepare, but the final `enter()` calls arbitrate publication and every loser rolls its private resources back. Each detach is bound to the exact entered object, so a stale disposer cannot remove a later same-id replacement. A detach requested during a synchronous creation notification waits for that dispatch to unwind, preserving created/disposed pairing. Teardown runs stop and drain (including outstanding idle-injection flushes) → detach agent → detach session → unwind scope; IDs become reusable at detach even if private scope cleanup is still finishing. Ordinary non-vetoing `agent/*` notifications go through `agentEvents(ctx, agent)`, per-step assembly goes through `assembleContextFor(agent)`, and turn-end durability checkpoints go through `ctx.sessions.flush(session)`.
- `ctx.agentLoop.create(id: string, options?: AgentOptions, meta?: { cwd?: string }): ReactLoopAgent` — synchronous no-setup create, used directly by programs and by `cordis.yml`-configured agents. It creates a fresh per-run session id `${id}-session-<uuid>` with optional metadata; the uuid avoids colliding with a prior durable log. Each call is a new session (a deliberate demo simplification — a real resume-or-create policy is a TODO). Disposed with the calling fiber.
- `ctx.agentLoop.create(id, options?, meta?)` synchronously creates a caller-fiber-owned agent with a fresh generated session id and optional cwd. Each call starts a new session rather than applying resume-or-create policy.
`AgentLoop` also implements the `AgentFactory` seam and registers itself via `ctx.agents.setFactory(this)`, so plugins create/resume agents through `ctx.agents` (the interface):
- `ctx.agents.create({ agentId, sessionId, meta?, seed?, agentOptions?, setup?, signal? }): Promise<AgentHandle>` — programmatic create on a caller-supplied `sessionId`, NOT `${id}-session`. It awaits the unpublished setup transaction before returning; `meta` carries cwd/lineage/seed-boundary metadata and `seed` reconstructs a forked child prefix after the session boundary validates and snapshots the durable values. `signal` applies only until this promise settles. The resolved [`AgentHandle`](../agent/README.md) owns exact teardown.
- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions?, setup?, signal? }): Promise<AgentHandle>` — load a persisted session via `ctx.sessionPersistence` ([session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md)), reconstruct its history, then await setup against a fresh unpublished agent scope before rollback-covered publication. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a clear error when persistence is absent). `signal` is creation-only. Returns an `AgentHandle`.
- `ctx.agents.create({ agentId, sessionId, meta?, seed?, agentOptions?, setup?, signal? })` validates and snapshots durable seed and metadata, awaits optional composition while unpublished, creates on the supplied session id, and returns an owned [`AgentHandle`](../agent/README.md). Its signal applies only until publication.
- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions?, setup?, signal? })` loads through optional [session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md), continues stored history and turn numbering under the resumed session id, and follows the same unpublished setup and creation-only cancellation boundary. It rejects when no persistence backend is mounted.
The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loop fiber (it discards the handle). For a programmatic agent, the handle holder is the only consumer-facing teardown capability; AgentLoop provider unload is the independent structural teardown edge, not another handle exposed to application code.
@@ -40,7 +38,7 @@ interface Config {
}
```
Agents listed in config are auto-created at startup. `cwd` applies only to fresh config-created sessions; `resumeSessionId` keeps the persisted session header. Config agents have no per-agent persona field: they use `dsh-system-prompt`'s deployment default, while programmatic factory callers can register an agent-scoped `deployment:persona` shadow in `setup`. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from `assembleContextFor(agent)` — the helper couples the typed agent with its matching scope selector. These are runtime facts of the agents THIS loop drives, unlike the `harness:identity` and default `deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
Configured agents start automatically. `cwd` applies only to fresh sessions; `resumeSessionId` retains persisted metadata. They use the deployment persona. Programmatic setup can shadow it per agent. This plugin supplies the per-agent `model` and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
### Exported concrete class
@@ -50,55 +48,11 @@ Agents listed in config are auto-created at startup. `cwd` applies only to fresh
### Loop lifecycle (`loop.ts`)
The internal loop driver runs one agent for its whole lifetime:
The driver owns one agent for its lifetime. It records turn, step, request, stream, and tool boundaries in the session log; live extension events coordinate policy around those durable facts. The [architecture turn flow](../../../docs/architecture.md#turn-flow) and generated [event catalog](../../../docs/cordis-catalog/events.md) are the authoritative sequence and signatures.
```
create agent → emit agent/session-start(source) ⟵ once, before turn 1
forever:
wait for queued messages (idle)
TURN (error-contained):
'turn/start'
each queued: waterfall agent/prompt-submit → allow (→ session('user/message'),
inject each additionalContexts entry) | block (→ session('prompt/blocked'), drop)
if every prompt blocked: 'turn/end'(rejected), no step ⟵ zero-step turn
STEP loop:
drain steering
assembly = await systemPrompt.assemble(assembleContextFor(agent))
⟵ renderPrompt(assembly) IS the full prompt
prefix ??= waterfall agent/session-prefix ⟵ once per instance (first step): frozen
session prefix; on the header, never history
await serial agent/pre-step(…, prefix) ⟵ surface mutation (compaction) outside the step;
pressure gates see the prefix the request carries
boundary = session.deriveMessages() ⟵ reconstruction boundary: same sync frame,
session('step/start') strictly before step/start
config = waterfall agent/request ⟵ frozen seed; return a replacement to switch
session('request/header'[-delta]) ⟵ the header event this request owes the log
stream llm.stream(freeze({header..., messages: prefix+boundary})) → session('assistant/chunk')
message = waterfall agent/step-result
session('assistant/message')
each tool-call: session('tool/call')
→ tools.execute() [pre waterfall → monotonic guards → around dispatch → post waterfall → final notification]
→ session('tool/result')
append buffered deferred/post-execute additionalContexts as session('context/message')(s)
drain steering → session('steering/message')
cont = waterfall agent/turn-continuation → ContinuationDecision
({action:'continue', reason?} records reason as next-step steering)
pending steering can override an ordinary stop
terminal = serial agent/turn-stop → ContinuationStop | undefined
(after ordinary decision/reason/steering folding)
if terminal stop, or ordinary action==stop with no pending steering: break
session('turn/end')
await session/flush
terminal turn: discard steering added before/during close and flush; keep ordinary queued sends
ordinary turn: re-enqueue leftover steering as queued
idle unless more queued
```
Plugin failure ends the current turn, not the loop. Cancellation clears pending work and aborts the current step without leaking to the next prompt. Terminal continuation stops remain authoritative through turn close and durability flush.
Error containment: a throwing plugin ends the **turn**, never the loop. A throwing `agent/turn-stop` policy likewise fails the turn closed. A successful terminal stop stays authoritative through `turn/end` and `session/flush`, preventing their listeners from resurrecting steering through the late fallback. Dispose mid-turn emits `agent/status('disposed')` and ends with reason `disposed`. A step that hits the model's output-token ceiling makes the turn end `max-tokens` (the rule: any `max-tokens` step in the turn surfaces as `max-tokens`; `disposed`/`aborted`/`error` still take precedence) — distinct from a clean `completed` stop.
Cancellation: `agent.cancel()` is the single public stop primitive — it clears the queued + steering FIFOs, aborts the in-flight step, and drives a turn-scoped marker the driver checks at every point a turn could start or continue (right after the idle wait, after the `running` flip, before each step, and at the continuation gate) so a turn about to start is dropped. A cancelled turn ends `aborted`; a queued-but-not-started prompt never runs and cannot be batched into the cancelled turn. The marker is reset once per loop iteration, so a cancel governs exactly one turn and never leaks onto a later prompt. (The loop still aborts its own per-step `AbortController` directly on disposal and from `cancel()`; that controller is loop-internal, not a public verb.)
### What is NOT here
### What belongs to plugins
Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
- Hooks and policy: the relevant `agent/*` checkpoints plus the guarded `tools/pre-execute` → `tools/execute` → `tools/post-execute` → `tools/result` pipeline; exact signatures and modes live in the [generated event catalog](../../../docs/cordis-catalog/events.md)
@@ -107,3 +61,24 @@ Everything that goes beyond "call the model, run the tools, repeat" belongs to p
- Sub-agents: implemented outside the loop as `ctx.subagents` providers; in-process providers use `ctx.agents.create()` and owned `AgentHandle` teardown, while child streaming/progress and background/poll collection remain deferred.
- Persistence: `session/event` + `session/flush`
- UI: `session/event` (assistant token stream, boundaries, tool activity) + `agent/*` control events (`agent/status`, `agent/created`/`agent/disposed`)
## Model Experience
### Complete conversation request
**What the model sees**: For each step, the loop sends the rendered per-agent system prompt, visible tool schemas, the frozen session prefix, and the session's derived messages. It supplies `model` and `cwd` variable values but no additional fixed prose.
**Token effect**: System text, schemas, and prefix are paid again on every step. Per-agent scoping chooses the initial contributions, while the authoritative assembly waterfall can alter the final request and makes its listener responsible for protocol coherence.
### Retained message history
**What the model sees**: Accepted user messages, assistant messages, tool calls and results, injected context, and steering are logged and sent on later steps. Raw stream chunks, lifecycle boundaries, and other log-only events are excluded.
**Token effect**: Input grows with every surface message until a compaction replacement shadows older nodes; a multi-step tool turn resends the accumulated prefix and history each step.
## Known Limitations and Deferred Work
- **Tool calls within a step execute sequentially** — parallel execution waits on concurrency-safety metadata in the tool contract (see `dsh-tools`).
- **No resume-or-create policy on the config path** — config-driven `create()` starts a fresh `${id}-session-<uuid>` every run (`TODO(demo)`), and a config `resumeSessionId` whose resume fails logs a warning and creates no agent.
- **Config agents have no per-agent persona field or setup hook** — they use the deployment persona; scoped persona/tool composition is available only through the programmatic `ctx.agents.create()` / `resume()` factory options.
- **No built-in turn budget** — the default continuation is `continue` whenever a step had tool calls or steering; bounding a runaway turn requires an `agent/turn-continuation` force-stop plugin.

89
packages/core/agent-loop/src/agent.ts
View File

@@ -48,10 +48,8 @@ export interface PreparedReactLoopAgent {
}
/**
* Construct one concrete agent together with unforgeable, instance-bound
* lifecycle controls. The package surface deliberately exposes neither source
* subpaths nor this helper: setup code may identify the concrete class, but it
* cannot publish or start the factory's unpublished instance.
* Construct an unpublished concrete agent with instance-bound lifecycle
* controls. Only those paired controls can publish or start this instance.
* @param ctx - the agent-loop service context used for driving and events.
* @param id - the concrete agent identity.
* @param options - loop options for the agent.
@@ -131,16 +129,7 @@ export class ReactLoopAgent implements Agent {
* leave it set to wrongly drop a later prompt.
*/
private cancelRequested = false
/**
* The resolved reason for the pending {@link cancel} (`reason ?? 'cancelled'`),
* read by the driver loop's marker branches so a turn dropped in a
* marker-only window (pre-step / continuation, where no `AbortController`
* carries the reason) ends with the SAME `{kind:'aborted', reason}` the
* mid-step abort path produces from `abort.signal.reason`. Without this the
* caller's `cancel(reason)` would be silently replaced by the literal
* 'cancelled' whenever the cancel landed outside a running step — making the
* logged reason race-dependent and the public `reason?` param half-effective.
*/
/** Pending cancellation reason, preserved even outside an active step signal. */
private cancelReason = 'cancelled'
private disposed: Promise<void>
private resolveDisposed!: () => void
@@ -179,11 +168,7 @@ export class ReactLoopAgent implements Agent {
private setStatus(status: AgentStatus): void {
if (this._status === status || this._status === 'disposed') return
this._status = status
// Release quiescence waiters on a transition OUT of running BEFORE emitting
// (the disposer handles the disposed transition separately). Settling first
// means a throwing `agent/status` subscriber cannot starve a `whenIdle()`
// waiter (docs/defensive-patterns.md "contain callback exceptions" — a lifecycle await must
// not hang on one bad listener).
// Settle first so a throwing status listener cannot starve quiescence waiters.
if (status !== 'running') this.settleIdleWaiters()
agentEvents(this.loopCtx, this).emit('agent/status', status)
}
@@ -275,18 +260,8 @@ export class ReactLoopAgent implements Agent {
// Decide the durability checkpoint from the log: an accepted one-shot
// turn must be flushed even when its message append was the failing step.
const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
// Checkpoint the one-shot turn for durability, exactly as the loop does at
// every turn/end. The loop is NOT running (we are idle), so nothing else
// will flush this turn. Fire-and-forget with error containment: inject()
// is synchronous, and a persistence backend failing must not throw into
// the caller (e.g. a tool-bash task-done callback). Disposal still drains
// independently, so a slow flush is safe. The task is tracked until it
// settles: driver disposal awaits every pending idle-injection checkpoint
// before unregistering the agent or detaching the session. A flush failure
// is reported via agent/error (step 0 — the idle-injection convention,
// there is no real step) AND the logger, mirroring the loop's post-turn/end
// flush path so plugins monitoring agent/error see idle-injection
// persistence failures too. A throwing agent/error listener is contained.
// Keep inject() synchronous: report checkpoint failures live instead of
// rejecting the caller, and track the task so disposal still drains it.
if (turnRecorded) {
// Through the store's flush (the carrier owner), never a raw parallel.
const flush = this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
@@ -296,10 +271,7 @@ export class ReactLoopAgent implements Agent {
agentEvents(this.loopCtx, this).emit('agent/error', turn, 0, err)
})
this.pendingIdleFlushes.add(flush)
// Attach the same retirement callback to both settlement arms so even a
// logger failure in the catch above cannot become an unhandled rejection.
// Teardown uses allSettled for the same reason: a reporting failure must
// not strand ownership.
// Retire on either settlement path.
const retire = (): void => { this.pendingIdleFlushes.delete(flush) }
void flush.then(retire, retire)
}
@@ -307,15 +279,7 @@ export class ReactLoopAgent implements Agent {
}
cancel(reason?: string): void {
// Arm-gate: only mark a cancellation when there is actually work to cancel —
// a running turn, an in-flight step, or queued/steering work. An idle cancel
// with nothing pending is a true no-op; arming the marker then would wrongly
// drop the NEXT legitimate prompt (the marker is consumed only at the loop's
// turn-decision points, which an idle parked loop does not reach until woken
// by a real send()). Note the gate canNOT be `status === 'running'` alone:
// the pre-step window (a send() queued but the loop not yet flipped to
// running) has status `idle` with `hasQueued` true, and the marker exists
// precisely to cover it.
// Arm only for current work; an idle marker would cancel the next prompt.
if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
this.cancelRequested = true
// Capture the resolved reason for the marker-only windows (pre-step /
@@ -335,29 +299,14 @@ export class ReactLoopAgent implements Agent {
}
/**
* Resolve once the agent has reached quiescence after settling out of
* `running`. If it is already disposed, awaits {@link done} (the loop-exit
* promise) — `agent/status('disposed')` fires in the disposer BEFORE the
* driver loop has unwound, so it is NOT itself a quiescence signal. If it is
* idle AND has no queued work, resolves immediately. Otherwise queues an
* internal waiter (see {@link idleWaiters}) released on the next
* running→idle/disposed transition, resolving on `idle` directly (the turn
* fully ended) or chaining {@link done} on `disposed` (wait for the loop to
* actually exit). Implements the {@link Agent.whenIdle} contract: a non-owner
* quiescence-observation hook, distinct from teardown (a lifecycle owner stops
* and unregisters via `AgentHandle.dispose()`, whose driver boundary awaits
* both {@link done} and outstanding idle-injection flushes, not through this).
* Resolve immediately when idle with no queued work, on the next quiescent
* idle transition otherwise, or after driver exit when already disposed.
* This observes quiescence; it does not own teardown.
*/
whenIdle(): Promise<void> {
if (this._status === 'disposed') return this.done
if (this._status !== 'running' && !this.#inbox.hasQueued) return Promise.resolve()
// Register an internal waiter (resolved by settleIdleWaiters on the next
// running→idle/disposed transition), NOT an effect-scoped `ctx.on` listener:
// a concurrent fiber disposal runs this agent's listener disposers, which
// could remove a `ctx.on` waiter before the `disposed` transition fires and
// hang the promise. On disposal the disposer settles the waiter AND we chain
// `done` here for true loop-exit quiescence (status flips to disposed before
// the loop unwinds); a plain idle transition resolves directly.
// Agent-owned waiters survive concurrent fiber disposal.
return new Promise<void>((resolve) => {
this.idleWaiters.push(() => {
resolve(this._status === 'disposed' ? this.done : undefined)
@@ -393,12 +342,7 @@ export class ReactLoopAgent implements Agent {
isCancelled: () => this.cancelRequested,
cancelReason: () => this.cancelReason,
clearCancel: () => { this.cancelRequested = false },
// Settle whenIdle() waiters WITHOUT a status transition — the pre-step
// cancel-skip path drops the about-to-run turn and re-parks without ever
// flipping running→idle, so a waiter registered in the pre-step window
// (status idle, hasQueued was true) would otherwise hang. This emits no
// agent/status, so an ACP agent/status listener never sees a spurious idle
// that would resolve a freshly-queued prompt as cancelled.
// Pre-step cancellation re-parks without emitting a status transition.
settleIdle: () => { this.settleIdleWaiters() },
})
}
@@ -438,11 +382,8 @@ export class ReactLoopAgent implements Agent {
// cleanup. The normal loop contains turn failures itself; allSettled is the
// final lifecycle backstop for anything outside those boundaries.
await Promise.allSettled([this.done])
// No new inject() can start after the synchronous disposed transition.
// Loop because settled tasks retire themselves in promise reactions that
// may run beside this continuation; either the set is empty or this waits
// the exact remaining quiescence boundary. allSettled keeps a failure in
// error reporting from skipping registry/session/scope disposers.
// Repeat because settled flushes retire in adjacent promise reactions;
// allSettled keeps reporting failures from skipping ownership teardown.
while (this.pendingIdleFlushes.size > 0) {
await Promise.allSettled([...this.pendingIdleFlushes])
}

9
packages/core/agent-loop/src/index.ts
View File

@@ -74,12 +74,9 @@ function signalAbortError(id: AgentId, signal: AbortSignal): Error {
}
/**
* One create/resume transaction from caller ownership through unpublished
* setup, rollback-covered publication, and final quiescent teardown.
*
* The class deliberately owns the state machine in one place. Registries only
* arbitrate identity at their final `enter()` calls; before that point every
* resource is private to this transaction.
* Caller-owned create/resume transaction through rollback-covered publication
* and quiescent teardown. Resources remain private until the final registry
* entry arbitrates identity.
*/
class AgentCreationTransaction {
private active = true

482
packages/core/agent-loop/src/loop.ts
View File

@@ -1,9 +1,7 @@
/**
* The agent loop driver: one `runLoop()` invocation drives one agent for its
* whole lifetime. Error-contained at the turn level — a throwing plugin ends
* the turn, never kills the loop. See the JSDoc on `runLoop()` for the full
* lifecycle pseudo-code.
*
* Drives one agent across queued durable turns. Turn failures are contained so
* later work can run; the session log, not this driver, owns conversation state.
* See docs/rfc/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md.
* @module dsh-agent-loop/loop
*/
@@ -25,33 +23,12 @@ import type { Inbox } from './inbox.ts'
/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
type CodedError = Error & { code?: string }
/**
* Normalize an arbitrary thrown value into a coded Error. A real Error passes
* through (its `code`, if any, is preserved by {@link errorData}); a non-Error
* throw is wrapped in a {@link HarnessError} with code `UNKNOWN` and the
* original value chained as `cause`, so a bad throw still carries a routable
* code instead of degrading to a bare message.
*/
/** Normalize thrown values while preserving an existing error code. */
function toError(error: unknown): CodedError {
return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
}
/**
* Map a model-call {@link FinishReason} to the step error it should raise, or
* `undefined` when the step completed normally.
*
* Adapters report provider/transport failures one of two sanctioned ways (see
* the StreamChunk contract in dsh-llm): throw from `stream()` (handled by the
* caller's try/catch), OR end the stream with a finish-error/aborted chunk
* (the only option for adapters that can't throw mid-stream, e.g.
* library-backed ones). This translates the latter into a thrown step error
* so the turn ends error/aborted (the failure recorded on `turn/end.reason`),
* never as a normal `completed` assistant message.
*
* `FinishReason` is merge-extensible (plugins/adapters can add `kind`s), so
* the switch handles the known terminal-failure kinds and treats every other
* kind — `stop`, `tool-calls`, `max-tokens`, future additions — as success.
*/
/** Convert terminal failure finishes into step errors; unknown extensible finishes remain successful. */
function finishError(finish: FinishReason): CodedError | undefined {
switch (finish.kind) {
case 'error': {
@@ -78,19 +55,7 @@ function errorData(err: CodedError): { message: string; code?: string } {
return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
}
/**
* The turn-end contribution of a step's *successful* finish, or `undefined`
* when the step finished ordinarily (a plain `completed`).
*
* {@link finishError} has already converted `error`/`aborted` finishes into
* thrown step errors, so the finishes that reach here are `stop`,
* `tool-calls`, `max-tokens`, or a future merge-extensible kind. Only
* `max-tokens` carries forward as a distinct {@link TurnEndReason}: a step that
* hit the output-token ceiling ended the turn cut-short rather than by the
* model's choice. `stop`/`tool-calls`/unknown kinds contribute nothing beyond
* the default `completed`. {@link runTurn} applies this with the rule "any
* `max-tokens` step in the turn makes the turn end `max-tokens`".
*/
/** Map a successful max-token finish onto the turn reason; other successful finishes add nothing. */
function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
switch (finish.kind) {
case 'max-tokens':
@@ -103,11 +68,7 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
}
}
/**
* Ambient handles the loop driver receives from the agent. Decouples the
* pure function `runLoop` from the mutable ReactLoopAgent fields, making the
* loop testable without a real agent.
*/
/** Mutable agent controls supplied to the loop driver. */
export interface LoopHandle {
/** Native-private agent inbox handed to the driver only at internal startup. */
readonly inbox: Inbox
@@ -116,122 +77,37 @@ export interface LoopHandle {
/** Resolves when the agent is disposed — unblocks the idle wait. */
disposed: Promise<void>
isDisposed(): boolean
/**
* Whether a `cancel()` is pending for the current turn. The driver checks this
* at every decision point where a turn could start or continue (right after
* the idle wait, after the `running` flip, before each step, and at the
* continuation gate) and drops the about-to-run / continuing turn. Reset once
* per loop iteration via {@link clearCancel} after the turn returns, so the
* marker governs exactly one cancellation and never leaks to a later prompt.
*/
/** Whether cancellation is pending for the current loop iteration. */
isCancelled(): boolean
/**
* The resolved reason for the pending cancel (`reason ?? 'cancelled'`), read
* by the marker branches (pre-step / continuation) so a turn dropped where no
* `AbortController` carries the reason still records the caller's
* `cancel(reason)` value — matching the mid-step abort path. Only meaningful
* when {@link isCancelled} is true.
*/
/** Resolved pending-cancellation reason; meaningful only while {@link isCancelled} is true. */
cancelReason(): string
/** Clear the cancel marker (called once per iteration after the turn returns). */
clearCancel(): void
/**
* Settle pending `whenIdle()` waiters WITHOUT a status transition. Used by the
* pre-step cancel-skip path: it drops the about-to-run turn and re-parks at the
* idle wait, so no `running→idle` transition fires to settle a `whenIdle()`
* waiter that was registered in the pre-step window — this settles it directly
* (it emits no `agent/status`, so an ACP `agent/status` listener never sees a
* spurious idle that would resolve a freshly-queued prompt as cancelled).
*/
/** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
settleIdle(): void
}
/**
* The agent loop. One invocation drives one agent for its whole lifetime:
*
* ```
* create agent → emit agent/session-start(source) ⟵ once, before turn 1
* forever:
* wait for queued messages (idle)
* TURN (error-contained — a throwing plugin ends the turn, never the loop):
* 'turn/start'; each queued msg: waterfall agent/prompt-submit ⟵ durable turn boundary (no agent/* mirror)
* allow → session('user/message'…) (+ inject additionalContexts) | block → drop
* every prompt blocked → 'turn/end'(rejected), 0 steps
* STEP loop:
* drain steering → session('steering/message') ⟵ catches late steering
* assembly = ctx.systemPrompt.assemble(assembleContextFor(agent)) ⟵ waterfall system-prompt/assemble
* (scope-filtered; scoped sections/tools join); renderPrompt
* (persona section + {{variables}}) IS the full prompt
* prefix ??= waterfall agent/session-prefix ⟵ once per loop instance (first step): frozen
* session prefix; logged on the header, never
* session history (scope-filtered, fused dispatch)
* await events.serial('agent/pre-step', …, prefix) ⟵ surface mutation (compaction) OUTSIDE the step;
* pressure gates see the prefix the request carries
* boundary = session.deriveMessages() ⟵ the reconstruction boundary: snapshot in the
* session('step/start') same sync frame, strictly before step/start
* config = waterfall agent/request(config) ⟵ frozen seed; a returned replacement switches
* session('request/header'|'request/header-delta') ⟵ the header event this request owes the
* log (initial/resume anchor, delta, fallback)
* req = freeze({header..., messages: prefix+boundary, sessionId, signal})
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks, frozen req)
* session('assistant/chunk')
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
* session('assistant/message' {content, usage?}) session records what actually ran
* each tool-call in msg (sequential, abort-checked):
* session('tool/call'); ctx.tools.execute() ⟵ tools/pre-execute (allow/deny/ask)
* → dispatch → tools/post-execute
* session('tool/result')
* append buffered deferred/post-execute contexts → session('context/message')(s)
* drain steering → session('steering/message')
* session('step/end') ⟵ durable step boundary (no agent/* mirror)
* cont = waterfall agent/turn-continuation ⟵ ContinuationDecision; default
* {action: hadToolCalls||steered ? 'continue':'stop'}; a continue.reason is
* recorded as next-step steering
* if action==stop && steering arrived (step/end/continuation listeners): continue anyway
* terminal = serial agent/turn-stop ⟵ stop or abstain; after all ordinary
* continuation and steering folding
* if terminal: discard pending steering and break
* if action==stop: break
* session('turn/end') ⟵ durable turn boundary (no agent/* mirror)
* await ctx.sessions.flush(session) ⟵ durability checkpoint (store-owned carrier)
* re-enqueue leftover steering as queued ⟵ steering is never stranded
* idle (emit agent/status) unless more queued
* ```
* Drive queued batches as durable turns until disposal. Plugin failures end the
* current turn without terminating the driver.
* @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
* @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
* @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
*/
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
// Per-instance transmission bookkeeping: whether THIS loop instance has
// anchored the log's header fold yet (its first request logs a
// 'initial'/'resume' request/header snapshot). Everything else the request
// needs is read from the session log itself — the loop holds no
// conversation state (the reconstructability RFC).
// Per-instance prefix and request-header state; conversation history remains in the session log.
const transmission = createTransmissionLog()
const { session } = agent
// The fused agent-subject dispatcher: every agent/* dispatch below carries
// the agent's scope (an `agent.ctx` listener hears only this agent) with
// the subject injected — one spelling, checked by the dev invariants.
// Fused subject and scope carrier for every agent event below.
const events = agentEvents(ctx, agent)
while (!handle.isDisposed()) {
await handle.inbox.waitForQueued(handle.disposed)
if (handle.isDisposed()) break
// Pre-step cancel (window 1): a `cancel()` landed after a `send()` woke the
// idle wait but before we flip to `running`. The cancelled queued/steering
// work is already cleared by `cancel()`. Clear the marker, then:
// - if NOTHING new is queued, drop the about-to-run turn and re-park,
// settling any `whenIdle()` waiter DIRECTLY (no running→idle transition
// fires here to settle it) and WITHOUT emitting `agent/status` (an ACP
// listener must not see a spurious idle that resolves a freshly-queued
// prompt as cancelled);
// - if a NEW prompt was queued AFTER the cancel (a send() that raced in
// before the loop resumed), the marker was for the cancelled work only —
// fall through and run the new prompt's turn. Do NOT settle waiters here:
// a whenIdle() waiter must wait for that new turn's running→idle, not
// resolve before it runs (the quiescence contract).
// Cancellation between wake and `running` skips only the cancelled work;
// a replacement prompt still runs and owns the eventual idle transition.
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
@@ -242,18 +118,8 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
handle.setStatus('running')
// Pre-step cancel (window 2): `setStatus('running')` emits `agent/status`
// SYNCHRONOUSLY, so a `running` listener can `cancel()` in the gap between the
// check above and `runTurn`. Mirror window 1: clear the marker, then
// - if NOTHING new is queued, drop the about-to-run turn and transition
// back to `idle` (`running` was already emitted, so a real idle
// transition balances the status AND settles `whenIdle()` waiters);
// - if a NEW prompt was queued AFTER the cancel (a `running` listener that
// cancels then sends), the marker was for the cancelled work only — fall
// through and run the new prompt's turn (status is already `running`), so
// a `whenIdle()` waiter resolves on THAT turn's running→idle, not before
// it runs. Settling here would resolve quiescence while the replacement
// is still queued and unrun (the same early-resolve race window 1 fixes).
// A synchronous `running` listener can cancel before `runTurn`; balance the
// status only when no replacement prompt was queued by that listener.
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
@@ -262,24 +128,13 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
}
}
// Re-derive the turn number from the log each iteration (do NOT keep a local
// counter): an idle `agent.inject()` can append its own one-shot turn while
// the loop waits above, so the next real turn must continue from whatever
// turn number is actually last in the log — a stale counter would collide.
// Idle injection can add a turn, so derive the next number from the log.
const turn = lastTurnNumber(session) + 1
let terminalStopped = false
try {
terminalStopped = await runTurn(ctx, events, agent, handle, turn, transmission)
} catch (error: unknown) {
// Backstop: runTurn rethrows only a PRE-turn throw (the invariant guard
// before turn/start) — no turn/start was appended, so no turn is open and
// none is owed. A session `error` here would land outside any turn (after
// the previous turn/end), where the persistence backend drops it as a
// crash tail (the turn-enclosure RFC). Report via agent/error + the logger only; the
// driver survives and moves on.
// Acceptance and internal dispatch validation can reject before
// turn/start commits. Report that supported pre-turn failure without
// inventing a turn/end for a turn that never opened.
// Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
try {
@@ -287,21 +142,10 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
} catch { /* contained: a throwing agent/error listener must not kill the driver */ }
}
// Reset the cancel marker UNCONDITIONALLY here, after the turn returns and
// before the next iteration's idle wait. NOT gated on the idle transition
// below: a `send()` that lands during the cancelled turn's flush window makes
// `hasQueued` true at the `setStatus('idle')` guard, so an idle-gated reset
// would never fire and the stale marker would wrongly drop that next prompt's
// turn. Resetting per iteration scopes the marker to exactly the turn that was
// cancelled.
// Reset per iteration, including when a prompt arrives during the flush window.
handle.clearCancel()
// Steering that arrived too late to join an ordinary turn (turn-end
// listeners, flush) becomes queued input so it is never stranded. A
// terminal-stop owner is the deliberate exception: discard the steering
// again after the close + flush window so terminal policy cannot be undone
// after its in-turn drain. Ordinary queued sends live in a separate FIFO and
// remain untouched.
// Late steering becomes queued input unless terminal policy stopped the turn.
for (const message of handle.inbox.drainSteering()) {
if (!terminalStopped) handle.inbox.enqueue(message)
}
@@ -315,10 +159,7 @@ async function runTurn(
): Promise<boolean> {
const { session } = agent
// --- Pre-turn. A throw here (the invariant guard) is owed NO turn/end —
// turn/start has not been appended — so it propagates to runLoop's backstop
// untouched. The queued messages are drained here but appended AFTER
// turn/start (below), so every event in the log lives inside a turn.
// Drain before opening the turn, but append only after `turn/start`.
const queued = handle.inbox.drainQueued()
const first = queued[0]
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
@@ -331,28 +172,17 @@ async function runTurn(
let errorReported = false
let terminalStopped = false
// Close the open step exactly once (idempotent via stepOpen). Post-commit
// session/event observers are contained by Session; a pre-commit validator
// failure still escapes so the outer recovery path may retry the boundary or
// fail loudly without pretending an uncommitted step/end exists.
// Close the committed step once; pre-commit validation failure still escapes.
const closeStep = (): void => {
if (!stepOpen) return
session.append('step/end', { turn, step })
stepOpen = false
}
// Record a step/turn failure exactly once: set the error reason (carrying the
// failing `step` — the durable failure lives entirely on turn/end.reason, there
// is no separate session error event) and emit agent/error (contained — trap: a
// throwing agent/error listener must not re-escape and strand the turn).
// Disposal and abort set `reason` directly without calling this (they are not
// failures).
// Record the durable turn failure once and contain the live error notification.
const failTurn = (err: CodedError): void => {
if (errorReported) return
errorReported = true
// The turn is still open here. Post-commit observers cannot escape append,
// and a pre-commit turn/end veto leaves no closing boundary to overwrite.
// Set the reason that the next successful closeTurn will append.
reason = { kind: 'error', step, ...errorData(err) }
try {
events.emit('agent/error', turn, step, err)
@@ -362,9 +192,7 @@ async function runTurn(
}
}
// Close the turn. Post-commit observer failures are contained by Session;
// pre-commit validation failures escape to recovery instead of being mistaken
// for a committed boundary. Turn boundaries are durable session events only.
// Pre-commit validation failure escapes rather than masquerading as a committed boundary.
const closeTurn = (): void => {
session.append('turn/end', { turn, reason })
}
@@ -419,11 +247,7 @@ async function runTurn(
}
while (true) {
// A fully-blocked batch (every prompt vetoed by prompt-submit) opens a
// zero-step turn that ends `rejected`: break BEFORE the first step so the
// boundary stays balanced (turn/start → turn/end) and the block is a
// durable in-turn fact. `anyAllowed` never changes inside the loop, so this
// only ever fires on the first iteration.
// A fully blocked batch closes its zero-step turn as rejected.
if (!anyAllowed) {
reason = { kind: 'rejected', reason: lastBlockReason }
break
@@ -442,48 +266,20 @@ async function runTurn(
const abort = new AbortController()
handle.setAbort(abort)
// Assemble the system prompt for this step. Done HERE (before step/start)
// because the pre-step seam needs it: compaction measures token pressure
// against the system prompt (it counts toward the budget). runStep reuses
// this same assembly for the request, so the prompt is assembled once per
// step. renderPrompt IS the full prompt — the persona is the order-0
// section (owned by dsh-system-prompt) and `{{variable}}`
// interpolation happens in the render, so there is no separate join.
// Assemble once before pre-step so pressure checks and the request share the same prompt.
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
const fullSystemPrompt = renderPrompt(assembly)
// Interruption landing after assembly: dispose() or cancel() in a
// turn-start listener (or a listener whose promise resolved before the
// await above) arms either handle.isDisposed() or handle.isCancelled().
// The Abort was created first, so any concurrent abort also lands on it.
// Drop the about-to-start step WITHOUT running the seam — no step is open
// yet, so end the turn accordingly (disposed wins for an unambiguous
// reason).
// Cancellation or disposal during assembly ends the turn before any step opens.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
break
}
// Compose the session prefix ONCE per loop instance, lazily before the
// instance's first pre-step: request-only messages placed in front of
// the ENTIRE derived history on every request this instance sends. It
// MUST precede the pre-step seam so compaction gates on THIS instance's
// prefix — reading a previous instance's logged prefix would let a
// resumed/forked instance whose contributor grew skip compaction and
// ship an over-window first request. The result is deep-cloned
// (decoupled from listener-held references), deep-frozen, and cached on
// the transmission bookkeeping, so reuse is structural — the prefix
// cannot change mid-session and the provider prefix cache holds by
// construction (resume = a new instance = a recompose, anchored by its
// 'resume' snapshot). The prefix is not session history — the header
// event in runStep is its only durable record
// (EpochHeader.messagePrefix). The frozen empty seed serves both the
// listener chain and the no-listener fallback: a contribution is a
// RETURNED extension of `await next()`, never an in-place push. This
// runs OUTSIDE the step, before the boundary snapshot: a composing
// listener's session append lands before the boundary and joins the
// CURRENT request.
// Compose the request-only prefix once per loop instance before pressure
// checks. It precedes all derived history and is recorded only in the
// request header, not as session history.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
const composed = await events.waterfall(
@@ -491,16 +287,7 @@ async function runTurn(
() => Promise.resolve(emptyPrefix),
)
// Interruption landing during prefix composition: mirror the assembly
// window above — drop the about-to-start step without running the
// seam, and DISCARD the composition instead of caching it. An
// abort-aware listener may have returned a degraded fallback under
// the firing signal; committing it would ship a prefix no request
// ever used (and no header ever logged) on this instance's next real
// request. The next turn recomposes under a live signal — the cache
// only ever holds a fully composed prefix. The cache-hit path needs
// no such check: nothing awaits between the assembly check above and
// the pre-step seam.
// Never cache an interrupted composition; the next turn recomposes it.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
@@ -509,19 +296,7 @@ async function runTurn(
transmission.sessionPrefix = deepFreeze(structuredClone(composed))
}
// Pre-step surface-mutation checkpoint (compaction), fired OUTSIDE the
// step: after `turn/start` (and the prior step's close) but before
// `step/start`, so a compaction's log-only `compact/*` records and its
// replacement node land cleanly outside any step (honest structure that
// crash-safety relies on — a dangling `compact/start` sits before the
// synthetic `turn/end` repair appends). Serial (awaited, in order, no
// veto): each listener completes its surface mutation before the next, so
// concurrent listeners cannot interleave their `session.append`s. A
// throwing listener escapes to the outer catch, which closes the (not-yet-
// open) step as a no-op and ends the turn via failTurn — a broken
// pre-step plugin ends the turn, not the loop. The composed session
// prefix rides along so token-pressure listeners count everything the
// request will actually carry.
// Await surface mutations outside the step; pressure checks receive the pending prefix.
await events.serial('agent/pre-step', turn, step, fullSystemPrompt, transmission.sessionPrefix, abort.signal)
// Interruption landing during the pre-step seam: do not open an empty step.
@@ -531,16 +306,8 @@ async function runTurn(
break
}
// The reconstruction boundary (the reconstructability RFC): the request's
// messages are snapshotted HERE, in the same synchronous frame as the
// step/start append directly below — so the snapshot is exactly the
// derivation over the log prefix strictly before step/start's seq.
// Anything appended later by the request-window inject seam or a
// concurrent task lands after the boundary and joins the NEXT request.
// session/event itself is observe-only: append reentrancy is rejected
// until the current callback list drains. An external reconstructor
// recovers these exact messages by folding the surface over
// events[0..stepStartSeq).
// Snapshot the exact log prefix before step/start: the reconstruction
// boundary. Appends after this synchronous snapshot join the next request.
const boundaryMessages = session.deriveMessages()
session.append('step/start', { turn, step })
@@ -587,13 +354,7 @@ async function runTurn(
break
}
// The successful step's finish reason carries forward: a `max-tokens`
// step makes the whole turn end `max-tokens` (the ACP RFC's rule "any
// max-tokens step surfaces as max-tokens"). `stepFinishReason` returns
// `max-tokens` or `undefined`, so a later ordinary step never resets a
// max-tokens turn back to completed, and a never-truncated turn keeps the
// default `completed`. The disposal/abort/error branches above and the
// continuation-window disposal check below override this — they win.
// Preserve max-token completion unless a later disposal, abort, or error wins.
const stepReason = stepFinishReason(stepOutcome.finish)
if (stepReason) reason = stepReason
@@ -615,24 +376,16 @@ async function runTurn(
break
}
// A forced `continue` may carry model-facing context: record it as
// next-STEP steering (the steering channel), so the continued turn's next
// iteration drains it before its request — the typed twin of the /goal
// step/end-steer pattern.
// A continuation reason becomes next-step steering.
if (decision.action === 'continue' && decision.reason) {
handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source })
}
let shouldContinue = decision.action === 'continue'
// Steering from step/end session-event or continuation listeners (the
// /goal pattern) demands the model see it — it overrides a stop decision;
// the next iteration's drain records it.
// Pending steering overrides an ordinary stop.
if (!shouldContinue && handle.inbox.hasSteering) shouldContinue = true
// Terminal policy runs only AFTER the extensible continuation waterfall,
// its optional reason, and late steering have all been folded. Unlike the
// waterfall, this serial seam is monotonic: the first stop bail wins, and
// no later listener or steering override can resurrect the turn.
// Terminal policy is monotonic and runs after ordinary continuation folding.
let terminalStop = false
try {
const stop = await events.serial('agent/turn-stop', turn)
@@ -645,19 +398,12 @@ async function runTurn(
}
if (terminalStop) {
terminalStopped = true
// A continuation reason or listener may have queued steering before the
// terminal checkpoint. Discard only steering (never ordinary queued
// prompts) so it cannot become a next step or be re-enqueued as a fresh
// turn by runLoop's late-steering fallback.
// Terminal stop discards steering but preserves ordinary queued prompts.
handle.inbox.drainSteering()
shouldContinue = false
}
// A cancel that landed during the continuation window — after the step's
// AbortController was cleared (setAbort(undefined)) but before the next
// step starts — has no controller to observe it, so the turn-scoped marker
// ends the turn here. cancel() also cleared the steering FIFO, so the
// override above did not re-arm continuation.
// The marker catches cancellation after the step controller was cleared.
if (handle.isCancelled()) {
reason = { kind: 'aborted', reason: handle.cancelReason() }
break
@@ -673,19 +419,11 @@ async function runTurn(
// Normal / inline-error loop exit: close the turn.
closeTurn()
} catch (error: unknown) {
// Decide whether this turn opened from the LOG, not a speculative flag. A
// pre-commit validator or acceptance failure leaves no turn/start and owes
// no turn/end, so it propagates to runLoop's backstop. Once turn/start is
// present, this path balances any committed step and records the failure.
// Close only a turn whose start committed to the log.
const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
if (!turnStartLogged) throw error
closeStep()
// Choose the close reason. Disposal wins only if no error was already
// reported: a turn disposed mid-step sets reason=disposed in the step-error
// branch (without reporting an error), so preserve disposed rather than
// overwrite it. Otherwise a mid-step throw on a live agent is a real
// failure → failTurn. (errorReported is mutated only inside the failTurn
// closure, which the analyzer can't follow, hence the inline lint-disable.)
// Preserve an established disposal reason; otherwise report the failure.
if (handle.isDisposed() && !errorReported) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition
reason = { kind: 'disposed' }
} else {
@@ -694,19 +432,11 @@ async function runTurn(
closeTurn()
}
// Durability checkpoint: persistence plugins drain write-behind buffers.
// A failing persistence plugin is reported but doesn't kill the agent.
// Through the store's flush (the carrier owner), never a raw parallel.
// Flush through the store-owned durability checkpoint without killing the driver on failure.
try {
await ctx.sessions.flush(session)
} catch (error: unknown) {
// The turn is already closed (turn/end appended above) and flush must run
// AFTER turn/end to be a checkpoint — so there is no in-turn position left
// for a session `error` event. Appending one here would land it after the
// last turn/end, where the persistence backend treats it as a crash tail
// and drops it on resume (the turn-enclosure RFC: every event is turn-enclosed). Report
// the failure via agent/error + the logger only; persistence keeps the
// buffered events for the next flush/dispose, so nothing is lost.
// The turn is closed, so report the failed flush live rather than append outside a turn.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
try {
@@ -727,13 +457,12 @@ function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boole
return messages.length > 0
}
/** One step: build the request from the boundary snapshot + the step's
* header → compose the session prefix if this instance has none yet → log
* the header event the request owes → stream model → record → execute
* tools. The caller assembles the
* system prompt, fires the `agent/pre-step` seam, snapshots the derivation,
* and opens the step BEFORE calling this, so `boundaryMessages` is exactly
* the surface prefix at step/start and already reflects any compaction. */
/**
* Run one committed step: transform call config, log the request header, build
* the request from the cached prefix plus the step-boundary snapshot, stream and
* record the response, then execute tools. The caller has already assembled the
* prompt, run `agent/pre-step`, snapshotted history, and opened the step.
*/
async function runStep(
ctx: Context,
events: AgentEventDispatch,
@@ -748,40 +477,23 @@ async function runStep(
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
const { session, options } = agent
// Seed the call config: the first request of THIS loop instance seeds from
// current AgentOptions — explicit options always win over the logged
// baseline, which is what keeps fork model-overrides and resume-time
// reconfiguration correct. Later steps seed from the log's folded header,
// which by then is exactly what this instance last logged.
// One deep-cloned, frozen seed serves BOTH the listener chain and the
// no-listener fallback: structuredClone decouples it from the session's
// cached header fold (a raw reference would let a delegating listener
// mutate the fold in place and silently skip the delta log), and the freeze
// makes in-place shaping unrepresentable — a switch is a RETURNED
// replacement, which the header event below records.
// Seed the first request from agent options and later requests from the logged header;
// detach and freeze so listeners must return an attributable replacement.
const seedConfig: LlmCallConfig = deepFreeze(structuredClone(transmission.loggedHeader
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- loggedHeader ⟹ a snapshot is in the log
? session.requestHeader()!.config
: { model: options.model ?? '' }))
// Shape the call config: listeners return a replacement to switch model or
// sampling (the seed is frozen — content shaping is not expressible here;
// model-visible content flows through the log channels). The header event
// below records whatever the request ACTUALLY uses, so a listener's switch
// is a logged, reconstructable fact, never silent drift.
// Listener replacements are recorded in the request header before dispatch.
const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
if (!config.model) {
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
}
// The session prefix was composed (once per instance) before this step's
// pre-step seam — the caller guarantees it, so the cache is always set here.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
const sessionPrefix = transmission.sessionPrefix!
// The request header (the log's request/header* vocabulary): canonical form,
// recorded before dispatch so the log always explains the request —
// including the session prefix, which no other event carries.
// Record the canonical header, including the otherwise-unlogged prefix, before dispatch.
const header = canonicalHeader({
config,
...system ? { system } : {},
@@ -790,11 +502,7 @@ async function runStep(
})
recordRequestHeader(session, transmission, header)
// Build and freeze: the request is a pure function of (boundary snapshot,
// logged header) — llm/stream listeners and adapters read it, mutation
// throws. sessionId + frozen is the loop-built marker the dev invariant
// keys on. Message order: header.messagePrefix, then the boundary
// snapshot — the reconstruction equation the invariant recomputes.
// Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
const request: GenerateOptions = deepFreeze({
model: header.config.model,
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
@@ -818,26 +526,16 @@ async function runStep(
assembler.push(chunk)
}
// Adapters report provider/transport failures one of two sanctioned ways
// (see the StreamChunk contract in dsh-llm): throw from stream() — already
// handled by the caller's try/catch — OR end the stream with a
// finish-error/aborted chunk. finishError() maps the latter to the step
// error to raise (turn ends error/aborted, not a normal completed message).
// Normalize failure finish chunks into the same path as thrown stream errors.
const stepError = finishError(assembler.finish)
if (stepError) throw stepError
if (assembler.finish.kind === 'max-tokens') {
let message: Message = withoutToolCalls(assembler.message())
message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
// Fire the assistant/message when there is content OR usage: a max-tokens
// step can be cut off with empty content but still carry token accounting,
// and assistant/message is the only host for usage (there is no standalone
// usage event). An empty-content assistant/message is skipped by
// deriveMessages(), so hosting usage on it never injects a spurious assistant
// turn into derived history.
// Preserve usage even when max-token truncation produced no content.
if (message.content.length > 0 || assembler.usage) {
// A max-tokens finish is itself a streamed `finish` chunk, so chunkSeqs is
// never empty here — pass the provenance unconditionally.
// The finish chunk guarantees non-empty provenance here.
session.append(
'assistant/message',
{ turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
@@ -847,20 +545,11 @@ async function runStep(
return { hadToolCalls: false, finish: assembler.finish }
}
// The step-result waterfall runs BEFORE the session append so the log (the
// source of truth for derived history and replay) records the message that
// tool dispatch actually uses.
// Record the post-waterfall message that tool dispatch uses.
let message: Message = assembler.message()
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
// Same content-or-usage guard as the max-tokens branch: a step that finishes
// with neither assembled content nor usage (e.g. a bare `stop` finish that
// streamed nothing) records no assistant/message — an empty-content message
// exists only to host usage, and deriveMessages() skips it either way, so
// appending one with no usage would be a pure trace-only row.
//
// sourceEventSeqs records the assistant/chunk provenance, but is omitted when
// no chunks streamed (the surface invariant rejects an empty sourceEventSeqs).
// Empty messages exist only to carry usage; omit empty provenance.
if (message.content.length > 0 || assembler.usage) {
session.append(
'assistant/message',
@@ -869,17 +558,9 @@ async function runStep(
)
}
// --- Tool execution (sequential; parallel execution is a TODO) ---
// If this becomes parallel, audit post-execute plugins that keep per-step
// pending state before their returned contexts are appended.
// ToolRegistry.execute converts tool failures (including aborts) into
// isError results, so abort is re-checked around every call here.
// Tool execution stays sequential; recheck abort around each normalized result.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Per-step buffer of contexts deferred by composite tools or attached by
// tools/post-execute listeners. Appended as context/message(s) only AFTER
// every tool/result for the step, so a multi-call step keeps adjacency
// (interleaving context between a call's result and the next call's would
// break the pairing the next model request relies on).
// Buffer context until all results are appended to preserve call/result adjacency.
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
@@ -891,12 +572,8 @@ async function runStep(
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): tools/pre-execute deliberately cannot rewrite
// `arguments` — tool/call (the audit record) and assistant/message (the
// model-history source) are logged BEFORE execute, and live consumers (ACP,
// tool-bash presentation) read the pre-execution args, so an execution-only
// rewrite would desync the UI from what ran. Designing that consistently is
// its own proposed RFC (docs/rfc/proposed/feature/…-pre-tool-input-rewrite.md).
// TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
// see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
@@ -906,33 +583,23 @@ async function runStep(
})
session.append('tool/result', {
turn, step,
// The correlation id MUST be the loop's authoritative call.id (the
// model-transcript id that deriveMessages turns into toolCallId), NOT
// result.callId — a post-execute waterfall listener returning a
// mismatched id would otherwise orphan the call↔result pairing in the
// next model request. A listener-internal id, if ever needed, belongs in
// a separate diagnostic field, never overloaded onto callId.
// Preserve transcript pairing even if a post-execute listener returns another id.
callId: call.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// The tool's private presentation payload (e.g. a result-time diff),
// persisted so a UI bridge reproduces the card on replay.
// Persist tool-owned presentation data for replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
// Buffer (don't append yet) every context carried by this call.
pendingContext.push(...result.additionalContexts ?? [])
// signal CAN flip during the await above (abort() inside a tool);
// the analyzer can't see through the await boundary.
// The signal may flip while the tool is awaited.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
/* v8 ignore stop */
}
// Append buffered post-execute context AFTER every tool/result, preserving
// tool-call/result adjacency across the whole batch. inject() appends into the
// open turn (a context/message at its chronological position).
// Append buffered context after the complete result batch.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
@@ -959,13 +626,8 @@ export function lastTurnNumber(session: Session): number {
}
/**
* Whether a turn is currently open in the session log (a `turn/start` with no
* matching later `turn/end`). Decided from the LOG, not agent status: status
* can be `running` while no turn is open (an `agent/status` listener firing
* before `turn/start`, or the post-`turn/end` flush window before status
* returns to idle), so status is not a reliable open-turn signal. Used by
* `inject()` to choose between appending into an open turn vs. wrapping the
* injection in its own one-shot turn (the turn-enclosure RFC).
* Whether the session log has an unmatched `turn/start`. Agent status is not
* sufficient during pre-start and post-end windows.
* @param session - the session whose log is inspected.
* @returns true when the log's last turn boundary is a `turn/start` with no matching `turn/end` yet.
*/

31
packages/core/agent-loop/src/request-log.ts
View File

@@ -1,12 +1,7 @@
/**
* Per-loop-instance transmission bookkeeping for the reconstructability
* contract: which header event to append before a request so the session log
* always explains the request (the reconstructability RFC). The loop is
* otherwise transmission-stateless — the comparison baseline is the log's own
* folded header (`Session.requestHeader()`), so resume and fork need no
* special path: a fresh loop instance simply logs a `'resume'` snapshot on
* its first request and deltas from there.
*
* Per-loop-instance request-header bookkeeping for reconstructability. The
* comparison baseline is the header folded from the session log, so a fresh
* loop instance needs no special resume or fork state.
* @module dsh-agent-loop/request-log
*/
@@ -37,22 +32,10 @@ export function createTransmissionLog(): TransmissionLog {
}
/**
* Append whatever header event this request owes the log, so folding the log
* reproduces the header the request was built under. Exactly one of four
* things happens:
*
* 1. This loop instance has not logged a header yet → a full `request/header`
* snapshot anchors the fold: reason `'initial'` when the log has no header
* events at all (a new conversation), `'resume'` when it does (process
* restart, fork seed — the boundary itself is a recorded fact, so the
* snapshot is appended even when nothing changed).
* 2. The header equals the folded baseline → nothing; the log already
* explains this request.
* 3. It differs and the delta round-trips (`applyHeaderDelta` on the baseline
* reproduces the header exactly) → a `request/header-delta`.
* 4. It differs and the delta encoding cannot express the change (a pure tool
* reordering) → a full snapshot with reason `'fallback'`; deltas are an
* encoding optimization, never a correctness dependency.
* Append whatever header event makes the log reproduce this request's header.
* The first request from an instance always records a full `initial` or `resume`
* snapshot. Later requests record nothing when unchanged, a round-tripping
* delta when expressible, or a full `fallback` snapshot otherwise.
*
* @param session - the session whose log explains the request.
* @param state - this loop instance's bookkeeping (mutated on first log).

33
packages/core/agent-loop/tests/agent.spec.ts
View File

@@ -167,10 +167,8 @@ describe('ReactLoopAgent', () => {
let flushes = 0
ctx.on('session/flush', () => { flushes += 1 })
// Non-serializable injected content makes Session.append throw AFTER
// turn/start was recorded. The turn/end must still be appended (finally),
// AND the durability checkpoint must still fire — the balanced turn is in
// memory and a crash before the next turn/dispose would otherwise lose it.
// Invalid injected content throws after turn/start. `finally` must still append turn/end and
// flush the balanced in-memory turn so a crash cannot lose it before the next checkpoint.
expect(() => {
agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
}).toThrow(/non-JSON-serializable/)
@@ -252,26 +250,20 @@ describe('ReactLoopAgent', () => {
})
it('disposer is idempotent (double-stop)', async () => {
// Create a bare ReactLoopAgent and start it through the package-internal
// test seam. Then call its disposer twice — the second call hits the
// early-return branch.
// The internal start seam exposes one idle driver's disposer for repeated invocation.
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('test'))
const prepared = prepareReactLoopAgent(ctx, AgentId('bare'), { model: 'mock' }, session)
const { agent } = prepared
// Start the loop to get the disposer; the agent waits for messages
// (idle, never-resolving cancel), so it will stay idle.
prepared.markPublished()
const dispose = prepared.startDriver()
// First dispose
const firstDisposal = dispose()
expect(agent.status).toBe('disposed')
await firstDisposal
// Second dispose — idempotent, no throw
await expect(dispose()).resolves.toBeUndefined()
expect(agent.status).toBe('disposed')
})
@@ -366,10 +358,8 @@ describe('ReactLoopAgent', () => {
})
it('whenIdle() subscribed while running resolves via done when the agent is then disposed', async () => {
// Covers the waiter's disposed arm: whenIdle() queues an internal waiter
// while running (not the fast path), then the disposer settles it and chains
// `done` (loop exit), not an eager resolve. A bare ReactLoopAgent + direct
// internal driver disposer keeps the emit synchronous.
// Queue the internal waiter while running, then dispose the bare driver. Its disposed branch
// must chain the loop's `done` promise rather than resolve before exit.
const ctx = new Context()
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
@@ -395,11 +385,8 @@ describe('ReactLoopAgent', () => {
})
it('whenIdle() subscribed while running survives a FIBER dispose (no hung promise)', async () => {
// The waiter is internal agent state, NOT an effect-scoped ctx.on listener:
// disposing the OWNING fiber runs the agent's listener disposers, which would
// have dropped a ctx.on-based waiter before the 'disposed' transition and
// hung the promise. With internal waiters, the fiber disposer still settles
// it. Regression for the round-3 whenIdle finding.
// The waiter is agent-owned state, not an effect-scoped listener that owner disposal would
// remove before the disposed transition. Fiber teardown must still settle it.
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: ReactLoopAgent
@@ -417,10 +404,8 @@ describe('ReactLoopAgent', () => {
})
it('whenIdle() on a disposed agent awaits the loop exit (done), not just the status flip', async () => {
// The disposer emits agent/status('disposed') BEFORE the driver loop
// unwinds, so whenIdle() must chain `done` (true quiescence) on the
// disposed path. Dispose a running agent, then assert whenIdle() resolves
// only after `done` — i.e. the loop has actually exited.
// Disposed status is emitted before the driver unwinds. `whenIdle()` must chain `done` so it
// resolves only after true loop exit.
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
let agent!: ReactLoopAgent

57
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -1,12 +1,9 @@
/**
* Tests for the queue-aware `Agent.cancel()` primitive. `cancel()` is the
* broad verb — it clears queued + steering work, aborts an in-flight step, and
* drops a turn about to start — whereas a bare step abort (the loop's private
* `AbortController`) kills only the current step and leaves the queue intact.
* These tests exercise every window where a cancel can land (idle, pre-step,
* mid-step, continuation) and the marker's arm/reset rules that keep a cancel
* from leaking to a later prompt or hanging `whenIdle()`.
*
* Tests for the queue-aware `Agent.cancel()` primitive. `cancel()` is the broad verb — it
* clears queued + steering work, aborts an in-flight step, and drops a turn about to start —
* whereas a bare step abort (the loop's private `AbortController`) kills only the current step
* and leaves the queue intact. The suite covers every landing window plus marker
* reset and `whenIdle()` quiescence.
* @module dsh-agent-loop/tests/cancel
*/
@@ -95,10 +92,8 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// Queue work, then register a whenIdle() waiter while in the pre-step window
// (status idle, hasQueued true) — it does NOT take the fast path. Then cancel.
// The skip path must settle this waiter directly (no running→idle transition
// ever fires), or it would hang forever.
// This waiter cannot rely on a running→idle transition because cancellation
// drops the turn before it runs; the skip path must settle it directly.
send(agent, 'q')
const idle = agent.whenIdle()
agent.cancel('pre-step')
@@ -234,12 +229,8 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// The first composition is interrupted mid-waterfall and — like an
// abort-aware listener bailing on a firing signal — contributes nothing.
// Caching that degraded result would silently strip the prefix from every
// later request of this instance; the loop must discard it and recompose
// on the next send, and the SECOND composition's value must be what the
// wire and the header log carry.
// The interrupted first composition must not cache its degraded empty value;
// the next prompt recomposes and logs/sends the fresh prefix.
const opener: Message = { role: 'user', content: [{ type: 'text', text: 'fresh opener' }] }
let compositions = 0
ctx.on('agent/session-prefix', async (_agent, _prefix, _signal, next): Promise<Message[]> => {
@@ -268,10 +259,8 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// A turn/start listener fires right after turn/start is appended, BEFORE any
// AbortController is installed for the step. Cancelling there must still drop
// the step (the turn-scoped marker, not the step AbortController, is what
// catches this) — no model step runs.
// A turn/start listener fires before a step controller exists, so the
// turn-scoped marker—not step abort—must drop the pending step.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('session/event', (session, event) => {
@@ -400,10 +389,8 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// setStatus('running') emits agent/status SYNCHRONOUSLY, so a running
// listener can cancel in the gap between the loop's pre-step check and
// runTurn. The second check (after the running flip) must drop the turn —
// runTurn would otherwise throw on the now-empty queue.
// `agent/status` is synchronous, so cancellation can land after the first
// pre-step check; the second check must drop the now-empty turn.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('agent/status', (subject, status) => {
@@ -421,11 +408,7 @@ describe('Agent.cancel()', () => {
})
it('window 2: whenIdle() does NOT resolve early when a running listener cancels then queues replacement work', async () => {
// The window-1 early-resolve race has a window-2 twin: a synchronous
// agent/status('running') listener cancels the about-to-run turn AND queues a
// replacement. window 2 must NOT settle waiters (via setStatus('idle')) while
// the replacement is still queued-and-unrun — it must fall through and run it,
// so whenIdle() resolves on the replacement turn's running→idle, not before.
// Cancellation must not settle idle while replacement work remains queued.
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -451,11 +434,8 @@ describe('Agent.cancel()', () => {
})
it('whenIdle() does NOT resolve early when a new prompt is queued during a pre-step cancel', async () => {
// The subtle race: a whenIdle() waiter is registered for prompt A; cancel()
// clears A; prompt B is queued BEFORE the loop resumes from the idle wait.
// The window-1 cancel branch must NOT settle the waiter while B is still
// queued-and-unrun — whenIdle() must wait for B's turn to actually run and
// settle (the quiescence contract), not resolve before B's first event.
// The subtle race: a whenIdle() waiter is registered for prompt A; cancel() clears A;
// prompt B is queued before the loop resumes from the idle wait.
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -465,9 +445,8 @@ describe('Agent.cancel()', () => {
agent.cancel('drop A') // arms marker, clears A
send(agent, 'B') // B races in before the loop resumes
// whenIdle() must resolve only AFTER B's turn fully ran — by which point B's
// user message and a turn/end are in the log. (Before the fix it resolved
// immediately, with zero events, then B ran afterward.)
// whenIdle() must resolve only after B's turn fully ran — by which point B's user message
// and a turn/end are in the log.
await idle
expect(userTexts(agent)).toContain('B')
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)

5
packages/core/agent-loop/tests/config-session-id.spec.ts
View File

@@ -101,9 +101,8 @@ describe('config-driven session id', () => {
await waitForIdle(ctx1, a1)
await ctx1.fiber.dispose()
// Run 2: a CONFIG agent with resumeSessionId continues that session. The
// resume is deferred until sessionPersistence loads (ctx.inject), so wait
// for the agent to appear, then assert it is on the resumed id with history.
// Resume waits for the injected persistence service, so poll until the
// config-created agent appears with its stored history.
const ctx2 = new Context()
await ctx2.plugin(LlmService)
await ctx2.plugin(SessionStore)

105
packages/core/agent-loop/tests/review-fixes.spec.ts → packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -39,7 +39,7 @@ function send(agent: ReactLoopAgent, text: string) {
agent.send([{ type: 'text', text }])
}
describe('HIGH: session log records what agent/step-result actually produced', () => {
describe('session log records what agent/step-result actually produced', () => {
it('a step-result rewrite is what the log, derived history, and tool dispatch all see', async () => {
const adapter = new MockAdapter([textResponse('original'), textResponse('done')])
const ctx = await harness(adapter)
@@ -89,7 +89,7 @@ describe('HIGH: session log records what agent/step-result actually produced', (
})
})
describe('HIGH: abort during tool execution ends the turn', () => {
describe('abort during tool execution ends the turn', () => {
it('aborting the in-flight step inside a tool prevents both remaining tools and the next model step', async () => {
const adapter = new MockAdapter([
// model asks for two tool calls in one step
@@ -141,7 +141,7 @@ describe('HIGH: abort during tool execution ends the turn', () => {
})
})
describe('HIGH: steering from late extension points is never stranded', () => {
describe('steering from late extension points is never stranded', () => {
it('steer() from an agent/turn-continuation listener overrides a stop decision', async () => {
const adapter = new MockAdapter([
textResponse('no tools, would stop here'),
@@ -168,21 +168,7 @@ describe('HIGH: steering from late extension points is never stranded', () => {
})
it('steer() from a step/end session-event listener forces a SAME-TURN next step (/goal pattern)', async () => {
// The /goal pattern steers from a step boundary so the model addresses a
// standing goal before stopping. Step boundaries have no agent/* mirror, so
// the surviving hook point is the durable step/end session event. With a
// no-tools first step the default continuation is stop; the steering queued
// here must force the `!shouldContinue && hasSteering` override so the SAME
// turn runs another step.
//
// The override is what this test guards, so it asserts the same-turn shape —
// NOT merely that the content reaches requests[1]. Without the override the
// turn would stop, and leftover steering is re-enqueued as a next-turn queued
// message, which ALSO lands in requests[1] (just one turn later). So a
// content-only assertion passes with the override disabled and guards
// nothing. The discriminator is the turn/step shape: override ⇒ ONE turn with
// TWO steps and the steering recorded as a `steering/message` BEFORE step 2;
// re-enqueue fallback ⇒ TWO turns.
// Assert the same-turn shape; content alone cannot distinguish re-enqueue.
const adapter = new MockAdapter([
textResponse('no tools, would stop'),
textResponse('after goal reminder'),
@@ -200,12 +186,10 @@ describe('HIGH: steering from late extension points is never stranded', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
// Same-turn continuation: the steering forced step 2 within turn 1.
const events = [...agent.session.events]
expect(events.filter(e => e.type === 'turn/start')).toHaveLength(1)
expect(events.filter(e => e.type === 'step/start')).toHaveLength(2)
// The steered content is recorded as steering (same turn), BEFORE step 2 —
// not as a fresh turn's user/message. This is the mechanism the override uses.
// Same-turn steering precedes the second step.
const steeringIdx = events.findIndex(e => e.type === 'steering/message')
const step2Idx = events.map(e => e.type).lastIndexOf('step/start')
expect(steeringIdx).toBeGreaterThanOrEqual(0)
@@ -263,7 +247,7 @@ describe('HIGH: steering from late extension points is never stranded', () => {
})
})
describe('HIGH: plugin exceptions are contained', () => {
describe('plugin exceptions are contained', () => {
it('a throwing agent/turn-continuation listener ends the turn with an error, loop survives', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
const ctx = await harness(adapter)
@@ -318,7 +302,7 @@ describe('HIGH: plugin exceptions are contained', () => {
})
})
describe('MEDIUM: disposed status is part of the agent/status contract', () => {
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 () => {
const adapter = new MockAdapter(['hang'])
const ctx = await harness(adapter)
@@ -365,7 +349,7 @@ describe('MEDIUM: disposed status is part of the agent/status contract', () => {
})
})
describe('MEDIUM: misc registry and config fixes', () => {
describe('adapter registration, routing, and accepted-input ownership', () => {
it('duplicate adapter registration is rejected', async () => {
const ctx = new Context()
await ctx.plugin(LlmService)
@@ -524,7 +508,7 @@ describe('MEDIUM: misc registry and config fixes', () => {
})
})
describe('MEDIUM: turn numbering continues across seeded (forked) sessions', () => {
describe('turn numbering continues across seeded sessions', () => {
it('a forked agent continues turn numbers after the seed log', async () => {
const first = new MockAdapter([textResponse('turn one')])
const ctx = await harness(first)
@@ -562,7 +546,7 @@ describe('MEDIUM: turn numbering continues across seeded (forked) sessions', ()
})
})
describe('LOW: discriminated SessionEvent narrows without casts', () => {
describe('discriminated SessionEvent narrows without casts', () => {
it('narrows event.data from event.type', () => {
const session = new Session(SessionId('s'))
const appended: SessionEvent = session.append('tool/call', {
@@ -580,12 +564,9 @@ describe('LOW: discriminated SessionEvent narrows without casts', () => {
})
})
describe('HIGH: a finish-error stream chunk ends the turn as error, not completed', () => {
describe('a finish-error stream chunk ends the turn as error, not completed', () => {
it('translates finish {kind:error} into a turn error with a logged error event', async () => {
// The second sanctioned adapter error path (besides throwing): an
// adapter that cannot throw mid-stream ends the stream with a
// finish-error chunk (e.g. the pi-ai adapter mapping a provider 401).
// The loop must NOT log a normal assistant/message + completed turn.
// A finish-error chunk must not produce a completed assistant turn.
const errorStream: StreamChunk[] = [
{ type: 'finish', reason: { kind: 'error', message: 'provider 401', code: 'AUTH' } },
]
@@ -606,7 +587,7 @@ describe('HIGH: a finish-error stream chunk ends the turn as error, not complete
// a standalone error event.
const turnEnd = events.find(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'error', step: 1, message: 'provider 401', code: 'AUTH' })
// Crucially: no assistant/message was logged for the failed step.
// A failed step must not synthesize an assistant message.
expect(events.some(event => event.type === 'assistant/message')).toBe(false)
})
@@ -652,10 +633,7 @@ describe('step boundary publication order', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-step-order'), { model: 'mock' })
// Session.append pushes the event BEFORE notifying session/event listeners,
// so a step/start listener always finds the matching event already in the
// log. (Step boundaries have no agent/* mirror — the session log is the live
// feed.)
// Append commits before observers run.
const observed: { turn: number; step: number; lastEventType: string | undefined; sawStepStart: boolean }[] = []
ctx.on('session/event', (subject, event) => {
if (subject !== agent.session || event.type !== 'step/start') return
@@ -678,10 +656,7 @@ describe('step boundary publication order', () => {
})
describe('turn and step boundary recovery', () => {
// Harness with the invariants plugin loaded as an oracle: it throws on
// append if the log goes unbalanced (turn/end while a step is open,
// turn/start while a turn is open, etc.), so a regression surfaces as an
// InvariantError on the NEXT turn's append rather than a silent imbalance.
// The invariants plugin makes an unbalanced log fail the test.
async function balancedHarness(adapter: MockAdapter) {
const ctx = new Context()
await ctx.plugin(LlmService)
@@ -833,9 +808,7 @@ describe('turn and step boundary recovery', () => {
})
it('a throwing agent/error listener during a step-error path still balances the turn, loop survives', async () => {
// First turn: model stream ends with a finish-error → step error path →
// failTurn emits agent/error, whose listener throws. The turn must still
// close balanced. Second turn proves the loop survived.
// Listener failure cannot interrupt error finalization or the next turn.
const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }]
const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
const ctx = await balancedHarness(adapter)
@@ -894,9 +867,7 @@ describe('turn and step boundary recovery', () => {
})
it('preserves reason disposed when a pre-step listener disposes then throws (outer-catch disposed branch)', async () => {
// A pre-step listener requests disposal and then throws before the ordinary
// post-listener disposal check. The outer catch sees disposal already won
// and must preserve reason=disposed rather than rewrite it as a plugin error.
// Disposal remains authoritative when the listener also throws.
const adapter = new MockAdapter([textResponse('never reached')])
const ctx = await balancedHarness(adapter)
let agent!: ReactLoopAgent
@@ -908,9 +879,6 @@ describe('turn and step boundary recovery', () => {
ctx.on('agent/pre-step', () => {
if (threw) return
threw = true
// Request disposal, then throw in the same synchronous tick: status flips
// to 'disposed' (the disposer aborts the step controller) and the throw
// drives control into the outer catch with isDisposed() already true.
void fiber.dispose()
throw new Error('boom pre-step during disposal')
})
@@ -1001,10 +969,7 @@ describe('turn and step boundary recovery', () => {
})
it('a throwing step/end observer cannot interrupt error finalization', async () => {
// A finish-error stream opens a step then fails it, driving finalization
// through closeStep() with the step open. Session contains the observer
// failure after committing step/end, so closeTurn still records the model
// failure and balances the turn.
// Observer failure after step/end commit cannot interrupt turn finalization.
const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }]
const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
const ctx = await harness(adapter)
@@ -1112,11 +1077,7 @@ describe('tool result call identity', () => {
describe('surface: assistant/message omits sourceEventSeqs when no chunks streamed', () => {
it('a step-result listener injecting content over an empty stream appends with surfaceOp but no sourceEventSeqs', async () => {
// An empty stream yields zero assistant/chunk events (finish defaults to
// `stop`), so chunkSeqs is empty. A step-result listener injects content, so
// the content-or-usage guard fires and an assistant/message is appended. Its
// sourceEventSeqs MUST be omitted (not `[]`) — the surface invariant rejects
// an empty sourceEventSeqs, and the dev invariants plugin would throw on it.
// Injected result content with no chunks must omit empty sourceEventSeqs.
const adapter = new MockAdapter([[]])
const ctx = await harness(adapter)
await ctx.plugin(Invariants)
@@ -1143,12 +1104,8 @@ describe('surface: assistant/message omits sourceEventSeqs when no chunks stream
describe('disposal and cancellation during pre-step assembly', () => {
it('disposal during system-prompt assembly drops the about-to-start step as disposed', { timeout: 30000 }, async () => {
// Block `system-prompt/assemble` on a promise. Start disposal (which
// calls stop() synchronously, setting status=disposed), then release the
// block. The loop must check isDisposed() after assembly and end the turn
// `disposed` — no LLM call. Don't await fiber.dispose() before releasing
// the blocker: the dispose chain awaits agent.done, which hangs until the
// loop unblocks.
// Start disposal, then release assembly. Do not await disposal first: it
// waits for the blocked driver to exit.
const adapter = new MockAdapter(['hang'])
let releaseAssemble!: () => void
const blocked = new Promise<void>(r => void (releaseAssemble = r))
@@ -1163,7 +1120,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
await ctx.plugin(Invariants)
ctx.llm.registerAdapter(['mock'], adapter)
// Blocking listener on the parent context (survives fiber disposal).
// Parent-owned listener survives agent-fiber disposal.
const unlisten = ctx.on('system-prompt/assemble', async function (_assembly, _context, next) {
await blocked
return next()
@@ -1181,28 +1138,22 @@ describe('disposal and cancellation during pre-step assembly', () => {
// Give the loop time to enter the step and reach assemble().
await new Promise(r => setTimeout(r, 50))
// Start disposal — stop() sets status=disposed synchronously, then the
// disposer's await agent.done hangs because the loop is blocked in the
// waterfall. Do NOT await yet; release the blocker first.
// Release assembly before awaiting disposal because disposal joins the blocked driver.
const disposalDone = fiber.dispose()
// Now release the blocked waterfall — the loop unblocks, checks
// isDisposed(), and exits, which resolves agent.done and disposalDone.
releaseAssemble()
await disposalDone
await agent.done
unlisten()
// Turn boundaries are durable rows; there is no `agent/*` mirror to assert.
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
// No step was opened, no LLM call was made.
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
// The durable turn/end record is the authoritative turn-boundary signal
// (turn boundaries have no agent/* mirror), so this asserts on the log.
})
it('cancel during system-prompt assembly drops the about-to-start step as aborted', { timeout: 30000 }, async () => {
@@ -1259,9 +1210,8 @@ describe('disposal and cancellation during pre-step assembly', () => {
})
it('disposal during agent/pre-step seam ends the turn disposed', { timeout: 15000 }, async () => {
// Block the `agent/pre-step` serial seam on a promise we control, then
// dispose the agent's fiber. When the block releases, the loop must see
// isDisposed() at the post-seam check and end the turn disposed.
// Start disposal, then release pre-step; awaiting disposal first would
// deadlock on the blocked driver.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
const blocker = new Promise<void>(r => void (releasePreStep = r))
@@ -1312,8 +1262,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
})
it('cancel during agent/pre-step seam ends the turn aborted', { timeout: 15000 }, async () => {
// Block `agent/pre-step`, then cancel() the agent. When the block releases,
// the post-seam check catches cancellation and ends the turn aborted.
// Release pre-step after cancellation to exercise the post-seam check.
const adapter = new MockAdapter(['hang'])
let releasePreStep!: () => void
const blocker = new Promise<void>(r => void (releasePreStep = r))

4
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -225,9 +225,7 @@ describe('disposed vs aborted branching', () => {
await fiber.dispose() // dispose during hang
await agent.done
// The review-fixes test for 'HIGH: disposed status' already covers
// this assertion path. The reason is 'disposed' because isDisposed() is
// checked before the abort signal check in the error path.
// Disposal wins abort classification because the error path checks it first.
expect(reasons).toContainEqual({ kind: 'disposed' })
})
})

19
packages/core/agent-loop/tests/inbox.spec.ts
View File

@@ -64,17 +64,12 @@ describe('Inbox', () => {
void inbox.waitForQueued(new Promise(() => {})) // first call, never resolved
void inbox.waitForQueued(p1) // second call overwrites wakeup
// Cancel p1 (the latest waiter's cancel) — the wakeup was overwritten
// to p1's resolve, so canceling p1 triggers the finally block which
// clears the wakeup if it matches.
// Cancelling the latest waiter clears the shared callback; enqueue must neither
// wake the stale waiter nor fail on the cleared callback.
r1()
await p1
// Now enqueue: the first waiter's wakeup (which was overwritten) won't
// fire, and the second waiter's wakeup was cleared by cancel.
// The enqueue calls wakeup?.() but wakeup was cleared — no crash, no hang.
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
// The overwrite path + finally cleanup are exercised
})
it('clears wakeup in finally handler when enqueue resolves', async () => {
@@ -88,23 +83,17 @@ describe('Inbox', () => {
})
it('finally handler does not clear wakeup when a different waiter overwrote it', async () => {
// First waiter's cancel resolves AFTER a second waiter overwrote wakeup.
// First waiter's finally sees wakeup !== its resolve → does not clear.
// A stale waiter's finally must not clear the replacement waiter.
const inbox = new Inbox()
const { promise: c1, resolve: r1 } = resolverPair()
void inbox.waitForQueued(c1) // wakeup = resolve1, c1.then(resolve1)
void inbox.waitForQueued(new Promise(() => {})) // wakeup = resolve2, cancel never resolves
// Resolve c1 (the first cancel). c1.then(resolve1) fires → resolve1() called
// → waiter1's promise resolves → finally: wakeup === resolve1? NO (it's resolve2)
// → wakeup is NOT cleared.
r1()
await c1
// Now enqueue: wakeup() calls resolve2 → waiter2 resolves
// But waiter2's cancel never resolves — that's fine, enqueue resolves it.
// The replacement remains registered and is resolved by enqueue.
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
// No need to await anything further — enqueue is synchronous wakeup
})
})

28
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -125,14 +125,8 @@ describe('agent/prompt-submit', () => {
})
it('a prompt-submit rewrite + additionalContexts is VISIBLE to the agent/pre-step seam (merged ordering)', async () => {
// The merge of the interception seams with master's compaction seam pins one
// ordering: `agent/prompt-submit` runs (rewriting the prompt and injecting
// context) BEFORE the step loop, and `agent/pre-step` fires INSIDE the step
// before the single deriveMessages(). So a compaction listener on
// `agent/pre-step` must observe the surface AFTER the prompt rewrite/inject —
// otherwise it would measure/compact stale history. This cross-test proves
// the two seams compose in the right order (each is covered in isolation
// elsewhere; this asserts they see each other's effects on the same turn).
// Prompt rewrites and injected context land before `agent/pre-step`, so a
// compaction listener measures the current surface before the single derive.
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -197,9 +191,8 @@ describe('agent/prompt-submit', () => {
})
it('a mixed batch records a prompt/blocked for the vetoed prompt while the allowed one runs', async () => {
// Two prompts queued into ONE turn: block "secret", allow "safe". The turn is
// NOT rejected (a prompt was allowed), so without a durable prompt/blocked the
// vetoed prompt and its reason would vanish from the log entirely.
// Blocking one prompt in a mixed batch must persist its reason even though
// the allowed prompt keeps the turn from ending rejected.
const adapter = new MockAdapter([textResponse('ran once')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -523,14 +516,13 @@ describe('agent/turn-continuation (ContinuationDecision)', () => {
await waitForIdle(ctx, agent)
const log = events(agent)
// same turn, two steps
// The continuation stays in the turn, is logged with provenance before step 2,
// and reaches that step's request.
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
expect(log.filter(e => e.type === 'step/start')).toHaveLength(2)
// the reason was recorded as steering BEFORE step 2, with its plugin source
const steering = log.find(e => e.type === 'steering/message')
expect(steering?.type === 'steering/message' && steering.data.content).toEqual([{ type: 'text', text: 'keep going on the goal' }])
expect(steering?.type === 'steering/message' && steering.data.source).toEqual({ kind: 'plugin', plugin: 'goal' })
// and reached the next request
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('keep going on the goal')
})
@@ -665,11 +657,9 @@ describe('tools/pre-execute gate (native-plugin permission pattern, end-to-end t
})
describe('worked example: a native hook plugin is just a cordis plugin on the seams', () => {
// The whole point of the interception taxonomy: a "native hook" needs no
// dsh-hook-protocol, no external command, no hook/* log — it is an ordinary
// cordis plugin subscribing to the canonical events and returning typed
// decisions. This proves all four seams compose end-to-end through the REAL
// loop, with NO hook/* SessionEvents involved (those belong to the bridge lib).
// The whole point of the interception taxonomy: a "native hook" needs no dsh-hook-protocol,
// no external command, no hook/* log — it is an ordinary cordis plugin subscribing to the
// canonical events and returning typed decisions.
const NativeGuard = {
name: 'native-guard',
apply(ctx: Context) {

39
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -183,11 +183,7 @@ describe('agent loop', () => {
})
it('contains a strict-variable render failure: the turn errors, the loop keeps serving turns', async () => {
// A persona claiming {{cwd}} on a session with NO cwd is a deployment
// authoring error — renderPrompt throws, the turn ends with an error, and
// the same agent must then RUN a later turn to completion (not merely
// report idle status): a rescue listener supplies the variable and the
// follow-up prompt reaches the model.
// A missing cwd variable must fail one turn without preventing a later valid turn.
const adapter = new MockAdapter([textResponse('ok after rescue')])
const ctx = await harness(adapter, 'In {{cwd}}.')
const errors: Error[] = []
@@ -548,9 +544,8 @@ describe('agent loop', () => {
})
it('agent/pre-step fires BEFORE the step it precedes opens (events land outside the step)', async () => {
// A listener appending a surface node in pre-step lands it BEFORE step/start
// in the log — proving the seam fires outside the step. The node is still in
// the derived request for that step (derive happens after step/start).
// The append lands before step/start, yet derive happens afterwards and the
// same step's request must include it.
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -583,10 +578,8 @@ describe('agent loop', () => {
})
it('a throwing agent/pre-step listener ends the turn (error), not the loop', async () => {
// The seam fires before step/start, so a throw escapes to runTurn's outer
// catch: the not-yet-open step closes as a no-op, the failure surfaces via
// agent/error, and the turn ends `error` (recorded on the durable turn/end).
// The loop survives and a follow-up prompt still runs.
// Before step/start, a pre-step throw reaches the turn catch: no step needs
// closing, the turn records error, and the loop remains available.
const adapter = new MockAdapter([textResponse('second turn ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
@@ -653,16 +646,14 @@ describe('agent loop', () => {
expect(adapter.requests).toHaveLength(1)
expect(reasons).toEqual([{ kind: 'max-tokens' }])
// and the reason is recorded in the log's turn/end event
// Assert the durable row, not only the live listener.
const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
expect(turnEnd!.data.reason).toEqual({ kind: 'max-tokens' })
})
it('a max-tokens step earlier in a turn still surfaces as max-tokens after a later completed step', async () => {
// Step 1 is cut off (max-tokens, no tool calls → would stop by default), so
// continuation must be FORCED to reach step 2 which finishes normally
// (stop). The rule "any max-tokens step surfaces as max-tokens" means the
// turn ends max-tokens even though the LAST step completed cleanly.
// Step 1 is cut off (max-tokens, no tool calls → would stop by default), so continuation
// must be FORCED to reach step 2 which finishes normally (stop).
const adapter = new MockAdapter([
maxTokensResponse('first half'),
textResponse('second half'),
@@ -744,11 +735,8 @@ describe('agent loop', () => {
expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
expect(reasons).toEqual([{ kind: 'max-tokens' }])
// No-data-loss: a max-tokens step whose only content was a dropped tool call
// has EMPTY assistant content, but its usage must still be represented. It
// rides on an (empty-content) assistant/message — there is no standalone
// usage event — and that empty message is skipped by deriveMessages(), so
// the derived history above is NOT corrupted by a spurious assistant turn.
// Empty content still needs an assistant/message to carry usage; derivation
// skips that host so it does not create a spurious assistant turn.
const assistantMessage = agent.session.events.find(e => e.type === 'assistant/message')
expect(assistantMessage?.type === 'assistant/message' && assistantMessage.data).toEqual({
turn: 1, step: 1, content: [], usage: { inputTokens: 10, outputTokens: 5 },
@@ -756,10 +744,9 @@ describe('agent loop', () => {
})
it('appends no assistant/message for a max-tokens step with empty content and no usage', async () => {
// A max-tokens step truncated to a dropped tool call AND with no usage chunk
// has nothing to record: empty content and no accounting → no assistant/message
// (the empty-content host exists only to carry usage). The turn still ends
// max-tokens.
// A max-tokens step truncated to a dropped tool call AND with no usage chunk has nothing to
// record: empty content and no accounting → no assistant/message (the empty-content host
// exists only to carry usage).
const callId = CallId('c1')
const adapter = new MockAdapter([[
{ type: 'block-start', index: 0, blockType: 'tool-call' },

17
packages/core/agent-loop/tests/properties.spec.ts
View File

@@ -1,12 +1,7 @@
/**
* Property-based tests for the agent loop's inbox/turn scheduling (the
* property-testing RFC). Deterministic by construction: schedules are driven
* through the `agent/status` settle signal (no wall-clock sleeps), so a flake
* is a finding, not timing noise.
*
* Invariants: every sent message appears exactly once in the log (none lost);
* turn numbers strictly increase; status transitions follow the legal machine
* idle→running→idle (and →disposed at teardown).
* Deterministic property tests for inbox scheduling: every sent message logs
* once, turn numbers increase, and status follows idle→running→idle/disposed.
* Schedules advance on status events rather than wall-clock sleeps.
*/
import { describe, expect, it } from 'vitest'
@@ -146,10 +141,8 @@ describe('agent loop scheduling properties', () => {
const ctx = await harness()
try {
const agent = ctx.agentLoop.create(AgentId('a'), { model: 'mock' })
// Capture an idle waiter before EACH send; the last one is guaranteed
// to resolve because the final send always triggers (or joins) a turn
// that ends idle. Awaiting an already-resolved waiter is a no-op, so a
// trailing settle step can't cause a hang.
// Capture before each send; the last waiter covers the final turn, and
// awaiting an already-settled earlier waiter is harmless.
let lastIdle: Promise<void> | undefined
for (const step of steps) {
const idle = nextIdle(ctx, agent)

16
packages/core/agent-loop/tests/request-cache.e2e.ts
View File

@@ -9,15 +9,13 @@ import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
/**
* With-key proof that log-derived requests translate into REAL provider cache
* hits: a multi-step tool turn (plus a follow-up turn) against the live
* DeepSeek API must report `cacheReadTokens > 0` on every request after the
* first — the adapter maps the provider's `prompt_cache_hit_tokens`, and the
* per-step usage recorded on `assistant/message` events is the production
* observable for cache behavior (the reconstructability RFC's measurement
* layer: prefix stability is corollary #1). Mocks prove the requests are
* append-extensions; only the real API proves those bytes actually hit the
* provider cache. Key-gated — skips entirely without $DEEPSEEK_API_KEY.
* With-key proof that log-derived requests translate into real provider cache hits: a
* multi-step tool turn (plus a follow-up turn) against the live DeepSeek API must report
* `cacheReadTokens > 0` on every request after the first — the adapter maps the provider's
* `prompt_cache_hit_tokens`, and the per-step usage recorded on `assistant/message` events is
* the production observable for cache behavior (the reconstructability RFC's measurement
* layer: prefix stability is corollary #1). Mocks establish append-extension;
* this key-gated test establishes a real provider cache hit.
*/
// Long enough that the shared request prefix comfortably spans the provider's

12
packages/core/agent-loop/tests/request-reconstruction.spec.ts
View File

@@ -1,11 +1,9 @@
/**
* Loop-level reconstructability: every request the loop sends is a pure
* function of the session log — messages are the derivation at the step/start
* boundary, the header is the fold of request/header* events — and every
* request is an append-extension of its predecessor unless a logged event
* (compaction replace, header change) explains the difference. The requests
* recorded by the mock adapter are the observable; the offline-rebuild test
* at the bottom is the theorem stated end-to-end.
* Loop-level reconstructability: every request the loop sends is a pure function of the
* session log — messages are the derivation at the step/start boundary, the header is the fold
* of request/header* events — and every request is an append-extension of its predecessor
* unless a logged event (compaction replace, header change) explains the difference. Mock-adapter
* requests are the observable, and the final offline rebuild states the full contract end to end.
*/
import { describe, expect, it } from 'vitest'

12
packages/core/agent-loop/tests/resume.spec.ts
View File

@@ -460,10 +460,8 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
})
it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => {
// Lifecycle 1: run a turn, then inject context while idle. The idle inject
// wraps its context/message in a one-shot turn AND checkpoints it (the turn-enclosure RFC)
// — without an explicit flush or clean dispose, the notice must still reach
// disk, since a crash before the next turn would otherwise lose it.
// Idle injection creates and flushes a one-shot turn. No explicit flush or
// clean disposal follows, so disk presence proves its own checkpoint ran.
const adapter1 = new MockAdapter([textResponse('answer')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const a1 = (await ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent as ReactLoopAgent
@@ -485,10 +483,8 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
})
it('an idle inject() survives persist + resume (turn-enclosed, not dropped as crash tail)', async () => {
// Lifecycle 1: run a turn, then inject context while idle. The idle inject
// wraps its context/message in a one-shot turn so it is turn-enclosed —
// otherwise scanLog would treat the trailing context as a crash tail and
// drop it on reload (the bug this guards).
// Turn enclosure keeps idle context out of crash-tail repair, so it must
// survive persistence and resume.
const adapter1 = new MockAdapter([textResponse('answer')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const a1 = (await ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent as ReactLoopAgent

8
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
View File

@@ -934,11 +934,9 @@ describe('agent scope lifecycle', () => {
order.push(`session-still-stored=${ctx.sessions.get(SessionId('o1-s')) !== undefined}`)
})
// Open a turn so the drain has real work: the loop must finish it BEFORE
// the registry entry goes away (the agent/disposed contract: "its fiber
// and any in-flight turn have been torn down"). Wait for the turn to be
// OPEN in the log — a dispose landing in the pre-step window would drop
// the queued prompt without ever opening a turn.
// Open a turn so disposal must drain real work before registry removal.
// Waiting for turn/start avoids pre-step disposal dropping the queued prompt
// before a turn opens.
const turnOpen = new Promise<void>((resolve) => {
const off = ctx.on('session/event', (_s, event) => {
if (event.type === 'turn/start') { off(); resolve() }

17
packages/core/agent-loop/tests/tool-order.spec.ts
View File

@@ -1,10 +1,9 @@
/**
* Loop-level tool-order determinism: the request/header event — and therefore
* the frozen request the adapter receives — carries the assembly's canonical
* tool order (system-prompt's `toolOrder` config, or lexicographic name
* order), regardless of the order tool plugins happened to register in.
* Registration order is a plugin-load artifact (concurrent dynamic imports
* race), so nothing downstream of the registry may depend on it.
* Loop-level tool-order determinism: the request/header event — and therefore the frozen
* request the adapter receives — carries the assembly's canonical tool order (system-prompt's
* `toolOrder` config, or lexicographic name order), regardless of the order tool plugins
* happened to register in. Registration order is a concurrent loading artifact
* and must not leak downstream.
*/
import { describe, expect, it } from 'vitest'
@@ -93,11 +92,7 @@ describe('loop-level canonical tool order', () => {
})
it('fails the turn — no model request — when toolOrder names an unregistered tool', async () => {
// The assemble rejection escapes to runTurn's outer catch: the open turn
// closes with an `error` reason (agent/error mirrors it), no step opens,
// no request/header is logged, the adapter never sees a request, and the
// agent returns to idle — a misconfigured deployment fails every turn
// deterministically instead of silently reordering nothing.
// Unknown tool order fails before step or request creation and returns the agent to idle.
const adapter = new MockAdapter([textResponse('never sent')])
const ctx = await harness(adapter, ['ghost', TOOL_ORDER_REST])
registerNamed(ctx, 'alpha')