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

refactor(agent-loop): start waking work directly

Browse Source
This commit is contained in:
_Kerman
2026-08-04 21:02:28 +08:00
parent 674278cb17
commit e50c4aca03
23 changed files with 103 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,7 +41,6 @@ type Phase =
lastTurn: number
wakeRequested: boolean
}
| { kind: 'collecting'; abort: AbortController; lastTurn: number }
| { kind: 'running'; abort: AbortController; turn: number; step: number }
type StepEndReason = Extract<TurnEndReason, { kind: 'completed' | 'max-tokens' }>
@@ -109,7 +108,7 @@ export class ReactLoopAgent implements Agent {
const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted
const resolvedTarget = wakingAfterAbort ? 'next-turn' : target
this.inbox.splice(resolvedTarget, Infinity, 0, [message])
if (wakeup) this.scheduleKick()
if (wakeup) this.wakeDriver()
}
followup(input: UserMessage): void {
@@ -148,14 +147,14 @@ export class ReactLoopAgent implements Agent {
return await task(maintenance.abort.signal)
} finally {
this.setPhase({ kind: 'idle', lastTurn: maintenance.lastTurn })
if (maintenance.wakeRequested) this.scheduleKick()
if (maintenance.wakeRequested) this.wakeDriver()
done.resolve()
}
})()
}
/** Schedule one driver, or remember its wake behind maintenance. */
private scheduleKick(): void {
/** Start one driver, or remember its wake behind maintenance. */
private wakeDriver(): void {
if (this.phase.kind === 'maintenance') {
if (!this.phase.abort.signal.aborted) this.phase.wakeRequested = true
return
@@ -163,10 +162,8 @@ export class ReactLoopAgent implements Agent {
if (this.phase.kind !== 'idle') return
const driver = Promise.withResolvers<void>()
this.activityDone = driver.promise
this.setPhase({ kind: 'collecting', abort: new AbortController(), lastTurn: this.phase.lastTurn })
queueMicrotask(() => {
this.loopCtx.agents.withInitiator(this, () => this.kick()).then(driver.resolve, driver.reject)
})
this.setPhase({ kind: 'running', abort: new AbortController(), turn: this.phase.lastTurn, step: 0 })
this.loopCtx.agents.withInitiator(this, () => this.kick()).then(driver.resolve, driver.reject)
}
async whenIdle(): Promise<void> {
@@ -221,14 +218,11 @@ export class ReactLoopAgent implements Agent {
/** Open one turn before claiming its first proposed step. */
private async turn(): Promise<boolean> {
if (this.phase.kind === 'idle' || this.phase.kind === 'maintenance') {
if (this.phase.kind !== 'running') {
this.throwError(new Error(`agent "${this.id}": turn without driver reservation`))
}
const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController()
const { signal } = abort
const lastTurn = this.phase.kind === 'collecting' ? this.phase.lastTurn : this.phase.turn
const phase = { kind: 'running' as const, abort, turn: lastTurn, step: 0 }
this.setPhase(phase)
const phase = this.phase
const { signal } = phase.abort
signal.throwIfAborted()
const turn = phase.turn + 1
try {
@@ -301,7 +295,10 @@ export class ReactLoopAgent implements Agent {
this.throwError(error)
}
}
return this.inbox.hasPending
if (!this.inbox.hasPending) return false
phase.abort = new AbortController()
phase.step = 0
return true
}
private async step(assembly: PromptAssembly): Promise<StepEndReason | null> {

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

@@ -72,8 +72,8 @@ describe('Agent.cancel()', () => {
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
})
it('cancel({ keepInbox: true }) preserves queued work and emits no discard', async () => {
const adapter = new MockAdapter([textResponse('preserved reply'), textResponse('wake reply')])
it('cancel({ keepInbox: true }) does not restore work already claimed by a waking send', async () => {
const adapter = new MockAdapter([textResponse('wake reply')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -81,21 +81,23 @@ describe('Agent.cancel()', () => {
content: [{ type: 'text', text: 'preserved' }],
source: { kind: 'user' },
}))
// Abort the collecting activity while preserving its queued item.
// A waking send starts and claims synchronously, so keepInbox has no
// pending item to preserve by the time this cancellation runs.
agent.cancel({ kind: 'user' }, { keepInbox: true })
expect(agent.session.events.some(event =>
event.type === 'agent/inbox/spliced' && event.data.outcome === 'canceled')).toBe(false)
await agent.whenIdle()
expect(agent.inbox.nextTurn).toHaveLength(1)
expect(agent.inbox.nextTurn).toHaveLength(0)
expect(userTexts(agent)).toEqual([])
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.findLast(event => event.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'aborted', reason: { kind: 'user' } })
// The preserved item still runs once a later follow-up wakes the driver.
const idle = waitForIdle(ctx, agent)
send(agent, 'wake it')
await idle
expect(userTexts(agent)).toEqual(['preserved', 'wake it'])
expect(adapter.requests).toHaveLength(2)
expect(userTexts(agent)).toEqual(['wake it'])
expect(adapter.requests).toHaveLength(1)
})
it('cancel({ keepInbox: true }) parks queued work after an active turn aborts', async () => {
@@ -124,23 +126,22 @@ describe('Agent.cancel()', () => {
expect(adapter.requests).toHaveLength(3)
})
it('pre-step cancel drops the about-to-start turn (no turn is opened)', async () => {
it('cancel after waking send closes its synchronously opened turn without a step', async () => {
const adapter = new MockAdapter([textResponse('should not run')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// followup() queues synchronously (status still idle, loop microtask not yet
// resumed). Cancel in that pre-step window: the queued turn must not run.
send(agent, 'drop me first')
send(agent, 'drop me second')
agent.cancel({ kind: 'user' })
// Give the loop a chance to wake and process the cancel.
await new Promise(r => setTimeout(r, 30))
// No turn was opened — the queued prompt was dropped, never recorded.
expect(userTexts(agent)).toEqual([])
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
expect(agent.session.events.filter(event => event.type === 'step/start')).toHaveLength(0)
expect(agent.session.events.findLast(event => event.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'aborted', reason: { kind: 'user' } })
expect(agent.status).toBe('idle')
})
@@ -218,7 +219,7 @@ describe('Agent.cancel()', () => {
await expect(Promise.race([
replacementObservation,
new Promise((_resolve, reject) => setTimeout(() => { reject(new Error('whenIdle hung after idle-listener cancel')) }, 1000)),
])).resolves.toEqual({ status: 'idle', requests: 1, turns: 1 })
])).resolves.toEqual({ status: 'idle', requests: 1, turns: 2 })
const idle = waitForIdle(ctx, agent)
send(agent, 'later')
@@ -511,10 +512,10 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
send(agent, 'A') // queues A (status still idle, loop microtask pending)
const idle = agent.whenIdle() // registers a waiter (idle + hasQueued → no fast path)
agent.cancel({ kind: 'user' }) // arms marker, clears A
send(agent, 'B') // B races in before the loop resumes
send(agent, 'A')
const idle = agent.whenIdle()
agent.cancel({ kind: 'user' })
send(agent, 'B')
await idle
expect(userTexts(agent)).toEqual([])
@@ -524,7 +525,7 @@ describe('Agent.cancel()', () => {
send(agent, 'C')
await replacementIdle
expect(userTexts(agent)).toEqual(['B', 'C'])
expect(agent.session.events.filter(event => event.type === 'turn/end')).toHaveLength(2)
expect(agent.session.events.filter(event => event.type === 'turn/end')).toHaveLength(3)
})
it("cancel clears the turn's steering — it is not re-enqueued as a fresh turn", async () => {

7
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -191,10 +191,11 @@ describe('abort during tool execution ends the turn', () => {
const adapter = new MockAdapter([textResponse('must not run')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-empty-batch'), { provider: 'mock', model: 'mock' })
ctx.on('agent/pre-step', (subject, _messages, _context, next) => {
if (subject !== agent) return next()
return Promise.resolve({ kind: 'enter', messages: [] })
})
send(agent, 'go')
// The wake microtask has not run yet: remove the only pending message so
// the admission batch is empty.
agent.inbox.remove(agent.inbox.nextTurn[0]!.id)
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.filter(event => event.type === 'turn/start'

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

@@ -25,11 +25,7 @@ async function harness(adapter: MockAdapter, persona = '') {
return ctx
}
/**
* Wait for the agent's NEXT transition to idle. Always event-based: callers
* invoke this right after send(), when the loop hasn't woken yet (status is
* still 'idle' synchronously), so polling the current status would lie.
*/
/** Wait for the agent's next transition to idle after a waking send. */
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
@@ -527,13 +523,15 @@ describe('agent loop', () => {
expect(flat).toContain('change of plans')
})
it('coalesces same-tick idle steering into one turn', async () => {
const adapter = new MockAdapter([textResponse('first')])
it('starts idle steering synchronously and enters later steering at the next step', async () => {
const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const idle = waitForIdle(ctx, agent)
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'first idle steer' }], source: { kind: 'user' } }))
expect(agent.status).toBe('running')
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'second idle steer' }], source: { kind: 'user' } }))
await idle
@@ -544,9 +542,10 @@ describe('agent loop', () => {
[{ type: 'text', text: 'first idle steer' }],
[{ type: 'text', text: 'second idle steer' }],
])
expect(adapter.requests).toHaveLength(1)
expect(adapter.requests).toHaveLength(2)
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('first idle steer')
expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('second idle steer')
expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('second idle steer')
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('second idle steer')
})
it('stops after a throwing pre-step listener and retains later steering until a wakeup', async () => {