Merge remote-tracking branch 'origin/master' into session-surface
Reconcile the session-surface feature with master's package reorg and simplifications: - Adopt master's folded usage (assistant/message.usage; standalone `usage` event dropped) and re-attach surface metadata (surfaceOp/sourceEventSeqs). - Add surface opts to master's new max-tokens assistant/message append. - Port surface columns onto the coordinator-refactored SQLite backend at its new path; drop the dead v1->v2 migration (bump-and-reject, no migration per pre-release policy). - Move the session-surface RFC into implemented/architecture/ and refresh its stale body (no migration, SESSION_FORMAT_VERSION=0, renamed package paths). - Update the core-data-structures catalog SessionEvent blocks for the two new surface fields; regenerate the cordis catalog. - Re-harvest ACP snapshot fixtures (keyless replay) to carry surface metadata.
This commit is contained in:
Hypatia May
433
packages/core/agent-loop/tests/agent.spec.ts
Normal file
433
packages/core/agent-loop/tests/agent.spec.ts
Normal file
@@ -0,0 +1,433 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import LlmService 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 AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
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 waitForStatus(ctx: Context, agent: ReactLoopAgent, expected: ReactLoopAgent['status']): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === expected) {
|
||||
dispose()
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function send(agent: ReactLoopAgent, text: string) {
|
||||
agent.send([{ type: 'text', text }])
|
||||
}
|
||||
|
||||
describe('ReactLoopAgent', () => {
|
||||
it('send() throws after disposal', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await fiber.dispose()
|
||||
await agent.done
|
||||
|
||||
expect(() => { agent.send([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
|
||||
})
|
||||
|
||||
it('steer() throws after disposal', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await fiber.dispose()
|
||||
await agent.done
|
||||
|
||||
expect(() => { agent.steer([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
|
||||
})
|
||||
|
||||
it('inject() throws after disposal', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await fiber.dispose()
|
||||
await agent.done
|
||||
|
||||
expect(() => { agent.inject([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
|
||||
})
|
||||
|
||||
it('inject() decides enclosure from the LOG (open turn), not agent status', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// Simulate an OPEN turn in the log while the agent is idle (status is not a
|
||||
// reliable open-turn signal). inject must append into that open turn, NOT
|
||||
// wrap a new one.
|
||||
agent.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
agent.inject([{ type: 'text', text: 'mid' }], { source: { kind: 'plugin', plugin: 'p' } })
|
||||
expect(agent.session.events.filter(e => e.type === 'turn/start')).toHaveLength(1)
|
||||
expect(agent.session.events.at(-1)!.type).toBe('context/message')
|
||||
|
||||
// Close the turn; now inject must wrap its own one-shot injection turn.
|
||||
agent.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
agent.inject([{ type: 'text', text: 'after' }], { source: { kind: 'plugin', plugin: 'p' } })
|
||||
const starts = agent.session.events.filter(e => e.type === 'turn/start')
|
||||
expect(starts).toHaveLength(2)
|
||||
const last = starts[1]!
|
||||
expect(last.type === 'turn/start' && last.data.trigger.kind).toBe('injection')
|
||||
expect(agent.session.events.at(-1)!.type).toBe('turn/end') // turn-enclosed
|
||||
})
|
||||
|
||||
it('idle inject() contains a failing flush (logs, does not throw into the caller)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
// A persistence-like listener whose flush rejects.
|
||||
ctx.on('session/flush', () => { throw new Error('disk gone') })
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// inject() is synchronous and fires a fire-and-forget flush; a rejecting
|
||||
// flush must be contained (logged), never thrown into the caller.
|
||||
expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
|
||||
await new Promise(r => setTimeout(r, 20)) // let the contained flush settle
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed'))
|
||||
warn.mockRestore()
|
||||
})
|
||||
|
||||
it('idle inject() closes its one-shot turn AND still checkpoints even if the append throws', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
let flushes = 0
|
||||
ctx.on('session/flush', () => { flushes += 1 })
|
||||
|
||||
// Non-serializable injected content makes Session.append throw AFTER
|
||||
// turn/start was recorded. The turn/end must still be appended (finally),
|
||||
// AND the durability checkpoint must still fire — the balanced turn is in
|
||||
// memory and a crash before the next turn/dispose would otherwise lose it.
|
||||
expect(() => {
|
||||
agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
|
||||
}).toThrow(/non-JSON-serializable/)
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
expect(types).toEqual(['turn/start', 'turn/end']) // balanced, no open turn
|
||||
await new Promise(r => setTimeout(r, 10)) // let the fire-and-forget flush run
|
||||
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(AgentId('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(AgentId('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)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// A non-serializable source makes the turn/start append throw BEFORE the
|
||||
// event is pushed (Session.append validates before push), so NO turn opens.
|
||||
// The finally's isTurnOpen() guard sees no open turn and appends nothing —
|
||||
// the log stays empty, not left with a dangling turn/start.
|
||||
expect(() => {
|
||||
agent.inject([{ type: 'text', text: 'x' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
|
||||
}).toThrow(/non-JSON-serializable/)
|
||||
expect(agent.session.events).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('steer() when idle falls through to send() and starts a turn', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// steer while idle delegates to send
|
||||
agent.steer([{ type: 'text', text: 'steer idle' }], { source: { kind: 'plugin', plugin: 'test' } })
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The message was recorded as a user-level message (send path)
|
||||
expect(agent.session.events.some(e => e.type === 'user/message')).toBe(true)
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('disposer is idempotent (double-stop)', async () => {
|
||||
// Create a bare ReactLoopAgent and call start() directly to get the disposer.
|
||||
// Then call it twice — the second call hits the early-return branch.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('test'))
|
||||
const agent = new ReactLoopAgent(ctx, AgentId('bare'), { model: 'mock' }, session)
|
||||
|
||||
// Start the loop to get the disposer; the agent waits for messages
|
||||
// (idle, never-resolving cancel), so it will stay idle.
|
||||
const dispose = agent.start()
|
||||
|
||||
// First dispose
|
||||
dispose()
|
||||
expect(agent.status).toBe('disposed')
|
||||
|
||||
// Second dispose — idempotent, no throw
|
||||
expect(() => { dispose() }).not.toThrow()
|
||||
expect(agent.status).toBe('disposed')
|
||||
})
|
||||
|
||||
it('setting the same status does not emit agent/status again', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const statuses: string[] = []
|
||||
ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent) statuses.push(status)
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// After the turn, agent is idle. Send again to trigger another attempt
|
||||
// to go idle — but it's already idle, so no emission.
|
||||
const idleTransitionCount = statuses.filter(s => s === 'idle').length
|
||||
expect(idleTransitionCount).toBe(1) // only the final transition from running
|
||||
})
|
||||
|
||||
it('whenIdle() resolves immediately when the agent is not running', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// Fresh agent is idle — whenIdle() takes the not-running fast path and
|
||||
// resolves without subscribing. await must not hang.
|
||||
await agent.whenIdle()
|
||||
expect(agent.status).not.toBe('running')
|
||||
})
|
||||
|
||||
it('whenIdle() waits for queued work that has not flipped status yet', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'queued')
|
||||
let settled = false
|
||||
const idle = agent.whenIdle().then(() => { settled = true })
|
||||
await Promise.resolve()
|
||||
expect(settled).toBe(false)
|
||||
|
||||
await waitForStatus(ctx, agent, 'running')
|
||||
agent.cancel('done')
|
||||
await idle
|
||||
expect(settled).toBe(true)
|
||||
expect(agent.status).toBe('idle')
|
||||
})
|
||||
|
||||
it('whenIdle() awaits the running→idle transition, ignoring other subjects/running events', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok'), textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
const other = ctx.agentLoop.create(AgentId('a2'), { model: 'mock' })
|
||||
|
||||
// Drive `agent` into `running`, then await whenIdle() — it subscribes to
|
||||
// agent/status and resolves on the first transition out of running.
|
||||
const running = new Promise<void>((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'running') { dispose(); resolve() }
|
||||
})
|
||||
})
|
||||
send(agent, 'go')
|
||||
await running
|
||||
expect(agent.status).toBe('running')
|
||||
|
||||
// While `agent`'s whenIdle is pending, churn `other` through running→idle:
|
||||
// every status event it emits hits whenIdle's guard with `subject !== this`,
|
||||
// so the wait must ignore them and only resolve on `agent`'s own idle.
|
||||
send(other, 'go')
|
||||
|
||||
await agent.whenIdle()
|
||||
expect(agent.status).toBe('idle')
|
||||
})
|
||||
|
||||
it('whenIdle() subscribed while running resolves via done when the agent is then disposed', async () => {
|
||||
// Covers the waiter's disposed arm: whenIdle() queues an internal waiter
|
||||
// while running (not the fast path), then the disposer settles it and chains
|
||||
// `done` (loop exit), not an eager resolve. A bare ReactLoopAgent + direct
|
||||
// start() disposer keeps the emit synchronous.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const session = ctx.sessions.create(SessionId('bare'))
|
||||
const agent = new ReactLoopAgent(ctx, AgentId('bare'), { model: 'mock' }, session)
|
||||
const dispose = agent.start()
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
|
||||
const idle = agent.whenIdle() // queues an internal waiter (running)
|
||||
dispose() // settles the waiter synchronously; whenIdle chains done
|
||||
await idle
|
||||
expect(agent.status).toBe('disposed')
|
||||
await agent.done
|
||||
})
|
||||
|
||||
it('whenIdle() subscribed while running survives a FIBER dispose (no hung promise)', async () => {
|
||||
// The waiter is internal agent state, NOT an effect-scoped ctx.on listener:
|
||||
// disposing the OWNING fiber runs the agent's listener disposers, which would
|
||||
// have dropped a ctx.on-based waiter before the 'disposed' transition and
|
||||
// hung the promise. With internal waiters, the fiber disposer still settles
|
||||
// it. Regression for the round-3 whenIdle finding.
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
|
||||
const idle = agent.whenIdle() // queued while running
|
||||
await fiber.dispose() // tears the fiber down (drops agent listeners)
|
||||
await idle // must resolve, not hang
|
||||
expect(agent.status).toBe('disposed')
|
||||
})
|
||||
|
||||
it('whenIdle() on a disposed agent awaits the loop exit (done), not just the status flip', async () => {
|
||||
// The disposer emits agent/status('disposed') BEFORE the driver loop
|
||||
// unwinds, so whenIdle() must chain `done` (true quiescence) on the
|
||||
// disposed path. Dispose a running agent, then assert whenIdle() resolves
|
||||
// only after `done` — i.e. the loop has actually exited.
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
|
||||
let doneResolved = false
|
||||
void agent.done.then(() => { doneResolved = true })
|
||||
await fiber.dispose() // sets status disposed, aborts, drains the loop
|
||||
expect(agent.status).toBe('disposed')
|
||||
|
||||
// whenIdle() must not resolve before `done` has — chaining `done` is the
|
||||
// quiescence guarantee. By here dispose() awaited the loop, so done is
|
||||
// settled; whenIdle resolves and done is observed resolved.
|
||||
await agent.whenIdle()
|
||||
expect(doneResolved).toBe(true)
|
||||
})
|
||||
|
||||
it('contains a throwing agent/status listener on the running transition', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
ctx.on('agent/status', (_subject, status) => {
|
||||
if (status === 'running') throw new Error('bad running listener')
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.status).toBe('idle')
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/status listener threw on running'))
|
||||
warn.mockRestore()
|
||||
})
|
||||
|
||||
it('contains a throwing agent/status listener on the idle transition', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
ctx.on('agent/status', (_subject, status) => {
|
||||
if (status === 'idle') throw new Error('bad idle listener')
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await agent.whenIdle()
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(agent.status).toBe('idle')
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/status listener threw on idle'))
|
||||
warn.mockRestore()
|
||||
})
|
||||
})
|
||||
338
packages/core/agent-loop/tests/cancel.spec.ts
Normal file
338
packages/core/agent-loop/tests/cancel.spec.ts
Normal file
@@ -0,0 +1,338 @@
|
||||
/**
|
||||
* Tests for the queue-aware `Agent.cancel()` primitive. `cancel()` is the
|
||||
* broad verb — it clears queued + steering work, aborts an in-flight step, and
|
||||
* drops a turn about to start — whereas a bare step abort (the loop's private
|
||||
* `AbortController`) kills only the current step and leaves the queue intact.
|
||||
* These tests exercise every window where a cancel can land (idle, pre-step,
|
||||
* mid-step, continuation) and the marker's arm/reset rules that keep a cancel
|
||||
* from leaking to a later prompt or hanging `whenIdle()`.
|
||||
*
|
||||
* @module dsh-agent-loop/tests/cancel
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
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 send(agent: ReactLoopAgent, text: string) {
|
||||
agent.send([{ type: 'text', text }])
|
||||
}
|
||||
|
||||
/** Resolve on the agent's next idle transition (event-based, not status poll). */
|
||||
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() }
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/** All user-message texts recorded in the log (to assert what actually ran). */
|
||||
function userTexts(agent: ReactLoopAgent): string[] {
|
||||
return agent.session.events
|
||||
.filter(e => e.type === 'user/message')
|
||||
.flatMap(e => e.type === 'user/message' ? e.data.content : [])
|
||||
.flatMap(b => b.type === 'text' ? [b.text] : [])
|
||||
}
|
||||
|
||||
describe('Agent.cancel()', () => {
|
||||
it('cancel() on an idle agent with nothing queued is a no-op; the next prompt runs (F2 leak guard)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('reply')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// The loop is parked at the idle wait with nothing queued. A cancel here must
|
||||
// NOT arm the marker — otherwise the next legitimate prompt would be dropped.
|
||||
agent.cancel('nothing to cancel')
|
||||
|
||||
send(agent, 'real prompt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The prompt ran: its user message is in the log and one turn completed.
|
||||
expect(userTexts(agent)).toEqual(['real prompt'])
|
||||
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
|
||||
})
|
||||
|
||||
it('pre-step cancel drops the about-to-start turn (no turn is opened)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not run')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// send() 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')
|
||||
agent.cancel('pre-step')
|
||||
|
||||
// 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.status).toBe('idle')
|
||||
})
|
||||
|
||||
it('a whenIdle() waiter registered BEFORE a pre-step cancel resolves (F1 hang guard)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('x')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// Queue work, then register a whenIdle() waiter while in the pre-step window
|
||||
// (status idle, hasQueued true) — it does NOT take the fast path. Then cancel.
|
||||
// The skip path must settle this waiter directly (no running→idle transition
|
||||
// ever fires), or it would hang forever.
|
||||
send(agent, 'q')
|
||||
const idle = agent.whenIdle()
|
||||
agent.cancel('pre-step')
|
||||
|
||||
// Must resolve (not hang). A timeout makes the failure a clear test failure.
|
||||
await Promise.race([
|
||||
idle,
|
||||
new Promise((_r, reject) => setTimeout(() => { reject(new Error('whenIdle hung after pre-step cancel')) }, 1000)),
|
||||
])
|
||||
expect(agent.status).toBe('idle')
|
||||
})
|
||||
|
||||
it('cancel() mid-step aborts the in-flight model call; the turn ends aborted', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
agent.cancel('mid-step')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'mid-step' }])
|
||||
})
|
||||
|
||||
it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
agent.cancel() // no reason → default 'cancelled'
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'cancelled' }])
|
||||
})
|
||||
|
||||
it('a prompt sent AFTER a cancelled turn settles runs normally (marker reset)', async () => {
|
||||
const adapter = new MockAdapter(['hang', textResponse('second reply')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// First turn hangs; cancel it mid-step.
|
||||
send(agent, 'first')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
agent.cancel('cancel first')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// The marker must have been reset after the cancelled turn — a fresh prompt
|
||||
// runs to completion rather than being dropped by a stale marker.
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(userTexts(agent)).toContain('second')
|
||||
// The second turn completed (its reply was streamed).
|
||||
const reasons = agent.session.events.filter(e => e.type === 'turn/end')
|
||||
expect(reasons.length).toBe(2)
|
||||
})
|
||||
|
||||
it('cancel from a synchronous agent/turn-start 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.
|
||||
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')
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_a, _t, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
dispose()
|
||||
|
||||
// No step streamed (the model never ran), and the turn ended aborted with
|
||||
// the CALLER's reason — the marker carries `cancel(reason)` through even
|
||||
// though no AbortController observed it in this window.
|
||||
expect(streamed).toBe(false)
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from turn-start' }])
|
||||
})
|
||||
|
||||
it('cancel during the continuation window ends the turn aborted and runs no further step', async () => {
|
||||
// A continuation-waterfall listener cancels DURING the continuation decision
|
||||
// (the finished step's AbortController is already cleared), and votes to
|
||||
// continue — but the turn-scoped marker checked right after must end the turn
|
||||
// `aborted` and run NO second step.
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
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))
|
||||
|
||||
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 next()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Only ONE step ran (the second was cancelled in the continuation window),
|
||||
// and the turn ended aborted with the CALLER's reason (carried by the
|
||||
// marker, since the finished step's AbortController was already cleared).
|
||||
expect(steps).toBe(1)
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'from continuation' }])
|
||||
})
|
||||
|
||||
it('cancel from a synchronous agent/status(running) listener drops the turn (window 2)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not run')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
// setStatus('running') emits agent/status SYNCHRONOUSLY, so a running
|
||||
// listener can cancel in the gap between the loop's pre-step check and
|
||||
// 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 })
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'running') agent.cancel('from running listener')
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
dispose()
|
||||
|
||||
// No turn opened, no step streamed, and a later prompt still runs (the marker
|
||||
// was reset).
|
||||
expect(streamed).toBe(false)
|
||||
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
|
||||
})
|
||||
|
||||
it('window 2: whenIdle() does NOT resolve early when a running listener cancels then queues replacement work', async () => {
|
||||
// The window-1 early-resolve race has a window-2 twin: a synchronous
|
||||
// agent/status('running') listener cancels the about-to-run turn AND queues a
|
||||
// replacement. window 2 must NOT settle waiters (via setStatus('idle')) while
|
||||
// the replacement is still queued-and-unrun — it must fall through and run it,
|
||||
// so whenIdle() resolves on the replacement turn's running→idle, not before.
|
||||
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let replaced = false
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject !== agent || status !== 'running' || replaced) return
|
||||
replaced = true
|
||||
agent.cancel('drop A')
|
||||
send(agent, 'B')
|
||||
})
|
||||
|
||||
send(agent, 'A')
|
||||
const idle = agent.whenIdle()
|
||||
await idle
|
||||
dispose()
|
||||
|
||||
// whenIdle() resolved only AFTER B's turn ran: B's user message + a turn/end
|
||||
// are in the log, and A was dropped.
|
||||
expect(userTexts(agent)).toContain('B')
|
||||
expect(userTexts(agent)).not.toContain('A')
|
||||
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
|
||||
})
|
||||
|
||||
it('whenIdle() does NOT resolve early when a new prompt is queued during a pre-step cancel', async () => {
|
||||
// The subtle race: a whenIdle() waiter is registered for prompt A; cancel()
|
||||
// clears A; prompt B is queued BEFORE the loop resumes from the idle wait.
|
||||
// The window-1 cancel branch must NOT settle the waiter while B is still
|
||||
// queued-and-unrun — whenIdle() must wait for B's turn to actually run and
|
||||
// settle (the quiescence contract), not resolve before B's first event.
|
||||
const adapter = new MockAdapter([textResponse('A reply'), textResponse('B reply')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { 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('drop A') // arms marker, clears A
|
||||
send(agent, 'B') // B races in before the loop resumes
|
||||
|
||||
// whenIdle() must resolve only AFTER B's turn fully ran — by which point B's
|
||||
// user message and a turn/end are in the log. (Before the fix it resolved
|
||||
// immediately, with zero events, then B ran afterward.)
|
||||
await idle
|
||||
expect(userTexts(agent)).toContain('B')
|
||||
expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
|
||||
// A was dropped (never ran); only B's turn is recorded.
|
||||
expect(userTexts(agent)).not.toContain('A')
|
||||
})
|
||||
|
||||
it("cancel clears the turn's steering — it is not re-enqueued as a fresh turn", async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
// Steer (joins the running turn's steering FIFO), then cancel: the steering
|
||||
// must be dropped, NOT re-enqueued as a new queued turn.
|
||||
agent.steer([{ type: 'text', text: 'steer text' }])
|
||||
agent.cancel('cancel with steering')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// After the cancelled turn settles, the agent is idle with NO follow-up turn
|
||||
// started from the dropped steering.
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('idle')
|
||||
const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
|
||||
expect(turnStarts.length).toBe(1) // only the original (cancelled) turn
|
||||
// The steering text was dropped — it never reached the log.
|
||||
const flat = agent.session.events
|
||||
.filter(e => e.type === 'steering/message')
|
||||
.flatMap(e => e.type === 'steering/message' ? e.data.content : [])
|
||||
.flatMap(b => b.type === 'text' ? [b.text] : [])
|
||||
expect(flat).not.toContain('steer text')
|
||||
})
|
||||
})
|
||||
137
packages/core/agent-loop/tests/config-session-id.spec.ts
Normal file
137
packages/core/agent-loop/tests/config-session-id.spec.ts
Normal file
@@ -0,0 +1,137 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import LlmService 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 AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
const dirs: string[] = []
|
||||
afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
|
||||
|
||||
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() }
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
describe('config-driven session id', () => {
|
||||
it('config-driven create uses a fresh ${id}-session-<uuid> per run (restart-safe)', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-session-'))
|
||||
dirs.push(root)
|
||||
const idPattern = /^cfg-session-[0-9a-f-]{36}$/
|
||||
// Run 1: a config agent persists a turn under a generated session id.
|
||||
const ctx1 = new Context()
|
||||
await ctx1.plugin(LlmService)
|
||||
await ctx1.plugin(SessionStore)
|
||||
await ctx1.plugin(SystemPrompt)
|
||||
await ctx1.plugin(ToolRegistry)
|
||||
await ctx1.plugin(AgentRegistry)
|
||||
await ctx1.plugin(AgentLoop, { agents: [{ id: AgentId('cfg'), model: 'mock', systemPrompt: '' }] })
|
||||
await ctx1.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('cfg')]))
|
||||
const a1 = ctx1.agents.get(AgentId('cfg')) as ReactLoopAgent
|
||||
expect(a1.session.id).toMatch(idPattern)
|
||||
a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Run 2 over the SAME root: a fresh id means no on-disk collision (a fixed
|
||||
// ${id}-session would crash here with "already has a persisted log").
|
||||
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: [{ id: AgentId('cfg'), model: 'mock', systemPrompt: '' }] })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('cfg2')]))
|
||||
const a2 = ctx2.agents.get(AgentId('cfg')) as ReactLoopAgent
|
||||
expect(a2.session.id).toMatch(idPattern)
|
||||
expect(a2.session.id).not.toBe(a1.session.id)
|
||||
a2.send([{ type: 'text', text: 'q2' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx2, a2)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resumeSessionId continues a persisted session (env-var resume)', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-'))
|
||||
dirs.push(root)
|
||||
|
||||
// Run 1: a programmatically-created agent on a KNOWN session id persists a
|
||||
// completed turn, so run 2 has a concrete id to resume.
|
||||
const ctx1 = new Context()
|
||||
await ctx1.plugin(LlmService)
|
||||
await ctx1.plugin(SessionStore)
|
||||
await ctx1.plugin(SystemPrompt)
|
||||
await ctx1.plugin(ToolRegistry)
|
||||
await ctx1.plugin(AgentRegistry)
|
||||
await ctx1.plugin(AgentLoop, { agents: [] })
|
||||
await ctx1.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')]))
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('main'), sessionId: SessionId('sticky-1') }).agent as ReactLoopAgent
|
||||
a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Run 2: a CONFIG agent with resumeSessionId continues that session. The
|
||||
// resume is deferred until sessionPersistence loads (ctx.inject), so wait
|
||||
// for the agent to appear, then assert it is on the resumed id with history.
|
||||
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: [{ id: AgentId('main'), model: 'mock', systemPrompt: '', resumeSessionId: SessionId('sticky-1') }] })
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('second')]))
|
||||
|
||||
// The deferred resume runs on a microtask after the backend is available.
|
||||
let resumed: ReactLoopAgent | undefined
|
||||
for (let i = 0; i < 50 && !resumed; i++) {
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
resumed = ctx2.agents.get(AgentId('main')) as ReactLoopAgent | undefined
|
||||
}
|
||||
expect(resumed).toBeDefined()
|
||||
// The live session id IS the resumed id (NOT a fresh ${id}-session-<uuid>),
|
||||
// and the prior turn's user message is in the derived history.
|
||||
expect(resumed!.session.id).toBe('sticky-1')
|
||||
const derived = resumed!.session.deriveMessages()
|
||||
expect(JSON.stringify(derived)).toContain('remember me')
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('config-driven resume of a missing session is contained: logs a warning, no agent, no crash', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-miss-'))
|
||||
dirs.push(root)
|
||||
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: [{ id: AgentId('main'), model: 'mock', systemPrompt: '', resumeSessionId: SessionId('does-not-exist') }] })
|
||||
const warn = vi.spyOn((ctx.agentLoop as unknown as { ctx: { logger: { warn: (...a: unknown[]) => void } } }).ctx.logger, 'warn')
|
||||
.mockImplementation(() => undefined)
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')]))
|
||||
|
||||
// The deferred resume fails (no such session on disk). It must be contained:
|
||||
// a warning is logged, no 'main' agent is registered, and the app stays up.
|
||||
await new Promise(r => setTimeout(r, 200))
|
||||
expect(ctx.agents.get(AgentId('main'))).toBeUndefined()
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('config-driven resume of "does-not-exist" failed'))
|
||||
warn.mockRestore()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
332
packages/core/agent-loop/tests/coverage-edges.spec.ts
Normal file
332
packages/core/agent-loop/tests/coverage-edges.spec.ts
Normal file
@@ -0,0 +1,332 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId, LlmError, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { 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 AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
|
||||
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 }])
|
||||
}
|
||||
|
||||
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
|
||||
// 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(AgentId('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', () => {
|
||||
it('passes through non-JSON arguments string without crashing', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
// model emits tool-call with malformed arguments (not valid JSON)
|
||||
[
|
||||
{ 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: 'not json' } },
|
||||
{ type: 'finish' as const, reason: { kind: 'tool-calls' as const } },
|
||||
] satisfies StreamChunk[],
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: 'echo tool',
|
||||
parameters: { input: { type: 'string' } },
|
||||
async execute(args: unknown) {
|
||||
return [{ type: 'text', text: typeof args === 'string' ? `raw: ${args}` : JSON.stringify(args) }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'use tool')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// tool/call event should have recorded the raw arguments string
|
||||
const callEvent = agent.session.events.find(e => e.type === 'tool/call')
|
||||
expect(callEvent).toBeDefined()
|
||||
if (callEvent!.type === 'tool/call') {
|
||||
expect(callEvent!.data.arguments).toBe('not json')
|
||||
}
|
||||
// the loop did not crash — a result was produced
|
||||
expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
|
||||
})
|
||||
|
||||
it('uses empty object when tool-call arguments are empty string', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
[
|
||||
{ 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: 'noarg', arguments: '' } },
|
||||
{ type: 'finish' as const, reason: { kind: 'tool-calls' as const } },
|
||||
] satisfies StreamChunk[],
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'noarg',
|
||||
description: 'no-arg tool',
|
||||
parameters: {},
|
||||
async execute() {
|
||||
return [{ type: 'text', text: 'ran with empty args' }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'use tool')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('toError normalization', () => {
|
||||
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(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/turn-start', () => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw 'naked string error' // non-Error throw, normalized via toError
|
||||
}
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]!.message).toBe('naked string error')
|
||||
// A non-Error throw is wrapped in a HarnessError with code UNKNOWN, so the
|
||||
// turn-end error reason carries a routable code instead of degrading.
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error' && turnEnd.data.reason.code).toBe('UNKNOWN')
|
||||
})
|
||||
|
||||
it('normalizes non-Error throws from agent/request waterfall via inline toError in runStep catch', async () => {
|
||||
const adapter = new MockAdapter([textResponse('irrelevant')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, _options, _next) => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw { code: 500 } // non-Error throw, goes through runStep catch
|
||||
}
|
||||
return _next()
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(errors).toHaveLength(1)
|
||||
// String() of { code: 500 } is '[object Object]'
|
||||
expect(errors[0]!.message).toBe('[object Object]')
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error' && turnEnd.data.reason.code).toBe('UNKNOWN')
|
||||
})
|
||||
})
|
||||
|
||||
describe('coded error data emission', () => {
|
||||
it('errorData includes code when a coded error (LlmError) is thrown from a plugin', async () => {
|
||||
const adapter = new MockAdapter([textResponse('turn 1')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let threwOnce = false
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, _options, next) => {
|
||||
if (!threwOnce) {
|
||||
threwOnce = true
|
||||
throw new LlmError('server overloaded', 'RATE_LIMIT')
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]!.message).toBe('server overloaded')
|
||||
|
||||
// turn-end error reason includes the code
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd).toBeDefined()
|
||||
if (turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error') {
|
||||
expect(turnEnd.data.reason.code).toBe('RATE_LIMIT')
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('disposed vs aborted branching', () => {
|
||||
it('handles dispose during model streaming producing reason "disposed"', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
await fiber.dispose() // dispose during hang
|
||||
await agent.done
|
||||
|
||||
// The review-fixes test for 'HIGH: disposed status' already covers
|
||||
// this assertion path. The reason is 'disposed' because isDisposed() is
|
||||
// checked before the abort signal check in the error path.
|
||||
expect(reasons).toContainEqual({ kind: 'disposed' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('structured tool error propagation (the runtime-validation RFC, part 2)', () => {
|
||||
it('forwards a tool HarnessError onto the tool/result session event', async () => {
|
||||
const { HarnessError } = await import('@deepseek-ai/dsh-llm')
|
||||
// First model turn calls the tool; second turn (after the tool result is
|
||||
// fed back) ends with plain text so the loop settles.
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'boom', {}),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'boom',
|
||||
description: 'always fails',
|
||||
parameters: {},
|
||||
async execute() {
|
||||
throw new HarnessError('exploded', 'BOOM')
|
||||
},
|
||||
}))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const toolResult = agent.session.events.find(e => e.type === 'tool/result')
|
||||
expect(toolResult?.type === 'tool/result' && toolResult.data.isError).toBe(true)
|
||||
expect(toolResult?.type === 'tool/result' && toolResult.data.error)
|
||||
.toEqual({ name: 'HarnessError', code: 'BOOM' })
|
||||
})
|
||||
})
|
||||
110
packages/core/agent-loop/tests/inbox.spec.ts
Normal file
110
packages/core/agent-loop/tests/inbox.spec.ts
Normal file
@@ -0,0 +1,110 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Inbox } from '@deepseek-ai/dsh-agent-loop'
|
||||
|
||||
function resolverPair() {
|
||||
let r!: () => void
|
||||
const p = new Promise<void>((resolve) => { r = resolve })
|
||||
return { promise: p, resolve: r }
|
||||
}
|
||||
|
||||
describe('Inbox', () => {
|
||||
it('enqueues and drains queued messages in FIFO order', () => {
|
||||
const inbox = new Inbox()
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } })
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } })
|
||||
expect(inbox.hasQueued).toBe(true)
|
||||
|
||||
const drained = inbox.drainQueued()
|
||||
expect(drained).toHaveLength(2)
|
||||
expect(drained[0]!.content[0]).toMatchObject({ text: 'first' })
|
||||
expect(drained[1]!.content[0]).toMatchObject({ text: 'second' })
|
||||
expect(inbox.hasQueued).toBe(false)
|
||||
})
|
||||
|
||||
it('pushes and drains steering messages separately from queued', () => {
|
||||
const inbox = new Inbox()
|
||||
inbox.steer({ content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } })
|
||||
expect(inbox.hasQueued).toBe(false)
|
||||
expect(inbox.hasSteering).toBe(true)
|
||||
|
||||
const steering = inbox.drainSteering()
|
||||
expect(steering).toHaveLength(1)
|
||||
expect(inbox.hasSteering).toBe(false)
|
||||
})
|
||||
|
||||
it('waitForQueued returns immediately when a queued message is already present', async () => {
|
||||
const inbox = new Inbox()
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'ready' }], source: { kind: 'user' } })
|
||||
|
||||
const started = Date.now()
|
||||
await inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
|
||||
expect(Date.now() - started).toBeLessThan(50)
|
||||
})
|
||||
|
||||
it('waitForQueued resolves when a message is enqueued', async () => {
|
||||
const inbox = new Inbox()
|
||||
const waiter = inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
|
||||
// enqueue after starting the wait
|
||||
setTimeout(() => { inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } }) }, 5)
|
||||
await waiter
|
||||
})
|
||||
|
||||
it('waitForQueued resolves when the cancel promise resolves', async () => {
|
||||
const inbox = new Inbox()
|
||||
const { promise, resolve } = resolverPair()
|
||||
const waiter = inbox.waitForQueued(promise)
|
||||
resolve()
|
||||
await waiter
|
||||
})
|
||||
|
||||
it('waitForQueued overwrites the previous wakeup callback (only the latest waiter is notified)', async () => {
|
||||
const inbox = new Inbox()
|
||||
const { promise: p1, resolve: r1 } = resolverPair()
|
||||
|
||||
void inbox.waitForQueued(new Promise(() => {})) // first call, never resolved
|
||||
void inbox.waitForQueued(p1) // second call overwrites wakeup
|
||||
|
||||
// Cancel p1 (the latest waiter's cancel) — the wakeup was overwritten
|
||||
// to p1's resolve, so canceling p1 triggers the finally block which
|
||||
// clears the wakeup if it matches.
|
||||
r1()
|
||||
await p1
|
||||
|
||||
// Now enqueue: the first waiter's wakeup (which was overwritten) won't
|
||||
// fire, and the second waiter's wakeup was cleared by cancel.
|
||||
// The enqueue calls wakeup?.() but wakeup was cleared — no crash, no hang.
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
|
||||
// The overwrite path + finally cleanup are exercised
|
||||
})
|
||||
|
||||
it('clears wakeup in finally handler when enqueue resolves', async () => {
|
||||
const inbox = new Inbox()
|
||||
void inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
|
||||
// The wakeup is set. Now trigger it via enqueue → wakeup() calls resolve,
|
||||
// promise resolves, finally clears wakeup because wakeup === resolve.
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } })
|
||||
// No explicit await needed — enqueue is synchronous, and the microtask
|
||||
// (finally) runs. The key coverage hit is finally with wakeup === resolve.
|
||||
})
|
||||
|
||||
it('finally handler does not clear wakeup when a different waiter overwrote it', async () => {
|
||||
// First waiter's cancel resolves AFTER a second waiter overwrote wakeup.
|
||||
// First waiter's finally sees wakeup !== its resolve → does not clear.
|
||||
const inbox = new Inbox()
|
||||
const { promise: c1, resolve: r1 } = resolverPair()
|
||||
|
||||
void inbox.waitForQueued(c1) // wakeup = resolve1, c1.then(resolve1)
|
||||
void inbox.waitForQueued(new Promise(() => {})) // wakeup = resolve2, cancel never resolves
|
||||
|
||||
// Resolve c1 (the first cancel). c1.then(resolve1) fires → resolve1() called
|
||||
// → waiter1's promise resolves → finally: wakeup === resolve1? NO (it's resolve2)
|
||||
// → wakeup is NOT cleared.
|
||||
r1()
|
||||
await c1
|
||||
|
||||
// Now enqueue: wakeup() calls resolve2 → waiter2 resolves
|
||||
// But waiter2's cancel never resolves — that's fine, enqueue resolves it.
|
||||
inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
|
||||
// No need to await anything further — enqueue is synchronous wakeup
|
||||
})
|
||||
})
|
||||
704
packages/core/agent-loop/tests/loop.spec.ts
Normal file
704
packages/core/agent-loop/tests/loop.spec.ts
Normal file
@@ -0,0 +1,704 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService, { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { 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 AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, maxTokensResponse, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
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 }])
|
||||
}
|
||||
|
||||
describe('agent loop', () => {
|
||||
it('runs a simple turn: queued message → model → idle, with ordered events', async () => {
|
||||
const adapter = new MockAdapter([textResponse('hello there')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
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))
|
||||
}
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(order).toEqual(['agent/turn-start', 'agent/step-start', 'agent/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
|
||||
// it (every event is turn-enclosed), then the assembled message (carrying the
|
||||
// step's usage).
|
||||
expect(types[0]).toBe('turn/start')
|
||||
expect(types[1]).toBe('user/message')
|
||||
expect(types).toContain('assistant/message')
|
||||
const assistantMessage = agent.session.events.find(e => e.type === 'assistant/message')
|
||||
expect(assistantMessage?.type === 'assistant/message' && assistantMessage.data.usage).toEqual({ inputTokens: 10, outputTokens: 'hello there'.length })
|
||||
expect(types.at(-1)).toBe('turn/end')
|
||||
|
||||
// derived history: user + assistant
|
||||
const messages = agent.session.deriveMessages()
|
||||
expect(messages.map(m => m.role)).toEqual(['user', 'assistant'])
|
||||
expect(messages[1]!.content).toEqual([{ type: 'text', text: 'hello there' }])
|
||||
})
|
||||
|
||||
it('round-trips tool calls: model requests tool → executes → result in next request', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'ping' }, 'calling echo'),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: 'echo back',
|
||||
parameters: { text: { type: 'string' } },
|
||||
async execute(args) {
|
||||
return [{ type: 'text', text: `echo: ${args.text}` }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'use the tool')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// two model calls happened (tool-call step, then final step)
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
|
||||
// the second request's derived history contains the tool result
|
||||
const secondMessages = adapter.requests[1]!.messages
|
||||
const toolResultMessage = secondMessages.find(m =>
|
||||
m.content.some(b => b.type === 'tool-result'))
|
||||
expect(toolResultMessage).toBeDefined()
|
||||
const block = toolResultMessage!.content.find(b => b.type === 'tool-result')!
|
||||
expect(block).toMatchObject({ toolCallId: 'c1', isError: false })
|
||||
expect((block).content).toEqual([{ type: 'text', text: 'echo: ping' }])
|
||||
|
||||
// session log records call + result
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
expect(types).toContain('tool/call')
|
||||
expect(types).toContain('tool/result')
|
||||
})
|
||||
|
||||
it('passes assembled system prompt and tool schemas into the request', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.systemPrompt.section({ name: 'persona', order: 0, text: 'You are a test agent.' })
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'noop',
|
||||
description: 'does nothing',
|
||||
parameters: {},
|
||||
async execute() {
|
||||
return []
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock', systemPrompt: 'Agent-specific suffix.' })
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const request = adapter.requests[0]
|
||||
expect(request!.system).toBe('You are a test agent.\n\nAgent-specific suffix.')
|
||||
expect(request!.tools?.map(t => t.name)).toEqual(['noop'])
|
||||
})
|
||||
|
||||
it('records raw chunks for replay and emits agent/stream-chunk', 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] : [])
|
||||
.filter((c: StreamChunk): c is Extract<StreamChunk, { type: 'text-delta' }> => c.type === 'text-delta')
|
||||
.map(c => c.text)
|
||||
.join('')
|
||||
expect(deltaText).toBe('abc')
|
||||
})
|
||||
|
||||
it('injects steering between steps and continues the turn', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'slow', {}),
|
||||
textResponse('addressed the steering'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'slow',
|
||||
description: '',
|
||||
parameters: {},
|
||||
async execute() {
|
||||
// steer while the turn is running (during tool execution)
|
||||
agent.steer([{ type: 'text', text: 'change of plans' }])
|
||||
return [{ type: 'text', text: 'tool done' }]
|
||||
},
|
||||
}))
|
||||
|
||||
send(agent, 'start')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const types = agent.session.events.map(e => e.type)
|
||||
expect(types).toContain('steering/message')
|
||||
// steering recorded before the second step's request derived its history
|
||||
const steeringSeq = agent.session.events.find(e => e.type === 'steering/message')!.seq
|
||||
const secondStepStart = agent.session.events.filter(e => e.type === 'step/start')[1]
|
||||
expect(secondStepStart).toBeDefined()
|
||||
expect(steeringSeq).toBeLessThan(secondStepStart!.seq)
|
||||
|
||||
// the second model request saw the steering content
|
||||
const secondRequest = adapter.requests[1]
|
||||
const flat = JSON.stringify(secondRequest!.messages)
|
||||
expect(flat).toContain('change of plans')
|
||||
})
|
||||
|
||||
it('steering while idle behaves like send (starts a turn)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
agent.steer([{ type: 'text', text: 'hello' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(agent.session.events.some(e => e.type === 'user/message')).toBe(true)
|
||||
})
|
||||
|
||||
it('inject() while idle wraps context in a one-shot turn, visible to the next request', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
agent.inject([{ type: 'text', text: 'file changed: a.ts' }], { source: { kind: 'plugin', plugin: 'watcher' } })
|
||||
// The idle inject records a self-contained turn (turn/start → context/message
|
||||
// → turn/end) so the event stays turn-enclosed, but does NOT run the model.
|
||||
await new Promise(r => setTimeout(r, 20))
|
||||
expect(agent.status).toBe('idle')
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
const injectedTurn = agent.session.events.filter(e => e.type === 'turn/start')
|
||||
expect(injectedTurn).toHaveLength(1)
|
||||
const it0 = injectedTurn[0]!
|
||||
expect(it0.type === 'turn/start' && it0.data.trigger.kind).toBe('injection')
|
||||
expect(agent.session.events.at(-1)!.type).toBe('turn/end') // turn-enclosed
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
const flat = JSON.stringify(adapter.requests[0]!.messages)
|
||||
expect(flat).toContain('file changed: a.ts')
|
||||
expect(flat).toContain('<context source=\\"plugin\\">')
|
||||
})
|
||||
|
||||
it('inject() while running appends into the open turn (no extra synthetic turn)', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'noticer', {}, 'calling'),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
// A tool that injects mid-execution: at this point the agent is running, so
|
||||
// inject must append the context/message into the ALREADY-open turn rather
|
||||
// than wrap it in its own one-shot turn.
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'noticer',
|
||||
description: 'injects a notice',
|
||||
parameters: {},
|
||||
async execute() {
|
||||
agent.inject([{ type: 'text', text: 'mid-turn notice' }], { source: { kind: 'plugin', plugin: 'x' } })
|
||||
return [{ type: 'text', text: 'ok' }]
|
||||
},
|
||||
}))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
// Exactly ONE turn ran (no synthetic injection turn), and the mid-turn
|
||||
// context/message sits inside it.
|
||||
const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
|
||||
expect(turnStarts).toHaveLength(1)
|
||||
const ts0 = turnStarts[0]!
|
||||
expect(ts0.type === 'turn/start' && ts0.data.trigger.kind).toBe('message')
|
||||
expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
|
||||
})
|
||||
|
||||
it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {
|
||||
// force-continue: model never calls tools, but a plugin forces 3 steps
|
||||
const adapter = new MockAdapter([
|
||||
textResponse('step 1'),
|
||||
textResponse('step 2'),
|
||||
textResponse('step 3'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steps = 0
|
||||
ctx.on('agent/step-end', () => void steps++)
|
||||
ctx.on('agent/turn-continuation', async (_agent, _turn, _defaultDecision, next) => {
|
||||
if (steps < 3) return true
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(steps).toBe(3)
|
||||
expect(adapter.requests).toHaveLength(3)
|
||||
})
|
||||
|
||||
it('agent/turn-continuation can veto continuation despite tool calls (budget-guard pattern)', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'echo', { text: 'x' })])
|
||||
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' })
|
||||
|
||||
ctx.on('agent/turn-continuation', async () => false as const)
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
// only one model call despite the tool call requesting a follow-up
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
// tool still executed before the decision
|
||||
expect(agent.session.events.some(e => e.type === 'tool/result')).toBe(true)
|
||||
})
|
||||
|
||||
it('agent/request waterfall can rewrite the request (model-switch pattern)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.llm.registerAdapter(['other-model'], adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
ctx.on('agent/request', async (_agent, _turn, _step, options, next) => {
|
||||
options.model = 'other-model'
|
||||
return next()
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests[0]!.model).toBe('other-model')
|
||||
})
|
||||
|
||||
it('cancel() mid-stream ends the turn with reason aborted', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
// wait until the stream is hanging, then cancel
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
agent.cancel('user interrupt')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
|
||||
})
|
||||
|
||||
it('surfaces max-tokens as the turn-end reason when the last step is cut off', async () => {
|
||||
// A single step that ends with a max-tokens finish (no tool calls): the
|
||||
// turn stops by default and ends max-tokens, not completed.
|
||||
const adapter = new MockAdapter([maxTokensResponse('truncat')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
expect(reasons).toEqual([{ kind: 'max-tokens' }])
|
||||
// and the reason is recorded in the log's turn/end event
|
||||
const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
|
||||
expect(turnEnd!.data.reason).toEqual({ kind: 'max-tokens' })
|
||||
})
|
||||
|
||||
it('a max-tokens step earlier in a turn still surfaces as max-tokens after a later completed step', async () => {
|
||||
// Step 1 is cut off (max-tokens, no tool calls → would stop by default), so
|
||||
// continuation must be FORCED to reach step 2 which finishes normally
|
||||
// (stop). The rule "any max-tokens step surfaces as max-tokens" means the
|
||||
// turn ends max-tokens even though the LAST step completed cleanly.
|
||||
const adapter = new MockAdapter([
|
||||
maxTokensResponse('first half'),
|
||||
textResponse('second half'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let steps = 0
|
||||
ctx.on('agent/step-end', () => void 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
|
||||
return next()
|
||||
})
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(steps).toBe(2)
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
expect(adapter.requests[1]!.messages).toEqual([
|
||||
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: 'first half' }] },
|
||||
])
|
||||
expect(reasons).toEqual([{ kind: 'max-tokens' }])
|
||||
})
|
||||
|
||||
it('a completed step after no max-tokens keeps the turn completed (max-tokens does not leak across turns)', async () => {
|
||||
// Two consecutive turns: turn 1 is cut off (max-tokens), turn 2 is a clean
|
||||
// stop. The per-turn reason must be independent — turn 2 ends completed.
|
||||
const adapter = new MockAdapter([maxTokensResponse('cut'), textResponse('clean')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'max-tokens' }, { kind: 'completed' }])
|
||||
})
|
||||
|
||||
it('does not dispatch tool calls from a max-tokens-truncated step', async () => {
|
||||
const callId = CallId('c1')
|
||||
const adapter = new MockAdapter([[
|
||||
{ type: 'block-start', index: 0, blockType: 'tool-call' },
|
||||
{ type: 'tool-call-delta', index: 0, id: callId, name: 'echo', argumentsDelta: '{"text":"x"}' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: callId, name: 'echo', arguments: '{"text":"x"}' } },
|
||||
{ type: 'usage', usage: { inputTokens: 10, outputTokens: 5 } },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } },
|
||||
]])
|
||||
const ctx = await harness(adapter)
|
||||
let executions = 0
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: '',
|
||||
parameters: { text: { type: 'string' } },
|
||||
async execute() {
|
||||
executions += 1
|
||||
return [{ type: 'text', text: 'should not run' }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(executions).toBe(0)
|
||||
expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
|
||||
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
|
||||
expect(reasons).toEqual([{ kind: 'max-tokens' }])
|
||||
// No-data-loss: a max-tokens step whose only content was a dropped tool call
|
||||
// has EMPTY assistant content, but its usage must still be represented. It
|
||||
// rides on an (empty-content) assistant/message — there is no standalone
|
||||
// usage event — and that empty message is skipped by deriveMessages(), so
|
||||
// the derived history above is NOT corrupted by a spurious assistant turn.
|
||||
const assistantMessage = agent.session.events.find(e => e.type === 'assistant/message')
|
||||
expect(assistantMessage?.type === 'assistant/message' && assistantMessage.data).toEqual({
|
||||
turn: 1, step: 1, content: [], usage: { inputTokens: 10, outputTokens: 5 },
|
||||
})
|
||||
})
|
||||
|
||||
it('appends no assistant/message for a max-tokens step with empty content and no usage', async () => {
|
||||
// A max-tokens step truncated to a dropped tool call AND with no usage chunk
|
||||
// has nothing to record: empty content and no accounting → no assistant/message
|
||||
// (the empty-content host exists only to carry usage). The turn still ends
|
||||
// max-tokens.
|
||||
const callId = CallId('c1')
|
||||
const adapter = new MockAdapter([[
|
||||
{ type: 'block-start', index: 0, blockType: 'tool-call' },
|
||||
{ type: 'tool-call-delta', index: 0, id: callId, name: 'echo', argumentsDelta: '{"text":"x"}' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: callId, name: 'echo', arguments: '{"text":"x"}' } },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } },
|
||||
]])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'echo',
|
||||
description: '',
|
||||
parameters: { text: { type: 'string' } },
|
||||
async execute() { return [{ type: 'text', text: 'should not run' }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'max-tokens' }])
|
||||
expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
|
||||
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
|
||||
})
|
||||
|
||||
it('appends no assistant/message for a normal stop finish with empty content and no usage', async () => {
|
||||
// A clean `stop` finish that streamed nothing assembled (no blocks) and
|
||||
// carried no usage chunk has nothing to record: the content-or-usage guard
|
||||
// on the normal step path suppresses a pure trace-only empty assistant/message.
|
||||
const adapter = new MockAdapter([[{ type: 'finish', reason: { kind: 'stop' } }]])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
const reasons: TurnEndReason[] = []
|
||||
ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason))
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(reasons).toEqual([{ kind: 'completed' }])
|
||||
expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
|
||||
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
|
||||
})
|
||||
|
||||
it('keeps safe max-tokens assistant content while dropping truncated tool calls', async () => {
|
||||
const callId = CallId('c1')
|
||||
const adapter = new MockAdapter([[
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'text-delta', index: 0, text: 'partial text' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text: 'partial text' } },
|
||||
{ type: 'block-start', index: 1, blockType: 'tool-call' },
|
||||
{ type: 'tool-call-delta', index: 1, id: callId, name: 'echo', argumentsDelta: '{"text"' },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } },
|
||||
]])
|
||||
const ctx = await harness(adapter)
|
||||
let stepResults = 0
|
||||
ctx.on('agent/step-result', async (_agent, _turn, _step, message, next) => {
|
||||
stepResults += 1
|
||||
expect(message.content).toEqual([{ type: 'text', text: 'partial text' }])
|
||||
return next()
|
||||
})
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(stepResults).toBe(1)
|
||||
expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
|
||||
expect(agent.session.deriveMessages()).toEqual([
|
||||
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
|
||||
{ role: 'assistant', content: [{ type: 'text', text: 'partial text' }] },
|
||||
])
|
||||
})
|
||||
|
||||
it('stops the turn when agent/step-end listener failure has recorded an error', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'x' }),
|
||||
textResponse('should not run'),
|
||||
])
|
||||
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 threw = false
|
||||
ctx.on('agent/step-end', () => {
|
||||
if (!threw) { threw = true; throw new Error('bad step-end listener') }
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('error')
|
||||
})
|
||||
|
||||
it('chains queued messages into consecutive turns', async () => {
|
||||
const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
|
||||
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))
|
||||
|
||||
// queue two messages while idle — first starts turn 1 immediately;
|
||||
// queue the second during turn 1 via a stream-chunk hook
|
||||
let queued = false
|
||||
ctx.on('agent/stream-chunk', () => {
|
||||
if (!queued) {
|
||||
queued = true
|
||||
send(agent, 'second message')
|
||||
}
|
||||
})
|
||||
|
||||
send(agent, 'first message')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(turns).toEqual([1, 2])
|
||||
expect(adapter.requests).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('awaits session/flush at turn end (persistence checkpoint)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('ok')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
let flushed = 0
|
||||
let flushedBeforeIdle = false
|
||||
ctx.on('session/flush', async (session) => {
|
||||
await new Promise(r => setTimeout(r, 10))
|
||||
flushed++
|
||||
flushedBeforeIdle = agent.status !== 'idle'
|
||||
void session
|
||||
})
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(flushed).toBe(1)
|
||||
expect(flushedBeforeIdle).toBe(true)
|
||||
})
|
||||
|
||||
it('errors from the model surface as agent/error and end the turn', async () => {
|
||||
const adapter = new MockAdapter([]) // script exhausted → throws
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
|
||||
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))
|
||||
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(errors).toHaveLength(1)
|
||||
expect(errors[0]!.message).toContain('script exhausted')
|
||||
expect(reasons[0]).toMatchObject({ kind: 'error' })
|
||||
// The durable failure lives entirely on turn/end.reason (with the failing
|
||||
// step), not a standalone error event.
|
||||
const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
|
||||
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({ kind: 'error', step: 1 })
|
||||
})
|
||||
|
||||
it('disposing the loop fiber mid-turn stops the loop (HMR safety)', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
let agent!: ReactLoopAgent
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
agent = inner.agentLoop.create(AgentId('scoped'), { model: 'mock' })
|
||||
}, { inject: ['agentLoop'] }))
|
||||
|
||||
expect(ctx.agents.get(AgentId('scoped'))).toBe(agent)
|
||||
send(agent, 'go')
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
expect(agent.status).toBe('running')
|
||||
|
||||
await fiber.dispose()
|
||||
await agent.done
|
||||
|
||||
expect(agent.status).toBe('disposed')
|
||||
expect(ctx.agents.get(AgentId('scoped'))).toBeUndefined()
|
||||
expect(() => { send(agent, 'too late') }).toThrow('disposed')
|
||||
})
|
||||
|
||||
it('creates agents from config on startup', async () => {
|
||||
const adapter = new MockAdapter([textResponse('from config')])
|
||||
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: [{ id: AgentId('config-agent'), model: 'mock', systemPrompt: 'Config prompt' }],
|
||||
})
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
|
||||
const agent = ctx.agents.get(AgentId('config-agent'))! as ReactLoopAgent
|
||||
expect(agent).toBeDefined()
|
||||
expect(agent.id).toBe('config-agent')
|
||||
expect(agent.options.model).toBe('mock')
|
||||
|
||||
// the agent is alive: send triggers a turn
|
||||
send(agent, 'hi')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(adapter.requests).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('replays a session log into an identical derived history', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
toolCallResponse('c1', 'echo', { text: 'x' }),
|
||||
textResponse('done'),
|
||||
])
|
||||
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' })
|
||||
send(agent, 'run')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const replayed = ctx.sessions.create(SessionId('replayed'), { seed: [...agent.session.events] })
|
||||
expect(replayed.deriveMessages()).toEqual(agent.session.deriveMessages())
|
||||
// event-by-event identity of types
|
||||
expect(replayed.events.map(e => e.type)).toEqual(
|
||||
agent.session.events.map(e => e.type))
|
||||
})
|
||||
})
|
||||
90
packages/core/agent-loop/tests/mock-adapter.ts
Normal file
90
packages/core/agent-loop/tests/mock-adapter.ts
Normal file
@@ -0,0 +1,90 @@
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** Helpers to write scripted responses tersely. */
|
||||
export function textResponse(text: string): StreamChunk[] {
|
||||
return [
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
...Array.from(text, (char): StreamChunk => ({ type: 'text-delta', index: 0, text: char })),
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text } },
|
||||
{ type: 'usage', usage: { inputTokens: 10, outputTokens: text.length } },
|
||||
{ type: 'finish', reason: { kind: 'stop' } },
|
||||
]
|
||||
}
|
||||
|
||||
/**
|
||||
* Like {@link textResponse} but the stream ends with a `max-tokens` finish —
|
||||
* the model was cut off at the output-token ceiling (DeepSeek's `length`).
|
||||
* Used to exercise the turn-end `max-tokens` surfacing rule.
|
||||
*/
|
||||
export function maxTokensResponse(text: string): StreamChunk[] {
|
||||
return [
|
||||
{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
...Array.from(text, (char): StreamChunk => ({ type: 'text-delta', index: 0, text: char })),
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text } },
|
||||
{ type: 'usage', usage: { inputTokens: 10, outputTokens: text.length } },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } },
|
||||
]
|
||||
}
|
||||
|
||||
export function toolCallResponse(rawCallId: string, name: string, args: object, text?: string): StreamChunk[] {
|
||||
const callId = CallId(rawCallId)
|
||||
const argumentsJson = JSON.stringify(args)
|
||||
const chunks: StreamChunk[] = []
|
||||
let index = 0
|
||||
if (text) {
|
||||
chunks.push(
|
||||
{ type: 'block-start', index, blockType: 'text' },
|
||||
{ type: 'text-delta', index, text },
|
||||
{ type: 'block-end', index, block: { type: 'text', text } },
|
||||
)
|
||||
index += 1
|
||||
}
|
||||
chunks.push(
|
||||
{ type: 'block-start', index, blockType: 'tool-call' },
|
||||
{ type: 'tool-call-delta', index, id: callId, name, argumentsDelta: argumentsJson.slice(0, 5) },
|
||||
{ type: 'tool-call-delta', index, id: callId, argumentsDelta: argumentsJson.slice(5) },
|
||||
{
|
||||
type: 'block-end',
|
||||
index,
|
||||
block: { type: 'tool-call', id: callId, name, arguments: argumentsJson },
|
||||
},
|
||||
{ type: 'usage', usage: { inputTokens: 10, outputTokens: 5 } },
|
||||
{ type: 'finish', reason: { kind: 'tool-calls' } },
|
||||
)
|
||||
return chunks
|
||||
}
|
||||
|
||||
/**
|
||||
* Mock adapter driven by a script: each model call consumes the next entry.
|
||||
* Records every request it receives for assertions. An entry may be a
|
||||
* function to compute chunks from the request, or a 'hang' marker that
|
||||
* streams one chunk then waits until aborted.
|
||||
*/
|
||||
export class MockAdapter extends LlmAdapter {
|
||||
requests: GenerateOptions[] = []
|
||||
|
||||
constructor(private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang')[]) {
|
||||
super()
|
||||
}
|
||||
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
this.requests.push(options)
|
||||
const entry = this.script.shift()
|
||||
if (!entry) throw new Error('MockAdapter: script exhausted')
|
||||
if (entry === 'hang') {
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text: 'partial' }
|
||||
await new Promise<void>((_resolve, reject) => {
|
||||
if (options.signal?.aborted) { reject(new Error('aborted')); return }
|
||||
options.signal?.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
|
||||
})
|
||||
return
|
||||
}
|
||||
const chunks = typeof entry === 'function' ? entry(options) : entry
|
||||
for (const chunk of chunks) {
|
||||
if (options.signal?.aborted) throw new Error('aborted')
|
||||
yield chunk
|
||||
}
|
||||
}
|
||||
}
|
||||
176
packages/core/agent-loop/tests/properties.spec.ts
Normal file
176
packages/core/agent-loop/tests/properties.spec.ts
Normal file
@@ -0,0 +1,176 @@
|
||||
/**
|
||||
* Property-based tests for the agent loop's inbox/turn scheduling (the
|
||||
* property-testing RFC). Deterministic by construction: schedules are driven
|
||||
* through the `agent/status` settle signal (no wall-clock sleeps), so a flake
|
||||
* is a finding, not timing noise.
|
||||
*
|
||||
* Invariants: every sent message appears exactly once in the log (none lost);
|
||||
* turn numbers strictly increase; status transitions follow the legal machine
|
||||
* idle→running→idle (and →disposed at teardown).
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { LlmAdapter } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { type ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import fc from 'fast-check'
|
||||
|
||||
/** A never-exhausting adapter: every model call returns the same short reply. */
|
||||
class EchoAdapter extends LlmAdapter {
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
if (options.signal?.aborted) throw new Error('aborted')
|
||||
const text = 'ok'
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text }
|
||||
yield { type: 'block-end', index: 0, block: { type: 'text', text } }
|
||||
yield { type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } }
|
||||
yield { type: 'finish', reason: { kind: 'stop' } }
|
||||
}
|
||||
}
|
||||
|
||||
async function harness() {
|
||||
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'], new EchoAdapter())
|
||||
return ctx
|
||||
}
|
||||
|
||||
/** Resolve on the agent's next transition to idle (event-based, not polled). */
|
||||
function nextIdle(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()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
/** Record every status transition for the legal-machine assertion. Returns
|
||||
* the seen list plus a disposer for the listener (per the registry convention). */
|
||||
function recordStatus(ctx: Context, agent: ReactLoopAgent): { seen: string[]; dispose: () => void } {
|
||||
const seen: string[] = []
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent) seen.push(status)
|
||||
})
|
||||
return { seen, dispose }
|
||||
}
|
||||
|
||||
function userMessageTexts(agent: ReactLoopAgent): string[] {
|
||||
return agent.session.events
|
||||
.filter(e => e.type === 'user/message')
|
||||
.map(e => (e.data as { content: { type: string; text?: string }[] }).content.map(b => b.text ?? '').join(''))
|
||||
}
|
||||
|
||||
function turnNumbers(agent: ReactLoopAgent): number[] {
|
||||
return agent.session.events
|
||||
.filter(e => e.type === 'turn/start')
|
||||
.map(e => (e.data as { turn: number }).turn)
|
||||
}
|
||||
|
||||
/** Assert a status trace is a legal run: idle/running alternating, ending idle. */
|
||||
function assertLegalStatusTrace(trace: string[]): void {
|
||||
for (let i = 1; i < trace.length; i++) {
|
||||
expect(trace[i]).not.toBe(trace[i - 1]) // no repeats (setStatus dedups)
|
||||
}
|
||||
for (const s of trace) expect(['idle', 'running']).toContain(s)
|
||||
}
|
||||
|
||||
describe('agent loop scheduling properties', () => {
|
||||
it('a synchronous burst loses no message and uses strictly increasing turns', async () => {
|
||||
await fc.assert(fc.asyncProperty(
|
||||
fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 6 }),
|
||||
async (texts) => {
|
||||
const ctx = await harness()
|
||||
try {
|
||||
const agent = ctx.agentLoop.create(AgentId('a'), { model: 'mock' })
|
||||
const { seen: trace } = recordStatus(ctx, agent)
|
||||
const idle = nextIdle(ctx, agent)
|
||||
// Send all in one synchronous tick: they queue before the loop wakes.
|
||||
for (const text of texts) agent.send([{ type: 'text', text }])
|
||||
await idle
|
||||
|
||||
// No message lost: every send appears as a user/message, in order.
|
||||
expect(userMessageTexts(agent)).toEqual(texts)
|
||||
// A synchronous burst batches into exactly one turn.
|
||||
expect(turnNumbers(agent)).toEqual([1])
|
||||
assertLegalStatusTrace(trace)
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
},
|
||||
), { numRuns: 25, timeout: 2000 })
|
||||
})
|
||||
|
||||
it('sequential sends each get their own turn with increasing numbers', async () => {
|
||||
await fc.assert(fc.asyncProperty(
|
||||
fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 5 }),
|
||||
async (texts) => {
|
||||
const ctx = await harness()
|
||||
try {
|
||||
const agent = ctx.agentLoop.create(AgentId('a'), { model: 'mock' })
|
||||
for (const text of texts) {
|
||||
const idle = nextIdle(ctx, agent)
|
||||
agent.send([{ type: 'text', text }])
|
||||
await idle
|
||||
}
|
||||
// Each send was drained at a separate turn start: N turns, 1..N.
|
||||
expect(turnNumbers(agent)).toEqual(texts.map((_, i) => i + 1))
|
||||
expect(userMessageTexts(agent)).toEqual(texts)
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
},
|
||||
), { numRuns: 20, timeout: 2000 })
|
||||
})
|
||||
|
||||
it('mixed schedule (send, optionally settle) loses no message and orders turns', async () => {
|
||||
// Each step is a (text, settle?) pair: settle=true awaits idle before the
|
||||
// next send (own turn); settle=false sends in the same tick (batches).
|
||||
const stepArb = fc.record({ text: fc.string({ minLength: 1 }), settle: fc.boolean() })
|
||||
await fc.assert(fc.asyncProperty(
|
||||
fc.array(stepArb, { minLength: 1, maxLength: 6 }),
|
||||
async (steps) => {
|
||||
const ctx = await harness()
|
||||
try {
|
||||
const agent = ctx.agentLoop.create(AgentId('a'), { model: 'mock' })
|
||||
// Capture an idle waiter before EACH send; the last one is guaranteed
|
||||
// to resolve because the final send always triggers (or joins) a turn
|
||||
// that ends idle. Awaiting an already-resolved waiter is a no-op, so a
|
||||
// trailing settle step can't cause a hang.
|
||||
let lastIdle: Promise<void> | undefined
|
||||
for (const step of steps) {
|
||||
const idle = nextIdle(ctx, agent)
|
||||
lastIdle = idle
|
||||
agent.send([{ type: 'text', text: step.text }])
|
||||
if (step.settle) await idle
|
||||
}
|
||||
await lastIdle
|
||||
|
||||
// No message lost or reordered, regardless of batching.
|
||||
expect(userMessageTexts(agent)).toEqual(steps.map(s => s.text))
|
||||
// Turn numbers are a strictly increasing 1..N prefix (N = turn count).
|
||||
const turns = turnNumbers(agent)
|
||||
expect(turns).toEqual(turns.map((_, i) => i + 1))
|
||||
// Every message landed in some turn; turns never exceed messages.
|
||||
expect(turns.length).toBeLessThanOrEqual(steps.length)
|
||||
expect(turns.length).toBeGreaterThanOrEqual(1)
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
},
|
||||
), { numRuns: 25, timeout: 3000 })
|
||||
})
|
||||
})
|
||||
241
packages/core/agent-loop/tests/resume.spec.ts
Normal file
241
packages/core/agent-loop/tests/resume.spec.ts
Normal file
@@ -0,0 +1,241 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
const dirs: string[] = []
|
||||
afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
|
||||
|
||||
async function persistentHarness(adapter: MockAdapter): Promise<{ ctx: Context; root: string }> {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-resume-'))
|
||||
dirs.push(root)
|
||||
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: [] })
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return { ctx, root }
|
||||
}
|
||||
|
||||
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() }
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
|
||||
it('createAgent uses the caller-supplied sessionId (not ${id}-session)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('hi')])
|
||||
const { ctx } = await persistentHarness(adapter)
|
||||
const { agent } = ctx.agents.create({ agentId: AgentId('a1'), sessionId: SessionId('custom-session'), meta: { cwd: '/w' } })
|
||||
expect(agent.session.id).toBe('custom-session')
|
||||
expect(agent.session.header.cwd).toBe('/w')
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('createAgent rejects a duplicate agent id BEFORE creating the session (no orphan)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('hi')])
|
||||
const { ctx } = await persistentHarness(adapter)
|
||||
ctx.agents.create({ agentId: AgentId('dup'), sessionId: SessionId('sess-a') })
|
||||
// A second create with the SAME agent id but a fresh session id must reject
|
||||
// up front — and must NOT leave an orphaned 'sess-b' session behind.
|
||||
expect(() => ctx.agents.create({ agentId: AgentId('dup'), sessionId: SessionId('sess-b') })).toThrow(/already registered/)
|
||||
expect(ctx.sessions.get(SessionId('sess-b'))).toBeUndefined()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('createAgent works without meta (no cwd)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('hi')])
|
||||
const { ctx } = await persistentHarness(adapter)
|
||||
const { agent } = ctx.agents.create({ agentId: AgentId('a-nometa'), sessionId: SessionId('nometa-session') })
|
||||
expect(agent.session.id).toBe('nometa-session')
|
||||
expect(agent.session.header.cwd).toBeUndefined()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('resume of a session with no cwd carries an undefined cwd header', async () => {
|
||||
// Lifecycle 1: create a no-cwd session and run a turn.
|
||||
const adapter1 = new MockAdapter([textResponse('a')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('nocwd-sess') }).agent as ReactLoopAgent
|
||||
a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Lifecycle 2: resume it; the header cwd stays undefined (no-cwd branch).
|
||||
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 a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('nocwd-sess') })).agent as ReactLoopAgent
|
||||
expect(a2.session.header.cwd).toBeUndefined()
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('resume of a forked session preserves the parentSession lineage in the header', async () => {
|
||||
// Lifecycle 1: persist a FORKED session (carries parentSession in its
|
||||
// header) by creating it with a complete-turn seed — the write path
|
||||
// materializes the fork (header + seed) on disk.
|
||||
const seed: SessionEvent[] = [
|
||||
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
|
||||
{ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
|
||||
]
|
||||
const adapter1 = new MockAdapter([textResponse('a')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const forked = ctx1.sessions.create(SessionId('forked-sess'), { seed, meta: { cwd: '/w', parentSession: SessionId('parent-sess') } })
|
||||
await ctx1.parallel('session/flush', forked)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Lifecycle 2: resume it; the parentSession header survives the round-trip
|
||||
// (exercises resume's parentSession-present branch).
|
||||
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 a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('forked-sess') })).agent as ReactLoopAgent
|
||||
expect(a2.session.header.parentSession).toBe('parent-sess')
|
||||
expect(a2.session.header.cwd).toBe('/w')
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => {
|
||||
// Lifecycle 1: run a turn, then inject context while idle. The idle inject
|
||||
// wraps its context/message in a one-shot turn AND checkpoints it (the turn-enclosure RFC)
|
||||
// — without an explicit flush or clean dispose, the notice must still reach
|
||||
// disk, since a crash before the next turn would otherwise lose it.
|
||||
const adapter1 = new MockAdapter([textResponse('answer')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } }).agent as ReactLoopAgent
|
||||
a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
|
||||
// Let inject()'s fire-and-forget flush settle (NO explicit flush/dispose).
|
||||
await new Promise(r => setTimeout(r, 30))
|
||||
|
||||
// A SEPARATE backend reads the on-disk log — proving the inject persisted
|
||||
// itself, not a later dispose drain.
|
||||
const probe = new Context()
|
||||
await probe.plugin(SessionStore)
|
||||
await probe.plugin(SessionPersistenceJsonl, { root })
|
||||
const loaded = await probe.sessionPersistence.load(SessionId('inject-sess'))
|
||||
expect(JSON.stringify(loaded.events)).toContain('background task 42 finished')
|
||||
await probe.fiber.dispose()
|
||||
await ctx1.fiber.dispose()
|
||||
})
|
||||
|
||||
it('an idle inject() survives persist + resume (turn-enclosed, not dropped as crash tail)', async () => {
|
||||
// Lifecycle 1: run a turn, then inject context while idle. The idle inject
|
||||
// wraps its context/message in a one-shot turn so it is turn-enclosed —
|
||||
// otherwise scanLog would treat the trailing context as a crash tail and
|
||||
// drop it on reload (the bug this guards).
|
||||
const adapter1 = new MockAdapter([textResponse('answer')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } }).agent as ReactLoopAgent
|
||||
a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
|
||||
await ctx1.parallel('session/flush', a1.session)
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Lifecycle 2: resume; the injected context is still in the derived history.
|
||||
const adapter2 = new MockAdapter([textResponse('next')])
|
||||
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 a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('inject-sess') })).agent as ReactLoopAgent
|
||||
const flat = JSON.stringify(a2.session.deriveMessages())
|
||||
expect(flat).toContain('background task 42 finished')
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('resume reloads a persisted session: history + turn numbering continue, no duplicate seqs', async () => {
|
||||
// Lifecycle 1: run one full turn, persisting it.
|
||||
const adapter1 = new MockAdapter([textResponse('first answer')])
|
||||
const { ctx: ctx1, root } = await persistentHarness(adapter1)
|
||||
const a1 = ctx1.agents.create({ agentId: AgentId('main'), sessionId: SessionId('sess-resume'), meta: { cwd: '/w' } }).agent as ReactLoopAgent
|
||||
a1.send([{ type: 'text', text: 'first question' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx1, a1)
|
||||
const events1 = [...a1.session.events]
|
||||
const seqs1 = events1.map(e => e.seq)
|
||||
expect(seqs1).toEqual([...seqs1].sort((x, y) => x - y)) // contiguous
|
||||
await ctx1.fiber.dispose()
|
||||
|
||||
// Lifecycle 2: a brand-new context over the SAME root; resume the session.
|
||||
const adapter2 = new MockAdapter([textResponse('second answer')])
|
||||
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 a2 = (await ctx2.agents.resume({ agentId: AgentId('main'), resumeSessionId: SessionId('sess-resume') })).agent as ReactLoopAgent
|
||||
// The resumed session carries the prior history…
|
||||
expect(a2.session.id).toBe('sess-resume')
|
||||
expect(a2.session.events.length).toBe(events1.length)
|
||||
const replay = new Session(SessionId('replay'), events1)
|
||||
expect(a2.session.deriveMessages()).toEqual(replay.deriveMessages())
|
||||
|
||||
// …and a new turn continues numbering (turn 2) with contiguous seqs.
|
||||
a2.send([{ type: 'text', text: 'second question' }], { source: { kind: 'user' } })
|
||||
await waitForIdle(ctx2, a2)
|
||||
const allSeqs = a2.session.events.map(e => e.seq)
|
||||
expect(allSeqs).toEqual(allSeqs.map((_, i) => i)) // 0..N contiguous, no duplicates
|
||||
const turnStarts = a2.session.events.filter(e => e.type === 'turn/start')
|
||||
expect(turnStarts.map(e => e.type === 'turn/start' && e.data.turn)).toEqual([1, 2])
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('resume rejects when session persistence is not configured', async () => {
|
||||
// A harness WITHOUT the persistence plugin.
|
||||
const adapter = new MockAdapter([textResponse('x')])
|
||||
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)
|
||||
await expect(ctx.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('nope') }))
|
||||
.rejects.toThrow(/session persistence is not configured/)
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
1049
packages/core/agent-loop/tests/review-fixes.spec.ts
Normal file
1049
packages/core/agent-loop/tests/review-fixes.spec.ts
Normal file
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user