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

fix(agent-loop): decide turn balance + idle-injection flush from the log (review #32)

Browse Source 
Session.append pushes the event BEFORE notifying session/event listeners,
so a throwing listener leaves the event in the log while the line after
the append (a boolean flag) never runs. Both turn-balance decisions were
gated on such flags, so a throwing listener could strand an open turn or
skip a durability checkpoint.

- loop.ts: the outer catch decided "turn/end owed" from `turnStarted`.
  A throwing listener on the turn/start append left turn/start logged but
  the flag false → catch rethrew and skipped turn/end → permanently open
  turn (violating ADR 0017). Now decided from the log (this turn's
  turn/start present), so the turn is always balanced; only a genuine
  pre-push failure (non-serializable trigger — turn/start never logged) is
  rethrown to the runLoop backstop. Removed the now-dead `turnStarted`.

- agent.ts inject(): the idle one-shot-turn flush was gated on a
  `turnRecorded` flag set after append('turn/end'); a throwing turn/end
  listener skipped the flush, losing the balanced in-memory injection turn
  on crash. Now the flush decision is read from the log, the synthetic
  turn/end append contains a throwing listener (turn stays balanced), and
  a failing idle flush is reported via agent/error (step 0 convention) AND
  the logger — mirroring the loop's post-turn/end flush path — with a
  throwing agent/error listener contained.

Rewrote the test that encoded the old (buggy) "turn/start listener throw
is rethrown, no turn/end" semantics to assert the balanced-turn contract,
and added regressions for the throwing-turn/end-listener flush and the
agent/error report. Updated Agent.inject JSDoc.
This commit is contained in:
Tianyi Cui
2026-06-15 23:44:54 +08:00
parent da81709a31
commit 4535bfab75
6 changed files with 158 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/agent-loop/src/agent.ts
+32 -10
View File
@@ -93,29 +93,51 @@ export class LoopAgent implements Agent {
// open injection turn that would corrupt later turns/replay. (If the
// turn/start append throws BEFORE pushing — non-serializable trigger, which
// can't happen for our fixed trigger — no turn was opened and none is owed.)
let turnRecorded = false
try {
this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
this.session.append('context/message', { content, source })
} finally {
// A turn was recorded iff turn/start made it into the log. Close it and
// mark it for the durability checkpoint below — which must run even when
// an append's listener threw (the turn is balanced and in memory, so it
// still needs a flush or a crash before the next turn/dispose loses it).
// Close the turn if turn/start made it into the log. Contain a throwing
// turn/end listener: Session.append pushes before notifying, so a throw
// here still leaves turn/end in the log (the turn is balanced) — swallow
// it so it neither replaces the original exception nor skips the flush
// decision below. (It surfaces through the flush path is not needed; the
// turn-balance contract is what matters and it holds.)
if (isTurnOpen(this.session)) {
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
turnRecorded = true
try {
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
} catch {
// turn/end is already in the log (pushed before the listener threw),
// so the turn is balanced; the throw is the listener's bug.
}
}
// Decide the durability checkpoint from the LOG, not a flag: a turn was
// recorded iff this turn's turn/start is logged (it may have been closed
// by a throwing-listener turn/end above, which still counts). A
// `turnRecorded` boolean set after append('turn/end') would be skipped by
// a throwing turn/end listener, losing the flush for a balanced in-memory
// turn (crash before the next turn/dispose would drop the idle injection).
const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
// Checkpoint the one-shot turn for durability, exactly as the loop does at
// every turn/end. The loop is NOT running (we are idle), so nothing else
// will flush this turn. Fire-and-forget with error containment: inject()
// is synchronous, and a persistence backend failing must not throw into
// the caller (e.g. a tool-bash task-done callback). Disposal still drains
// independently, so a slow flush is safe. In the finally so it also runs
// when an append's listener threw (the turn is still balanced + durable).
// independently, so a slow flush is safe. A flush failure is reported via
// agent/error (step 0 — the idle-injection convention, there is no real
// step) AND the logger, mirroring the loop's post-turn/end flush path so
// plugins monitoring agent/error see idle-injection persistence failures
// too. A throwing agent/error listener is contained.
if (turnRecorded) {
void Promise.resolve(this.ctx.parallel('session/flush', this.session)).catch((error: unknown) => {
this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${String(error)}`)
const err = error instanceof Error ? error : new Error(String(error))
this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
try {
this.ctx.emit('agent/error', this, turn, 0, err)
} catch {
// contained: the failure is already logged; a throwing agent/error
// listener must not escape this fire-and-forget catch.
}
})
}
}
packages/agent-loop/src/loop.ts
+17 -9
View File
@@ -170,7 +170,6 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
let reason: TurnEndReason = { kind: 'completed' }
let step = 0
let turnStarted = false
let turnEnded = false
let stepOpen = false
let errorReported = false
@@ -236,9 +235,10 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
try {
// --- Turn boundary. Once turn/start is appended, a turn/end is owed no
// matter what throws below; the catch + closeTurn guarantee it.
// matter what throws below; the catch + closeTurn guarantee it (the catch
// decides "owed" from the log via isTurnOpen, so even a throwing turn/start
// listener — append pushes before notifying — still gets its turn/end).
session.append('turn/start', { turn, trigger })
turnStarted = true
// 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.
@@ -320,12 +320,20 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
// Normal / inline-error loop exit: close the turn and notify.
closeTurn(true)
} catch (error: unknown) {
// A pre-turn throw (turn/start append) is owed no turn/end — rethrow to
// the backstop. Otherwise a boundary emit (turn-start, step-start, the
// normal-path turn-end) or other unhandled throw escaped: close any open
// step, choose the reason (disposal wins only if no error was reported),
// record the error, and close the turn WITHOUT re-emitting agent/turn-end.
if (!turnStarted) throw error
// Decide whether this turn was ever opened from the LOG, not a flag.
// Session.append pushes the event BEFORE notifying session/event listeners,
// so a throwing listener on the `turn/start` append leaves turn/start in the
// log even though execution never reached the lines after that append.
// Gating on a "turn started" boolean would skip turn/end and leave a
// permanently OPEN turn that poisons the next turn/replay (ADR 0017). We
// check the log for THIS turn's turn/start: present means a turn/end is owed
// (or was already appended — closeTurn/failTurn are idempotent, so running
// them again is a safe no-op that still preserves the disposed/error reason
// chosen below). Absent means the turn/start append threw BEFORE its push (a
// non-serializable trigger — impossible for our fixed trigger); nothing was
// opened, so rethrow to the runLoop backstop.
const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
if (!turnStartLogged) throw error
closeStep()
// Choose the close reason. Disposal wins only if no error was already
// reported: a turn disposed mid-step sets reason=disposed in the step-error
packages/agent-loop/tests/agent.spec.ts
+44
View File
@@ -141,6 +141,50 @@ describe('LoopAgent', () => {
expect(flushes).toBe(1) // checkpoint fired despite the throw
})
it('idle inject() still checkpoints when a listener throws on the synthetic turn/end', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
let flushes = 0
ctx.on('session/flush', () => { flushes += 1 })
// A session/event listener that throws on the synthetic turn/end. Append
// pushes before notifying, so turn/end is in the log (turn balanced) but the
// throw must NOT skip the durability checkpoint — the flush decision is made
// from the log, not a flag set after the (throwing) append.
let threw = false
ctx.on('session/event', (_s, event) => {
if (!threw && event.type === 'turn/end') { threw = true; throw new Error('boom turn/end') }
})
expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
const types = agent.session.events.map(e => e.type)
expect(types).toEqual(['turn/start', 'context/message', 'turn/end']) // balanced
await new Promise(r => setTimeout(r, 10))
expect(flushes).toBe(1) // checkpoint fired despite the throwing turn/end listener
})
it('idle inject() reports a failing flush via agent/error (step 0) AND the logger', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
// A non-Error rejection exercises the String() normalization branch.
ctx.on('session/flush', () => { throw 'disk gone' })
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
const errors: { turn: number; step: number; message: string }[] = []
ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message }))
agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } })
await new Promise(r => setTimeout(r, 20)) // let the contained flush settle
// Reported via agent/error (step 0 — the idle-injection convention) so
// plugins monitoring agent/error see idle-injection persistence failures,
// mirroring the loop's post-turn/end flush path. A non-Error throw is
// normalized to an Error.
expect(errors).toEqual([{ turn: 1, step: 0, message: 'disk gone' }])
expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed'))
warn.mockRestore()
})
it('idle inject() with a non-serializable source opens no turn (nothing to close)', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
packages/agent-loop/tests/coverage-edges.spec.ts
+42 -7
View File
@@ -35,9 +35,11 @@ function send(agent: LoopAgent, text: string) {
agent.send([{ type: 'text', text }])
}
describe('loop backstop catch', () => {
it('a throwing turn-start listener is caught by the backstop and loop survives', async () => {
// The first turn will abort before the model call (turn-start throw).
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)
@@ -57,6 +59,9 @@ describe('loop backstop catch', () => {
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')
@@ -65,7 +70,7 @@ describe('loop backstop catch', () => {
expect(adapter.requests[0]!.messages.some(m => m.content.some(b => 'text' in b && b.text === 'second'))).toBe(true)
})
it('a throwing turn-end listener is caught by the backstop and loop survives', async () => {
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('a1', { model: 'mock' })
@@ -84,7 +89,8 @@ describe('loop backstop catch', () => {
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. The error is surfaced by the backstop.
// 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
@@ -92,6 +98,35 @@ describe('loop backstop catch', () => {
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
// never enters the log. runTurn sees no logged turn/start and rethrows; the
// runLoop backstop reports via agent/error (step 0) + the logger and the
// driver survives. This is the ONLY path that reaches the backstop.
const adapter = new MockAdapter([textResponse('turn 2')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
const errors: { turn: number; step: number; message: string }[] = []
ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message }))
// A non-serializable source (BigInt) on the queued message.
agent.send([{ type: 'text', text: 'first' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
await waitForIdle(ctx, agent)
expect(errors).toHaveLength(1)
expect(errors[0]!.step).toBe(0)
expect(errors[0]!.message).toMatch(/non-JSON-serializable/)
// No turn boundary was written (the turn/start append threw before push).
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
// loop survives: a well-formed second turn runs normally.
send(agent, 'second')
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(1)
})
})
describe('tool JSON parse', () => {
@@ -157,7 +192,7 @@ describe('tool JSON parse', () => {
})
describe('toError normalization', () => {
it('normalizes non-Error throws from turn-start listeners via toError in the backstop', async () => {
it('normalizes non-Error throws from turn-start listeners via toError', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a1', { model: 'mock' })
@@ -166,7 +201,7 @@ describe('toError normalization', () => {
ctx.on('agent/turn-start', () => {
if (!threwOnce) {
threwOnce = true
throw 'naked string error' // non-Error throw, goes through backstop's toError
throw 'naked string error' // non-Error throw, normalized via toError
}
})
packages/agent-loop/tests/review-fixes.spec.ts
+20 -10
View File
@@ -822,13 +822,15 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
expect(errorEmits).toHaveLength(0)
})
it('a throw at the turn/start append (before turnStarted) is rethrown to the runLoop backstop', async () => {
// A session/event listener that throws specifically on the turn/start
// event makes session.append('turn/start') throw while turnStarted is
// still false. runTurn must NOT try to close a turn it never opened — it
// rethrows, and the runLoop backstop records the error and survives.
// (Uses the plain harness — NOT the invariants oracle — because the
// throwing listener is itself a session/event subscriber.)
it('a throwing session/event listener on the turn/start append still balances the turn', async () => {
// Session.append pushes the event BEFORE notifying session/event listeners,
// so a listener throwing on turn/start leaves turn/start IN THE LOG. The
// loop must therefore still owe (and append) a turn/end — deciding "owed"
// from the log via isTurnOpen, not a "turn started" flag that the throw
// skipped. Otherwise the turn stays permanently open and poisons the next
// turn/replay (ADR 0017). (Uses the plain harness — NOT the invariants
// oracle — because the throwing listener is itself a session/event
// subscriber.)
const adapter = new MockAdapter([textResponse('turn 2')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create('a-preturn', { model: 'mock' })
@@ -843,10 +845,18 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
send(agent, 'go')
await waitForIdle(ctx, agent)
// The backstop logged exactly one error for the failed pre-turn append.
// The error was surfaced exactly once via agent/error.
expect(errors.map(e => e.message)).toEqual(['boom turn/start append'])
// No turn/end was appended (none is owed — the turn never opened).
expect([...agent.session.events].some(e => e.type === 'turn/end')).toBe(false)
// The turn is BALANCED: turn/start is in the log (it was pushed before the
// listener threw), so a turn/end was owed and appended — no open turn. The
// last turn-boundary event being turn/end is exactly the loop's isTurnOpen
// check (no open turn remains).
const types = [...agent.session.events].map(e => e.type)
expect(types.filter(t => t === 'turn/start')).toHaveLength(1)
expect(types.filter(t => t === 'turn/end')).toHaveLength(1)
const lastBoundary = [...agent.session.events].reverse().find(e => e.type === 'turn/start' || e.type === 'turn/end')
expect(lastBoundary?.type).toBe('turn/end')
expect(agent.session.events.at(-1)?.type).toBe('turn/end')
// loop survives: a second turn runs normally.
send(agent, 'second')
packages/agent/src/types.ts
+3 -1
View File
@@ -68,7 +68,9 @@ export interface Agent {
* an inject while idle wraps its `context/message` in a one-shot `injection`
* turn (`turn/start` → `context/message` → `turn/end`) and checkpoints it for
* durability, so every event stays inside a turn and a persistence backend
* never loses a between-turn notice.
* never loses a between-turn notice. The idle checkpoint is fire-and-forget
* (inject is synchronous): a failing flush is reported via `agent/error`
* (step `0`) and the logger, never thrown into the caller.
*
* TODO(review): exact envelope/rendering rules live in dsh-session and need
* review once a real adapter exists.