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

refactor(agent-loop): project runtime context before steps

Browse Source
This commit is contained in:
_Kerman
2026-08-01 20:39:10 +08:00
parent 1a09174987
commit d38c8bfaf3
11 changed files with 255 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,8 +1,8 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
/**
* Tests for the queue-aware `Agent.cancel()` primitive. The default clears
* queued and steering work, while `keepInbox` preserves pending input and
* resumes waking turns after the active turn reaches quiescence. The suite
* queued and steering work, while `keepInbox` preserves pending input for a
* later wake after the active turn reaches quiescence. The suite
* covers every landing window plus signal reset and `whenIdle()` quiescence.
* @module dsh-agent-loop/tests/cancel
*/
@@ -284,41 +284,6 @@ describe('Agent.cancel()', () => {
expect(adapter.requests).toHaveLength(1)
})
it('cancel({ keepInbox: true }) aborts the active turn and drains the queued tail in FIFO order', async () => {
const adapter = new MockAdapter([
'hang',
textResponse('second reply'),
textResponse('third reply'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('keep-inbox-running'), { provider: 'mock', model: 'mock' })
const reasons: TurnEndReason[] = []
const discards: unknown[] = []
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/end') reasons.push(event.data.reason)
})
ctx.on('agent/inbox/discard', (subject, items) => {
if (subject === agent) discards.push(items)
})
send(agent, 'active')
await new Promise(resolve => setTimeout(resolve, 30))
send(agent, 'queued second')
send(agent, 'queued third')
const idle = agent.whenIdle()
agent.cancel({ kind: 'user' }, { keepInbox: true })
await idle
expect(discards).toEqual([])
expect(userTexts(agent)).toEqual(['active', 'queued second', 'queued third'])
expect(reasons).toEqual([
{ kind: 'aborted' },
{ kind: 'completed' },
{ kind: 'completed' },
])
expect(adapter.requests).toHaveLength(3)
})
it('cancel from an assistant/message observer skips execution but balances replay', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'danger', {}),
@@ -770,7 +735,7 @@ describe('Agent.cancel()', () => {
agent.cancel({ kind: 'user' })
await idle
const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
if (stage === 'pre-step') {
if (stage === 'pre-step' || stage === 'system-prompt') {
expect(turnEnd).toBeUndefined()
} else {
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })

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

@@ -1122,7 +1122,7 @@ describe('tool result call identity', () => {
})
describe('disposal and cancellation during pre-step assembly', () => {
it('disposal during system-prompt assembly closes the started step as disposed', { timeout: 30000 }, async () => {
it('disposal during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => {
// Start disposal, then release assembly. Do not await disposal first: it
// waits for the blocked driver to exit.
const adapter = new MockAdapter(['hang'])
@@ -1154,7 +1154,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
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().
// Give the loop time to reach pre-step assembly.
await new Promise(r => setTimeout(r, 50))
// Release assembly before awaiting disposal because disposal joins the blocked driver.
@@ -1165,18 +1165,15 @@ describe('disposal and cancellation during pre-step assembly', () => {
await driverDone(agent)
unlisten()
// Turn boundaries are durable rows; there is no `agent/*` mirror to assert.
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'disposed' } })
expect(e.filter(x => x.type === 'step/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'step/end')).toHaveLength(1)
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'step/end')).toBe(false)
expect(e.some(x => x.type === 'assistant/chunk')).toBe(false)
})
it('cancel during system-prompt assembly closes the started step as aborted', { timeout: 30000 }, async () => {
it('cancel during system-prompt assembly prevents the turn from opening', { timeout: 30000 }, async () => {
const adapter = new MockAdapter([textResponse('should not appear')])
let releaseAssemble!: () => void
const blocker = new Promise<void>(r => void (releaseAssemble = r))
@@ -1215,16 +1212,14 @@ describe('disposal and cancellation during pre-step assembly', () => {
unlisten()
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
const turnEnd = e.findLast(x => x.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'aborted', reason: { kind: 'user' } })
expect(e.filter(x => x.type === 'step/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'step/end')).toHaveLength(1)
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
expect(e.some(x => x.type === 'step/start')).toBe(false)
expect(e.some(x => x.type === 'step/end')).toBe(false)
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)
expect(reasons).toEqual([{ kind: 'aborted', reason: { kind: 'user' } }])
expect(reasons).toEqual([])
})
it('disposal during pre-step prevents the turn from opening', { timeout: 15000 }, async () => {
@@ -1356,14 +1351,10 @@ describe('disposal and cancellation during pre-step assembly', () => {
await driverDone(agent)
const e = [...agent.session.events]
expect(e.filter(x => x.type === 'turn/start')).toHaveLength(1)
expect(e.filter(x => x.type === 'turn/end')).toHaveLength(1)
// The critical assertions: after disposal, the turn has no assistant
// artifacts — the turn ended disposed before the model was invoked.
expect(e.some(x => x.type === 'turn/start')).toBe(false)
expect(e.some(x => x.type === 'turn/end')).toBe(false)
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 turn-boundary record
// (turn boundaries have no agent/* mirror).
})
})

14
packages/core/agent-loop/tests/tool-order.spec.ts
View File

@@ -93,8 +93,7 @@ describe('loop-level canonical tool order', () => {
expect(Object.isFrozen(adapter.requests[0])).toBe(true)
})
it('fails the turn — no model request — when toolOrder names an unregistered tool', async () => {
// Unknown tool order fails before step or request creation and returns the agent to idle.
it('fails before opening a turn when toolOrder names an unregistered tool', async () => {
const adapter = new MockAdapter([textResponse('never sent')])
const ctx = await harness(adapter, ['ghost', TOOL_ORDER_REST])
registerNamed(ctx, 'alpha')
@@ -103,12 +102,9 @@ describe('loop-level canonical tool order', () => {
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(foldRequestHeader(agent.session.events)).toBeUndefined()
const end = agent.session.events.find(e => e.type === 'turn/end')
expect(end?.type === 'turn/end' && end.data.reason).toEqual({
kind: 'error',
error: 'toolOrder lists unregistered tool "ghost"; known tools: alpha',
})
expect(agent.session.events.filter(e => e.type === 'step/start')).toHaveLength(1)
expect(agent.session.events.filter(e => e.type === 'step/end')).toHaveLength(1)
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(false)
expect(agent.session.events.some(e => e.type === 'step/start')).toBe(false)
expect(agent.session.events.some(e => e.type === 'step/end')).toBe(false)
})
})

286
packages/core/agent-loop/tests/turn-admission.spec.ts
View File

@@ -1,286 +0,0 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { type Agent, type InboxItem } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import LlmService, { createUserMessage } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import { MockAdapter, textResponse } from './mock-adapter.ts'
async function harness(adapter: MockAdapter): Promise<Context> {
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 prompt(agent: Agent, text: string): void {
agent.followup(createUserMessage({
content: [{ type: 'text', text }],
source: { kind: 'user' },
}))
}
function itemText(item: InboxItem): string {
return item.message.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')
}
interface InboxRecording {
readonly events: string[]
readonly enqueued: InboxItem['id'][]
readonly dequeued: InboxItem['id'][]
readonly discarded: InboxItem['id'][]
}
/** Record the complete inbox lifecycle of one agent for order and identity assertions. */
function recordInbox(ctx: Context): InboxRecording {
const events: string[] = []
const enqueued: InboxItem['id'][] = []
const dequeued: InboxItem['id'][] = []
const discarded: InboxItem['id'][] = []
ctx.on('agent/inbox/enqueue', (_agent, item) => {
events.push(`enqueue:${item.placement}:${itemText(item)}`)
enqueued.push(item.id)
})
ctx.on('agent/inbox/dequeue', (_agent, item) => {
events.push(`dequeue:${itemText(item)}`)
dequeued.push(item.id)
})
ctx.on('agent/inbox/discard', (_agent, items) => {
events.push(`discard:${items.map(itemText).join(',')}`)
discarded.push(...items.map(item => item.id))
})
return { events, enqueued, dequeued, discarded }
}
/** Text of every ordinary prompt the log admitted, in durable order. */
function promptTexts(agent: Agent): string[] {
return agent.session.events.flatMap(event => event.type === 'user/message'
? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : [])
: [])
}
describe('idle turn admission reservation', () => {
it('holds later waking prompts in the FIFO until release', 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 inbox = recordInbox(ctx)
const release = agent.reserveTurnAdmission()
expect(release).toBeDefined()
prompt(agent, 'first prompt')
prompt(agent, 'second prompt')
expect(agent.acceptsNextStep).toBe(false)
await new Promise<void>((resolve) => { setTimeout(resolve, 5) })
expect(agent.status).toBe('idle')
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events).toHaveLength(0)
expect(inbox.events).toEqual([
'enqueue:queued:first prompt',
'enqueue:queued:second prompt',
])
release?.()
await agent.whenIdle()
expect(promptTexts(agent)).toEqual(['first prompt', 'second prompt'])
expect(agent.session.events.flatMap(event =>
event.type === 'turn/start' ? [event.data.turn] : [])).toEqual([1, 2])
expect(inbox.events).toEqual([
'enqueue:queued:first prompt',
'enqueue:queued:second prompt',
'dequeue:first prompt',
'dequeue:second prompt',
])
expect(inbox.dequeued).toEqual(inbox.enqueued)
expect(inbox.discarded).toEqual([])
})
it('refuses acquisition when an accepted waking prompt still owns the next turn', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
prompt(agent, 'accepted first')
expect(agent.status).toBe('idle')
expect(agent.reserveTurnAdmission()).toBeUndefined()
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
})
it('refuses acquisition while a turn is running', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const reserved: unknown[] = []
ctx.on('agent/step', () => {
reserved.push(agent.reserveTurnAdmission())
})
prompt(agent, 'running')
await agent.whenIdle()
expect(agent.status).toBe('idle')
expect(reserved).toEqual([undefined])
})
it('refuses a second reservation and releases idempotently', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const release = agent.reserveTurnAdmission()
expect(agent.reserveTurnAdmission()).toBeUndefined()
prompt(agent, 'queued behind the reservation')
release?.()
release?.()
await agent.whenIdle()
expect(promptTexts(agent)).toEqual(['queued behind the reservation'])
expect(adapter.requests).toHaveLength(1)
const second = agent.reserveTurnAdmission()
expect(second).toBeDefined()
second?.()
})
it('ignores a stale release once a later reservation owns the boundary', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const stale = agent.reserveTurnAdmission()
stale?.()
const live = agent.reserveTurnAdmission()
prompt(agent, 'held by the live reservation')
stale?.()
await new Promise<void>((resolve) => { setTimeout(resolve, 5) })
expect(adapter.requests).toHaveLength(0)
live?.()
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
})
it('acquires beside quiet queued work and leaves it queued', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.send(createUserMessage({
content: [{ type: 'text', text: 'quiet' }],
source: { kind: 'user' },
}), {
target: 'next-turn',
wakeup: false,
})
const release = agent.reserveTurnAdmission()
expect(release).toBeDefined()
release?.()
await agent.whenIdle()
expect(adapter.requests).toHaveLength(0)
})
it('makes whenIdle() wait for release without spinning on a settled promise', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const machine = agent as Agent & { done: Promise<void> }
let backing = machine.done
let reads = 0
Object.defineProperty(agent, 'done', {
configurable: true,
get(): Promise<void> {
reads += 1
return backing
},
set(value: Promise<void>) {
backing = value
},
})
const release = agent.reserveTurnAdmission()
prompt(agent, 'waiting for the reservation')
let settled = false
const idle = agent.whenIdle().then(() => { settled = true })
for (let tick = 0; tick < 5; tick += 1) {
await new Promise<void>((resolve) => { setTimeout(resolve, 1) })
}
expect(settled).toBe(false)
expect(reads).toBeLessThanOrEqual(2)
release?.()
await idle
expect(settled).toBe(true)
expect(adapter.requests).toHaveLength(1)
})
it('resolves whenIdle() after release with nothing queued', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const release = agent.reserveTurnAdmission()
let settled = false
const idle = agent.whenIdle().then(() => { settled = true })
await new Promise<void>((resolve) => { setTimeout(resolve, 5) })
expect(settled).toBe(false)
release?.()
await idle
expect(agent.status).toBe('idle')
})
it('lets cancellation discard held prompts and keeps the boundary quiet', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const inbox = recordInbox(ctx)
const release = agent.reserveTurnAdmission()
prompt(agent, 'discarded while held')
agent.cancel({ kind: 'user' })
expect(inbox.events).toEqual([
'enqueue:queued:discarded while held',
'discard:discarded while held',
])
expect(inbox.discarded).toEqual(inbox.enqueued)
expect(inbox.dequeued).toEqual([])
release?.()
await agent.whenIdle()
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events).toHaveLength(0)
})
it('disposes the agent without waiting for the reservation to be released', async () => {
const adapter = new MockAdapter([])
const ctx = await harness(adapter)
const handle = await ctx.agents.create({
sessionId: SessionId('a1'),
agentOptions: { provider: 'mock', model: 'mock' },
})
const { agent } = handle
const release = agent.reserveTurnAdmission()
prompt(agent, 'discarded by disposal')
await handle.dispose()
expect(ctx.agents.list()).toEqual([])
expect(adapter.requests).toHaveLength(0)
release?.()
})
})