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

Merge PR #224 updates into prose cleanup

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-12 23:36:49 +08:00
parent 75838e10b5 2ca806a1ec
commit 74ada5777c
165 changed files with 11693 additions and 6395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,43 +7,51 @@
*/
import type { Context } from 'cordis'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
import type { Scoped } from '@deepseek-ai/dsh-scope'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import type { Session } from '@deepseek-ai/dsh-session'
import { Inbox } from './inbox.ts'
import { snapshotJsonValue, type Session } from '@deepseek-ai/dsh-session'
import { Inbox, type InboxMessage } from './inbox.ts'
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
/** Agents whose rollback-covered publication enabled driving. */
const driveEnabledAgents = new WeakSet<ReactLoopAgent>()
/** Sessions already claimed by a concrete driver construction. */
const claimedDriverSessions = new WeakSet<Session>()
/** Module-private driver entry: its symbol is absent from the package surface. */
const startDriver = Symbol('dsh.agent-loop.start-driver')
/** Module-private quiescent stop, valid both before and after driver start. */
const stopDriver = Symbol('dsh.agent-loop.stop-driver')
/** Module-private context binding for the mutually referential agent scope. */
const bindContext = Symbol('dsh.agent-loop.bind-context')
/** Module-private publication marker. */
const publishAgent = Symbol('dsh.agent-loop.publish-agent')
/** Factory-owned controls that can operate only on the agent created with them. */
export interface PreparedReactLoopAgent {
/** The unpublished concrete agent. */
agent: ReactLoopAgent
/** Open its driving verbs at the rollback-covered publication boundary. */
enableDrive(): void
/** Mark the agent public so teardown emits its status lifecycle. */
markPublished(): void
/** Stop the prepared instance even when publication has not started its loop. */
dispose(): Promise<void> | void
/**
* Start its driver after publication and session-start notification.
* The returned disposer reaches quiescence for both the loop and every
* fire-and-forget idle-injection flush the agent started.
*/
startDriver(): () => Promise<void>
startDriver(): () => Promise<void> | void
}
/**
* 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 enable or start the factory's unpublished instance.
* cannot publish or start the factory's unpublished 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.
@@ -56,15 +64,32 @@ export function prepareReactLoopAgent(
if (claimedDriverSessions.has(session)) {
throw new Error(`session "${session.id}" already has a concrete agent driver`)
}
claimedDriverSessions.add(session)
const agent = new ReactLoopAgent(ctx, id, options, session)
claimedDriverSessions.add(session)
const dispose = () => agent[stopDriver]()
return {
agent,
enableDrive: () => { driveEnabledAgents.add(agent) },
startDriver: () => agent[startDriver](),
markPublished: () => { agent[publishAgent]() },
dispose,
startDriver: () => {
agent[startDriver]()
return dispose
},
}
}
/**
* Install the concrete agent's scope context exactly once. Construction and
* scope minting are mutually referential (the scope key is the agent), so the
* factory performs this one post-construction binding before setup receives
* the unpublished agent. The module-private binding rejects a second bind.
* @param agent - the unpublished concrete agent to bind.
* @param ctx - its fully extended agent scope context.
*/
export function bindReactLoopAgentContext(agent: ReactLoopAgent, ctx: Context): void {
agent[bindContext](ctx)
}
/**
* The concrete {@link Agent} implementation owned by the agent-loop plugin.
*
@@ -73,7 +98,7 @@ export function prepareReactLoopAgent(
* the agent/* event taxonomy — plugins never need this class.
*/
export class ReactLoopAgent implements Agent {
/** Queued + steering FIFOs; native-private so setup cannot bypass driving verbs. */
/** Queued + steering FIFOs; native-private so callers cannot bypass the public driving verbs. */
readonly #inbox = new Inbox()
/**
@@ -84,21 +109,20 @@ export class ReactLoopAgent implements Agent {
* context are mutually referential (the scope is keyed BY this agent), so
* neither can exist strictly before the other.
*/
ctx!: Context
private boundContext: Context | undefined
/**
* The dispatch carrier for this agent's own emits (`agent/status`,
* `agent/queued`, `agent/error`): keyed by the agent, base = the agent
* (listener `this` is the agent). Built lazily because it is self-referential.
*/
private get carrier(): Scoped<Agent> {
return (this.#carrier ??= scopeTarget(this, this))
/** The agent's scoped composition context, bound once by its factory. */
get ctx(): Context {
if (this.boundContext === undefined) throw new Error(`agent "${this.id}" context is not bound`)
return this.boundContext
}
#carrier: Scoped<Agent> | undefined
private _status: AgentStatus = 'idle'
private currentAbort: AbortController | undefined
/** Whether runLoop has been installed into {@link done}. */
private driverStarted = false
/** Whether registry publication began and status disposal is externally visible. */
private published = false
/**
* Turn-scoped cancel marker, set by {@link cancel} and read/cleared by the
* driver loop (via the LoopHandle) at every point a turn could start or
@@ -155,14 +179,13 @@ 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).
// 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).
if (status !== 'running') this.settleIdleWaiters()
try {
this.loopCtx.emit(this.carrier, 'agent/status', this, status)
} catch (error: unknown) {
this.loopCtx.logger.warn(`agent "${this.id}": agent/status listener threw on ${status}: ${String(error)}`)
}
agentEvents(this.loopCtx, this).emit('agent/status', status)
}
/**
@@ -180,32 +203,45 @@ export class ReactLoopAgent implements Agent {
return options?.source ?? { kind: 'user' }
}
/** Reject every driving verb while creation setup still owns the agent. */
private assertDriveEnabled(action: string): void {
if (driveEnabledAgents.has(this)) return
throw new Error(`agent "${this.id}" cannot ${action} before creation setup completes`)
/**
* Accept one public send/steer payload as the exact detached record shared by
* the live notification and inbox. Lossless-JSON materialization reads every
* nested field once; deep freeze prevents an observer from rewriting queued
* work before the loop drains it.
*/
private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const accepted = snapshotJsonValue({ content, source })
if (accepted === undefined) {
throw new TypeError('agent message content and source must be losslessly JSON-serializable')
}
return deepFreeze(accepted)
}
/** Reject a driving operation once teardown has synchronously closed the agent. */
private assertNotDisposed(): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
}
send(content: ContentBlock[], options?: SendOptions): void {
this.assertDriveEnabled('send')
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
const source = this.resolveSource(options)
this.#inbox.enqueue({ content, source })
this.loopCtx.emit(this.carrier, 'agent/queued', this, content, { source, steering: false })
this.assertNotDisposed()
const accepted = this.acceptInboxMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
}
steer(content: ContentBlock[], options?: SendOptions): void {
this.assertDriveEnabled('steer')
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
this.assertNotDisposed()
if (this._status !== 'running') { this.send(content, options); return }
const source = this.resolveSource(options)
this.#inbox.steer({ content, source })
this.loopCtx.emit(this.carrier, 'agent/queued', this, content, { source, steering: true })
const accepted = this.acceptInboxMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
}
inject(content: ContentBlock[], options?: SendOptions): void {
this.assertDriveEnabled('inject')
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
this.assertNotDisposed()
const source = this.resolveSource(options)
if (isTurnOpen(this.session)) {
// A turn is open in the LOG (decided from the log, not agent status —
@@ -217,44 +253,47 @@ export class ReactLoopAgent implements Agent {
// No turn open: wrap the injection in a one-shot turn so every event stays
// turn-enclosed (the durability/replay boundary is the turn).
const turn = lastTurnNumber(this.session) + 1
// Once turn/start enters the log, a turn/end is OWED no matter what — even if a throwing
// `session/event` listener escapes from the turn/start append (Session.append pushes the
// event before notifying listeners) or the context/message append throws (non-serializable
// content, throwing listener).
// Once turn/start enters the log, a turn/end is owed even if the message
// append fails acceptance or pre-commit validation. The finally re-checks
// the log and closes only a turn that actually opened; post-commit observers
// are contained by Session and cannot create a false append failure.
try {
this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
this.session.append('context/message', { content, source }, { surfaceOp: 'append' })
} finally {
// Close the turn if turn/start made it into the log.
// Close the turn if turn/start made it into the log. A pre-commit veto
// must escape rather than being mistaken for a committed turn/end.
if (isTurnOpen(this.session)) {
try {
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
} catch {
// turn/end is already in the log (pushed before the listener threw),
// so the turn is balanced; the throw is the listener's bug.
}
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
}
// Decide the durability checkpoint from the LOG, not a flag: a turn was recorded iff this
// turn's turn/start is logged (it may have been closed by a throwing-listener turn/end
// above, which still counts).
// 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.
// 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.
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) => {
const err = error instanceof Error ? error : new Error(String(error))
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
try {
this.loopCtx.emit(this.carrier, 'agent/error', this, turn, 0, err)
} catch {
// contained: the failure is already logged; a throwing agent/error
// listener must not escape this fire-and-forget catch.
}
const rendered = renderThrown(error)
const err = error instanceof Error ? error : new Error(rendered)
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${rendered}`)
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.
// 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.
const retire = (): void => { this.pendingIdleFlushes.delete(flush) }
void flush.then(retire, retire)
}
@@ -262,9 +301,15 @@ export class ReactLoopAgent implements Agent {
}
cancel(reason?: string): void {
this.assertDriveEnabled('cancel')
// Arm-gate: only mark a cancellation when there is actually work to cancel — a running
// turn, an in-flight step, or queued/steering work.
// 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.
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 /
@@ -272,8 +317,10 @@ export class ReactLoopAgent implements Agent {
// below; the marker path reads it via the LoopHandle's cancelReason().
this.cancelReason = reason ?? 'cancelled'
}
// Drop all pending queued + steering work (un-started prompts never run; the cancelled
// turn's steering is not re-enqueued).
// Drop all pending queued + steering work (un-started prompts never run; the
// cancelled turn's steering is not re-enqueued). Cleared directly even when
// the loop is parked in waitForQueued — there is no turn to stop and nothing
// left for the parked loop to run, so no wake is needed.
this.#inbox.clear()
// Interrupt an in-flight step immediately (the running turn observes the
// abort and ends `aborted`). The marker covers the windows where no step is
@@ -282,15 +329,29 @@ export class ReactLoopAgent implements Agent {
}
/**
* Resolve once the agent has reached quiescence after settling out of `running`.
* 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).
*/
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.
// 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.
return new Promise<void>((resolve) => {
this.idleWaiters.push(() => {
resolve(this._status === 'disposed' ? this.done : undefined)
@@ -298,18 +359,25 @@ export class ReactLoopAgent implements Agent {
})
}
/** Bind the mutually referential scope context once. */
private [bindContext](ctx: Context): void {
if (this.boundContext !== undefined) throw new Error(`agent "${this.id}" context is already bound`)
this.boundContext = ctx
}
/** Mark that public lifecycle publication began. */
private [publishAgent](): void {
this.published = true
}
/**
* Start the driver loop. Returns a disposer: calling it sets status to
* `disposed`, emits `agent/status('disposed')`, resolves the disposed
* promise (unblocking the idle wait), releases any `whenIdle` waiters, and
* aborts the current request if any. Its returned promise resolves only after
* the loop exits and every idle-injection flush started by this agent settles.
* @returns the disposer — idempotent, synchronously marks the agent disposed,
* and asynchronously reaches loop + flush quiescence without rejecting (it
* runs inside the fiber's LIFO disposal chain, where a rejection would skip
* later disposers).
* Start the driver loop. The prepared controller already owns its stable
* disposer, so teardown can mark the agent disposed even in the narrow
* publication window before this method runs.
*/
[startDriver](): () => Promise<void> {
[startDriver](): void {
if (this._status === 'disposed') return
this.driverStarted = true
this.done = runLoop(this.loopCtx, this, {
inbox: this.#inbox,
setStatus: (status) => { this.setStatus(status) },
@@ -319,41 +387,63 @@ 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.
// 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.
settleIdle: () => { this.settleIdleWaiters() },
})
// The disposer must be infallible: it runs inside the fiber's LIFO
// disposal chain, where a throw would skip later disposers (e.g. the
// registry unregistration) and leave `done` pending forever.
return async () => {
if (this._status !== 'disposed') {
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
// setStatus refuses transitions out of 'disposed', so emit directly —
// 'disposed' is part of the agent/status contract. Guarded: a throwing
// listener must not break the disposal chain.
try {
this.loopCtx.emit(this.carrier, 'agent/status', this, 'disposed')
} catch {
// listener error during disposal — nothing safe left to do with it
}
}
// An unexpected driver rejection must not skip registry/session/scope
// 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.
while (this.pendingIdleFlushes.size > 0) {
await Promise.allSettled([...this.pendingIdleFlushes])
}
/**
* Quiescent stop shared by pre-start rollback and live teardown. It marks the
* agent disposed synchronously, contains an unexpected loop rejection, and
* drains every idle-injection flush before resolving.
*/
private [stopDriver](): Promise<void> | void {
if (this._status !== 'disposed') {
this._status = 'disposed'
this.resolveDisposed()
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.
this.settleIdleWaiters()
this.currentAbort?.abort('disposed')
// An unpublished rollback has no public status lifecycle to announce.
// Once publication begins, disposed is part of the agent/status contract.
if (this.published) {
agentEvents(this.loopCtx, this).emit('agent/status', 'disposed')
}
}
// Before runLoop starts there is normally nothing asynchronous to drain;
// keep publication rollback synchronous so create() cannot throw while its
// session/agent entries are still briefly live. A session-start listener
// may have called inject(), however, so preserve
// its durability checkpoint as a real quiescence boundary.
if (!this.driverStarted && this.pendingIdleFlushes.size === 0) return
return this.drainDriver()
}
/** Await the loop (when started) and every outstanding idle flush. */
private async drainDriver(): Promise<void> {
// An unexpected driver rejection must not skip registry/session/scope
// 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.
while (this.pendingIdleFlushes.size > 0) {
await Promise.allSettled([...this.pendingIdleFlushes])
}
}
}
/** Render an ordinary thrown value for the error event and log. */
function renderThrown(value: unknown): string {
return value instanceof Error ? value.message : String(value)
}

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

@@ -1,8 +1,6 @@
/**
* THE concrete agent plugin: creates ReactLoopAgents, runs their loops, and
* registers them in ctx.agents. Deliberately thin — every behavior beyond
* "call the model, run the tools, repeat" belongs to plugins on the event
* taxonomy.
* Concrete agent-loop plugin: creates scoped ReactLoopAgents, publishes them
* through the agent/session registries, and owns their ordered teardown.
*
* @module @deepseek-ai/dsh-agent-loop
*/
@@ -13,62 +11,334 @@ import z from 'schemastery'
import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions, SessionStartSource } from '@deepseek-ai/dsh-agent'
import type {
AgentFactory,
AgentHandle,
AgentId,
AgentOptions,
CreateAgentOptions,
ResumeAgentOptions,
SessionStartSource,
} from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-llm'
import { SessionId, type SessionHeader } from '@deepseek-ai/dsh-session'
import type { Session } from '@deepseek-ai/dsh-session'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { Session, SessionHeader } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import { prepareReactLoopAgent, ReactLoopAgent } from './agent.ts'
import {
bindReactLoopAgentContext,
prepareReactLoopAgent,
ReactLoopAgent,
} from './agent.ts'
import type { PreparedReactLoopAgent } from './agent.ts'
export { ReactLoopAgent } from './agent.ts'
/** Fiber states that cannot own or serve a new lifecycle. */
const INACTIVE_STATES: ReadonlySet<FiberState> = new Set([
FiberState.UNLOADING,
FiberState.DISPOSED,
FiberState.FAILED,
])
/** Factory-level ownership of every preparing or live transaction. */
class FactoryOwnership {
private accepting = true
private transactions = new Set<AgentCreationTransaction>()
constructor(private readonly fiber: Context['fiber']) {}
isActive(): boolean {
return this.accepting && !INACTIVE_STATES.has(this.fiber.state)
}
track(transaction: AgentCreationTransaction): () => void {
this.transactions.add(transaction)
return () => { this.transactions.delete(transaction) }
}
async dispose(): Promise<void> {
this.accepting = false
const reason = new Error('agent loop is not active')
await Promise.all(
[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
)
}
}
/** Build the public cancellation error while preserving a caller-supplied cause. */
function signalAbortError(id: AgentId, signal: AbortSignal): Error {
if (signal.reason instanceof Error) return signal.reason
return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
}
/**
* 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.
*/
class AgentCreationTransaction {
private active = true
private failure: Error | undefined
private readonly deactivation = Promise.withResolvers<void>()
private readonly publication = Promise.withResolvers<void>()
private readonly torndown = Promise.withResolvers<void>()
private readonly wrapperCompletion = Promise.withResolvers<void>()
private preparing: Promise<void> | undefined
private driver: PreparedReactLoopAgent | undefined
private scope: Scope | undefined
private session: Session | undefined
private lifecycleDispose: (() => Promise<void> | void) | undefined
private detachSession: (() => void) | undefined
private detachAgent: (() => void) | undefined
private publishing = false
private cleanupTask: Promise<void> | undefined
private ownerFollowing = true
private readonly ownerDispose: () => Promise<void> | void
private readonly untrackFactory: () => void
private readonly abortListener: (() => void) | undefined
readonly ownerAgent: Context['agent']
readonly ownerFiber: Context['fiber']
constructor(
private readonly loopCtx: Context,
private readonly ownerCtx: Context,
private readonly ownership: FactoryOwnership,
readonly id: AgentId,
signal?: AbortSignal,
) {
ownerCtx.fiber.assertActive()
this.ownerAgent = ownerCtx.agent
this.ownerFiber = ownerCtx.fiber
if (!ownership.isActive()) throw new Error('agent loop is not active')
this.ownerDispose = ownerCtx.effect(() => () => {
if (!this.ownerFollowing) return
return this.dispose(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
}, `agentLoop.owner(${id})`)
this.untrackFactory = ownership.track(this)
if (signal === undefined) {
this.abortListener = undefined
} else {
this.abortListener = () => {
/* v8 ignore next 3 -- transaction teardown contains callback/driver failures; rejection is a future-drift backstop. */
void this.dispose(signalAbortError(id, signal)).catch((error: unknown) => {
this.loopCtx.logger.error(error)
})
}
signal.addEventListener('abort', this.abortListener, { once: true })
if (signal.aborted) this.deactivate(signalAbortError(id, signal))
}
this.signal = signal
}
private readonly signal: AbortSignal | undefined
/** Whether caller, provider, and optional parent-agent ownership remain live. */
isActive(): boolean {
return this.active
&& this.ownership.isActive()
&& this.ownerFiber.uid !== null
&& !INACTIVE_STATES.has(this.ownerFiber.state)
&& this.ownerAgent?.status !== 'disposed'
}
/** Fail synchronously at every real lifecycle boundary after deactivation. */
assertActive(): void {
if (this.isActive()) return
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
throw this.failure ?? new Error(`agent "${this.id}" setup aborted: owner disposed during setup`)
}
/** Race an external async operation against structural/signal deactivation. */
async waitFor<T>(operation: PromiseLike<T> | T): Promise<T> {
this.assertActive()
return await Promise.race([
Promise.resolve(operation),
this.deactivation.promise.then(() => {
/* v8 ignore next -- deactivate() assigns failure before resolving deactivation. */
throw this.failure ?? new Error(`agent "${this.id}" creation deactivated`)
}),
])
}
/** Construct the driver and scope, then install their complete ordered lifecycle. */
prepare(options: AgentOptions, session: Session): ReactLoopAgent {
this.assertActive()
const gate = Promise.withResolvers<void>()
this.preparing = gate.promise
try {
this.session = session
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session)
this.driver = driver
const agent = driver.agent
const scope = createScope(this.loopCtx, agent)
this.scope = scope
bindReactLoopAgentContext(agent, scope.ctx.extend({ agent }))
this.installLifecycle(scope, driver)
this.assertActive()
return agent
} catch (error: unknown) {
if (!this.isActive() && error instanceof Error && /inactive context/.test(error.message)) {
throw this.failure ?? this.disposalReason()
}
throw error
} finally {
gate.resolve()
this.preparing = undefined
}
}
/** Register the exact scope disposer inside the ordered transaction effect. */
private installLifecycle(scope: Scope, driver: PreparedReactLoopAgent): void {
this.lifecycleDispose = this.ownerCtx.effect(function* (this: AgentCreationTransaction) {
// First yielded, disposed last.
yield () => { this.finish() }
yield scope.rawDispose
yield () => {
this.detachSession?.()
this.detachSession = undefined
}
yield () => {
this.detachAgent?.()
this.detachAgent = undefined
}
// Last yielded, disposed first.
yield () => {
this.deactivate(this.disposalReason())
if (this.publishing) {
return this.publication.promise.then(() => driver.dispose())
}
return driver.dispose()
}
}.bind(this), `agentLoop.lifecycle(${this.id})`)
}
/** Publish the exact prepared objects and start the driver. */
publish(source: SessionStartSource): AgentHandle {
this.assertActive()
const driver = this.driver
/* v8 ignore next -- publish() is private and every caller invokes prepare() first. */
if (driver === undefined) throw new Error(`agent "${this.id}" is not prepared`)
const agent = driver.agent
const session = this.session
/* v8 ignore next -- prepare() assigns the session before it can produce the driver above. */
if (session === undefined) throw new Error(`agent "${this.id}" has no prepared session`)
this.publishing = true
try {
this.detachSession = agent.ctx.sessions.enter(session)
this.detachAgent = this.loopCtx.agents.enter(agent)
agent.ctx.sessions.announce(session)
this.assertActive()
this.loopCtx.agents.announce(agent)
this.assertActive()
driver.markPublished()
agentEvents(this.loopCtx, agent).emit('agent/session-start', source)
this.assertActive()
driver.startDriver()
return { agent, dispose: () => this.dispose() }
} finally {
this.publishing = false
this.publication.resolve()
}
}
/** Mark the transaction inactive exactly once and wake load/setup races. */
private deactivate(reason: Error): void {
if (!this.active) return
this.active = false
this.failure = reason
this.deactivation.resolve()
}
/** Choose the structural cause when an owner/factory effect starts teardown first. */
private disposalReason(): Error {
if (this.failure !== undefined) return this.failure
if (!this.ownership.isActive()) return new Error('agent loop is not active')
if (this.ownerFiber.uid === null || INACTIVE_STATES.has(this.ownerFiber.state) || this.ownerAgent?.status === 'disposed') {
return new Error(`agent "${this.id}" setup aborted: owner disposed during setup`)
}
return new Error(`agent "${this.id}" lifecycle disposed`)
}
/** Complete ownership bookkeeping after every resource reached quiescence. */
private finish(): void {
this.untrackFactory()
this.ownerFollowing = false
void this.ownerDispose()
this.torndown.resolve()
}
/**
* Deactivate and quiesce this transaction. The promise is memoized because
* Cordis effect disposers are single-shot while handles promise shared
* quiescence to every racing owner.
*/
dispose(reason = new Error(`agent "${this.id}" lifecycle disposed`)): Promise<void> {
this.deactivate(reason)
return (this.cleanupTask ??= (async () => {
if (this.preparing !== undefined) await this.preparing
if (this.lifecycleDispose !== undefined) {
await this.lifecycleDispose()
await this.torndown.promise
return
}
try {
await this.driver?.dispose()
} finally {
try {
await this.scope?.dispose()
} finally {
this.finish()
}
}
})())
}
/** Mark the public create/resume continuation settled and detach its creation-only signal. */
finishWrapper(): void {
if (this.signal !== undefined && this.abortListener !== undefined) {
this.signal.removeEventListener('abort', this.abortListener)
}
this.wrapperCompletion.resolve()
}
/** Factory shutdown joins both resource teardown and the public wrapper's deactivation continuation. */
async disposeForFactory(reason: Error): Promise<void> {
await this.dispose(reason)
await this.wrapperCompletion.promise
}
}
declare module 'cordis' {
interface Context {
agentLoop: AgentLoop
}
}
/**
* Plugin config: the agents to create — or resume, via `resumeSessionId` —
* declaratively at startup, so a cordis.yml deployment needs no code.
*/
/** Plugin configuration for declarative startup agents. */
export interface Config {
/** Agents created from configuration at startup. */
/** Agents created or resumed at plugin startup. */
agents: (AgentOptions & {
/** Agent id to register under; also seeds the fresh per-run session id (`${id}-session-<uuid>`). */
/** Registry identity for the live agent. */
id: AgentId
/** Optional workspace cwd for the config-created fresh session. */
/** Optional workspace for a fresh session. */
cwd?: string
/**
* If set, the config agent RESUMES this persisted session id instead of starting a fresh
* `${id}-session-<uuid>`.
*/
/** Persisted session to resume instead of creating a fresh session. */
resumeSessionId?: SessionId
})[]
}
/**
* The agent-loop plugin (`ctx.agentLoop`): creates {@link ReactLoopAgent}s, runs
* their loops, and registers them in `ctx.agents`. Also implements the
* {@link AgentFactory} seam, so plugins create/resume agents through
* `ctx.agents` (the interface) without depending on this concrete package.
*
* The loop itself is deliberately thin — every behavior beyond "call the
* model, run the tools, repeat" belongs to plugins listening on the event
* taxonomy declared in @deepseek-ai/dsh-agent.
*/
/** Concrete ReactLoopAgent factory and driver service. */
export class AgentLoop extends Service implements AgentFactory {
static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']
/** IDs held by unpublished async creation transactions. */
private pendingAgentIds = new Set<AgentId>()
private pendingSessionIds = new Set<SessionId>()
// The schema validates plain strings (cordis.yml config values are untyped at runtime); the
// {@link Config} TYPE declares the branded `id`/`resumeSessionId` because the config format
// is the boundary where an id enters.
/** Runtime schema for declarative agents. */
static Config = z.object({
agents: z.array(z.object({
id: z.string().required(),
@@ -78,337 +348,143 @@ export class AgentLoop extends Service implements AgentFactory {
})).default([]),
}) as unknown as z<Config>
private readonly ownership: FactoryOwnership
/** Plain holder prevents Cordis from re-tracing the factory's dependency context through a caller shadow. */
private readonly runtime: { ctx: Context }
constructor(ctx: Context, public config: Config) {
super(ctx, 'agentLoop')
// Provide the agent-creation factory to the registry (effect-scoped: the
// slot is cleared on dispose).
ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()')
// The prompt variables the shipped loop provides, registered once.
this.ownership = new FactoryOwnership(ctx.fiber)
this.runtime = { ctx }
ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
ctx.effect(() => ctx.agents.setFactory(this), 'agentLoop.setFactory()')
ctx.systemPrompt.variable('model', context => context.agent?.options.model)
ctx.systemPrompt.variable('cwd', context => context.agent?.session.header.cwd)
for (const { id, cwd, resumeSessionId, ...options } of config.agents) {
if (resumeSessionId !== undefined && resumeSessionId !== '') {
// Wait for a late persistence service before resuming the configured session.
ctx.effect(() => {
const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => {
void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options })
.catch((error: unknown) => {
this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
})
})
// Return the exact child-fiber disposer.
return fiber.dispose
}, `agentLoop.resume(${id})`)
} else {
if (resumeSessionId === undefined || resumeSessionId === '') {
this.create(id, options, cwd === undefined ? {} : { cwd })
continue
}
ctx.effect(() => {
const fiber = ctx.inject(['sessionPersistence'], (childCtx: Context) => {
void this.resumeWith(ctx, childCtx.sessionPersistence, {
agentId: id,
resumeSessionId,
agentOptions: options,
}).catch((error: unknown) => {
ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
})
})
return fiber.dispose
}, `agentLoop.resume(${id})`)
}
}
/**
* Create a config-driven agent with a unique session id for this run.
* @param id - agent id and generated-session prefix.
* @param options - loop options.
* @param meta - optional fresh-session metadata.
* @returns running agent owned by the calling fiber.
* Create an agent on a fresh per-run session, owned by the accessing fiber.
* Constructor-driven config calls use the loop fiber itself.
* @param id - agent registry id.
* @param options - concrete loop options.
* @param meta - optional fresh-session workspace metadata.
* @returns the published running agent.
*/
// TODO(demo): define a production resume-or-create policy for config-driven agents.
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
this.assertAgentIdFree(id)
// The calling fiber owns the prepared session and agent lifecycle.
const session = this.ctx.sessions.prepare(SessionId(`${id}-session-${randomUUID()}`), { meta })
const { agent } = this.start(id, options, session, 'startup')
return agent
}
/**
* Programmatic factory create ({@link AgentFactory}): an agent on a caller-supplied
* `sessionId` (not `${id}-session`), with optional session metadata (validated `cwd`,
* lineage) and an optional `seed` event prefix.
*
* @param options - agent id, caller-supplied session id, optional seed/meta,
* and agent options.
* @returns the handle whose dispose tears down exactly this agent.
*/
async createAgent(options: CreateAgentOptions): Promise<AgentHandle> {
// Snapshot every caller-owned field before the first async setup boundary.
const agentId = options.agentId
const sessionId = options.sessionId
const setup = options.setup
const agentOptions = structuredClone(options.agentOptions ?? {})
const seed = options.seed === undefined ? undefined : structuredClone(options.seed)
const meta = structuredClone(options.meta ?? {})
const release = this.reserve(agentId, sessionId)
const loopCtx = this.runtime.ctx
const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
try {
const session = this.ctx.sessions.prepare(sessionId, {
...seed !== undefined ? { seed } : {},
meta,
})
// A seeded (forked) create is still a fresh start, NOT a resume.
return await this.startOwned(agentId, agentOptions, session, 'startup', setup)
const sessionId = SessionId(`${id}-session-${randomUUID()}`)
const session = loopCtx.sessions.prepare(sessionId, { meta })
const agent = transaction.prepare(options, session)
transaction.publish('startup')
return agent
} catch (error: unknown) {
void transaction.dispose(error instanceof Error ? error : new Error(String(error)))
throw error
} finally {
release()
transaction.finishWrapper()
}
}
/**
* Resume an agent on a persisted session ({@link AgentFactory}). Loads the session log +
* metadata via `ctx.sessionPersistence`, reconstructs the live session with the loaded
* events (so `lastTurnNumber`/`deriveMessages` continue), and starts a fresh agent on it.
* The live session id is the resumed id, not `${agentId}-session`.
*
* @param options - the persisted session id to reload, plus agent id/options.
* @returns the handle for the agent resumed on the reconstructed session.
* Create an owned agent on a caller-supplied session id.
* @param ownerCtx - caller context that structurally owns the transaction.
* @param options - identities, session seed/metadata, loop options, setup, and cancellation.
* @returns the published handle.
*/
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
// Read the service through `ctx.get('sessionPersistence')` — a direct global-store lookup
// keyed by the isolate symbol — not `this.ctx.sessionPersistence`.
const persistence = this.ctx.get('sessionPersistence')
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
this.ownership,
options.agentId,
options.signal,
)
try {
const session = this.runtime.ctx.sessions.prepare(options.sessionId, {
...options.seed === undefined ? {} : { seed: options.seed },
...options.meta === undefined ? {} : { meta: options.meta },
})
const agent = transaction.prepare(options.agentOptions ?? {}, session)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('startup')
} catch (error: unknown) {
await transaction.dispose(error instanceof Error ? error : new Error(String(error)))
throw error
} finally {
transaction.finishWrapper()
}
}
/**
* Resume an owned agent from the configured persistence service.
* @param ownerCtx - caller context that owns load, setup, and the live lifecycle.
* @param options - persisted identity, loop options, setup, and cancellation.
* @returns the published handle.
*/
async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle> {
const persistence = this.runtime.ctx.get('sessionPersistence')
if (persistence === undefined) {
throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)')
}
return this.resumeWith(persistence, options)
return this.resumeWith(ownerCtx, persistence, options)
}
/**
* Resume against an EXPLICIT persistence handle. Factored out of {@link resume}
* so the config-driven path can pass the handle it obtained from a
* `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the
* service's own fiber) did not inject `sessionPersistence`, so reading it
* there from inside the inject child trips the cordis inject guard. The
* sessions store + registry are still read through `this.ctx` (both are in
* AgentLoop's static inject, so they resolve fine).
*/
private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise<AgentHandle> {
// Persistence is an async trust boundary. Reserve, load, reconstruct, and
// publish only the identities/options accepted at entry—never fields
// reread from a caller-owned object after the await.
const agentId = options.agentId
const sessionId = options.resumeSessionId
const agentOptions = structuredClone(options.agentOptions ?? {})
const setup = options.setup
const { promise: ownerDisposed, resolve: markOwnerDisposed } = Promise.withResolvers<void>()
const { promise: transactionSettled, resolve: markTransactionSettled } = Promise.withResolvers<void>()
let observingOwner = true
// Resume must observe its caller from before persistence I/O begins.
const disposeLoadSentinel = this.ctx.effect(() => () => {
if (!observingOwner) return
markOwnerDisposed()
// Owner-triggered teardown does not reach quiescence until the resume
// transaction has observed disposal and released both reservations.
return transactionSettled
}, `agentLoop.resumeLoad(${agentId})`)
try {
const release = this.reserve(agentId, sessionId)
try {
const loadTask = persistence.load(sessionId)
const { meta, events } = await Promise.race([
loadTask,
ownerDisposed.then(() => {
throw new Error(`agent "${agentId}" resume aborted: owner disposed during persistence load`)
}),
])
// An out-of-band direct registry/session insertion can still race this
// service's reservation, so the public enter primitives re-check exact
// liveness at publication.
const session = this.ctx.sessions.prepare(sessionId, {
seed: events,
meta: {
createdAt: meta.createdAt,
...meta.cwd !== undefined ? { cwd: meta.cwd } : {},
...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {},
...meta.seedLength !== undefined ? { seedLength: meta.seedLength } : {},
},
})
// Calling startOwned synchronously installs the complete lifecycle
// effect before it reaches its first setup await. Only then disarm the
// load sentinel: ownership passes directly from one effect to the other
// with no disposal gap.
const starting = this.startOwned(agentId, agentOptions, session, 'resume', setup)
observingOwner = false
await disposeLoadSentinel()
return await starting
} finally {
release()
}
} finally {
try {
// Manual handoff/removal must not return transactionSettled: awaiting that promise from
// inside this transaction would deadlock it.
observingOwner = false
await disposeLoadSentinel()
} finally {
markTransactionSettled()
}
}
}
/**
* Reject a duplicate agent id BEFORE the session is entered into the store, so
* a failed factory call never leaves an orphaned live session (and lazy
* persistence state) behind. `register()` enforces the same uniqueness, but
* only after the session has already entered the store.
*/
private assertAgentIdFree(id: AgentId): void {
if (this.ctx.agents.get(id) !== undefined || this.pendingAgentIds.has(id)) {
throw new Error(`agent "${id}" is already registered`)
}
}
/** Reserve both public identities for one unpublished async transaction. */
private reserve(agentId: AgentId, sessionId: SessionId): () => void {
this.assertAgentIdFree(agentId)
if (this.ctx.sessions.get(sessionId) !== undefined || this.pendingSessionIds.has(sessionId)) {
throw new Error(`session "${sessionId}" already exists`)
}
this.pendingAgentIds.add(agentId)
this.pendingSessionIds.add(sessionId)
return () => {
this.pendingAgentIds.delete(agentId)
this.pendingSessionIds.delete(sessionId)
}
}
/**
* Construct an unpublished agent and synchronously install its complete
* teardown skeleton before any setup await. The closures are assigned their
* session/registry/loop disposers only at publication, while the exact scope
* disposer is nested immediately. Therefore owner unload during setup flips
* `active`, unwinds the scope, and wins the race without any late Cordis
* effect collection.
*/
private prepareLifecycle(id: AgentId, options: AgentOptions, session: Session): {
agent: ReactLoopAgent
active: () => boolean
deactivated: Promise<void>
publish: (source: SessionStartSource) => void
disposeAgent: () => Promise<void>
} {
// When creation is invoked through an agent scope (subagents), the owner agent's disposed
// status flips synchronously at handle teardown—earlier than Cordis reaches nested scope
// effects.
const ownerAgent = this.ctx.agent
const ownerFiber = this.ctx.fiber
const driver = prepareReactLoopAgent(this.ctx, id, options, session)
const { agent } = driver
const scope: Scope = createScope(this.ctx, agent)
agent.ctx = scope.ctx.extend({ agent })
let active = true
let detachSession: (() => void) | undefined
let detachAgent: (() => void) | undefined
let stop: (() => Promise<void>) | undefined
const { promise: deactivated, resolve: markDeactivated } = Promise.withResolvers<void>()
const { promise: torndown, resolve: markTorndown } = Promise.withResolvers<void>()
const dispose = this.ctx.effect(function* () {
// First yielded, disposed last: every preceding teardown stage settled.
yield () => { markTorndown() }
// Exact identity moves the scope fiber out of the owner's concurrent
// sibling list and into this ordered transaction.
yield scope.rawDispose
yield () => {
detachSession?.()
detachSession = undefined
}
yield () => {
detachAgent?.()
detachAgent = undefined
}
// Last yielded, disposed first. Keep the pre-publication path
// synchronous: returning a Promise only after the loop actually began
// lets a failed announcement roll back registry/store before create's
// rejection is observed.
yield () => {
active = false
markDeactivated()
if (stop === undefined) return
return stop()
}
}, 'agentLoop.lifecycle()')
let disposing: Promise<void> | undefined
const disposeAgent = (): Promise<void> => (disposing ??= (async () => {
await dispose()
await torndown
})())
const publish = (source: SessionStartSource): void => {
// Publication is one synchronous, rollback-covered sequence. Setup has
// already completed, so its scoped listeners observe both announcements.
detachSession = agent.ctx.sessions.enter(session)
detachAgent = this.ctx.agents.enter(agent)
this.ctx.sessions.announce(session)
this.ctx.agents.announce(agent)
// Setup is over and both entries are live. Open the driving surface just
// before session-start so its listeners retain their supported ability to
// inject/queue, while setup itself can never drive an unpublished agent.
driver.enableDrive()
try {
agentEvents(this.ctx, agent).emit('agent/session-start', source)
} catch (error: unknown) {
this.ctx.logger.warn(`agent "${id}": agent/session-start listener threw: ${String(error)}`)
}
stop = driver.startDriver()
}
return {
agent,
active: () => active
&& ownerFiber.state !== FiberState.UNLOADING
&& ownerFiber.state !== FiberState.DISPOSED
&& ownerFiber.state !== FiberState.FAILED
&& ownerAgent?.status !== 'disposed',
deactivated,
publish,
disposeAgent,
}
}
/** Publish a no-setup config agent synchronously. */
private start(
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
const lifecycle = this.prepareLifecycle(id, options, session)
try {
lifecycle.publish(source)
return { agent: lifecycle.agent, disposeAgent: lifecycle.disposeAgent }
} catch (error: unknown) {
void lifecycle.disposeAgent()
throw error
}
}
/**
* Build an {@link AgentHandle} for a PREPARED session + a fresh agent.
*/
private async startOwned(
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
setup?: (agentCtx: Context) => Promise<void> | void,
/** Resume through an explicit persistence handle used by the deferred config path. */
private async resumeWith(
ownerCtx: Context,
persistence: SessionPersistence,
options: ResumeAgentOptions,
): Promise<AgentHandle> {
const lifecycle = this.prepareLifecycle(id, options, session)
const transaction = new AgentCreationTransaction(
this.runtime.ctx,
ownerCtx,
this.ownership,
options.agentId,
options.signal,
)
try {
// The owner-disposal branch makes a never-settling setup unable to hold
// the transaction or its ID reservations forever. Promise.race installs
// rejection observation on setup even if owner disposal wins first.
const setupTask = Promise.resolve(setup?.(lifecycle.agent.ctx))
await Promise.race([
setupTask,
lifecycle.deactivated.then(() => {
throw new Error(`agent "${id}" setup aborted: owner disposed during setup`)
}),
])
// Cordis begins a fiber unload synchronously but invokes nested effect disposers from its
// next microtask.
await Promise.resolve()
if (!lifecycle.active()) {
throw new Error(`agent "${id}" setup aborted: owner disposed during setup`)
}
lifecycle.publish(source)
return { agent: lifecycle.agent, dispose: lifecycle.disposeAgent }
const loaded = await transaction.waitFor(persistence.load(options.resumeSessionId))
transaction.assertActive()
const session = this.runtime.ctx.sessions.prepare(options.resumeSessionId, {
seed: loaded.events,
meta: {
createdAt: loaded.meta.createdAt,
...loaded.meta.cwd === undefined ? {} : { cwd: loaded.meta.cwd },
...loaded.meta.parentSession === undefined ? {} : { parentSession: loaded.meta.parentSession },
...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
},
})
const agent = transaction.prepare(options.agentOptions ?? {}, session)
await transaction.waitFor(options.setup?.(agent.ctx))
transaction.assertActive()
return transaction.publish('resume')
} catch (error: unknown) {
await lifecycle.disposeAgent()
await transaction.dispose(error instanceof Error ? error : new Error(String(error)))
throw error
} finally {
transaction.finishWrapper()
}
}
}

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

@@ -11,7 +11,7 @@ import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, ContinuationStop, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -37,22 +37,20 @@ function toError(error: unknown): CodedError {
}
/**
* Validate the runtime result of the terminal-stop serial event. Event types
* protect TypeScript listeners, but JavaScript and casts can still return an
* arbitrary bail value; accepting one as an implicit stop would hide a broken
* policy plugin.
*/
function assertContinuationStop(value: unknown): asserts value is ContinuationStop | undefined {
if (value === undefined) return
const candidate = Object(value) as { action?: unknown }
if (candidate.action !== 'stop') {
throw new Error('agent/turn-stop returned an invalid result; expected { action: \'stop\' } or undefined')
}
}
/**
* Map a model-call {@link FinishReason} to the step error it should raise, or `undefined` when
* the step completed normally.
* 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.
*/
function finishError(finish: FinishReason): CodedError | undefined {
switch (finish.kind) {
@@ -81,8 +79,17 @@ function errorData(err: CodedError): { message: string; code?: string } {
}
/**
* The turn-end contribution of a step's *successful* finish, or `undefined` when the step
* finished ordinarily (a plain `completed`).
* 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`".
*/
function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
switch (finish.kind) {
@@ -140,15 +147,66 @@ export interface LoopHandle {
}
/**
* The agent loop. One invocation drives one agent for its whole lifetime.
* 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 additionalContext) | 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 post-execute additionalContext → 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
* ```
* @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).
// 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).
const transmission = createTransmissionLog()
const { session } = agent
@@ -161,8 +219,19 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
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`.
// 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).
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
@@ -173,8 +242,18 @@ 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`.
// 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).
if (handle.isCancelled()) {
handle.clearCancel()
if (!handle.inbox.hasQueued) {
@@ -183,14 +262,24 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
}
}
// Re-derive turn numbers because idle injection can advance the log.
// 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.
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.
// 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.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
try {
@@ -198,12 +287,21 @@ 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.
// 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.
handle.clearCancel()
// Steering that arrived too late to join an ordinary turn (turn-end listeners, flush)
// becomes queued input so it is never stranded.
// 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.
for (const message of handle.inbox.drainSteering()) {
if (!terminalStopped) handle.inbox.enqueue(message)
}
@@ -217,7 +315,10 @@ async function runTurn(
): Promise<boolean> {
const { session } = agent
// --- Pre-turn.
// --- 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.
const queued = handle.inbox.drainQueued()
const first = queued[0]
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
@@ -230,36 +331,28 @@ async function runTurn(
let errorReported = false
let terminalStopped = false
// Close the open step exactly once (idempotent via stepOpen).
const closeStep = (): boolean => {
if (!stepOpen) return 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.
const closeStep = (): void => {
if (!stepOpen) return
session.append('step/end', { turn, step })
stepOpen = false
// Preserve step balance even when an event listener throws after append.
let failure: unknown
try {
session.append('step/end', { turn, step })
} catch (error: unknown) {
failure = error
}
// A throwing step/end session-event listener surfaces as a turn error via failTurn
// (idempotent).
if (failure !== undefined) {
failTurn(toError(failure))
return true
}
return 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).
// 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).
const failTurn = (err: CodedError): void => {
if (errorReported) return
errorReported = true
// The turn is always still open here: the only failure that can reach failTurn once
// turn/end is appended would be a throwing turn-boundary listener, and turn boundaries are
// durable session events with no agent/* mirror to throw.
// 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)
@@ -269,23 +362,23 @@ async function runTurn(
}
}
// Close the turn.
// 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.
const closeTurn = (): void => {
// Session.append pushes turn/end before notifying session/event listeners, so a throwing
// listener leaves turn/end in the log (the turn is balanced) but would otherwise escape —
// from the outer catch it would propagate to the runLoop backstop.
try {
session.append('turn/end', { turn, reason })
} catch (error: unknown) {
ctx.logger.warn(`agent "${agent.id}": session/event listener threw on turn/end at turn ${turn}: ${toError(error).message}`)
}
session.append('turn/end', { turn, reason })
}
try {
// --- Turn boundary.
// --- Turn boundary. Once turn/start is appended, a turn/end is owed no
// matter what throws below; the catch + closeTurn guarantee it. A pre-commit
// veto leaves no turn/start in the log and therefore owes no turn/end.
session.append('turn/start', { turn, trigger })
// Each drained queued message runs the `agent/prompt-submit` waterfall before it becomes a
// `user/message` — a hook can rewrite the prompt or block it.
// Each drained queued message runs the `agent/prompt-submit` waterfall before
// it becomes a `user/message` — a hook can rewrite the prompt or block it.
// Recorded INSIDE the turn (after turn/start) so every event is turn-enclosed;
// turn/end is now owed, so a throwing prompt-submit listener (the waterfall
// throws) is caught below and the turn still closes.
let anyAllowed = false
// Seeded with a floor (only observable if the batch were empty, which
// runTurn never allows — it is called with ≥1 queued message); each `block`
@@ -299,7 +392,13 @@ async function runTurn(
)
if (decision.kind === 'block') {
lastBlockReason = decision.reason
// Log each veto because turn/end cannot represent every blocked prompt.
// Record the veto durably: `PromptDecision.reason` is the durable record
// of why a prompt was blocked, but a fully-blocked batch's `rejected`
// turn/end only preserves the LAST reason, and a MIXED batch (this prompt
// blocked, another allowed) does not end `rejected` at all — so without
// this append a blocked prompt would vanish from the log whenever any
// sibling prompt is allowed. `prompt/blocked` sits in the open turn in
// place of the `user/message` this prompt would have become.
session.append('prompt/blocked', { content: message.content, source: message.source, reason: decision.reason })
continue
}
@@ -315,7 +414,11 @@ async function runTurn(
}
while (true) {
// A fully blocked batch ends as a balanced zero-step rejected turn.
// 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.
if (!anyAllowed) {
reason = { kind: 'rejected', reason: lastBlockReason }
break
@@ -326,29 +429,56 @@ async function runTurn(
// the request.
drainSteering(agent, handle.inbox, turn)
// The step's AbortController exists before any async pre-step work so a dispose() or
// cancel() — in a synchronous turn-start listener or an async listener whose effect fires
// before we block — always has an armed abort to cancel against. isDisposed below covers
// disposal, which does not set the cancel marker.
// The step's AbortController exists BEFORE any async pre-step work so a
// dispose() or cancel() — in a synchronous turn-start listener or an
// async listener whose effect fires before we block — always has an armed
// abort to cancel against. isDisposed below covers disposal, which does
// NOT set the cancel marker. Cleared on every exit path below.
const abort = new AbortController()
handle.setAbort(abort)
// Assemble the system prompt for this step.
// 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.
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().
// 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).
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.
// 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.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
const composed = await events.waterfall(
@@ -356,9 +486,16 @@ 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.
// 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.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
@@ -367,7 +504,19 @@ async function runTurn(
transmission.sessionPrefix = deepFreeze(structuredClone(composed))
}
// Run compaction between steps so its surface events remain outside step brackets.
// 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 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.
@@ -377,20 +526,28 @@ 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.
// 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).
const boundaryMessages = session.deriveMessages()
// Mark the step open before the append: Session.append pushes the event to the log before
// notifying session/event listeners, so a THROWING step/start listener leaves step/start
// in the log.
stepOpen = true
session.append('step/start', { turn, step })
// Only a committed step/start creates a balancing obligation. A
// pre-commit veto throws before this assignment; post-commit observers
// are contained inside Session.append().
stepOpen = true
// Cancel landing in the step-start window: a synchronous `session/event` step/start
// listener can cancel after the step is already open.
// Cancel landing in the step-start window: a synchronous `session/event`
// step/start listener can cancel after the step is already open. Check
// AFTER the step/start append and before `runStep`: drop the step, end the
// turn accordingly. closeStep balances the already-appended step/start.
if (handle.isCancelled() || handle.isDisposed()) {
handle.setAbort(undefined)
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
@@ -425,14 +582,20 @@ async function runTurn(
break
}
// Preserve max-tokens once any step reports it.
// 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.
const stepReason = stepFinishReason(stepOutcome.finish)
if (stepReason) reason = stepReason
// Steering that arrived during streaming/tool execution.
const steered = drainSteering(agent, handle.inbox, turn)
if (closeStep()) break
closeStep()
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
let decision: ContinuationDecision
@@ -461,12 +624,13 @@ async function runTurn(
// the next iteration's drain records it.
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.
// 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.
let terminalStop = false
try {
const stop = await events.strictSerial('agent/turn-stop', turn)
assertContinuationStop(stop)
const stop = await events.serial('agent/turn-stop', turn)
terminalStop = stop !== undefined
} catch (error: unknown) {
// A broken terminal policy is an ordinary continuation failure: fail
@@ -476,13 +640,19 @@ async function runTurn(
}
if (terminalStop) {
terminalStopped = true
// A continuation reason or listener may have queued steering before the terminal
// checkpoint.
// 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.
handle.inbox.drainSteering()
shouldContinue = false
}
// A turn-scoped marker catches cancellation between step controllers.
// 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.
if (handle.isCancelled()) {
reason = { kind: 'aborted', reason: handle.cancelReason() }
break
@@ -498,11 +668,19 @@ async function runTurn(
// Normal / inline-error loop exit: close the turn.
closeTurn()
} catch (error: unknown) {
// Decide whether this turn was ever opened from the LOG, not a flag.
// 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.
const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
if (!turnStartLogged) throw error
closeStep()
// Choose the close reason.
// 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.)
if (handle.isDisposed() && !errorReported) { // eslint-disable-line @typescript-eslint/no-unnecessary-condition
reason = { kind: 'disposed' }
} else {
@@ -517,8 +695,13 @@ async function runTurn(
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.
// 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.
const err = toError(error)
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
try {
@@ -560,17 +743,27 @@ 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.
// 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.
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).
// 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.
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`)
@@ -592,9 +785,11 @@ 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.
// 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.
const request: GenerateOptions = deepFreeze({
model: header.config.model,
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
@@ -618,16 +813,23 @@ async function runStep(
assembler.push(chunk)
}
// Normalize terminal error chunks into the same failure path as thrown adapter errors.
// 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).
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).
// 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.
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.
@@ -646,7 +848,14 @@ async function runStep(
let message: Message = assembler.message()
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
// Do not append an assistant message without content or usage; omit empty provenance too.
// 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).
if (message.content.length > 0 || assembler.usage) {
session.append(
'assistant/message',
@@ -659,7 +868,11 @@ async function runStep(
// ToolRegistry.execute converts tool failures (including aborts) into
// isError results, so abort is re-checked around every call here.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Per-step buffer of `additionalContext` attached by tools/post-execute listeners.
// Per-step buffer of `additionalContext` 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 tool-call/result adjacency
// (interleaving context between a call's result and the next call's would
// break the pairing the next model request relies on).
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
@@ -671,7 +884,12 @@ async function runStep(
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): arguments cannot change after their audit and history events are logged.
// 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).
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
@@ -681,10 +899,12 @@ 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.
// 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.
callId: call.id,
content: result.content,
isError: result.isError,
@@ -728,9 +948,13 @@ 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`).
*
* 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).
* @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.
*/