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

Merge branch 'codex/simp-session-dead-surface' into codex/simp-session-log-representation

Browse Source 
# Conflicts:
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/core-data-structures/core.md
#	docs/persistence-catalog.md
#	docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md
#	docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md
#	docs/rfc/implemented/feature/2026-07-06-explicit-tool-order.md
#	docs/rfc/implemented/feature/2026-07-06-sandbox.md
#	docs/rfc/implemented/feature/2026-07-07-session-prefix.md
#	packages/compact/compact-basic/src/index.ts
#	packages/compact/compact-basic/tests/compact-basic.spec.ts
#	packages/core/agent-loop/src/loop.ts
#	packages/core/agent-loop/src/request-log.ts
#	packages/core/agent-loop/tests/request-reconstruction.spec.ts
#	packages/core/agent/src/types.ts
#	packages/core/session/README.md
#	packages/core/session/src/request-header.ts
#	packages/core/session/src/surface.ts
#	packages/core/session/src/tool-pairing.ts
#	packages/core/session/src/types.ts
#	packages/core/session/tests/surface.spec.ts
#	packages/core/session/tests/tool-pairing.spec.ts
#	packages/llm/llm/src/call-config.ts
#	packages/support/acp-snapshot/src/suite.ts
#	packages/support/invariants/README.md
#	packages/ui/user-approval/README.md
This commit is contained in:
Tianyi Cui
2026-07-14 18:34:34 +08:00
parent 443526285e 34a059c9bc
commit 78276c52a7
539 changed files with 3964 additions and 10850 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
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,53 +48,9 @@ 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 additionalContext) | 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 post-execute additionalContext 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
```
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.)
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.
### What belongs to plugins
@@ -128,4 +82,3 @@ Everything that goes beyond "call the model, run the tools, repeat" belongs to p
- **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.
- **`runLoop`/`Inbox`/`InboxMessage` stay exported with no outside consumer** — [removal is proposed](../../../docs/rfc/proposed/simplification/2026-07-04-prune-dead-core-spine-surface.md).

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)
}
@@ -269,18 +254,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) => {
@@ -290,10 +265,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)
}
@@ -301,15 +273,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 /
@@ -329,29 +293,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)
@@ -387,12 +336,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() },
})
}
@@ -432,11 +376,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

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

@@ -117,14 +117,8 @@ describe('agent/prompt-submit', () => {
})
it('a prompt-submit rewrite + additionalContext 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' })
@@ -189,9 +183,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' })
@@ -515,14 +508,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')
})
@@ -618,11 +610,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[] = []
@@ -522,9 +518,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' })
@@ -557,10 +552,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' })
@@ -627,16 +620,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'),
@@ -718,11 +709,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 },
@@ -730,10 +718,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/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')