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

refactor(events): document event-domain semantics, drop step-boundary mirror emits

Browse Source 
Pin the three-domain rule (session = durable fact log, agent = live runtime
surface, tools = registry/exec): a durable replayable fact is a SessionEvent; a
live interception or transient/live-object signal is an agent/tools Cordis
event. A boundary that is both is mirrored as an agent/* emit ONLY where a live
consumer needs the Agent handle.

Apply it to the boundary twins: drop agent/step-start and agent/step-end (no
production consumer needs the live Agent at a step boundary — consumers read the
durable step/start/step/end session events). Keep agent/turn-start/turn-end (the
stdio UI labels output by agent.id). Tests that observed step boundaries via the
removed emits now observe the durable session events; the pinned behavior is
unchanged.

Conservative subset of the proposed "remove boundary mirror events"
simplification; foundation for the Hooks subsystem's canonical event surface.
This commit is contained in:
Tianyi Cui
2026-06-30 10:32:55 +08:00
parent 3f85f522ea
commit 05b75abbca
9 changed files with 176 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -204,7 +204,7 @@ describe('Agent.cancel()', () => {
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
let steps = 0
ctx.on('agent/step-start', () => { steps += 1 })
ctx.on('session/event', (_session, event) => { if (event.type === 'step/start') steps += 1 })
const reasons: TurnEndReason[] = []
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))

23
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -46,15 +46,21 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
// Turn boundaries are live agent/* emits; step boundaries are durable
// session events only (no agent/* mirror). Interleave both feeds 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) {
for (const name of ['agent/turn-start', 'agent/turn-end'] as const) {
ctx.on(name, () => void order.push(name))
}
ctx.on('session/event', (_session, event) => {
if (event.type === 'step/start' || event.type === 'step/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(['agent/turn-start', 'step/start', 'step/end', 'agent/turn-end'])
const types = agent.session.events.map(e => e.type)
// turn/start opens the turn, THEN the queued user message is recorded inside
@@ -269,7 +275,7 @@ 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
return next()
@@ -371,7 +377,7 @@ 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) => {
@@ -536,7 +542,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'),
@@ -552,8 +558,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')

105
packages/core/agent-loop/tests/review-fixes.spec.ts
View File

@@ -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'),
@@ -539,24 +509,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),
})
})
@@ -621,29 +593,35 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
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 fails the turn balanced (no step stranded open)', 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 throw fires
// INSIDE session.append('step/start') — before the loop marks the step open —
// so the loop never had an open step to close (no step/end is owed). The
// throw drives the outer catch, which fails the turn balanced. The invariants
// oracle (balancedHarness) rejects any imbalance, so a green run proves the
// turn/start..turn/end nesting holds with a lone step/start and no step/end.
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))
send(agent, 'go')
await waitForIdle(ctx, agent)
const e = [...agent.session.events]
const c = boundaryCounts(agent)
expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 1, stepEnd: 1, errors: 1 })
// step/start was appended (Session.append pushes before notifying), but the
// listener throw pre-empted the loop marking the step open, so no step/end is
// owed; the turn still closes exactly once with an error, balanced.
expect(c).toMatchObject({ turnStart: 1, turnEnd: 1, stepStart: 1, stepEnd: 0, errors: 1 })
expect(errors.map(x => x.message)).toEqual(['boom step-start'])
// step/end must precede 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')
expect(stepEndIdx).toBeGreaterThanOrEqual(0)
expect(stepEndIdx).toBeLessThan(turnEndIdx)
expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason.kind).toBe('error')
})
it('a throwing agent/error listener during a step-error path still balances the turn, loop survives', async () => {
@@ -829,17 +807,20 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
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))
@@ -903,18 +884,18 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
})
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') }