Merge origin/master into skill system branch
Resolve documentation split, tool presentation, and generated catalog changes from master while preserving the skill system integration.
This commit is contained in:
Yichen Jiang
@@ -46,10 +46,14 @@ Agents listed in config are auto-created at startup. `cwd` applies only to fresh
|
||||
One invocation of `runLoop()` 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):
|
||||
drain queued → 'turn/start' → session('user/message')
|
||||
'turn/start'
|
||||
each queued: waterfall agent/prompt-submit → allow (→ session('user/message'),
|
||||
inject additionalContext) | block (→ session('prompt/blocked'), drop)
|
||||
if every prompt blocked: 'turn/end'(rejected), no step ⟵ zero-step turn
|
||||
STEP loop:
|
||||
drain steering
|
||||
assembly = systemPrompt.assemble()
|
||||
@@ -59,10 +63,14 @@ forever:
|
||||
stream llm.stream(request) → session('assistant/chunk')
|
||||
message = waterfall agent/step-result
|
||||
session('assistant/message')
|
||||
each tool-call: session('tool/call') → tools.execute() → session('tool/result')
|
||||
each tool-call: session('tool/call')
|
||||
→ tools.execute() [waterfall tools/pre-execute → dispatch → tools/post-execute]
|
||||
→ session('tool/result')
|
||||
append buffered post-execute additionalContext as session('context/message')(s)
|
||||
drain steering → session('steering/message')
|
||||
cont = waterfall agent/turn-continuation
|
||||
if !cont: break
|
||||
cont = waterfall agent/turn-continuation → ContinuationDecision
|
||||
({action:'continue', reason?} records reason as next-step steering)
|
||||
if action==stop (and no pending steering): break
|
||||
session('turn/end')
|
||||
await session/flush
|
||||
re-enqueue leftover steering as queued
|
||||
@@ -76,9 +84,9 @@ Cancellation: `agent.cancel()` is the single public stop primitive — it clears
|
||||
### What is NOT here
|
||||
|
||||
Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
|
||||
- Hooks: `agent/pre-step`, `agent/request`, `agent/step-result`, `tools/execute`, `agent/turn-continuation`
|
||||
- Hooks: `agent/session-start`, `agent/prompt-submit`, `agent/pre-step`, `agent/request`, `agent/step-result`, `tools/pre-execute`, `tools/post-execute`, `agent/turn-continuation`
|
||||
- Compaction: `agent/pre-step`
|
||||
- Sandbox, permission, plan mode: `tools/execute`
|
||||
- Sandbox, permission, plan mode: `tools/pre-execute` (deny/ask gate), `tools/post-execute`
|
||||
- Sub-agents: implemented outside the loop as `ctx.subagents` providers; in-process providers use `ctx.agents.create()` and owned `AgentHandle` teardown, while child streaming/progress and background/poll collection remain deferred.
|
||||
- Persistence: `session/event` + `session/flush`
|
||||
- UI: `agent/stream-chunk` + `agent/*` events
|
||||
- UI: `session/event` (assistant token stream, boundaries, tool activity) + `agent/*` control events (`agent/status`, `agent/created`/`agent/disposed`)
|
||||
|
||||
@@ -79,7 +79,7 @@ export class ReactLoopAgent implements Agent {
|
||||
// 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
|
||||
// waiter (docs/defensive-patterns.md "contain callback exceptions" — a lifecycle await must
|
||||
// not hang on one bad listener).
|
||||
if (status !== 'running') this.settleIdleWaiters()
|
||||
try {
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
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 { 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'
|
||||
@@ -124,6 +124,10 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* 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.
|
||||
* @param id - the agent id; also seeds the generated session id.
|
||||
* @param options - loop options (model, limits, …); defaults applied per option.
|
||||
* @param meta - optional session metadata for the fresh session.
|
||||
* @returns the running agent, owned by the calling fiber (no handle).
|
||||
*/
|
||||
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
|
||||
this.assertAgentIdFree(id)
|
||||
@@ -132,7 +136,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
// 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)
|
||||
const { agent } = this.start(id, options, session, 'startup')
|
||||
return agent
|
||||
}
|
||||
|
||||
@@ -145,6 +149,9 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* `seed` (a balanced completed-turn prefix of the parent's log) so the child
|
||||
* starts with the parent's context. Returns an {@link AgentHandle} the owner
|
||||
* disposes to tear down exactly this agent.
|
||||
* @param options - agent id, caller-supplied session id, optional seed/meta,
|
||||
* and agent options.
|
||||
* @returns the handle whose dispose tears down exactly this agent.
|
||||
*/
|
||||
createAgent(options: CreateAgentOptions): AgentHandle {
|
||||
// Check the agent id BEFORE preparing the session: register() would reject a
|
||||
@@ -155,7 +162,9 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
...options.seed !== undefined ? { seed: options.seed } : {},
|
||||
meta: options.meta ?? {},
|
||||
})
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
|
||||
// A seeded (forked) create is still a fresh start, NOT a resume — `resume`
|
||||
// is reserved for reloading a PERSISTED session via resume()/resumeWith().
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'startup')
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -169,6 +178,8 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* 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.
|
||||
* @param options - the persisted session id to reload, plus agent id/options.
|
||||
* @returns the handle for the agent resumed on the reconstructed session.
|
||||
*/
|
||||
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
|
||||
// Read the service through `ctx.get('sessionPersistence')` — a direct
|
||||
@@ -227,7 +238,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
...meta.seedLength !== undefined ? { seedLength: meta.seedLength } : {},
|
||||
},
|
||||
})
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session)
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'resume')
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -264,14 +275,33 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* so a throwing `session/created`/`agent/created` listener unwinds the
|
||||
* already-yielded disposers instead of leaking.
|
||||
*
|
||||
* `source` says why the session began ({@link SessionStartSource}); it is
|
||||
* emitted as `agent/session-start` once, AFTER the agent is registered (so a
|
||||
* listener can resolve the agent via `ctx.agents.get(id)` and `inject()` into
|
||||
* it) and BEFORE the loop starts its first turn. The emit is contained: a
|
||||
* throwing session-start listener must not abort agent construction — it is
|
||||
* logged, and the agent still starts. (Unlike a turn-boundary throw, there is
|
||||
* no open turn here to balance; the durable evidence of a session-start hook
|
||||
* is whatever it `inject()`ed.)
|
||||
*
|
||||
* Returns the agent plus the composite effect's disposer (`disposeAgent`).
|
||||
*/
|
||||
private start(id: AgentId, options: AgentOptions, session: Session): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
|
||||
private start(
|
||||
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
|
||||
): { 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)
|
||||
// Fire AFTER register (a listener can ctx.agents.get(id) + inject()) and
|
||||
// BEFORE the loop's first turn. Contained: a throwing listener is logged,
|
||||
// never aborts construction (no open turn to balance here).
|
||||
try {
|
||||
this.ctx.emit('agent/session-start', agent, source)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`agent "${id}": agent/session-start listener threw: ${String(error)}`)
|
||||
}
|
||||
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,
|
||||
@@ -298,8 +328,8 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* `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)
|
||||
private startOwned(id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource): AgentHandle {
|
||||
const { agent, disposeAgent } = this.start(id, options, session, source)
|
||||
let disposing: Promise<void> | undefined
|
||||
return { agent, dispose: () => (disposing ??= disposeAgent()) }
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
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 { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
@@ -142,30 +143,37 @@ export interface LoopHandle {
|
||||
* 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):
|
||||
* drain queued → 'turn/start' → session('user/message'…) → emit agent/turn-start
|
||||
* '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() ⟵ waterfall system-prompt/assemble
|
||||
* await ctx.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
|
||||
* session('step/start'); emit agent/step-start ⟵ append before emit (the event-sourcing RFC)
|
||||
* session('step/start') ⟵ durable step boundary (no agent/* mirror)
|
||||
* req = {model, system, tools, messages: session.deriveMessages(), signal}
|
||||
* req = waterfall agent/request ⟵ hooks/model-switch
|
||||
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
|
||||
* session('assistant/chunk'); emit agent/stream-chunk
|
||||
* 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() ⟵ waterfall tools/execute
|
||||
* session('tool/call'); ctx.tools.execute() ⟵ tools/pre-execute (allow/deny/ask)
|
||||
* → dispatch → tools/post-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
|
||||
* 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
|
||||
* if action==stop: break
|
||||
* session('turn/end') ⟵ durable turn boundary (no agent/* mirror)
|
||||
* 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
|
||||
@@ -277,37 +285,32 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
|
||||
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).
|
||||
// Close the open step exactly once (idempotent via stepOpen). Step boundaries
|
||||
// are durable session events only — there is no agent/* step emit to mirror
|
||||
// them (see the agent event-domain rule). A throwing step/end session-event
|
||||
// listener must not abort finalization and strand the turn open (turn/end
|
||||
// balance > notifying one bad listener); it is contained and surfaced as a
|
||||
// turn error below.
|
||||
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.
|
||||
// error below.
|
||||
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.
|
||||
// A throwing step/end session-event 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
|
||||
@@ -324,47 +327,38 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
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}`)
|
||||
}
|
||||
// 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. A throwing `turn/end` session-event listener is already
|
||||
// contained inside closeTurn (append pushes before notifying, so the
|
||||
// boundary is durable). So set the error reason for closeTurn to append.
|
||||
reason = { kind: 'error', step, ...errorData(err) }
|
||||
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.
|
||||
// contained: the error is already captured on `reason`; 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
|
||||
// Close the turn. Called exactly once per turn — the normal loop exit and the
|
||||
// outer catch are mutually exclusive paths, and this never throws (the append
|
||||
// is contained below), so there is no re-entry to guard against (unlike
|
||||
// closeStep, which the cancel branches and the outer catch can both reach).
|
||||
// Turn boundaries are durable session events only — there is no agent/* turn
|
||||
// emit to mirror them (see the agent event-domain rule).
|
||||
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'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.)
|
||||
// but would otherwise escape — from the outer catch it would propagate to
|
||||
// the runLoop backstop. Contain it: the boundary is durable either way, and
|
||||
// finalization must not abort on a bad listener.
|
||||
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 {
|
||||
@@ -373,20 +367,60 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
// 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.
|
||||
// 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`
|
||||
// decision carries a required `reason` and overwrites it, so a fully-blocked
|
||||
// batch always reports the last vetoing reason.
|
||||
let lastBlockReason = 'prompt blocked by hook'
|
||||
for (const message of queued) {
|
||||
session.append('user/message', { content: message.content, source: message.source }, { surfaceOp: 'append' })
|
||||
const decision = await ctx.waterfall(
|
||||
'agent/prompt-submit', agent, message.content, message.source,
|
||||
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
|
||||
)
|
||||
if (decision.kind === 'block') {
|
||||
lastBlockReason = decision.reason
|
||||
// Record the veto durably: `PromptDecision.reason` is the durable record
|
||||
// of why a prompt was blocked, but a fully-blocked batch's `rejected`
|
||||
// turn/end only preserves the LAST reason, and a MIXED batch (this prompt
|
||||
// blocked, another allowed) does not end `rejected` at all — so without
|
||||
// this append a blocked prompt would vanish from the log whenever any
|
||||
// sibling prompt is allowed. `prompt/blocked` sits in the open turn in
|
||||
// place of the `user/message` this prompt would have become.
|
||||
session.append('prompt/blocked', { content: message.content, source: message.source, reason: decision.reason })
|
||||
continue
|
||||
}
|
||||
anyAllowed = true
|
||||
// `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
|
||||
const content = decision.content ?? message.content
|
||||
session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
|
||||
// `allow.additionalContext` is a SEPARATE context/message the next request
|
||||
// also sees. The turn is open, so inject() appends it into THIS turn.
|
||||
if (decision.additionalContext) {
|
||||
agent.inject(decision.additionalContext.content, { source: decision.additionalContext.source })
|
||||
}
|
||||
}
|
||||
ctx.emit('agent/turn-start', agent, turn)
|
||||
|
||||
while (true) {
|
||||
// A fully-blocked batch (every prompt vetoed by prompt-submit) opens a
|
||||
// zero-step turn that ends `rejected`: break BEFORE the first step so the
|
||||
// boundary stays balanced (turn/start → turn/end) and the block is a
|
||||
// durable in-turn fact. `anyAllowed` never changes inside the loop, so this
|
||||
// only ever fires on the first iteration.
|
||||
if (!anyAllowed) {
|
||||
reason = { kind: 'rejected', reason: lastBlockReason }
|
||||
break
|
||||
}
|
||||
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)
|
||||
// Steering from the previous round's continuation listeners joins before
|
||||
// the request.
|
||||
drainSteering(agent, 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
|
||||
@@ -432,24 +466,25 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
// pre-step plugin ends the turn, not the loop.
|
||||
await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal)
|
||||
|
||||
// Interruption landing during the pre-step seam: do not open an empty
|
||||
// step. `agent/step-start` listeners get their own check below because
|
||||
// they necessarily run after step/start is appended/emitted.
|
||||
// Interruption landing during the pre-step seam: do not open an empty step.
|
||||
if (handle.isCancelled() || handle.isDisposed()) {
|
||||
handle.setAbort(undefined)
|
||||
reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
|
||||
break
|
||||
}
|
||||
|
||||
session.append('step/start', { turn, step })
|
||||
// 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. Setting stepOpen first
|
||||
// means the outer catch's closeStep() then appends the balancing step/end
|
||||
// (turn stays enclosed) instead of stranding an open step under turn/end.
|
||||
stepOpen = true
|
||||
ctx.emit('agent/step-start', agent, turn, step)
|
||||
session.append('step/start', { turn, step })
|
||||
|
||||
// Cancel landing in the step-start window: a synchronous
|
||||
// `agent/step-start` listener can cancel after the step is already open.
|
||||
// Check AFTER step/start append + emit and before `runStep`: drop the
|
||||
// step, end the turn accordingly. closeStep balances the already-appended
|
||||
// step/start.
|
||||
// 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() }
|
||||
@@ -494,14 +529,14 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
if (stepReason) reason = stepReason
|
||||
|
||||
// Steering that arrived during streaming/tool execution.
|
||||
const steered = drainSteering(ctx, agent, turn)
|
||||
const steered = drainSteering(agent, turn)
|
||||
|
||||
if (closeStep()) break
|
||||
|
||||
const defaultDecision = stepOutcome.hadToolCalls || steered
|
||||
let shouldContinue: boolean
|
||||
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
|
||||
let decision: ContinuationDecision
|
||||
try {
|
||||
shouldContinue = await ctx.waterfall(
|
||||
decision = await ctx.waterfall(
|
||||
'agent/turn-continuation', agent, turn, defaultDecision,
|
||||
() => Promise.resolve(defaultDecision),
|
||||
)
|
||||
@@ -511,9 +546,18 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
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.
|
||||
// A forced `continue` may carry model-facing context: record it as
|
||||
// next-STEP steering (the steering channel), so the continued turn's next
|
||||
// iteration drains it before its request — the typed twin of the /goal
|
||||
// step/end-steer pattern.
|
||||
if (decision.action === 'continue' && decision.reason) {
|
||||
agent.inbox.steer({ content: decision.reason.content, source: decision.reason.source })
|
||||
}
|
||||
let shouldContinue = decision.action === 'continue'
|
||||
|
||||
// Steering from step/end session-event or continuation listeners (the
|
||||
// /goal pattern) demands the model see it — it overrides a stop decision;
|
||||
// the next iteration's drain records it.
|
||||
if (!shouldContinue && agent.inbox.hasSteering) shouldContinue = true
|
||||
|
||||
// A cancel that landed during the continuation window — after the step's
|
||||
@@ -533,8 +577,8 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
}
|
||||
}
|
||||
|
||||
// Normal / inline-error loop exit: close the turn and notify.
|
||||
closeTurn(true)
|
||||
// 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.
|
||||
// Session.append pushes the event BEFORE notifying session/event listeners,
|
||||
@@ -543,28 +587,29 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
// 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.
|
||||
// and the normal-exit `closeTurn()` did NOT run (we are here because a throw
|
||||
// preceded it — the two `closeTurn()` sites are on mutually exclusive paths),
|
||||
// so this catch appends turn/end with the disposed/error reason chosen below.
|
||||
// `closeStep()` IS idempotent (guarded by `stepOpen`) — it may have run
|
||||
// already in a step branch, so running it again is a safe no-op. Absent
|
||||
// turn/start means the 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.)
|
||||
// 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 {
|
||||
failTurn(toError(error))
|
||||
}
|
||||
closeTurn(false)
|
||||
closeTurn()
|
||||
}
|
||||
|
||||
// Durability checkpoint: persistence plugins drain write-behind buffers.
|
||||
@@ -590,11 +635,10 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
|
||||
}
|
||||
|
||||
/** Drain the steering queue into the session. Returns whether any arrived. */
|
||||
function drainSteering(ctx: Context, agent: ReactLoopAgent, turn: number): boolean {
|
||||
function drainSteering(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 }, { surfaceOp: 'append' })
|
||||
ctx.emit('agent/steering', agent, turn, message.content, message.source)
|
||||
}
|
||||
return messages.length > 0
|
||||
}
|
||||
@@ -636,7 +680,6 @@ async function runStep(
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
|
||||
chunkSeqs.push(chunkEvent.seq)
|
||||
ctx.emit('agent/stream-chunk', agent, turn, step, chunk)
|
||||
assembler.push(chunk)
|
||||
}
|
||||
|
||||
@@ -695,6 +738,12 @@ 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. 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 */
|
||||
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
|
||||
@@ -705,6 +754,12 @@ async function runStep(
|
||||
} catch {
|
||||
parsedArguments = call.arguments
|
||||
}
|
||||
// TODO(pre-tool-input-rewrite): tools/pre-execute deliberately cannot rewrite
|
||||
// `arguments` — tool/call (the audit record) and assistant/message (the
|
||||
// model-history source) are logged BEFORE execute, and live consumers (ACP,
|
||||
// tool-bash presentation) read the pre-execution args, so an execution-only
|
||||
// rewrite would desync the UI from what ran. Designing that consistently is
|
||||
// its own proposed RFC (docs/rfc/proposed/feature/…-pre-tool-input-rewrite.md).
|
||||
const result = await ctx.tools.execute({
|
||||
callId: call.id,
|
||||
name: call.name,
|
||||
@@ -716,7 +771,7 @@ async function runStep(
|
||||
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
|
||||
// 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.
|
||||
@@ -724,7 +779,12 @@ async function runStep(
|
||||
content: result.content,
|
||||
isError: result.isError,
|
||||
...result.error ? { error: result.error } : {},
|
||||
// The tool's private presentation payload (e.g. a result-time diff),
|
||||
// persisted so a UI bridge reproduces the card on replay.
|
||||
...result.meta !== undefined ? { meta: result.meta } : {},
|
||||
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
|
||||
// Buffer (don't append yet) any post-execute additionalContext for this call.
|
||||
if (result.additionalContext) pendingContext.push(result.additionalContext)
|
||||
// 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 */
|
||||
@@ -733,6 +793,13 @@ async function runStep(
|
||||
/* v8 ignore stop */
|
||||
}
|
||||
|
||||
// Append buffered post-execute context AFTER every tool/result, preserving
|
||||
// tool-call/result adjacency across the whole batch. inject() appends into the
|
||||
// open turn (a context/message at its chronological position).
|
||||
for (const context of pendingContext) {
|
||||
agent.inject(context.content, { source: context.source })
|
||||
}
|
||||
|
||||
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
|
||||
}
|
||||
|
||||
|
||||
@@ -117,7 +117,7 @@ describe('Agent.cancel()', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
@@ -134,7 +134,7 @@ describe('Agent.cancel()', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
@@ -166,22 +166,23 @@ describe('Agent.cancel()', () => {
|
||||
expect(reasons.length).toBe(2)
|
||||
})
|
||||
|
||||
it('cancel from a synchronous agent/turn-start listener drops the step (step-start window)', async () => {
|
||||
it('cancel from a synchronous turn/start session-event listener drops the step (step-start window)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not stream')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// A turn-start listener fires BEFORE any AbortController is installed for the
|
||||
// step. Cancelling there must still drop the step (the turn-scoped marker,
|
||||
// not the step AbortController, is what catches this) — no model step runs.
|
||||
// A turn/start listener fires right after turn/start is appended, BEFORE any
|
||||
// AbortController is installed for the step. Cancelling there must still drop
|
||||
// the step (the turn-scoped marker, not the step AbortController, is what
|
||||
// catches this) — no model step runs.
|
||||
let streamed = false
|
||||
ctx.on('agent/stream-chunk', () => { streamed = true })
|
||||
const dispose = ctx.on('agent/turn-start', (subject) => {
|
||||
if (subject === agent) agent.cancel('from turn-start')
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
||||
const dispose = ctx.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'turn/start') agent.cancel('from turn-start')
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -194,23 +195,23 @@ describe('Agent.cancel()', () => {
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from turn-start' }])
|
||||
})
|
||||
|
||||
it('cancel from a synchronous agent/step-start listener drops the step (post-step-start window)', async () => {
|
||||
it('cancel from a synchronous step/start session-event listener drops the step (post-step-start window)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not stream')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// A step-start listener fires AFTER step/start is appended (and after the
|
||||
// pre-step seam), so cancelling there lands in the SECOND cancel check (the
|
||||
// one that must closeStep() to balance the already-open step) — distinct
|
||||
// from a turn-start cancel, which is caught before the step opens.
|
||||
// A step/start session-event listener fires AFTER step/start is appended
|
||||
// (and after the pre-step seam), so cancelling there lands in the SECOND
|
||||
// cancel check (the one that must closeStep() to balance the already-open
|
||||
// step) — distinct from a turn-start cancel, caught before the step opens.
|
||||
let streamed = false
|
||||
ctx.on('agent/stream-chunk', () => { streamed = true })
|
||||
const dispose = ctx.on('agent/step-start', (subject) => {
|
||||
if (subject === agent) agent.cancel('from step-start')
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
||||
const dispose = ctx.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'step/start') agent.cancel('from step-start')
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -224,7 +225,7 @@ describe('Agent.cancel()', () => {
|
||||
expect(types.filter(t => t === 'step/start').length).toBe(types.filter(t => t === 'step/end').length)
|
||||
})
|
||||
|
||||
it('disposal from a synchronous agent/step-start listener closes the open step as disposed', async () => {
|
||||
it('disposal from a synchronous step/start session-event listener closes the open step as disposed', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not stream')])
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
@@ -244,9 +245,9 @@ describe('Agent.cancel()', () => {
|
||||
|
||||
let disposalDone: Promise<void> | undefined
|
||||
let streamed = false
|
||||
ctx.on('agent/stream-chunk', () => { streamed = true })
|
||||
ctx.on('agent/step-start', (subject) => {
|
||||
if (subject === agent) disposalDone = handle.dispose()
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
||||
ctx.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'step/start') disposalDone = handle.dispose()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
@@ -271,16 +272,18 @@ describe('Agent.cancel()', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steps = 0
|
||||
ctx.on('agent/step-start', () => { steps += 1 })
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_session, event) => {
|
||||
if (event.type === 'step/start') steps += 1
|
||||
if (event.type === 'turn/end') reasons.push(event.data.reason)
|
||||
})
|
||||
|
||||
let continued = false
|
||||
ctx.on('agent/turn-continuation', async (subject, _turn, _default, next) => {
|
||||
if (subject === agent && !continued) {
|
||||
continued = true
|
||||
agent.cancel('from continuation')
|
||||
return true // vote to continue — the post-waterfall marker check must override
|
||||
return { action: 'continue' as const } // vote to continue — the post-waterfall marker check must override
|
||||
}
|
||||
return next()
|
||||
})
|
||||
@@ -305,7 +308,7 @@ describe('Agent.cancel()', () => {
|
||||
// runTurn. The second check (after the running flip) must drop the turn —
|
||||
// runTurn would otherwise throw on the now-empty queue.
|
||||
let streamed = false
|
||||
ctx.on('agent/stream-chunk', () => { streamed = true })
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'running') agent.cancel('from running listener')
|
||||
})
|
||||
|
||||
@@ -36,69 +36,6 @@ function send(agent: ReactLoopAgent, text: string) {
|
||||
}
|
||||
|
||||
describe('turn boundary listener throws (handled in-turn, loop survives)', () => {
|
||||
it('a throwing agent/turn-start listener surfaces via agent/error and the loop survives', async () => {
|
||||
// The agent/turn-start emit happens AFTER turn/start is appended to the log,
|
||||
// so a throwing listener is handled inside runTurn (the turn is balanced and
|
||||
// closed via failTurn → agent/error), NOT rethrown to the runLoop backstop.
|
||||
// The second turn should proceed normally and consume the first script entry.
|
||||
const adapter = new MockAdapter([textResponse('turn 2')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/turn-start', () => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw new Error('broken turn-start listener')
|
||||
}
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(errors.map(e => e.message)).toEqual(['broken turn-start listener'])
|
||||
// The turn is balanced: its turn/start was logged, so a turn/end was owed
|
||||
// and appended (decided from the log, not a flag).
|
||||
expect(agent.session.events.at(-1)?.type).toBe('turn/end')
|
||||
|
||||
// loop survives: second turn works fine and makes the model call
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(adapter.requests[0]!.messages.some(m => m.content.some(b => 'text' in b && b.text === 'second'))).toBe(true)
|
||||
})
|
||||
|
||||
it('a throwing agent/turn-end listener surfaces via agent/error and the loop survives', async () => {
|
||||
const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/turn-end', () => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw new Error('broken turn-end listener')
|
||||
}
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
// The turn-end throw happens after the model call is complete, so turn 1's
|
||||
// request is consumed. turn/end is already in the log (append pushes before
|
||||
// notifying), so the turn is balanced; the error is surfaced via agent/error.
|
||||
expect(errors.map(e => e.message)).toEqual(['broken turn-end listener'])
|
||||
|
||||
// loop survives: second turn works fine
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('a pre-push turn/start failure (non-serializable source) is rethrown to the runLoop backstop', async () => {
|
||||
// A non-serializable message source makes the turn/start append throw BEFORE
|
||||
// the event is pushed (Session.append validates before push), so turn/start
|
||||
@@ -192,14 +129,14 @@ describe('tool JSON parse', () => {
|
||||
})
|
||||
|
||||
describe('toError normalization', () => {
|
||||
it('normalizes non-Error throws from turn-start listeners via toError', async () => {
|
||||
it('normalizes non-Error throws from a turn/start session-event listener via toError', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/turn-start', () => {
|
||||
if (!threwOnce) {
|
||||
ctx.on('session/event', (_session, event) => {
|
||||
if (event.type === 'turn/start' && !threwOnce) {
|
||||
threwOnce = true
|
||||
throw 'naked string error' // non-Error throw, normalized via toError
|
||||
}
|
||||
@@ -287,7 +224,7 @@ describe('disposed vs aborted branching', () => {
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
|
||||
529
packages/core/agent-loop/tests/interception.spec.ts
Normal file
529
packages/core/agent-loop/tests/interception.spec.ts
Normal file
@@ -0,0 +1,529 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineTool, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, {
|
||||
AgentId,
|
||||
type ContinuationDecision,
|
||||
type PromptDecision,
|
||||
type SessionStartSource,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
|
||||
/**
|
||||
* The interception seams introduced by the hooks taxonomy: `agent/prompt-submit`,
|
||||
* `agent/session-start`, the reshaped `agent/turn-continuation`
|
||||
* ({@link ContinuationDecision}), and the `tools/pre-execute` / `tools/post-execute`
|
||||
* split with `additionalContext` buffering. These verify the canonical event
|
||||
* surface a hook bridge (or a native plugin) programs against, WITHOUT any
|
||||
* external protocol — a native plugin uses the typed decisions directly.
|
||||
*/
|
||||
|
||||
async function harness(adapter: MockAdapter) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return ctx
|
||||
}
|
||||
|
||||
function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'idle') {
|
||||
dispose()
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function send(agent: ReactLoopAgent, text: string) {
|
||||
agent.send([{ type: 'text', text }])
|
||||
}
|
||||
|
||||
function events(agent: ReactLoopAgent): SessionEvent[] {
|
||||
return [...agent.session.events]
|
||||
}
|
||||
|
||||
describe('agent/prompt-submit', () => {
|
||||
it('allow (default via next) records the user/message unchanged', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const seen: string[] = []
|
||||
ctx.on('agent/prompt-submit', async (_agent, content, _source, next) => {
|
||||
seen.push(content.map(b => (b.type === 'text' ? b.text : '')).join(''))
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'hello')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(seen).toEqual(['hello'])
|
||||
const userMsg = events(agent).find(e => e.type === 'user/message')
|
||||
expect(userMsg?.type === 'user/message' && userMsg.data.content).toEqual([{ type: 'text', text: 'hello' }])
|
||||
})
|
||||
|
||||
it('allow with content REWRITES the prompt before it is recorded', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
|
||||
({ kind: 'allow', content: [{ type: 'text', text: 'REWRITTEN' }] }))
|
||||
|
||||
send(agent, 'original')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const userMsg = events(agent).find(e => e.type === 'user/message')
|
||||
expect(userMsg?.type === 'user/message' && userMsg.data.content).toEqual([{ type: 'text', text: 'REWRITTEN' }])
|
||||
// the rewritten prompt is what reached the model
|
||||
expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('REWRITTEN')
|
||||
expect(JSON.stringify(adapter.requests[0]!.messages)).not.toContain('original')
|
||||
})
|
||||
|
||||
it('allow with additionalContext injects a separate context/message into the turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
|
||||
({
|
||||
kind: 'allow',
|
||||
additionalContext: { content: [{ type: 'text', text: 'extra ctx' }], source: { kind: 'plugin', plugin: 'test' } },
|
||||
}))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const log = events(agent)
|
||||
const userMsg = log.find(e => e.type === 'user/message')
|
||||
const ctxMsg = log.find(e => e.type === 'context/message')
|
||||
expect(userMsg).toBeDefined()
|
||||
expect(ctxMsg?.type === 'context/message' && ctxMsg.data.content).toEqual([{ type: 'text', text: 'extra ctx' }])
|
||||
expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
|
||||
// both the prompt and the injected context reach the model
|
||||
const sent = JSON.stringify(adapter.requests[0]!.messages)
|
||||
expect(sent).toContain('extra ctx')
|
||||
})
|
||||
|
||||
it('a prompt-submit rewrite + additionalContext is VISIBLE to the agent/pre-step seam (merged ordering)', async () => {
|
||||
// The merge of the interception seams with master's compaction seam pins one
|
||||
// ordering: `agent/prompt-submit` runs (rewriting the prompt and injecting
|
||||
// context) BEFORE the step loop, and `agent/pre-step` fires INSIDE the step
|
||||
// before the single deriveMessages(). So a compaction listener on
|
||||
// `agent/pre-step` must observe the surface AFTER the prompt rewrite/inject —
|
||||
// otherwise it would measure/compact stale history. This cross-test proves
|
||||
// the two seams compose in the right order (each is covered in isolation
|
||||
// elsewhere; this asserts they see each other's effects on the same turn).
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
|
||||
({
|
||||
kind: 'allow',
|
||||
content: [{ type: 'text', text: 'REWRITTEN prompt' }],
|
||||
additionalContext: { content: [{ type: 'text', text: 'injected ctx' }], source: { kind: 'plugin', plugin: 'test' } },
|
||||
}))
|
||||
|
||||
// The pre-step seam (where compaction lives) derives the surface it would act
|
||||
// on. Capture what it sees on the first step.
|
||||
let preStepDerived: string | undefined
|
||||
ctx.on('agent/pre-step', (subject, _turn, step) => {
|
||||
if (subject === agent && step === 1) preStepDerived = JSON.stringify(subject.session.deriveMessages())
|
||||
})
|
||||
|
||||
send(agent, 'ORIGINAL prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The pre-step seam ran and saw BOTH the rewrite (not the original) and the
|
||||
// injected context — i.e. the prompt-submit effects landed before it.
|
||||
expect(preStepDerived).toBeDefined()
|
||||
expect(preStepDerived).toContain('REWRITTEN prompt')
|
||||
expect(preStepDerived).toContain('injected ctx')
|
||||
expect(preStepDerived).not.toContain('ORIGINAL prompt')
|
||||
})
|
||||
|
||||
it('block drops the (only) prompt → zero-step turn ends rejected, model never called', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not run')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
|
||||
({ kind: 'block', reason: 'blocked by policy' }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'do something')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// the model was never called
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
// the turn opened and closed balanced, with no user/message and no step
|
||||
const log = events(agent)
|
||||
expect(log.some(e => e.type === 'turn/start')).toBe(true)
|
||||
expect(log.some(e => e.type === 'turn/end')).toBe(true)
|
||||
expect(log.some(e => e.type === 'user/message')).toBe(false)
|
||||
expect(log.some(e => e.type === 'step/start')).toBe(false)
|
||||
// the veto is recorded durably as a prompt/blocked in the open turn
|
||||
const blocked = log.find(e => e.type === 'prompt/blocked')
|
||||
expect(blocked?.type === 'prompt/blocked' && blocked.data).toMatchObject({
|
||||
content: [{ type: 'text', text: 'do something' }],
|
||||
reason: 'blocked by policy',
|
||||
})
|
||||
// ended rejected with the block reason
|
||||
expect(reasons).toEqual([{ kind: 'rejected', reason: 'blocked by policy' }])
|
||||
const turnEnd = log.findLast(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'rejected', reason: 'blocked by policy' })
|
||||
})
|
||||
|
||||
it('a mixed batch records a prompt/blocked for the vetoed prompt while the allowed one runs', async () => {
|
||||
// Two prompts queued into ONE turn: block "secret", allow "safe". The turn is
|
||||
// NOT rejected (a prompt was allowed), so without a durable prompt/blocked the
|
||||
// vetoed prompt and its reason would vanish from the log entirely.
|
||||
const adapter = new MockAdapter([textResponse('ran once')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/prompt-submit', async (_agent, content, _source, next): Promise<PromptDecision> => {
|
||||
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
|
||||
return text === 'secret' ? { kind: 'block', reason: 'policy: no secrets' } : next()
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
// both sends land before the loop drains → one batched turn
|
||||
send(agent, 'secret')
|
||||
send(agent, 'safe')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const log = events(agent)
|
||||
// the allowed prompt became a user/message and drove exactly one model call
|
||||
const userMsgs = log.filter(e => e.type === 'user/message')
|
||||
expect(userMsgs).toHaveLength(1)
|
||||
expect(userMsgs[0]?.type === 'user/message' && userMsgs[0].data.content).toEqual([{ type: 'text', text: 'safe' }])
|
||||
expect(adapter.requests.length).toBeGreaterThanOrEqual(1)
|
||||
// the blocked prompt is durably recorded, with its content + reason
|
||||
const blocked = log.filter(e => e.type === 'prompt/blocked')
|
||||
expect(blocked).toHaveLength(1)
|
||||
expect(blocked[0]?.type === 'prompt/blocked' && blocked[0].data).toMatchObject({
|
||||
content: [{ type: 'text', text: 'secret' }],
|
||||
reason: 'policy: no secrets',
|
||||
})
|
||||
// the turn did NOT reject — a sibling was allowed — so the boundary reason
|
||||
// alone would not have preserved the block
|
||||
expect(reasons.some(r => r.kind === 'rejected')).toBe(false)
|
||||
})
|
||||
|
||||
it('a throwing prompt-submit listener ends the turn balanced (error), loop survives', async () => {
|
||||
const adapter = new MockAdapter([textResponse('after')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threw = false
|
||||
ctx.on('agent/prompt-submit', async () => {
|
||||
if (!threw) { threw = true; throw new Error('prompt hook broke') }
|
||||
return { kind: 'allow' as const }
|
||||
})
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(errors.map(e => e.message)).toEqual(['prompt hook broke'])
|
||||
// turn balanced
|
||||
const log = events(agent)
|
||||
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
|
||||
expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1)
|
||||
|
||||
// loop survives: a second prompt runs normally
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests.length).toBeGreaterThanOrEqual(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('agent/session-start', () => {
|
||||
it('fires once with source "startup" for a fresh create, before the first turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
const sources: SessionStartSource[] = []
|
||||
ctx.on('agent/session-start', (_agent, source) => void sources.push(source))
|
||||
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
// fires synchronously at create, before any turn
|
||||
expect(sources).toEqual(['startup'])
|
||||
expect(events(agent).some(e => e.type === 'turn/start')).toBe(false)
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
// still only one session-start
|
||||
expect(sources).toEqual(['startup'])
|
||||
})
|
||||
|
||||
it('a session-start listener can inject context the first request sees', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
ctx.on('agent/session-start', (agent) => {
|
||||
agent.inject([{ type: 'text', text: 'session preamble' }], { source: { kind: 'plugin', plugin: 'test' } })
|
||||
})
|
||||
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// the injected context reached the model on the first (only) request
|
||||
expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('session preamble')
|
||||
// and is recorded with the plugin source, never mislabeled as a user prompt
|
||||
const ctxMsg = events(agent).find(e => e.type === 'context/message')
|
||||
expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
|
||||
})
|
||||
|
||||
it('a throwing session-start listener does not abort agent construction', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
ctx.on('agent/session-start', () => { throw new Error('session-start hook broke') })
|
||||
|
||||
// create must not throw — the listener error is contained/logged
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
expect(agent.id).toBe(AgentId('a1'))
|
||||
|
||||
// and the agent still runs
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('agent/turn-continuation (ContinuationDecision)', () => {
|
||||
it('a continue decision with a reason records next-step steering in the same turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('step 1 no tools'), textResponse('step 2')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let forced = false
|
||||
ctx.on('agent/turn-continuation', async (_agent, _turn, _default, next): Promise<ContinuationDecision> => {
|
||||
if (!forced) {
|
||||
forced = true
|
||||
return { action: 'continue', reason: { content: [{ type: 'text', text: 'keep going on the goal' }], source: { kind: 'plugin', plugin: 'goal' } } }
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const log = events(agent)
|
||||
// same turn, two steps
|
||||
expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
|
||||
expect(log.filter(e => e.type === 'step/start')).toHaveLength(2)
|
||||
// the reason was recorded as steering BEFORE step 2, with its plugin source
|
||||
const steering = log.find(e => e.type === 'steering/message')
|
||||
expect(steering?.type === 'steering/message' && steering.data.content).toEqual([{ type: 'text', text: 'keep going on the goal' }])
|
||||
expect(steering?.type === 'steering/message' && steering.data.source).toEqual({ kind: 'plugin', plugin: 'goal' })
|
||||
// and reached the next request
|
||||
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('keep going on the goal')
|
||||
})
|
||||
|
||||
it('a stop decision ends the turn even when the step had tool calls', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'echo', { text: 'hi' })])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo', description: 'echo', parameters: { text: { type: 'string' } },
|
||||
async execute(args) { return [{ type: 'text', text: String(args.text) }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/turn-continuation', async (): Promise<ContinuationDecision> => ({ action: 'stop' }))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// default would have continued (had tool calls), but the stop decision wins
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(events(agent).some(e => e.type === 'tool/result')).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('tools/post-execute additionalContext buffering across a multi-call step', () => {
|
||||
it('appends each call\'s additionalContext only AFTER all tool/results, preserving adjacency', async () => {
|
||||
// One assistant step with TWO tool calls; the second model response stops.
|
||||
const twoCalls = [
|
||||
{ type: 'block-start' as const, index: 0, blockType: 'tool-call' as const },
|
||||
{ type: 'block-end' as const, index: 0, block: { type: 'tool-call' as const, id: CallId('c1'), name: 'echo', arguments: '{"text":"a"}' } },
|
||||
{ type: 'block-start' as const, index: 1, blockType: 'tool-call' as const },
|
||||
{ type: 'block-end' as const, index: 1, block: { type: 'tool-call' as const, id: CallId('c2'), name: 'echo', arguments: '{"text":"b"}' } },
|
||||
{ type: 'usage' as const, usage: { inputTokens: 5, outputTokens: 5 } },
|
||||
{ type: 'finish' as const, reason: { kind: 'tool-calls' as const } },
|
||||
]
|
||||
const adapter = new MockAdapter([twoCalls, textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo', description: 'echo', parameters: { text: { type: 'string' } },
|
||||
async execute(args) { return [{ type: 'text', text: String(args.text) }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// Each call attaches additionalContext naming itself.
|
||||
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
|
||||
({ kind: 'accept', additionalContext: { content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } } }))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Event order in the log: both tool/results, THEN both context/messages —
|
||||
// never interleaved (which would break tool-call/result adjacency).
|
||||
const types = events(agent).map(e => e.type)
|
||||
const firstResult = types.indexOf('tool/result')
|
||||
const lastResult = types.lastIndexOf('tool/result')
|
||||
const firstCtx = types.indexOf('context/message')
|
||||
expect(firstResult).toBeGreaterThanOrEqual(0)
|
||||
expect(lastResult).toBeGreaterThan(firstResult) // two results
|
||||
expect(firstCtx).toBeGreaterThan(lastResult) // context only after ALL results
|
||||
// both contexts present
|
||||
const ctxTexts = events(agent)
|
||||
.filter(e => e.type === 'context/message')
|
||||
.flatMap(e => (e.type === 'context/message' ? e.data.content : []))
|
||||
.map(b => (b.type === 'text' ? b.text : ''))
|
||||
expect(ctxTexts).toEqual(['ctx-c1', 'ctx-c2'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('tools/pre-execute gate (native-plugin permission pattern, end-to-end through the loop)', () => {
|
||||
it('deny short-circuits dispatch into an isError result the model sees', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'danger', {}), textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
let ran = false
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'danger', description: 'danger', parameters: {},
|
||||
async execute() { ran = true; return [{ type: 'text', text: 'should not run' }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
|
||||
if (exec.name === 'danger') return { kind: 'deny', reason: 'blocked dangerous tool' }
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(ran).toBe(false)
|
||||
const result = events(agent).find(e => e.type === 'tool/result')
|
||||
expect(result?.type === 'tool/result' && result.data.isError).toBe(true)
|
||||
expect(result?.type === 'tool/result'
|
||||
&& result.data.content.some(b => b.type === 'text' && b.text.includes('blocked dangerous tool'))).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('worked example: a native hook plugin is just a cordis plugin on the seams', () => {
|
||||
// The whole point of the interception taxonomy: a "native hook" needs no
|
||||
// dsh-hook-protocol, no external command, no hook/* log — it is an ordinary
|
||||
// cordis plugin subscribing to the canonical events and returning typed
|
||||
// decisions. This proves all four seams compose end-to-end through the REAL
|
||||
// loop, with NO hook/* SessionEvents involved (those belong to the bridge lib).
|
||||
const NativeGuard = {
|
||||
name: 'native-guard',
|
||||
apply(ctx: Context) {
|
||||
// 1. SessionStart: seed a standing instruction.
|
||||
ctx.on('agent/session-start', (agent, source) => {
|
||||
agent.inject(
|
||||
[{ type: 'text', text: `policy active (started: ${source})` }],
|
||||
{ source: { kind: 'plugin', plugin: 'native-guard' } },
|
||||
)
|
||||
})
|
||||
// 2. PromptSubmit: block a forbidden prompt, annotate the rest.
|
||||
ctx.on('agent/prompt-submit', async (_agent, content, _source, next): Promise<PromptDecision> => {
|
||||
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
|
||||
if (text.includes('rm -rf')) return { kind: 'block', reason: 'destructive prompt blocked' }
|
||||
return next()
|
||||
})
|
||||
// 3. PreToolUse: deny a dangerous tool by name.
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
|
||||
if (exec.name === 'danger') return { kind: 'deny', reason: 'danger tool denied' }
|
||||
return next()
|
||||
})
|
||||
// 4. PostToolUse: attach context after a tool runs.
|
||||
ctx.on('tools/post-execute', async (_exec, _result, next): Promise<PostToolDecision> => {
|
||||
const decision = await next()
|
||||
if (decision.kind === 'accept') {
|
||||
return { kind: 'accept', additionalContext: { content: [{ type: 'text', text: 'audited' }], source: { kind: 'plugin', plugin: 'native-guard' } } }
|
||||
}
|
||||
return decision
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
it('all four seams fire for a real allowed turn with a tool call', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'echo', { text: 'hi' }), textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
await ctx.plugin(NativeGuard)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo', description: 'echo', parameters: { text: { type: 'string' } },
|
||||
async execute(args) { return [{ type: 'text', text: String(args.text) }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'please echo hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const log = events(agent)
|
||||
// session-start preamble injected
|
||||
expect(log.some(e => e.type === 'context/message'
|
||||
&& e.data.content.some(b => b.type === 'text' && b.text.includes('policy active (started: startup)')))).toBe(true)
|
||||
// prompt allowed → user/message recorded
|
||||
expect(log.some(e => e.type === 'user/message')).toBe(true)
|
||||
// tool ran (echo allowed) and post-execute attached "audited" context
|
||||
expect(log.some(e => e.type === 'tool/result' && !e.data.isError)).toBe(true)
|
||||
expect(log.some(e => e.type === 'context/message'
|
||||
&& e.data.content.some(b => b.type === 'text' && b.text === 'audited'))).toBe(true)
|
||||
// NO hook/* events — a native plugin needs none
|
||||
expect(log.some(e => e.type.startsWith('hook/'))).toBe(false)
|
||||
})
|
||||
|
||||
it('the same plugin blocks a destructive prompt → rejected turn, model never called', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not run')])
|
||||
const ctx = await harness(adapter)
|
||||
await ctx.plugin(NativeGuard)
|
||||
const agent = ctx.agentLoop.create(AgentId('a2'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'run rm -rf /')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(reasons).toEqual([{ kind: 'rejected', reason: 'destructive prompt blocked' }])
|
||||
})
|
||||
|
||||
it('HMR-safety: disposing the plugin fiber removes all four listeners', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const fiber = await ctx.plugin(NativeGuard)
|
||||
await fiber.dispose()
|
||||
|
||||
// After disposal, a destructive prompt is NOT blocked (the listener is gone).
|
||||
const agent = ctx.agentLoop.create(AgentId('a3'), { model: 'mock' })
|
||||
send(agent, 'run rm -rf /')
|
||||
await waitForIdle(ctx, agent)
|
||||
// the prompt ran (not rejected) — proving the prompt-submit listener was disposed
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(events(agent).some(e => e.type === 'user/message')).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -46,15 +46,20 @@ describe('agent loop', () => {
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// All boundaries — turn and step — are durable session events on the
|
||||
// session/event feed (no agent/* mirror). Record them in fire order to
|
||||
// assert the full boundary nesting.
|
||||
const order: string[] = []
|
||||
for (const name of ['agent/turn-start', 'agent/step-start', 'agent/step-end', 'agent/turn-end'] as const) {
|
||||
ctx.on(name, () => void order.push(name))
|
||||
}
|
||||
ctx.on('session/event', (_session, event) => {
|
||||
if (event.type === 'turn/start' || event.type === 'step/start' || event.type === 'step/end' || event.type === 'turn/end') {
|
||||
order.push(event.type)
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(order).toEqual(['agent/turn-start', 'agent/step-start', 'agent/step-end', 'agent/turn-end'])
|
||||
expect(order).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end'])
|
||||
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
// turn/start opens the turn, THEN the queued user message is recorded inside
|
||||
@@ -110,6 +115,32 @@ describe('agent loop', () => {
|
||||
expect(types).toContain('tool/result')
|
||||
})
|
||||
|
||||
it('threads a tool-attached meta (execute object return) onto the tool/result event', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'writer', { path: 'a.txt' }, 'writing'),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
// A tool that returns the { content, meta } object form: the loop must
|
||||
// persist `meta` on the tool/result event so a UI reproduces the card on replay.
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'writer',
|
||||
description: 'writes a file',
|
||||
parameters: { path: { type: 'string' } },
|
||||
async execute() {
|
||||
return { content: [{ type: 'text', text: 'ok' }], meta: { diffs: [{ path: 'a.txt', oldText: null, newText: 'x' }] } }
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'use the tool')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const toolResult = agent.session.events.find(e => e.type === 'tool/result')
|
||||
expect(toolResult?.type === 'tool/result' && toolResult.data.meta)
|
||||
.toEqual({ diffs: [{ path: 'a.txt', oldText: null, newText: 'x' }] })
|
||||
})
|
||||
|
||||
it('passes assembled system prompt and tool schemas into the request', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
@@ -132,21 +163,17 @@ describe('agent loop', () => {
|
||||
expect(request!.tools?.map(t => t.name)).toEqual(['noop'])
|
||||
})
|
||||
|
||||
it('records raw chunks for replay and emits agent/stream-chunk', async () => {
|
||||
it('records raw chunks for replay as assistant/chunk session events', async () => {
|
||||
const adapter = new MockAdapter([textResponse('abc')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const streamed: StreamChunk[] = []
|
||||
ctx.on('agent/stream-chunk', (_agent, _turn, _step, chunk) => void streamed.push(chunk))
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const chunkEvents = agent.session.events.filter(e => e.type === 'assistant/chunk')
|
||||
// textResponse('abc') = block-start + 3 deltas + block-end + usage + finish = 7
|
||||
expect(chunkEvents).toHaveLength(7)
|
||||
expect(streamed).toHaveLength(7)
|
||||
// replay: chunk events alone re-assemble to the recorded assistant message
|
||||
const deltaText = chunkEvents
|
||||
.flatMap(e => e.type === 'assistant/chunk' ? [e.data.chunk] : [])
|
||||
@@ -269,9 +296,9 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steps = 0
|
||||
ctx.on('agent/step-end', () => void steps++)
|
||||
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
|
||||
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, next) => {
|
||||
if (steps < 3) return true
|
||||
if (steps < 3) return { action: 'continue' as const }
|
||||
return next()
|
||||
})
|
||||
|
||||
@@ -294,7 +321,7 @@ describe('agent loop', () => {
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/turn-continuation', async () => false as const)
|
||||
ctx.on('agent/turn-continuation', async () => ({ action: 'stop' }) as const)
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -430,7 +457,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
// wait until the stream is hanging, then cancel
|
||||
@@ -450,7 +477,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -475,16 +502,16 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steps = 0
|
||||
ctx.on('agent/step-end', () => void steps++)
|
||||
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
|
||||
// Force exactly one continuation (step 1 → step 2), then defer to default
|
||||
// (step 2 is a plain stop with no tool calls → stops).
|
||||
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, next) => {
|
||||
if (steps < 2) return true
|
||||
if (steps < 2) return { action: 'continue' as const }
|
||||
return next()
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -506,7 +533,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -539,7 +566,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -581,7 +608,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -600,7 +627,7 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -640,7 +667,7 @@ describe('agent loop', () => {
|
||||
])
|
||||
})
|
||||
|
||||
it('stops the turn when agent/step-end listener failure has recorded an error', async () => {
|
||||
it('stops the turn when a step/end session-event listener failure has recorded an error', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'x' }),
|
||||
textResponse('should not run'),
|
||||
@@ -656,8 +683,11 @@ describe('agent loop', () => {
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
let threw = false
|
||||
ctx.on('agent/step-end', () => {
|
||||
if (!threw) { threw = true; throw new Error('bad step-end listener') }
|
||||
// A throwing step/end session-event listener is the surviving boundary-listener
|
||||
// failure path (step boundaries have no agent/* mirror): closeStep contains it
|
||||
// and surfaces it as a turn error rather than stranding the turn open.
|
||||
ctx.on('session/event', (_session, event) => {
|
||||
if (event.type === 'step/end' && !threw) { threw = true; throw new Error('bad step/end listener') }
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
@@ -674,13 +704,13 @@ describe('agent loop', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const turns: number[] = []
|
||||
ctx.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
|
||||
|
||||
// queue two messages while idle — first starts turn 1 immediately;
|
||||
// queue the second during turn 1 via a stream-chunk hook
|
||||
// queue the second during turn 1 when the first assistant chunk streams
|
||||
let queued = false
|
||||
ctx.on('agent/stream-chunk', () => {
|
||||
if (!queued) {
|
||||
ctx.on('session/event', (_s, event) => {
|
||||
if (event.type === 'assistant/chunk' && !queued) {
|
||||
queued = true
|
||||
send(agent, 'second message')
|
||||
}
|
||||
@@ -721,7 +751,7 @@ describe('agent loop', () => {
|
||||
const errors: Error[] = []
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
@@ -94,6 +94,36 @@ describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('agent/session-start fires "startup" for createAgent and "resume" for resume()', async () => {
|
||||
// Lifecycle 1: a fresh createAgent emits session-start with source 'startup'.
|
||||
const adapter1 = new MockAdapter([textResponse('a')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const sources1: string[] = []
|
||||
ctx1.on('agent/session-start', (_agent, source) => void sources1.push(source))
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('s'), sessionId: SessionId('start-sess') }).agent as ReactLoopAgent
|
||||
expect(sources1).toEqual(['startup'])
|
||||
a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Lifecycle 2: resuming the persisted session emits session-start 'resume'.
|
||||
const adapter2 = new MockAdapter([textResponse('b')])
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(LlmService)
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SystemPrompt)
|
||||
await ctx2.plugin(ToolRegistry)
|
||||
await ctx2.plugin(AgentRegistry)
|
||||
await ctx2.plugin(AgentLoop, { agents: [] })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx2.llm.registerAdapter(['mock'], adapter2)
|
||||
const sources2: string[] = []
|
||||
ctx2.on('agent/session-start', (_agent, source) => void sources2.push(source))
|
||||
await ctx2.agents.resume({ agentId: AgentId('s'), resumeSessionId: SessionId('start-sess') })
|
||||
expect(sources2).toEqual(['resume'])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('resume of a forked session preserves the parentSession lineage and seed boundary in the header', async () => {
|
||||
// Lifecycle 1: persist a FORKED session (carries parentSession + seedLength
|
||||
// in its header) by creating it with a complete-turn seed — the write path
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import * as Invariants from '@deepseek-ai/dsh-invariants'
|
||||
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
@@ -132,7 +132,7 @@ describe('HIGH: abort during tool execution ends the turn', () => {
|
||||
}))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -144,36 +144,6 @@ describe('HIGH: abort during tool execution ends the turn', () => {
|
||||
})
|
||||
|
||||
describe('HIGH: steering from late extension points is never stranded', () => {
|
||||
it('steer() from an agent/step-end listener reaches the next request (/goal pattern)', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'x' }),
|
||||
textResponse('after steering'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: '',
|
||||
parameters: { text: { type: 'string' } },
|
||||
async execute(args) {
|
||||
return [{ type: 'text', text: String(args.text) }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steeredOnce = false
|
||||
ctx.on('agent/step-end', () => {
|
||||
if (steeredOnce) return
|
||||
steeredOnce = true
|
||||
agent.steer([{ type: 'text', text: 'goal reminder from step-end' }])
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('goal reminder from step-end')
|
||||
})
|
||||
|
||||
it('steer() from an agent/turn-continuation listener overrides a stop decision', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
textResponse('no tools, would stop here'),
|
||||
@@ -199,20 +169,69 @@ describe('HIGH: steering from late extension points is never stranded', () => {
|
||||
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('one more thing')
|
||||
})
|
||||
|
||||
it('steer() from an agent/turn-end listener becomes a queued message for the next turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
|
||||
it('steer() from a step/end session-event listener forces a SAME-TURN next step (/goal pattern)', async () => {
|
||||
// The /goal pattern steers from a step boundary so the model addresses a
|
||||
// standing goal before stopping. Step boundaries have no agent/* mirror, so
|
||||
// the surviving hook point is the durable step/end session event. With a
|
||||
// no-tools first step the default continuation is stop; the steering queued
|
||||
// here must force the `!shouldContinue && hasSteering` override so the SAME
|
||||
// turn runs another step.
|
||||
//
|
||||
// The override is what this test guards, so it asserts the same-turn shape —
|
||||
// NOT merely that the content reaches requests[1]. Without the override the
|
||||
// turn would stop, and leftover steering is re-enqueued as a next-turn queued
|
||||
// message, which ALSO lands in requests[1] (just one turn later). So a
|
||||
// content-only assertion passes with the override disabled and guards
|
||||
// nothing. The discriminator is the turn/step shape: override ⇒ ONE turn with
|
||||
// TWO steps and the steering recorded as a `steering/message` BEFORE step 2;
|
||||
// re-enqueue fallback ⇒ TWO turns.
|
||||
const adapter = new MockAdapter([
|
||||
textResponse('no tools, would stop'),
|
||||
textResponse('after goal reminder'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steeredOnce = false
|
||||
ctx.on('agent/turn-end', () => {
|
||||
if (steeredOnce) return
|
||||
ctx.on('session/event', (subject, event) => {
|
||||
if (subject !== agent.session || event.type !== 'step/end' || steeredOnce) return
|
||||
steeredOnce = true
|
||||
agent.steer([{ type: 'text', text: 'too late for this turn' }])
|
||||
agent.steer([{ type: 'text', text: 'goal reminder from step/end' }])
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Same-turn continuation: the steering forced step 2 within turn 1.
|
||||
const events = [...agent.session.events]
|
||||
expect(events.filter(e => e.type === 'turn/start')).toHaveLength(1)
|
||||
expect(events.filter(e => e.type === 'step/start')).toHaveLength(2)
|
||||
// The steered content is recorded as steering (same turn), BEFORE step 2 —
|
||||
// not as a fresh turn's user/message. This is the mechanism the override uses.
|
||||
const steeringIdx = events.findIndex(e => e.type === 'steering/message')
|
||||
const step2Idx = events.map(e => e.type).lastIndexOf('step/start')
|
||||
expect(steeringIdx).toBeGreaterThanOrEqual(0)
|
||||
expect(steeringIdx).toBeLessThan(step2Idx)
|
||||
// and it reached the next model request.
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('goal reminder from step/end')
|
||||
})
|
||||
|
||||
it('steer() from a turn/end session-event listener becomes a queued message for the next turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const turns: number[] = []
|
||||
ctx.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
|
||||
let steeredOnce = false
|
||||
ctx.on('session/event', (subject, event) => {
|
||||
if (subject !== agent.session) return
|
||||
if (event.type === 'turn/start') turns.push(event.data.turn)
|
||||
if (event.type === 'turn/end' && !steeredOnce) {
|
||||
steeredOnce = true
|
||||
agent.steer([{ type: 'text', text: 'too late for this turn' }])
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -253,12 +272,12 @@ describe('HIGH: plugin exceptions are contained', () => {
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/turn-continuation', async (): Promise<boolean> => {
|
||||
ctx.on('agent/turn-continuation', async (): Promise<ContinuationDecision> => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw new Error('broken continuation plugin')
|
||||
}
|
||||
return false
|
||||
return { action: 'stop' }
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
@@ -314,7 +333,7 @@ describe('MEDIUM: disposed status is part of the agent/status contract', () => {
|
||||
const statuses: string[] = []
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/status', (_agent, status) => void statuses.push(status))
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
@@ -391,7 +410,7 @@ describe('MEDIUM: misc registry and config fixes', () => {
|
||||
expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }])
|
||||
})
|
||||
|
||||
it('agent/queued carries the resolved source; agent/steering carries its source', async () => {
|
||||
it('agent/queued carries the resolved source; steering/message records its source', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
@@ -406,15 +425,16 @@ describe('MEDIUM: misc registry and config fixes', () => {
|
||||
}))
|
||||
|
||||
const queuedSources: { source: MessageSource; steering: boolean }[] = []
|
||||
const steeringSources: MessageSource[] = []
|
||||
ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
|
||||
ctx.on('agent/steering', (_agent, _turn, _content, source) => void steeringSources.push(source))
|
||||
|
||||
send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, steering: false })
|
||||
expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, steering: true })
|
||||
// The drain appends the durable steering/message with the caller's source
|
||||
// intact — the log, not a transient emit, is where consumers read it.
|
||||
const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
|
||||
expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
|
||||
})
|
||||
})
|
||||
@@ -443,7 +463,7 @@ describe('MEDIUM: turn numbering continues across seeded (forked) sessions', ()
|
||||
ctx2.effect(() => forked.start())
|
||||
|
||||
const turns: number[] = []
|
||||
ctx2.on('agent/turn-start', (_agent, turn) => void turns.push(turn))
|
||||
ctx2.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
|
||||
forked.send([{ type: 'text', text: 'continue' }])
|
||||
await new Promise<void>((resolve) => {
|
||||
ctx2.on('agent/status', (subject, status) => {
|
||||
@@ -487,7 +507,7 @@ describe('HIGH: a finish-error stream chunk ends the turn as error, not complete
|
||||
const agent = ctx.agentLoop.create(AgentId('a-finish-error'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -512,7 +532,7 @@ describe('HIGH: a finish-error stream chunk ends the turn as error, not complete
|
||||
const agent = ctx.agentLoop.create(AgentId('a-finish-aborted'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -530,7 +550,7 @@ describe('HIGH: a finish-error stream chunk ends the turn as error, not complete
|
||||
const agent = ctx.agentLoop.create(AgentId('a-finish-error-nocode'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
@@ -539,24 +559,26 @@ describe('HIGH: a finish-error stream chunk ends the turn as error, not complete
|
||||
})
|
||||
})
|
||||
|
||||
describe('P1-6: step/start is appended before agent/step-start is emitted', () => {
|
||||
it('a step-start listener sees the step/start event already in session.events', async () => {
|
||||
describe('P1-6: a step/start session-event listener sees the event already in the log', () => {
|
||||
it('the step/start event is in session.events when its session/event listener fires', async () => {
|
||||
const adapter = new MockAdapter([textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-step-order'), { model: 'mock' })
|
||||
|
||||
// Capture, at the moment agent/step-start fires, whether the matching
|
||||
// step/start event is already in the log (append-before-emit, the event-sourcing RFC).
|
||||
// Session.append pushes the event BEFORE notifying session/event listeners,
|
||||
// so a step/start listener always finds the matching event already in the
|
||||
// log. (Step boundaries have no agent/* mirror — the session log is the live
|
||||
// feed.)
|
||||
const observed: { turn: number; step: number; lastEventType: string | undefined; sawStepStart: boolean }[] = []
|
||||
ctx.on('agent/step-start', (subject, turn, step) => {
|
||||
if (subject !== agent) return
|
||||
const events = [...subject.session.events]
|
||||
ctx.on('session/event', (subject, event) => {
|
||||
if (subject !== agent.session || event.type !== 'step/start') return
|
||||
const events = [...subject.events]
|
||||
const last = events.at(-1)
|
||||
observed.push({
|
||||
turn,
|
||||
step,
|
||||
turn: event.data.turn,
|
||||
step: event.data.step,
|
||||
lastEventType: last?.type,
|
||||
sawStepStart: events.some(e => e.type === 'step/start' && e.data.turn === turn && e.data.step === step),
|
||||
sawStepStart: events.some(e => e.type === 'step/start' && e.data.turn === event.data.turn && e.data.step === event.data.step),
|
||||
})
|
||||
})
|
||||
|
||||
@@ -599,35 +621,23 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
}
|
||||
}
|
||||
|
||||
it('a throwing agent/turn-start listener still closes the turn with exactly one error and one turn/end, no step', async () => {
|
||||
const adapter = new MockAdapter([textResponse('never reached')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-turnstart'), { model: 'mock' })
|
||||
|
||||
let threw = false
|
||||
ctx.on('agent/turn-start', () => { if (!threw) { threw = true; throw new Error('boom turn-start') } })
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const c = boundaryCounts(agent)
|
||||
// turn opened and closed; no step ran; exactly one error turn-end + emitted.
|
||||
expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 0, stepEnd: 0, errors: 1 })
|
||||
expect(errors.map(e => e.message)).toEqual(['boom turn-start'])
|
||||
expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toEqual({ kind: 'error', step: 0, message: 'boom turn-start' })
|
||||
// model was never called (we threw before the step's request).
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('a throwing agent/step-start listener closes the open step then the turn (step/end before turn/end)', async () => {
|
||||
it('a throwing step/start session-event listener closes the open step then the turn (step/end before turn/end)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('never reached')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-stepstart'), { model: 'mock' })
|
||||
|
||||
// Step boundaries have no agent/* mirror; a throwing step/start session-event
|
||||
// listener is the surviving step-boundary-listener failure. The loop marks
|
||||
// the step open BEFORE appending step/start (Session.append pushes before
|
||||
// notifying, so a post-push listener throw still leaves stepOpen=true), so
|
||||
// the outer catch's closeStep() appends the balancing step/end — the turn
|
||||
// stays enclosed. The invariants oracle (balancedHarness) rejects any
|
||||
// imbalance, so a green run proves turn/start → step/start → step/end →
|
||||
// turn/end nesting holds.
|
||||
let threw = false
|
||||
ctx.on('agent/step-start', () => { if (!threw) { threw = true; throw new Error('boom step-start') } })
|
||||
ctx.on('session/event', (_s, event) => {
|
||||
if (event.type === 'step/start' && !threw) { threw = true; throw new Error('boom step-start') }
|
||||
})
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
@@ -638,7 +648,7 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
const c = boundaryCounts(agent)
|
||||
expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 1, stepEnd: 1, errors: 1 })
|
||||
expect(errors.map(x => x.message)).toEqual(['boom step-start'])
|
||||
// step/end must precede turn/end (the invariants oracle would reject
|
||||
// step/end precedes turn/end (the invariants oracle would reject
|
||||
// turn/end-while-step-open, but assert the order explicitly too).
|
||||
const stepEndIdx = e.findIndex(x => x.type === 'step/end')
|
||||
const turnEndIdx = e.findIndex(x => x.type === 'turn/end')
|
||||
@@ -690,7 +700,7 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
@@ -707,46 +717,46 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
expect(e.some(x => x.type === 'turn/end' && x.data.reason.kind === 'error')).toBe(false)
|
||||
})
|
||||
|
||||
it('preserves reason disposed when the turn-end emit throws during disposal (outer-catch disposed branch)', async () => {
|
||||
// Dispose mid-step → the step-error branch sets reason=disposed (no error
|
||||
// reported). closeTurn(true) then emits agent/turn-end, whose listener
|
||||
// throws → control reaches the outer catch with isDisposed() && !errorReported,
|
||||
// which must PRESERVE disposed rather than overwrite it with the listener's
|
||||
// throw. This is the only path that exercises that catch sub-branch.
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
it('preserves reason disposed when a pre-step listener disposes then throws (outer-catch disposed branch)', async () => {
|
||||
// Reach the OUTER catch while disposed: an `agent/pre-step` listener requests
|
||||
// disposal AND throws. The throw escapes the pre-step `await` (line ~419) to
|
||||
// the loop's outer catch — BEFORE the post-pre-step disposal check at ~422
|
||||
// gets to run — so the catch sees `isDisposed() && !errorReported` and must
|
||||
// PRESERVE reason=disposed rather than overwrite it with the listener's throw
|
||||
// (disposal is not a failure). This is the surviving path to that sub-branch
|
||||
// now that there is no turn-boundary emit to throw from.
|
||||
const adapter = new MockAdapter([textResponse('never reached')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('a-dispose-emit-throw'), { model: 'mock' })
|
||||
agent = inner.agentLoop.create(AgentId('a-prestep-dispose-throw'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
// The FIRST agent/turn-end emit throws (the disposal-driven turn end).
|
||||
let threw = false
|
||||
ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end during disposal') } })
|
||||
// Collect agent/error emissions to prove none is surfaced through that
|
||||
// channel either (the listener throw must be fully contained).
|
||||
ctx.on('agent/pre-step', () => {
|
||||
if (threw) return
|
||||
threw = true
|
||||
// Request disposal, then throw in the same synchronous tick: status flips
|
||||
// to 'disposed' (the disposer aborts the step controller) and the throw
|
||||
// drives control into the outer catch with isDisposed() already true.
|
||||
void fiber.dispose()
|
||||
throw new Error('boom pre-step during disposal')
|
||||
})
|
||||
const errorEmits: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errorEmits.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await fiber.dispose() // dispose during the hanging step
|
||||
await agent.done
|
||||
|
||||
// The throwing turn-end listener actually fired — proving the outer-catch
|
||||
// path was exercised, not skipped.
|
||||
expect(threw).toBe(true)
|
||||
|
||||
const e = [...agent.session.events]
|
||||
// Exactly one turn/start and one turn/end (balanced); the turn/end carries
|
||||
// the disposed reason, NOT an error reason from the throwing listener.
|
||||
// Balanced: one turn/start, one turn/end carrying disposed (NOT error).
|
||||
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
|
||||
const turnEnd = e.findLast(x => x.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
|
||||
// The throwing turn-end listener is contained: the turn/end carries the
|
||||
// disposed reason (not an error) and no agent/error is emitted (disposal is
|
||||
// not a failure; the throw is swallowed).
|
||||
expect(e.some(x => x.type === 'turn/end' && x.data.reason.kind === 'error')).toBe(false)
|
||||
// No step opened (the throw was before step/start) and disposal is not a
|
||||
// failure, so no agent/error for the contained throw.
|
||||
expect(e.some(x => x.type === 'step/start')).toBe(false)
|
||||
expect(errorEmits).toHaveLength(0)
|
||||
})
|
||||
|
||||
@@ -792,54 +802,20 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('a throwing turn-end listener on a SUCCESSFUL turn leaves no event after turn/end (loadable log)', async () => {
|
||||
// Regression: a normal turn completes, closeTurn(true) appends turn/end and
|
||||
// emits agent/turn-end whose listener throws. The error must NOT be appended
|
||||
// as a session event after turn/end — that would sit past the commit
|
||||
// boundary and be dropped as a crash tail on resume (the turn-enclosure RFC). It is
|
||||
// surfaced via agent/error instead, and the log's last event is turn/end.
|
||||
const adapter = new MockAdapter([textResponse('done'), textResponse('next ok')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-tend'), { model: 'mock' })
|
||||
|
||||
let threw = false
|
||||
ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end') } })
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const c = boundaryCounts(agent)
|
||||
expect(c.turnEnd).toBe(1)
|
||||
expect(c.errors).toBe(0) // NO session error event (it would be post-turn/end)
|
||||
expect(agent.session.events.at(-1)?.type).toBe('turn/end') // last event is the boundary
|
||||
expect(errors.map(e => e.message)).toEqual(['boom turn-end']) // surfaced via agent/error
|
||||
// The late throw is also logged directly: failTurn's turn-already-ended
|
||||
// branch warns so a throwing turn-end listener after turn/end never vanishes.
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/turn-end listener threw after turn 1 closed'))
|
||||
// The whole log is loadable (nothing dropped): a fresh replay sees the turn.
|
||||
const replay = new Session(SessionId('replay'), [...agent.session.events])
|
||||
expect(replay.deriveMessages().map(m => m.role)).toEqual(['user', 'assistant'])
|
||||
|
||||
// loop survives.
|
||||
send(agent, 'again')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(boundaryCounts(agent).turnEnd).toBe(2)
|
||||
})
|
||||
|
||||
it('a throwing agent/step-end listener during a successful step ends the turn as error, not completed', async () => {
|
||||
// closeStep() must surface a throwing step-end listener via failTurn so the
|
||||
it('a throwing step/end session-event listener during a successful step ends the turn as error, not completed', async () => {
|
||||
// closeStep() must surface a throwing step/end listener via failTurn so the
|
||||
// turn ends with reason error, not a silent "completed" with the throw
|
||||
// swallowed. Regression test for the closeStep() catch that previously
|
||||
// swallowed the throw in the normal (no-tool, no-steering) path.
|
||||
// swallowed the throw in the normal (no-tool, no-steering) path. (Step
|
||||
// boundaries have no agent/* mirror; the session-event listener is the path.)
|
||||
const adapter = new MockAdapter([textResponse('all good'), textResponse('turn 2 ok')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-stepend-throw'), { model: 'mock' })
|
||||
|
||||
let threw = false
|
||||
ctx.on('agent/step-end', () => { if (!threw) { threw = true; throw new Error('boom step-end') } })
|
||||
ctx.on('session/event', (_s, event) => {
|
||||
if (event.type === 'step/end' && !threw) { threw = true; throw new Error('boom step-end') }
|
||||
})
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
@@ -869,52 +845,19 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
expect(c2.stepStart).toBe(c2.stepEnd)
|
||||
})
|
||||
|
||||
it('a step error followed by a throwing turn-end listener logs the error exactly once (no double-report)', async () => {
|
||||
// The step fails (finish-error) → failTurn records ONE error and sets the
|
||||
// error reason. closeTurn(true) then appends turn/end and emits
|
||||
// agent/turn-end, whose listener throws → the outer catch calls failTurn
|
||||
// again, but its errorReported guard makes it a no-op. Trap #1: exactly one
|
||||
// error, the turn stays balanced.
|
||||
const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider down' } }]
|
||||
const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
|
||||
const ctx = await balancedHarness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-double'), { model: 'mock' })
|
||||
|
||||
let threw = false
|
||||
ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end') } })
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const c = boundaryCounts(agent)
|
||||
// exactly one error turn-end + one agent/error emit, despite two failTurn calls.
|
||||
expect(c.errors).toBe(1)
|
||||
expect(errors.map(e => e.message)).toEqual(['provider down'])
|
||||
expect(c.turnStart).toBe(1)
|
||||
expect(c.turnEnd).toBe(1) // single turn/end, balanced
|
||||
expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toMatchObject({ kind: 'error', step: 1, message: 'provider down' })
|
||||
|
||||
// loop survives the compound failure.
|
||||
send(agent, 'again')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(boundaryCounts(agent).turnEnd).toBe(2)
|
||||
})
|
||||
|
||||
it('a throwing session/event listener on step/end during finalization still appends turn/end', async () => {
|
||||
// A throwing agent/step-start listener drives the outer catch, which calls
|
||||
// closeStep() during finalization. closeStep appends step/end; a
|
||||
// A finish-error stream opens a step then fails it, driving finalization
|
||||
// through closeStep() with the step open. closeStep appends step/end; a
|
||||
// session/event listener throwing on THAT must not abort the catch before
|
||||
// closeTurn(false) — step/end is already logged (balance holds) and the
|
||||
// throw is contained + surfaced via failTurn, so turn/end is still appended.
|
||||
const adapter = new MockAdapter([textResponse('never reached')])
|
||||
// closeTurn — step/end is already logged (balance holds) and the throw is
|
||||
// contained + surfaced via failTurn, so turn/end is still appended. (The
|
||||
// failed step itself also routes through failTurn; the step/end-listener
|
||||
// throw is the second, contained, failure.)
|
||||
const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }]
|
||||
const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-stependthrow'), { model: 'mock' })
|
||||
|
||||
// Open a step, then make the agent/step-start emit throw (boundary throw →
|
||||
// outer catch → closeStep during finalization).
|
||||
ctx.on('agent/step-start', () => { throw new Error('boom step-start') })
|
||||
let threw = false
|
||||
ctx.on('session/event', (_s, event) => {
|
||||
if (!threw && event.type === 'step/end') { threw = true; throw new Error('boom step/end listener') }
|
||||
@@ -941,11 +884,10 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
it('a throwing session/event listener on turn/end is contained (turn still balanced, loop survives)', async () => {
|
||||
// closeTurn appends turn/end; Session.append pushes it BEFORE notifying
|
||||
// session/event listeners, so a throwing listener leaves turn/end in the log
|
||||
// (the turn is balanced) but must not escape — from the normal-path
|
||||
// closeTurn(true) it would otherwise propagate; the append is contained so
|
||||
// the turn/end emit + loop continue. (A throwing agent/turn-end LISTENER is
|
||||
// a separate, already-tested path; here the session/event append notify is
|
||||
// what throws.)
|
||||
// (the turn is balanced) but must not escape — from the normal-path closeTurn
|
||||
// it would otherwise propagate; the append is contained so the loop continues.
|
||||
// Turn boundaries are durable session events only (no agent/* mirror), so this
|
||||
// session/event append-notify throw is the sole turn-end-listener failure path.
|
||||
const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a-turnendappend'), { model: 'mock' })
|
||||
@@ -972,7 +914,7 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
|
||||
})
|
||||
|
||||
describe('P1-7: tool/result is logged under the originating call.id, not result.callId', () => {
|
||||
it('a tools/execute listener returning a mismatched callId cannot orphan the call↔result pairing', async () => {
|
||||
it('the loop records tool/result under the model call.id even when a post-execute listener replaces content', async () => {
|
||||
// Model emits a tool-call with id "c1", then a final text turn.
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { x: 1 }),
|
||||
@@ -986,12 +928,13 @@ describe('P1-7: tool/result is logged under the originating call.id, not result.
|
||||
async execute() { return [{ type: 'text', text: 'ok' }] },
|
||||
}))
|
||||
|
||||
// A waterfall listener short-circuits with a result carrying the WRONG
|
||||
// callId (a listener-internal/proxy id). The loop must still record the
|
||||
// tool/result under the model's authoritative call.id.
|
||||
ctx.on('tools/execute', (exec) => {
|
||||
// A post-execute listener transforms the result (accept-with-replacement).
|
||||
// The loop must still record the tool/result under the model's authoritative
|
||||
// call.id (the loop ignores result.callId — which the registry always sets to
|
||||
// exec.callId anyway — and uses call.id, the model-transcript id).
|
||||
ctx.on('tools/post-execute', (exec, _result) => {
|
||||
expect(exec.callId).toBe(CallId('c1')) // the loop passed the real id in
|
||||
return Promise.resolve({ callId: CallId('wrong-proxy-id'), content: [{ type: 'text', text: 'ok' }], isError: false })
|
||||
return Promise.resolve({ kind: 'accept', content: [{ type: 'text', text: 'ok' }] })
|
||||
}, { prepend: true })
|
||||
|
||||
const agent = ctx.agentLoop.create(AgentId('a-callid'), { model: 'mock' })
|
||||
@@ -1084,7 +1027,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
// Give the loop time to enter the step and reach assemble().
|
||||
@@ -1110,10 +1053,8 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
// No step was opened, no LLM call was made.
|
||||
expect(e.some(x => x.type === 'step/start')).toBe(false)
|
||||
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
|
||||
// agent/turn-end may not fire when disposal happens during assembly: the
|
||||
// fiber's disposer (stop→status=disposed) runs before closeTurn(true)'s
|
||||
// emit, and the LIFO chain disposes effects in reverse registration order.
|
||||
// The turn/end durable record is the one that matters.
|
||||
// The durable turn/end record is the authoritative turn-boundary signal
|
||||
// (turn boundaries have no agent/* mirror), so this asserts on the log.
|
||||
})
|
||||
|
||||
it('cancel during system-prompt assembly drops the about-to-start step as aborted', { timeout: 30000 }, async () => {
|
||||
@@ -1142,7 +1083,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 50))
|
||||
@@ -1197,7 +1138,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 50))
|
||||
@@ -1218,9 +1159,8 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'disposed' })
|
||||
expect(e.some(x => x.type === 'step/start')).toBe(false)
|
||||
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
|
||||
// agent/turn-end may not fire when disposal happens during pre-step: the
|
||||
// fiber's disposer runs before closeTurn(true)'s emit. The durable turn/end
|
||||
// is the authoritative record.
|
||||
// The durable turn/end record is the authoritative turn-boundary signal
|
||||
// (turn boundaries have no agent/* mirror).
|
||||
})
|
||||
|
||||
it('cancel during agent/pre-step seam ends the turn aborted', { timeout: 15000 }, async () => {
|
||||
@@ -1250,7 +1190,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
@@ -1315,7 +1255,7 @@ describe('disposal/cancel honored during pre-step assembly (P1-1)', () => {
|
||||
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
|
||||
expect(e.some(x => x.type === 'assistant/message')).toBe(false)
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
// The durable turn/end reason is the authoritative record; agent/turn-end
|
||||
// may not fire when disposal interleaves with closeTurn(true)'s emit.
|
||||
// The durable turn/end reason is the authoritative turn-boundary record
|
||||
// (turn boundaries have no agent/* mirror).
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user