Merge remote-tracking branch 'origin/master' into feat/adr0016-type-build-check
This commit is contained in:
imccyu
299
packages/core/agent-loop/src/agent.ts
Normal file
299
packages/core/agent-loop/src/agent.ts
Normal file
@@ -0,0 +1,299 @@
|
||||
/**
|
||||
* The concrete Agent implementation: ReactLoopAgent plus its inbox. Everything
|
||||
* observable happens through session events and the agent/* event taxonomy —
|
||||
* plugins never need this class.
|
||||
*
|
||||
* @module dsh-agent-loop/agent
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import { Inbox } from './inbox'
|
||||
import { isTurnOpen, lastTurnNumber, runLoop } from './loop'
|
||||
|
||||
/**
|
||||
* The concrete {@link Agent} implementation owned by the agent-loop plugin.
|
||||
*
|
||||
* Owns the inbox (queued + steering FIFOs), the per-step AbortController, and
|
||||
* the loop driver. Everything observable happens through session events and
|
||||
* the agent/* event taxonomy — plugins never need this class.
|
||||
*/
|
||||
export class ReactLoopAgent implements Agent {
|
||||
readonly inbox = new Inbox()
|
||||
|
||||
private _status: AgentStatus = 'idle'
|
||||
private currentAbort: AbortController | undefined
|
||||
/**
|
||||
* 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
|
||||
* continue. Armed ONLY when there is something to cancel (a running turn, an
|
||||
* in-flight step, or queued/steering work), so an idle no-op cancel cannot
|
||||
* leave it set to wrongly drop a later prompt.
|
||||
*/
|
||||
private cancelRequested = false
|
||||
/**
|
||||
* The resolved reason for the pending {@link cancel} (`reason ?? 'cancelled'`),
|
||||
* read by the driver loop's marker branches so a turn dropped in a
|
||||
* marker-only window (pre-step / continuation, where no `AbortController`
|
||||
* carries the reason) ends with the SAME `{kind:'aborted', reason}` the
|
||||
* mid-step abort path produces from `abort.signal.reason`. Without this the
|
||||
* caller's `cancel(reason)` would be silently replaced by the literal
|
||||
* 'cancelled' whenever the cancel landed outside a running step — making the
|
||||
* logged reason race-dependent and the public `reason?` param half-effective.
|
||||
*/
|
||||
private cancelReason = 'cancelled'
|
||||
private disposed: Promise<void>
|
||||
private resolveDisposed!: () => void
|
||||
/** Resolves when the driver loop has fully exited (tests/disposal). */
|
||||
done: Promise<void> = Promise.resolve()
|
||||
/**
|
||||
* Pending {@link whenIdle} waiters, resolved by {@link settleIdleWaiters} when
|
||||
* the agent next settles out of `running`. Kept as internal agent state (NOT
|
||||
* an effect-scoped `ctx.on` listener) so a concurrent fiber disposal — which
|
||||
* runs the agent's own listeners' disposers — cannot drop the waiter before
|
||||
* the `disposed` transition fires and leave the promise hanging.
|
||||
*/
|
||||
private idleWaiters: (() => void)[] = []
|
||||
|
||||
constructor(
|
||||
private ctx: Context,
|
||||
public readonly id: AgentId,
|
||||
public readonly options: AgentOptions,
|
||||
public readonly session: Session,
|
||||
) {
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
this.disposed = promise
|
||||
this.resolveDisposed = resolve
|
||||
}
|
||||
|
||||
get status(): AgentStatus {
|
||||
return this._status
|
||||
}
|
||||
|
||||
private setStatus(status: AgentStatus): void {
|
||||
if (this._status === status || this._status === 'disposed') return
|
||||
this._status = status
|
||||
// Release quiescence waiters on a transition OUT of running BEFORE emitting
|
||||
// (the disposer handles the disposed transition separately). Settling first
|
||||
// means a throwing `agent/status` subscriber cannot starve a `whenIdle()`
|
||||
// waiter (AGENTS.md "contain callback exceptions" — a lifecycle await must
|
||||
// not hang on one bad listener).
|
||||
if (status !== 'running') this.settleIdleWaiters()
|
||||
try {
|
||||
this.ctx.emit('agent/status', this, status)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`agent "${this.id}": agent/status listener threw on ${status}: ${String(error)}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve and clear all pending {@link whenIdle} waiters. Called on a
|
||||
* running→idle transition (from {@link setStatus}) and on disposal (from the
|
||||
* {@link start} disposer, which chains `done` for true loop-exit quiescence).
|
||||
*/
|
||||
private settleIdleWaiters(): void {
|
||||
const waiters = this.idleWaiters
|
||||
this.idleWaiters = []
|
||||
for (const resolve of waiters) resolve()
|
||||
}
|
||||
|
||||
private resolveSource(options?: SendOptions): MessageSource {
|
||||
return options?.source ?? { kind: 'user' }
|
||||
}
|
||||
|
||||
send(content: ContentBlock[], options?: SendOptions): void {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
const source = this.resolveSource(options)
|
||||
this.inbox.enqueue({ content, source })
|
||||
this.ctx.emit('agent/queued', this, content, { source, steering: false })
|
||||
}
|
||||
|
||||
steer(content: ContentBlock[], options?: SendOptions): void {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
if (this._status !== 'running') { this.send(content, options); return }
|
||||
const source = this.resolveSource(options)
|
||||
this.inbox.steer({ content, source })
|
||||
this.ctx.emit('agent/queued', this, content, { source, steering: true })
|
||||
}
|
||||
|
||||
inject(content: ContentBlock[], options?: SendOptions): void {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
const source = this.resolveSource(options)
|
||||
if (isTurnOpen(this.session)) {
|
||||
// A turn is open in the LOG (decided from the log, not agent status —
|
||||
// status can be `running` with no turn open): the context/message is
|
||||
// turn-enclosed by that turn, so append it directly.
|
||||
this.session.append('context/message', { content, source })
|
||||
return
|
||||
}
|
||||
// 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). The finally re-checks the log via isTurnOpen() and closes the
|
||||
// turn if one was actually opened, so the log never carries a permanently
|
||||
// open injection turn that would corrupt later turns/replay. (If the
|
||||
// turn/start append throws BEFORE pushing — non-serializable trigger, which
|
||||
// can't happen for our fixed trigger — no turn was opened and none is owed.)
|
||||
try {
|
||||
this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
|
||||
this.session.append('context/message', { content, source })
|
||||
} finally {
|
||||
// Close the turn if turn/start made it into the log. Contain a throwing
|
||||
// turn/end listener: Session.append pushes before notifying, so a throw
|
||||
// here still leaves turn/end in the log (the turn is balanced) — swallow
|
||||
// it so it neither replaces the original exception nor skips the flush
|
||||
// decision below. (It surfaces through the flush path is not needed; the
|
||||
// turn-balance contract is what matters and it holds.)
|
||||
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.
|
||||
}
|
||||
}
|
||||
// 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). A
|
||||
// `turnRecorded` boolean set after append('turn/end') would be skipped by
|
||||
// a throwing turn/end listener, losing the flush for a balanced in-memory
|
||||
// turn (crash before the next turn/dispose would drop the idle injection).
|
||||
const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
|
||||
// Checkpoint the one-shot turn for durability, exactly as the loop does at
|
||||
// every turn/end. The loop is NOT running (we are idle), so nothing else
|
||||
// will flush this turn. Fire-and-forget with error containment: inject()
|
||||
// is synchronous, and a persistence backend failing must not throw into
|
||||
// the caller (e.g. a tool-bash task-done callback). Disposal still drains
|
||||
// independently, so a slow flush is safe. 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) {
|
||||
void Promise.resolve(this.ctx.parallel('session/flush', this.session)).catch((error: unknown) => {
|
||||
const err = error instanceof Error ? error : new Error(String(error))
|
||||
this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
|
||||
try {
|
||||
this.ctx.emit('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.
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
cancel(reason?: string): void {
|
||||
// Arm-gate: only mark a cancellation when there is actually work to cancel —
|
||||
// a running turn, an in-flight step, or queued/steering work. An idle cancel
|
||||
// with nothing pending is a true no-op; arming the marker then would wrongly
|
||||
// drop the NEXT legitimate prompt (the marker is consumed only at the loop's
|
||||
// turn-decision points, which an idle parked loop does not reach until woken
|
||||
// by a real send()). Note the gate canNOT be `status === 'running'` alone:
|
||||
// the pre-step window (a send() queued but the loop not yet flipped to
|
||||
// running) has status `idle` with `hasQueued` true, and the marker exists
|
||||
// precisely to cover it.
|
||||
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 /
|
||||
// continuation). The mid-step path reads it from abort.signal.reason
|
||||
// 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). 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
|
||||
// running (pre-step, continuation).
|
||||
this.currentAbort?.abort(reason ?? 'cancelled')
|
||||
}
|
||||
|
||||
/**
|
||||
* 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()`, which awaits {@link done}
|
||||
* directly, 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. 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)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. The returned `agent.done` promise
|
||||
* resolves once the loop exits.
|
||||
*/
|
||||
start(): () => void {
|
||||
this.done = runLoop(this.ctx, this, {
|
||||
setStatus: (status) => { this.setStatus(status) },
|
||||
setAbort: controller => void (this.currentAbort = controller),
|
||||
disposed: this.disposed,
|
||||
isDisposed: () => this._status === 'disposed',
|
||||
isCancelled: () => this.cancelRequested,
|
||||
cancelReason: () => this.cancelReason,
|
||||
clearCancel: () => { this.cancelRequested = false },
|
||||
// Settle whenIdle() waiters WITHOUT a status transition — the pre-step
|
||||
// cancel-skip path drops the about-to-run turn and re-parks without ever
|
||||
// flipping running→idle, so a waiter registered in the pre-step window
|
||||
// (status idle, hasQueued was true) would otherwise hang. This emits no
|
||||
// agent/status, so an ACP agent/status listener never sees a spurious idle
|
||||
// that would resolve a freshly-queued prompt as cancelled.
|
||||
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 () => {
|
||||
if (this._status === 'disposed') return
|
||||
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.ctx.emit('agent/status', this, 'disposed')
|
||||
} catch {
|
||||
// listener error during disposal — nothing safe left to do with it
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
75
packages/core/agent-loop/src/inbox.ts
Normal file
75
packages/core/agent-loop/src/inbox.ts
Normal file
@@ -0,0 +1,75 @@
|
||||
/**
|
||||
* Per-agent message inbox: queued and steering FIFOs. Purely an in-memory
|
||||
* mechanism of the loop driver — the public surface is `Agent.send()` and
|
||||
* `Agent.steer()`.
|
||||
*
|
||||
* @module dsh-agent-loop/inbox
|
||||
*/
|
||||
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** One message waiting in an agent's inbox. */
|
||||
export interface InboxMessage {
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-agent inbox: a queued FIFO (drained at turn start) and a steering FIFO
|
||||
* (drained between steps of a running turn). Purely an in-memory mechanism of
|
||||
* the loop — the public surface is `Agent.send()` / `Agent.steer()`.
|
||||
*/
|
||||
export class Inbox {
|
||||
private queuedMessages: InboxMessage[] = []
|
||||
private steeringMessages: InboxMessage[] = []
|
||||
private wakeup: (() => void) | undefined
|
||||
|
||||
/** Resolves when a queued message arrives (used by the idle loop). */
|
||||
get hasQueued(): boolean {
|
||||
return this.queuedMessages.length > 0
|
||||
}
|
||||
|
||||
get hasSteering(): boolean {
|
||||
return this.steeringMessages.length > 0
|
||||
}
|
||||
|
||||
enqueue(message: InboxMessage): void {
|
||||
this.queuedMessages.push(message)
|
||||
this.wakeup?.()
|
||||
}
|
||||
|
||||
steer(message: InboxMessage): void {
|
||||
this.steeringMessages.push(message)
|
||||
}
|
||||
|
||||
/** Drain all queued messages (turn start). */
|
||||
drainQueued(): InboxMessage[] {
|
||||
return this.queuedMessages.splice(0)
|
||||
}
|
||||
|
||||
/** Drain all steering messages (between steps). */
|
||||
drainSteering(): InboxMessage[] {
|
||||
return this.steeringMessages.splice(0)
|
||||
}
|
||||
|
||||
/**
|
||||
* Discard all pending messages (queued + steering) without delivering them —
|
||||
* used by `cancel()`, which drops un-started work rather than draining it into
|
||||
* a turn. Unlike `drainQueued`/`drainSteering`, the messages are thrown away.
|
||||
*/
|
||||
clear(): void {
|
||||
this.queuedMessages.length = 0
|
||||
this.steeringMessages.length = 0
|
||||
}
|
||||
|
||||
/** Wait until a queued message arrives or `cancel` resolves. */
|
||||
waitForQueued(cancel: Promise<void>): Promise<void> {
|
||||
if (this.hasQueued) return Promise.resolve()
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
this.wakeup = resolve
|
||||
void cancel.then(resolve)
|
||||
return promise.finally(() => {
|
||||
if (this.wakeup === resolve) this.wakeup = undefined
|
||||
})
|
||||
}
|
||||
}
|
||||
300
packages/core/agent-loop/src/index.ts
Normal file
300
packages/core/agent-loop/src/index.ts
Normal file
@@ -0,0 +1,300 @@
|
||||
/**
|
||||
* 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.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-agent-loop
|
||||
*/
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import z from 'schemastery'
|
||||
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent'
|
||||
import type {} from '@deepseek-ai/dsh-llm'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session } 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 { ReactLoopAgent } from './agent'
|
||||
|
||||
export { ReactLoopAgent } from './agent'
|
||||
export { Inbox, type InboxMessage } from './inbox'
|
||||
export { runLoop } from './loop'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
agentLoop: AgentLoop
|
||||
}
|
||||
}
|
||||
|
||||
export interface Config {
|
||||
/** Agents created from configuration at startup. */
|
||||
agents: (AgentOptions & {
|
||||
id: AgentId
|
||||
/**
|
||||
* If set, the config agent RESUMES this persisted session id instead of
|
||||
* starting a fresh `${id}-session-<uuid>`. Sourced from an env var in
|
||||
* cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a
|
||||
* demo can continue a prior conversation without code changes. Requires a
|
||||
* `dsh-session-persistence` backend; the resume is deferred until that
|
||||
* service is available (via `ctx.inject`) and the loaded session's events
|
||||
* seed the live session so history continues.
|
||||
*
|
||||
* The schema accepts a plain string at runtime (cordis.yml values are
|
||||
* untyped); the brand is compile-time only — the config format is the
|
||||
* boundary where an id enters, so the TYPE declares the brand here.
|
||||
*/
|
||||
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.
|
||||
*/
|
||||
export class AgentLoop extends Service implements AgentFactory {
|
||||
static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']
|
||||
|
||||
// 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. The brand is a
|
||||
// zero-cost compile-time cast, so the runtime schema stays string-based and we
|
||||
// assert the branded view once here — the single schema boundary.
|
||||
static Config = z.object({
|
||||
agents: z.array(z.object({
|
||||
id: z.string().required(),
|
||||
model: z.string(),
|
||||
systemPrompt: z.string(),
|
||||
resumeSessionId: z.string(),
|
||||
})).default([]),
|
||||
}) as unknown as z<Config>
|
||||
|
||||
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()')
|
||||
for (const { id, resumeSessionId, ...options } of config.agents) {
|
||||
if (resumeSessionId !== undefined && resumeSessionId !== '') {
|
||||
// Resume a prior session instead of starting fresh. resume() needs
|
||||
// `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml
|
||||
// lists the backend later). `ctx.inject(['sessionPersistence'], cb)`
|
||||
// runs `cb` with a child ctx once the service exists; the child reads
|
||||
// the persistence and hands it to resumeWith (which uses this.ctx — the
|
||||
// parent — for sessions/registry, all in AgentLoop's static inject). A
|
||||
// failed resume is contained + logged: startup must not crash.
|
||||
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 () => void fiber.dispose()
|
||||
}, `agentLoop.resume(${id})`)
|
||||
} else {
|
||||
this.create(id, options)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Config-driven create: an agent on a FRESH, non-colliding session id per run
|
||||
* (`${id}-session-<uuid>`, no cwd). Used for `cordis.yml`-configured agents
|
||||
* and as the shared core for the programmatic factory {@link createAgent}.
|
||||
*
|
||||
* Why a per-run id, not a fixed `${id}-session`: once a durable persistence
|
||||
* backend is loaded, a fixed id collides on the second run — the backend
|
||||
* refuses to re-create an id whose log already exists on disk (the SessionId
|
||||
* is the identity). A fresh id means each run is a new session.
|
||||
*
|
||||
* TODO(demo): each run starting a brand-new session is fine for demos but is
|
||||
* NOT real conversation continuity. A production config-driven agent needs a
|
||||
* deliberate resume-or-create policy (resume the prior session if one exists,
|
||||
* else start fresh) or an explicit caller-chosen session id — revisit when the
|
||||
* UI/ACP path owns session selection.
|
||||
*
|
||||
* TODO(sub-agents): spawn/fork land here — accept a parent agent reference;
|
||||
* fork seeds the new Session with the parent's event log, spawn starts
|
||||
* fresh; the child is returned as a regular Agent handle.
|
||||
*/
|
||||
create(id: AgentId, options: AgentOptions = {}): ReactLoopAgent {
|
||||
this.assertAgentIdFree(id)
|
||||
// Config/programmatic path: prepare the session and let start() fold its
|
||||
// lifecycle into the agent's composite effect (so a fiber unload tears the
|
||||
// session + agent down as one ordered chain, capturing the loop's closing
|
||||
// flush). The whole effect is owned by THIS fiber; no AgentHandle is needed.
|
||||
const session = this.ctx.sessions.prepare(SessionId(`${id}-session-${randomUUID()}`), { meta: {} })
|
||||
const { agent } = this.start(id, options, session)
|
||||
return agent
|
||||
}
|
||||
|
||||
/**
|
||||
* Programmatic factory create ({@link AgentFactory}): an agent on a
|
||||
* caller-supplied `sessionId` (NOT `${id}-session`), with optional session
|
||||
* metadata (validated `cwd`, lineage). The ACP bridge uses this so the
|
||||
* client-generated session id becomes the live/persisted session id. Returns
|
||||
* an {@link AgentHandle} the owner disposes to tear down exactly this agent.
|
||||
*/
|
||||
createAgent(options: CreateAgentOptions): AgentHandle {
|
||||
// Check the agent id BEFORE preparing the session: register() would reject a
|
||||
// duplicate id only AFTER the session enters the store, leaving an orphaned
|
||||
// live session (and lazy persistence state) that blocks reuse of that id.
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const session = this.ctx.sessions.prepare(options.sessionId, { meta: options.meta ?? {} })
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
|
||||
}
|
||||
|
||||
/**
|
||||
* 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`.
|
||||
*
|
||||
* Requires `ctx.sessionPersistence`; rejects with a clear error if it is not
|
||||
* configured. NOT hard-injected (that would make non-persistent demos pend
|
||||
* forever) — callers that need resume (ACP) inject `sessionPersistence`, so
|
||||
* by the time this runs the service exists.
|
||||
*/
|
||||
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`. AgentLoop deliberately does NOT inject
|
||||
// `sessionPersistence` (injecting it would pend non-persistent demos
|
||||
// forever). The `ctx.<name>` property proxy resolves a service by an
|
||||
// ancestor-only walk of the current fiber's parent chain; from AgentLoop's
|
||||
// own fiber (which lacks the inject) that walk never reaches the sibling
|
||||
// backend fiber and throws "cannot get property … without inject". Worse,
|
||||
// when the call arrives via a traceable shadow (e.g. the ACP bridge child
|
||||
// fiber → `ctx.agents.resume()` → `this.factory.resume()`), the walk starts
|
||||
// at the shadow's origin fiber and fails the same way. `ctx.get(name)`
|
||||
// sidesteps the fiber walk entirely (a store lookup by the global isolate
|
||||
// key), so resume works from any caller fiber. It is strict by default: a
|
||||
// backend that is not ACTIVE (absent, or mid-teardown) reads as undefined
|
||||
// and we reject below, rather than handing back an unusable handle.
|
||||
const persistence = this.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)
|
||||
}
|
||||
|
||||
/**
|
||||
* 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> {
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
const { meta, events } = await persistence.load(options.resumeSessionId)
|
||||
// Re-check the agent id AFTER the await: the pre-load check above can go
|
||||
// stale while load() is pending (a concurrent resume/create may register the
|
||||
// same id). Re-checking immediately before prepare()/start keeps the
|
||||
// "no orphaned session on a duplicate id" guarantee under concurrency.
|
||||
this.assertAgentIdFree(options.agentId)
|
||||
// Reconstruct the live session with the FULL persisted header (createdAt,
|
||||
// cwd, lineage) so resume preserves identity, not just the cwd. The seed
|
||||
// events make lastTurnNumber/deriveMessages continue; the backend already
|
||||
// has state (cursor) from the load above, so onCreated is a no-op and the
|
||||
// seed is not re-persisted. prepare() (not create()) so the session
|
||||
// lifecycle folds into the agent's composite effect (ordered teardown).
|
||||
const session = this.ctx.sessions.prepare(options.resumeSessionId, {
|
||||
seed: events,
|
||||
meta: {
|
||||
createdAt: meta.createdAt,
|
||||
...meta.cwd !== undefined ? { cwd: meta.cwd } : {},
|
||||
...meta.parentSession !== undefined ? { parentSession: meta.parentSession } : {},
|
||||
},
|
||||
})
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
|
||||
}
|
||||
|
||||
/**
|
||||
* 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) {
|
||||
throw new Error(`agent "${id}" is already registered`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shared: construct a ReactLoopAgent over a PREPARED (not-yet-entered)
|
||||
* session, then build the ONE composite effect that owns the whole agent
|
||||
* lifecycle — session entry, registry registration, and the loop. Keeping all
|
||||
* three in a SINGLE effect (not sibling effects) is load-bearing: a fiber
|
||||
* unload disposes sibling effects CONCURRENTLY (`Promise.all`), which would
|
||||
* race the session detach against the loop's closing flush and drop the
|
||||
* closing `turn/end`. Inside one effect the disposers run as an ORDERED LIFO
|
||||
* chain — the runtime awaits each disposer's returned promise before the next:
|
||||
*
|
||||
* yield session-detach (disposed LAST — detach onAppend + remove entry)
|
||||
* yield register (disposed 2nd — unregister)
|
||||
* yield stop-and-drain (disposed FIRST — request loop stop, await agent.done)
|
||||
*
|
||||
* So on teardown: the loop is stopped and AWAITED to exit (its final
|
||||
* `session/flush` + `turn/end` fire through the still-attached `onAppend`),
|
||||
* THEN the agent is unregistered, THEN the session is detached — capturing the
|
||||
* closing events before detach, whether the trigger is the handle's `dispose()`
|
||||
* OR a fiber unload. Rollback safety: each yield runs before the next mutation,
|
||||
* so a throwing `session/created`/`agent/created` listener unwinds the
|
||||
* already-yielded disposers instead of leaking.
|
||||
*
|
||||
* Returns the agent plus the composite effect's disposer (`disposeAgent`).
|
||||
*/
|
||||
private start(id: AgentId, options: AgentOptions, session: Session): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
|
||||
const agent = new ReactLoopAgent(this.ctx, id, options, session)
|
||||
const dispose = this.ctx.effect(function* (this: AgentLoop) {
|
||||
yield this.ctx.sessions.enter(session)
|
||||
this.ctx.sessions.announce(session)
|
||||
yield this.ctx.agents.register(agent)
|
||||
const stop = agent.start()
|
||||
// Disposed FIRST (LIFO): request loop stop (sync), then AWAIT the loop's
|
||||
// actual exit so its closing flush lands while onAppend (yielded above,
|
||||
// disposed later) is still attached.
|
||||
yield async () => { stop(); await agent.done }
|
||||
}.bind(this), 'agentLoop.start()')
|
||||
return { agent, disposeAgent: async () => { await dispose() } }
|
||||
}
|
||||
|
||||
/**
|
||||
* Build an {@link AgentHandle} for a PREPARED session + a fresh agent. The
|
||||
* handle's `dispose()` runs the composite effect's disposer (see
|
||||
* {@link start}) — which stops the loop, awaits its exit (final flush
|
||||
* captured), unregisters the agent, and detaches the session, in that order.
|
||||
* The same composite effect is what a fiber unload disposes, so both teardown
|
||||
* triggers honor the ordering identically.
|
||||
*
|
||||
* `dispose()` is MEMOIZED: the underlying cordis effect disposer is
|
||||
* single-shot (a second call returns immediately because the effect's epoch is
|
||||
* already cleared, NOT awaiting the in-flight teardown), so concurrent/repeated
|
||||
* `dispose()` calls would otherwise resolve before the first call's
|
||||
* `await agent.done` + final flush completed. Memoizing the promise makes every
|
||||
* caller observe the SAME quiescence boundary, honoring the
|
||||
* `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
|
||||
* helper).
|
||||
*/
|
||||
private startOwned(id: AgentId, options: AgentOptions, session: Session): AgentHandle {
|
||||
const { agent, disposeAgent } = this.start(id, options, session)
|
||||
let disposing: Promise<void> | undefined
|
||||
return { agent, dispose: () => (disposing ??= disposeAgent()) }
|
||||
}
|
||||
}
|
||||
|
||||
export default AgentLoop
|
||||
695
packages/core/agent-loop/src/loop.ts
Normal file
695
packages/core/agent-loop/src/loop.ts
Normal file
@@ -0,0 +1,695 @@
|
||||
/**
|
||||
* The agent loop driver: one `runLoop()` invocation drives one agent for its
|
||||
* whole lifetime. Error-contained at the turn level — a throwing plugin ends
|
||||
* the turn, never kills the loop. See the JSDoc on `runLoop()` for the full
|
||||
* lifecycle pseudo-code.
|
||||
*
|
||||
* @module dsh-agent-loop/loop
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import type { FinishReason, GenerateOptions, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { BlockAssembler, HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import type { ReactLoopAgent } from './agent'
|
||||
|
||||
/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
|
||||
type CodedError = Error & { code?: string }
|
||||
|
||||
/**
|
||||
* Normalize an arbitrary thrown value into a coded Error. A real Error passes
|
||||
* through (its `code`, if any, is preserved by {@link errorData}); a non-Error
|
||||
* throw is wrapped in a {@link HarnessError} with code `UNKNOWN` and the
|
||||
* original value chained as `cause`, so a bad throw still carries a routable
|
||||
* code instead of degrading to a bare message.
|
||||
*/
|
||||
function toError(error: unknown): CodedError {
|
||||
return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
|
||||
}
|
||||
|
||||
/**
|
||||
* Map a model-call {@link FinishReason} to the step error it should raise, or
|
||||
* `undefined` when the step completed normally.
|
||||
*
|
||||
* Adapters report provider/transport failures one of two sanctioned ways (see
|
||||
* the StreamChunk contract in dsh-llm): throw from `stream()` (handled by the
|
||||
* caller's try/catch), OR end the stream with a finish-error/aborted chunk
|
||||
* (the only option for adapters that can't throw mid-stream, e.g.
|
||||
* library-backed ones). This translates the latter into a thrown step error
|
||||
* so the turn ends error/aborted (the failure recorded on `turn/end.reason`),
|
||||
* never as a normal `completed` assistant message.
|
||||
*
|
||||
* `FinishReason` is merge-extensible (plugins/adapters can add `kind`s), so
|
||||
* the switch handles the known terminal-failure kinds and treats every other
|
||||
* kind — `stop`, `tool-calls`, `max-tokens`, future additions — as success.
|
||||
*/
|
||||
function finishError(finish: FinishReason): CodedError | undefined {
|
||||
switch (finish.kind) {
|
||||
case 'error': {
|
||||
const error: CodedError = new Error(finish.message)
|
||||
if (finish.code !== undefined) error.code = finish.code
|
||||
return error
|
||||
}
|
||||
case 'aborted': {
|
||||
const error: CodedError = new Error('model stream aborted')
|
||||
error.code = 'ABORTED'
|
||||
return error
|
||||
}
|
||||
// stop / tool-calls / max-tokens / plugin-added kinds → not a failure.
|
||||
default:
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the `{ message, code? }` part of an error payload, omitting the
|
||||
* `code` key entirely when absent (exactOptionalPropertyTypes-correct).
|
||||
*/
|
||||
function errorData(err: CodedError): { message: string; code?: string } {
|
||||
return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
|
||||
}
|
||||
|
||||
/**
|
||||
* The turn-end contribution of a step's *successful* finish, or `undefined`
|
||||
* when the step finished ordinarily (a plain `completed`).
|
||||
*
|
||||
* {@link finishError} has already converted `error`/`aborted` finishes into
|
||||
* thrown step errors, so the finishes that reach here are `stop`,
|
||||
* `tool-calls`, `max-tokens`, or a future merge-extensible kind. Only
|
||||
* `max-tokens` carries forward as a distinct {@link TurnEndReason}: a step that
|
||||
* hit the output-token ceiling ended the turn cut-short rather than by the
|
||||
* model's choice. `stop`/`tool-calls`/unknown kinds contribute nothing beyond
|
||||
* the default `completed`. {@link runTurn} applies this with the rule "any
|
||||
* `max-tokens` step in the turn makes the turn end `max-tokens`".
|
||||
*/
|
||||
function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
|
||||
switch (finish.kind) {
|
||||
case 'max-tokens':
|
||||
return { kind: 'max-tokens' }
|
||||
// stop / tool-calls / plugin-added kinds → no turn-end contribution
|
||||
// beyond the default `completed`. FinishReason is merge-extensible, so a
|
||||
// default (not assertNever) handles unknown kinds as ordinary success.
|
||||
default:
|
||||
return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ambient handles the loop driver receives from the agent. Decouples the
|
||||
* pure function `runLoop` from the mutable ReactLoopAgent fields, making the
|
||||
* loop testable without a real agent.
|
||||
*/
|
||||
export interface LoopHandle {
|
||||
setStatus(status: 'idle' | 'running'): void
|
||||
setAbort(controller: AbortController | undefined): void
|
||||
/** Resolves when the agent is disposed — unblocks the idle wait. */
|
||||
disposed: Promise<void>
|
||||
isDisposed(): boolean
|
||||
/**
|
||||
* Whether a `cancel()` is pending for the current turn. The driver checks this
|
||||
* at every decision point where a turn could start or continue (right after
|
||||
* the idle wait, after the `running` flip, before each step, and at the
|
||||
* continuation gate) and drops the about-to-run / continuing turn. Reset once
|
||||
* per loop iteration via {@link clearCancel} after the turn returns, so the
|
||||
* marker governs exactly one cancellation and never leaks to a later prompt.
|
||||
*/
|
||||
isCancelled(): boolean
|
||||
/**
|
||||
* The resolved reason for the pending cancel (`reason ?? 'cancelled'`), read
|
||||
* by the marker branches (pre-step / continuation) so a turn dropped where no
|
||||
* `AbortController` carries the reason still records the caller's
|
||||
* `cancel(reason)` value — matching the mid-step abort path. Only meaningful
|
||||
* when {@link isCancelled} is true.
|
||||
*/
|
||||
cancelReason(): string
|
||||
/** Clear the cancel marker (called once per iteration after the turn returns). */
|
||||
clearCancel(): void
|
||||
/**
|
||||
* Settle pending `whenIdle()` waiters WITHOUT a status transition. Used by the
|
||||
* pre-step cancel-skip path: it drops the about-to-run turn and re-parks at the
|
||||
* idle wait, so no `running→idle` transition fires to settle a `whenIdle()`
|
||||
* waiter that was registered in the pre-step window — this settles it directly
|
||||
* (it emits no `agent/status`, so an ACP `agent/status` listener never sees a
|
||||
* spurious idle that would resolve a freshly-queued prompt as cancelled).
|
||||
*/
|
||||
settleIdle(): void
|
||||
}
|
||||
|
||||
/**
|
||||
* The agent loop. One invocation drives one agent for its whole lifetime:
|
||||
*
|
||||
* ```
|
||||
* forever:
|
||||
* wait for queued messages (idle)
|
||||
* TURN (error-contained — a throwing plugin ends the turn, never the loop):
|
||||
* drain queued → 'turn/start' → session('user/message'…) → emit agent/turn-start
|
||||
* STEP loop:
|
||||
* drain steering → session('steering/message') ⟵ catches late steering
|
||||
* session('step/start'); emit agent/step-start ⟵ append before emit (the event-sourcing RFC)
|
||||
* assembly = ctx.systemPrompt.assemble() ⟵ waterfall system-prompt/assemble
|
||||
* req = {model, system, tools, messages: session.deriveMessages(), signal}
|
||||
* req = waterfall agent/request ⟵ hooks/compaction/model-switch
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
|
||||
* session('assistant/chunk'); emit agent/stream-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() ⟵ waterfall tools/execute
|
||||
* session('tool/result')
|
||||
* drain steering → session('steering/message'); emit agent/steering
|
||||
* emit agent/step-end
|
||||
* cont = waterfall agent/turn-continuation(default = hadToolCalls || steered)
|
||||
* if !cont && steering arrived from step-end/continuation listeners: cont = true
|
||||
* if !cont: break
|
||||
* session('turn/end'); emit agent/turn-end
|
||||
* await ctx.parallel('session/flush', session) ⟵ durability checkpoint
|
||||
* re-enqueue leftover steering as queued ⟵ steering is never stranded
|
||||
* idle (emit agent/status) unless more queued
|
||||
* ```
|
||||
*/
|
||||
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
|
||||
const { session } = agent
|
||||
|
||||
while (!handle.isDisposed()) {
|
||||
await agent.inbox.waitForQueued(handle.disposed)
|
||||
if (handle.isDisposed()) break
|
||||
|
||||
// Pre-step cancel (window 1): a `cancel()` landed after a `send()` woke the
|
||||
// idle wait but before we flip to `running`. The cancelled queued/steering
|
||||
// work is already cleared by `cancel()`. Clear the marker, then:
|
||||
// - if NOTHING new is queued, drop the about-to-run turn and re-park,
|
||||
// settling any `whenIdle()` waiter DIRECTLY (no running→idle transition
|
||||
// fires here to settle it) and WITHOUT emitting `agent/status` (an ACP
|
||||
// listener must not see a spurious idle that resolves a freshly-queued
|
||||
// prompt as cancelled);
|
||||
// - if a NEW prompt was queued AFTER the cancel (a send() that raced in
|
||||
// before the loop resumed), the marker was for the cancelled work only —
|
||||
// fall through and run the new prompt's turn. Do NOT settle waiters here:
|
||||
// a whenIdle() waiter must wait for that new turn's running→idle, not
|
||||
// resolve before it runs (the quiescence contract).
|
||||
if (handle.isCancelled()) {
|
||||
handle.clearCancel()
|
||||
if (!agent.inbox.hasQueued) {
|
||||
handle.settleIdle()
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
handle.setStatus('running')
|
||||
|
||||
// Pre-step cancel (window 2): `setStatus('running')` emits `agent/status`
|
||||
// SYNCHRONOUSLY, so a `running` listener can `cancel()` in the gap between the
|
||||
// check above and `runTurn`. Mirror window 1: clear the marker, then
|
||||
// - if NOTHING new is queued, drop the about-to-run turn and transition
|
||||
// back to `idle` (`running` was already emitted, so a real idle
|
||||
// transition balances the status AND settles `whenIdle()` waiters);
|
||||
// - if a NEW prompt was queued AFTER the cancel (a `running` listener that
|
||||
// cancels then sends), the marker was for the cancelled work only — fall
|
||||
// through and run the new prompt's turn (status is already `running`), so
|
||||
// a `whenIdle()` waiter resolves on THAT turn's running→idle, not before
|
||||
// it runs. Settling here would resolve quiescence while the replacement
|
||||
// is still queued and unrun (the same early-resolve race window 1 fixes).
|
||||
if (handle.isCancelled()) {
|
||||
handle.clearCancel()
|
||||
if (!agent.inbox.hasQueued) {
|
||||
handle.setStatus('idle')
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
try {
|
||||
await runTurn(ctx, agent, handle, turn)
|
||||
} catch (error: unknown) {
|
||||
// Backstop: runTurn rethrows only a PRE-turn throw (the invariant guard
|
||||
// before turn/start) — no turn/start was appended, so no turn is open and
|
||||
// none is owed. A session `error` here would land outside any turn (after
|
||||
// the previous turn/end), where the persistence backend drops it as a
|
||||
// crash tail (the turn-enclosure RFC). Report via agent/error + the logger only; the
|
||||
// driver survives and moves on.
|
||||
const err = toError(error)
|
||||
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, 0, err)
|
||||
} catch { /* contained: a throwing agent/error listener must not kill the driver */ }
|
||||
}
|
||||
|
||||
// Reset the cancel marker UNCONDITIONALLY here, after the turn returns and
|
||||
// before the next iteration's idle wait. NOT gated on the idle transition
|
||||
// below: a `send()` that lands during the cancelled turn's flush window makes
|
||||
// `hasQueued` true at the `setStatus('idle')` guard, so an idle-gated reset
|
||||
// would never fire and the stale marker would wrongly drop that next prompt's
|
||||
// turn. Resetting per iteration scopes the marker to exactly the turn that was
|
||||
// cancelled.
|
||||
handle.clearCancel()
|
||||
|
||||
// Steering that arrived too late to join this turn (turn-end listeners,
|
||||
// flush) becomes a queued message — it must never be stranded. (A cancelled
|
||||
// turn already cleared its steering, so there is nothing to re-enqueue.)
|
||||
for (const message of agent.inbox.drainSteering()) {
|
||||
agent.inbox.enqueue(message)
|
||||
}
|
||||
|
||||
if (!agent.inbox.hasQueued) handle.setStatus('idle')
|
||||
}
|
||||
}
|
||||
|
||||
async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, turn: number): Promise<void> {
|
||||
const { session } = agent
|
||||
|
||||
// --- Pre-turn. A throw here (the invariant guard) is owed NO turn/end —
|
||||
// turn/start has not been appended — so it propagates to runLoop's backstop
|
||||
// untouched. The queued messages are drained here but appended AFTER
|
||||
// turn/start (below), so every event in the log lives inside a turn.
|
||||
const queued = agent.inbox.drainQueued()
|
||||
const first = queued[0]
|
||||
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
|
||||
if (!first) throw new Error('runTurn invariant violated: no queued message at turn start')
|
||||
const trigger: TurnTrigger = { kind: 'message', source: first.source }
|
||||
|
||||
let reason: TurnEndReason = { kind: 'completed' }
|
||||
let step = 0
|
||||
let turnEnded = false
|
||||
let stepOpen = false
|
||||
let errorReported = false
|
||||
|
||||
// Close the open step exactly once (idempotent via stepOpen). The
|
||||
// agent/step-end emit is contained: a throwing step-end listener must not
|
||||
// abort finalization and strand the turn open (turn/end balance > notifying
|
||||
// one bad listener). Appended before the emit (the event-sourcing RFC append-before-emit).
|
||||
const closeStep = (): boolean => {
|
||||
if (!stepOpen) return false
|
||||
stepOpen = false
|
||||
// Session.append pushes step/end BEFORE notifying session/event listeners,
|
||||
// so a throwing listener leaves step/end in the log (balance holds) but
|
||||
// would otherwise abort finalization. Contain it and surface it as a turn
|
||||
// error below — the same outcome as a throwing agent/step-end listener.
|
||||
let failure: unknown
|
||||
try {
|
||||
session.append('step/end', { turn, step })
|
||||
} catch (error: unknown) {
|
||||
failure = error
|
||||
}
|
||||
try {
|
||||
ctx.emit('agent/step-end', agent, turn, step)
|
||||
} catch (error: unknown) {
|
||||
failure ??= error
|
||||
}
|
||||
// A throwing step/end session-event listener OR a throwing agent/step-end
|
||||
// listener surfaces as a turn error via failTurn (idempotent). This prevents
|
||||
// a throwing listener from producing a silent "completed" turn when the step
|
||||
// itself succeeded, AND keeps finalization going when closeStep runs from
|
||||
// the outer catch.
|
||||
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).
|
||||
// Disposal and abort set `reason` directly without calling this (they are not
|
||||
// failures).
|
||||
const failTurn = (err: CodedError): void => {
|
||||
if (errorReported) return
|
||||
errorReported = true
|
||||
// Set the error reason ONLY while the turn is still open — closeTurn appends
|
||||
// turn/end with it. If the turn has already ended (the only way here: a
|
||||
// throwing agent/turn-end listener after closeTurn(true) already appended
|
||||
// turn/end), the reason can no longer affect the durable log, so log the late
|
||||
// throw directly instead — otherwise the listener exception would vanish.
|
||||
if (!turnEnded) {
|
||||
reason = { kind: 'error', step, ...errorData(err) }
|
||||
} else {
|
||||
ctx.logger.warn(`agent "${agent.id}": agent/turn-end listener threw after turn ${turn} closed: ${err.message}`)
|
||||
}
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, step, err)
|
||||
} catch {
|
||||
// contained: the error is already captured (on `reason`, or via the logger
|
||||
// above); a throwing agent/error listener must not prevent the turn from
|
||||
// closing.
|
||||
}
|
||||
}
|
||||
|
||||
// Close the turn exactly once (idempotent via turnEnded). `emit` is false on
|
||||
// the error path (the failure was already surfaced via agent/error) and true
|
||||
// on the normal/inline-error path. A throwing agent/turn-end listener on the
|
||||
// normal path escapes to the outer catch, which surfaces it via failTurn —
|
||||
// turn/end is already appended, so balance holds either way.
|
||||
const closeTurn = (emit: boolean): void => {
|
||||
if (turnEnded) return
|
||||
turnEnded = true
|
||||
// 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's closeTurn(false) it
|
||||
// would propagate to the runLoop backstop, and from the normal-path
|
||||
// closeTurn(true) it would skip the agent/turn-end emit. Contain it: the
|
||||
// boundary is durable either way, and finalization must not abort on a bad
|
||||
// listener. (On the normal path the outer catch also re-runs closeTurn,
|
||||
// which is an idempotent no-op once turnEnded is set.)
|
||||
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}`)
|
||||
}
|
||||
if (emit) ctx.emit('agent/turn-end', agent, turn, reason)
|
||||
}
|
||||
|
||||
try {
|
||||
// --- Turn boundary. Once turn/start is appended, a turn/end is owed no
|
||||
// matter what throws below; the catch + closeTurn guarantee it (the catch
|
||||
// decides "owed" from the log via isTurnOpen, so even a throwing turn/start
|
||||
// listener — append pushes before notifying — still gets its turn/end).
|
||||
session.append('turn/start', { turn, trigger })
|
||||
// Record the queued user messages INSIDE the turn (after turn/start), so
|
||||
// every event in the log is turn-enclosed. turn/end is now owed, so a throw
|
||||
// while appending these is caught below and the turn is still closed.
|
||||
for (const message of queued) {
|
||||
session.append('user/message', { content: message.content, source: message.source })
|
||||
}
|
||||
ctx.emit('agent/turn-start', agent, turn)
|
||||
|
||||
while (true) {
|
||||
step += 1
|
||||
|
||||
// Steering from the previous round's step-end/continuation listeners
|
||||
// (or turn-start listeners on the first step) joins before the request.
|
||||
drainSteering(ctx, agent, turn)
|
||||
|
||||
session.append('step/start', { turn, step })
|
||||
stepOpen = true
|
||||
ctx.emit('agent/step-start', agent, turn, step)
|
||||
|
||||
const abort = new AbortController()
|
||||
handle.setAbort(abort)
|
||||
|
||||
// Cancel landing in the step-start window: a synchronous `agent/turn-start`
|
||||
// or `agent/step-start` listener (both fire before this point) can have
|
||||
// called `cancel()`, and `runStep` would otherwise run a full extra step
|
||||
// with no AbortController having observed it. Check the marker AFTER
|
||||
// setAbort (so the next-iteration drain sees a clean controller) and before
|
||||
// `runStep`: drop the step, end the turn `aborted`. closeStep balances the
|
||||
// already-appended step/start.
|
||||
if (handle.isCancelled()) {
|
||||
handle.setAbort(undefined)
|
||||
reason = { kind: 'aborted', reason: handle.cancelReason() }
|
||||
closeStep()
|
||||
break
|
||||
}
|
||||
|
||||
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
|
||||
try {
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, abort.signal)
|
||||
} catch (error: unknown) {
|
||||
stepOutcome = { error: toError(error) }
|
||||
} finally {
|
||||
handle.setAbort(undefined)
|
||||
}
|
||||
|
||||
if ('error' in stepOutcome) {
|
||||
// Steering that arrived during the failed step stays in the inbox —
|
||||
// runLoop re-enqueues it as a queued message, so an abort-then-steer
|
||||
// starts a fresh turn instead of being silently consumed.
|
||||
closeStep()
|
||||
const { error } = stepOutcome
|
||||
if (handle.isDisposed()) {
|
||||
reason = { kind: 'disposed' }
|
||||
} else if (abort.signal.aborted) {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
|
||||
} else {
|
||||
failTurn(error)
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
// The successful step's finish reason carries forward: a `max-tokens`
|
||||
// step makes the whole turn end `max-tokens` (the ACP RFC's rule "any
|
||||
// max-tokens step surfaces as max-tokens"). `stepFinishReason` returns
|
||||
// `max-tokens` or `undefined`, so a later ordinary step never resets a
|
||||
// max-tokens turn back to completed, and a never-truncated turn keeps the
|
||||
// default `completed`. The disposal/abort/error branches above and the
|
||||
// continuation-window disposal check below override this — they win.
|
||||
const stepReason = stepFinishReason(stepOutcome.finish)
|
||||
if (stepReason) reason = stepReason
|
||||
|
||||
// Steering that arrived during streaming/tool execution.
|
||||
const steered = drainSteering(ctx, agent, turn)
|
||||
|
||||
if (closeStep()) break
|
||||
|
||||
const defaultDecision = stepOutcome.hadToolCalls || steered
|
||||
let shouldContinue: boolean
|
||||
try {
|
||||
shouldContinue = await ctx.waterfall(
|
||||
'agent/turn-continuation', agent, turn, defaultDecision,
|
||||
() => Promise.resolve(defaultDecision),
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
// A broken continuation plugin ends the turn, not the loop.
|
||||
failTurn(toError(error))
|
||||
break
|
||||
}
|
||||
|
||||
// Steering from step-end/continuation listeners (the /goal pattern)
|
||||
// demands the model see it — it overrides a negative decision; the
|
||||
// next iteration's drain records it.
|
||||
if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
if (!shouldContinue || handle.isDisposed()) {
|
||||
/* v8 ignore next -- disposal during continuation-decision window is a narrow race; error-path disposal is covered elsewhere */
|
||||
if (handle.isDisposed()) reason = { kind: 'disposed' }
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Normal / inline-error loop exit: close the turn and notify.
|
||||
closeTurn(true)
|
||||
} catch (error: unknown) {
|
||||
// Decide whether this turn was ever opened from the LOG, not a flag.
|
||||
// Session.append pushes the event BEFORE notifying session/event listeners,
|
||||
// so a throwing listener on the `turn/start` append leaves turn/start in the
|
||||
// log even though execution never reached the lines after that append.
|
||||
// Gating on a "turn started" boolean would skip turn/end and leave a
|
||||
// permanently OPEN turn that poisons the next turn/replay (the turn-enclosure RFC). We
|
||||
// check the log for THIS turn's turn/start: present means a turn/end is owed
|
||||
// (or was already appended — closeTurn/failTurn are idempotent, so running
|
||||
// them again is a safe no-op that still preserves the disposed/error reason
|
||||
// chosen below). Absent means the turn/start append threw BEFORE its push (a
|
||||
// non-serializable trigger — impossible for our fixed trigger); nothing was
|
||||
// opened, so rethrow to the runLoop backstop.
|
||||
const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
|
||||
if (!turnStartLogged) throw error
|
||||
closeStep()
|
||||
// Choose the close reason. Disposal wins only if no error was already
|
||||
// reported: a turn disposed mid-step sets reason=disposed in the step-error
|
||||
// branch (without reporting an error), and if closeTurn(true)'s turn-end
|
||||
// emit then throws, we land here and must PRESERVE disposed rather than
|
||||
// overwrite it with the listener's throw. Otherwise a boundary-emit 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 {
|
||||
failTurn(toError(error))
|
||||
}
|
||||
closeTurn(false)
|
||||
}
|
||||
|
||||
// Durability checkpoint: persistence plugins drain write-behind buffers.
|
||||
// A failing persistence plugin is reported but doesn't kill the agent.
|
||||
try {
|
||||
await ctx.parallel('session/flush', session)
|
||||
} catch (error: unknown) {
|
||||
// The turn is already closed (turn/end appended above) and flush must run
|
||||
// AFTER turn/end to be a checkpoint — so there is no in-turn position left
|
||||
// for a session `error` event. Appending one here would land it after the
|
||||
// last turn/end, where the persistence backend treats it as a crash tail
|
||||
// and drops it on resume (the turn-enclosure RFC: every event is turn-enclosed). Report
|
||||
// the failure via agent/error + the logger only; persistence keeps the
|
||||
// buffered events for the next flush/dispose, so nothing is lost.
|
||||
const err = toError(error)
|
||||
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, step, err)
|
||||
} catch {
|
||||
// contained: a throwing agent/error listener must not escape the loop.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Drain the steering queue into the session. Returns whether any arrived. */
|
||||
function drainSteering(ctx: Context, agent: ReactLoopAgent, turn: number): boolean {
|
||||
const messages = agent.inbox.drainSteering()
|
||||
for (const message of messages) {
|
||||
agent.session.append('steering/message', { turn, content: message.content, source: message.source })
|
||||
ctx.emit('agent/steering', agent, turn, message.content, message.source)
|
||||
}
|
||||
return messages.length > 0
|
||||
}
|
||||
|
||||
/** One step: assemble request → stream model → record → execute tools. */
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
signal: AbortSignal,
|
||||
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
|
||||
const { session, options } = agent
|
||||
|
||||
// --- Request assembly ---
|
||||
const assembly = await ctx.systemPrompt.assemble()
|
||||
const system = [renderPrompt(assembly), options.systemPrompt ?? '']
|
||||
.filter(text => text.length > 0)
|
||||
.join('\n\n')
|
||||
|
||||
let request: GenerateOptions = {
|
||||
model: options.model ?? '',
|
||||
messages: session.deriveMessages(),
|
||||
...system ? { system } : {},
|
||||
...assembly.tools.length > 0 ? { tools: assembly.tools } : {},
|
||||
signal,
|
||||
}
|
||||
request = await ctx.waterfall('agent/request', agent, turn, step, request, () => Promise.resolve(request))
|
||||
if (!request.model) {
|
||||
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
|
||||
}
|
||||
|
||||
// --- Model call (streaming-first; raw chunks are the replay record) ---
|
||||
const assembler = new BlockAssembler()
|
||||
for await (const chunk of ctx.llm.stream(request)) {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
session.append('assistant/chunk', { turn, step, chunk })
|
||||
ctx.emit('agent/stream-chunk', agent, turn, step, chunk)
|
||||
assembler.push(chunk)
|
||||
}
|
||||
|
||||
// Adapters report provider/transport failures one of two sanctioned ways
|
||||
// (see the StreamChunk contract in dsh-llm): throw from stream() — already
|
||||
// handled by the caller's try/catch — OR end the stream with a
|
||||
// finish-error/aborted chunk. finishError() maps the latter to the step
|
||||
// error to raise (turn ends error/aborted, not a normal completed message).
|
||||
const stepError = finishError(assembler.finish)
|
||||
if (stepError) throw stepError
|
||||
|
||||
if (assembler.finish.kind === 'max-tokens') {
|
||||
let message: Message = withoutToolCalls(assembler.message())
|
||||
message = withoutToolCalls(await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message)))
|
||||
// Fire the assistant/message when there is content OR usage: a max-tokens
|
||||
// step can be cut off with empty content but still carry token accounting,
|
||||
// and assistant/message is the only host for usage (there is no standalone
|
||||
// usage event). An empty-content assistant/message is skipped by
|
||||
// deriveMessages(), so hosting usage on it never injects a spurious assistant
|
||||
// turn into derived history.
|
||||
if (message.content.length > 0 || assembler.usage) {
|
||||
session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
|
||||
}
|
||||
return { hadToolCalls: false, finish: assembler.finish }
|
||||
}
|
||||
|
||||
// The step-result waterfall runs BEFORE the session append so the log (the
|
||||
// source of truth for derived history and replay) records the message that
|
||||
// tool dispatch actually uses.
|
||||
let message: Message = assembler.message()
|
||||
message = await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message))
|
||||
|
||||
// Same content-or-usage guard as the max-tokens branch: a step that finishes
|
||||
// with neither assembled content nor usage (e.g. a bare `stop` finish that
|
||||
// streamed nothing) records no assistant/message — an empty-content message
|
||||
// exists only to host usage, and deriveMessages() skips it either way, so
|
||||
// appending one with no usage would be a pure trace-only row.
|
||||
if (message.content.length > 0 || assembler.usage) {
|
||||
session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
|
||||
}
|
||||
|
||||
// --- Tool execution (sequential; parallel execution is a TODO) ---
|
||||
// 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')
|
||||
for (const call of toolCalls) {
|
||||
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
|
||||
let parsedArguments: unknown
|
||||
try {
|
||||
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
|
||||
} catch {
|
||||
parsedArguments = call.arguments
|
||||
}
|
||||
const result = await ctx.tools.execute({
|
||||
callId: call.id,
|
||||
name: call.name,
|
||||
arguments: parsedArguments,
|
||||
agent,
|
||||
signal,
|
||||
})
|
||||
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 tools/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,
|
||||
...result.error ? { error: result.error } : {},
|
||||
})
|
||||
// signal CAN flip during the await above (abort() inside a tool);
|
||||
// the analyzer can't see through the await boundary.
|
||||
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
|
||||
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
|
||||
}
|
||||
|
||||
function withoutToolCalls(message: Message): Message {
|
||||
return { ...message, content: message.content.filter(block => block.type !== 'tool-call') }
|
||||
}
|
||||
|
||||
/** The last turn number in a (possibly seeded) session log, or 0. */
|
||||
export function lastTurnNumber(session: Session): number {
|
||||
const lastStart = session.events.findLast(event => event.type === 'turn/start')
|
||||
return lastStart?.data.turn ?? 0
|
||||
}
|
||||
|
||||
/**
|
||||
* 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).
|
||||
*/
|
||||
export function isTurnOpen(session: Session): boolean {
|
||||
const last = session.events.findLast(e => e.type === 'turn/start' || e.type === 'turn/end')
|
||||
return last?.type === 'turn/start'
|
||||
}
|
||||
Reference in New Issue
Block a user