From 97111cb5b32a3d7975ddad7c5282fcf37a5026d4 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Thu, 30 Jul 2026 17:56:02 +0800 Subject: [PATCH] fix: align lifecycle consumers with durable inbox semantics --- packages/core/agent-loop/tests/agent.spec.ts | 112 +--------------- .../agent-loop/tests/interception.spec.ts | 123 +++++++----------- .../tests/request-reconstruction.spec.ts | 38 ++---- packages/core/agent-loop/tests/resume.spec.ts | 12 +- .../core/agent-loop/tests/tool-calls.spec.ts | 13 +- packages/goal/goal-session/src/index.ts | 37 ++++-- .../goal-session/tests/goal-session.spec.ts | 80 ++++-------- 7 files changed, 124 insertions(+), 291 deletions(-) diff --git a/packages/core/agent-loop/tests/agent.spec.ts b/packages/core/agent-loop/tests/agent.spec.ts index 295117d7d3..589f993c63 100644 --- a/packages/core/agent-loop/tests/agent.spec.ts +++ b/packages/core/agent-loop/tests/agent.spec.ts @@ -26,50 +26,17 @@ function send(agent: Agent, text: string): void { } describe('Agent', () => { - it('does not echo caller-owned message identities from delivery methods', async () => { - const adapter = new MockAdapter([ - textResponse('one'), - textResponse('two'), - textResponse('three'), - ]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const message = (text: string) => createUserMessage({ - content: [{ type: 'text' as const, text }], - source: { kind: 'user' as const }, - }) - const call = (method: 'send' | 'inject' | 'followup' | 'steer', args: unknown[]): unknown => { - const implementation: unknown = Reflect.get(agent, method) - if (typeof implementation !== 'function') throw new Error(`missing Agent.${method}`) - return Reflect.apply(implementation, agent, args) - } - - expect(call('send', [message('quiet'), { - target: 'next-turn', - wakeup: false, - }])).toBeUndefined() - expect(call('inject', [message('context')])).toBeUndefined() - expect(call('followup', [message('followup')])).toBeUndefined() - expect(call('steer', [message('steering')])).toBeUndefined() - await agent.whenIdle() - - expect(adapter.requests).toHaveLength(3) - }) - - it('idle inject() appends context without opening a turn or requesting a flush', async () => { + it('idle inject() durably stages context without opening a turn', async () => { const adapter = new MockAdapter([textResponse('ok')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - let flushes = 0 - ctx.on('session/flush', () => { flushes += 1 }) agent.inject(createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'p' } })) - expect(agent.session.events.map(event => event.type)).toEqual(['user/message']) + expect(agent.session.events.map(event => event.type)).toEqual(['agent/inbox/spliced']) expect(agent.status).toBe('idle') expect(adapter.requests).toHaveLength(0) await agent.whenIdle() - expect(flushes).toBe(0) }) it('inject() preserves an explicitly empty plugin source', async () => { @@ -79,7 +46,7 @@ describe('Agent', () => { agent.inject(createUserMessage({ content: [{ type: 'text', text: 'empty plugin source' }], source: { kind: 'plugin', plugin: '' } })) const injected = agent.session.events.at(-1) - expect(injected?.type === 'user/message' && injected.data.source) + expect(injected?.type === 'agent/inbox/spliced' && injected.data.inserted[0]?.source) .toEqual({ kind: 'plugin', plugin: '' }) }) @@ -119,79 +86,6 @@ describe('Agent', () => { expect(statuses).toEqual(['running', 'idle']) }) - it('awaits the turn-end checkpoint before claiming the next queued turn', async () => { - const adapter = new MockAdapter([textResponse('one'), textResponse('two')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const firstFlush = Promise.withResolvers() - const flushedTurns: number[] = [] - ctx.on('session/flush', async (session) => { - const turnEnd = session.events.findLast(event => event.type === 'turn/end') - flushedTurns.push(turnEnd?.data.turn ?? 0) - if (turnEnd?.data.turn === 1) await firstFlush.promise - }) - - send(agent, 'first') - send(agent, 'second') - - await vi.waitFor(() => { expect(flushedTurns).toEqual([1]) }) - expect(adapter.requests).toHaveLength(1) - firstFlush.resolve(undefined) - await agent.whenIdle() - - expect(adapter.requests).toHaveLength(2) - expect(flushedTurns).toEqual([1, 2]) - }) - - it('keeps whenIdle pending through the final turn checkpoint', async () => { - const ctx = await harness(new MockAdapter([textResponse('done')])) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const flush = Promise.withResolvers() - let flushStarted = false - ctx.on('session/flush', () => { - flushStarted = true - return flush.promise - }) - - send(agent, 'go') - await vi.waitFor(() => { expect(flushStarted).toBe(true) }) - let idleSettled = false - const idle = agent.whenIdle().then(() => { idleSettled = true }) - await Promise.resolve() - expect(idleSettled).toBe(false) - - flush.resolve(undefined) - await idle - expect(agent.status).toBe('idle') - }) - - it('reports a rejected turn-end checkpoint and continues queued work', async () => { - const adapter = new MockAdapter([textResponse('one'), textResponse('two')]) - const ctx = await harness(adapter) - const warning = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) - const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const failure = new Error('disk unavailable') - const errors: { turn: number; step: number; error: unknown }[] = [] - let flushes = 0 - ctx.on('session/flush', () => { - flushes += 1 - if (flushes === 1) throw failure - }) - ctx.on('agent/error', (subject, turn, step, error) => { - if (subject === agent) errors.push({ turn, step, error }) - }) - - send(agent, 'first') - send(agent, 'second') - await agent.whenIdle() - - expect(adapter.requests).toHaveLength(2) - expect(flushes).toBe(2) - expect(errors).toEqual([{ turn: 1, step: 1, error: failure }]) - expect(warning).toHaveBeenCalledWith(expect.stringContaining('session/flush failed at turn 1: disk unavailable')) - warning.mockRestore() - }) - it('whenIdle() resolves immediately without active work', async () => { const ctx = await harness(new MockAdapter([textResponse('ok')])) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) diff --git a/packages/core/agent-loop/tests/interception.spec.ts b/packages/core/agent-loop/tests/interception.spec.ts index 832181adb5..6998ccb7f4 100644 --- a/packages/core/agent-loop/tests/interception.spec.ts +++ b/packages/core/agent-loop/tests/interception.spec.ts @@ -119,7 +119,7 @@ describe('agent/prompt-submit', () => { await idle expect(observed).toHaveLength(1) - expect(observed[0]).toBe(input) + expect(observed[0]).not.toBe(input) expect(observed[0]).toMatchObject({ content: [{ type: 'text', text: 'accepted text' }], source: { kind: 'plugin', plugin: 'accepted source' }, @@ -237,7 +237,10 @@ describe('agent/prompt-submit', () => { const entered = Promise.withResolvers() const decision = Promise.withResolvers() let claimed: UserMessage[] = [] + let firstAdmission = true ctx.on('agent/prompt-submit', async (_agent, messages) => { + if (!firstAdmission) return { kind: 'allow', messages } + firstAdmission = false claimed = messages entered.resolve(undefined) return decision.promise @@ -271,21 +274,24 @@ describe('agent/prompt-submit', () => { 'turn/start', 'user/message', 'user/message', - 'steering/message', + 'user/message', ]) expect(staged[1]?.type === 'user/message' && staged[1].data.content) .toEqual([{ type: 'text', text: 'admitted prompt' }]) expect(staged[2]?.type === 'user/message' && staged[2].data.content) .toEqual([{ type: 'text', text: 'attached context' }]) - expect(staged[3]?.type === 'steering/message' && staged[3].data.message.content) + expect(staged[3]?.type === 'user/message' && staged[3].data.content) .toEqual([{ type: 'text', text: 'admission steering' }]) - const request = JSON.stringify(adapter.requests[0]?.messages) - expect(request).toContain('admitted prompt') - expect(request).toContain('attached context') - expect(request).toContain('admission steering') + const firstRequest = JSON.stringify(adapter.requests[0]?.messages) + expect(firstRequest).toContain('admitted prompt') + expect(firstRequest).not.toContain('attached context') + expect(firstRequest).not.toContain('admission steering') + const nextRequest = JSON.stringify(adapter.requests[1]?.messages) + expect(nextRequest).toContain('attached context') + expect(nextRequest).toContain('admission steering') }) - it('keeps admission-time outbox input staged when admission is blocked', async () => { + it('cancels admission-time input when admission is blocked', async () => { const adapter = new MockAdapter([textResponse('retried')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('blocked-admission-outbox'), { provider: 'mock', model: 'mock' }) @@ -307,8 +313,8 @@ describe('agent/prompt-submit', () => { decision.resolve({ kind: 'block', reason: 'policy' }) await blockedIdle - expect(agent.inbox.nextStep).toHaveLength(2) - expect(events(agent)).toEqual([]) + expect(agent.inbox.nextStep).toHaveLength(0) + expect(events(agent).some(event => event.type === 'turn/start')).toBe(false) expect(adapter.requests).toEqual([]) disposeBlock() @@ -317,17 +323,13 @@ describe('agent/prompt-submit', () => { const staged = events(agent).filter(event => event.type === 'user/message' || event.type === 'steering/message') - expect(staged.map(event => event.type)).toEqual([ - 'user/message', - 'steering/message', - 'user/message', - ]) + expect(staged.map(event => event.type)).toEqual(['user/message']) expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt') - expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged context') - expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering') + expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged context') + expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged steering') }) - it('orders rejected-admission outbox input before a later admitted prompt', async () => { + it('cancels later queued work when an admission is blocked', async () => { const adapter = new MockAdapter([textResponse('continued')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('rejected-admission-order'), { @@ -361,23 +363,12 @@ describe('agent/prompt-submit', () => { send(agent, 'later prompt') await idle - const staged = events(agent).filter(event => - event.type === 'turn/start' || event.type === 'user/message' || event.type === 'steering/message') - expect(staged.map(event => event.type)).toEqual([ - 'turn/start', - 'user/message', - 'steering/message', - 'user/message', - ]) - expect(staged[1]?.type === 'user/message' && staged[1].data.content) - .toEqual([{ type: 'text', text: 'earlier state change' }]) - expect(staged[2]?.type === 'steering/message' && staged[2].data.message.content) - .toEqual([{ type: 'text', text: 'earlier steering' }]) - expect(staged[3]?.type === 'user/message' && staged[3].data.content) - .toEqual([{ type: 'text', text: 'later prompt' }]) + expect(events(agent).some(event => event.type === 'turn/start')).toBe(false) + expect(agent.inbox.hasPending).toBe(false) + expect(adapter.requests).toEqual([]) }) - it('commits context-only injection when admission closes without a turn', async () => { + it('cancels context-only injection when admission closes without a turn', async () => { const adapter = new MockAdapter([]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('blocked-admission-context'), { provider: 'mock', model: 'mock' }) @@ -399,51 +390,31 @@ describe('agent/prompt-submit', () => { await idle const log = events(agent) - expect(log.map(event => event.type)).toEqual(['user/message']) - expect(log[0]?.type === 'user/message' && log[0].data.content) - .toEqual([{ type: 'text', text: 'independent context' }]) + expect(log.some(event => event.type === 'user/message')).toBe(false) + expect(agent.inbox.hasPending).toBe(false) expect(adapter.requests).toEqual([]) }) - it('retains rejected-admission context when its idle append fails', async () => { - const adapter = new MockAdapter([textResponse('retried')]) + it('leaves inbox state unchanged when its durable append fails', async () => { + const adapter = new MockAdapter([]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('blocked-admission-append-failure'), { provider: 'mock', model: 'mock', }) - const warned = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) vi.spyOn(agent.session, 'append').mockImplementationOnce(() => { throw new Error('append unavailable') }) - const entered = Promise.withResolvers() - const decision = Promise.withResolvers() - const disposeBlock = ctx.on('agent/prompt-submit', async () => { - entered.resolve(undefined) - return decision.promise - }) - - agent.followup(createUserMessage({ content: [{ type: 'text', text: 'blocked prompt' }], source: { kind: 'user' } })) - await entered.promise - agent.inject(createUserMessage({ - content: [{ type: 'text', text: 'retained context' }], - source: { kind: 'plugin', plugin: 'test' }, - })) - decision.resolve({ kind: 'block', reason: 'policy' }) - await agent.whenIdle() + expect(() => { + send(agent, 'blocked prompt') + }).toThrow('append unavailable') expect(events(agent)).toEqual([]) - expect(warned).toHaveBeenCalledWith(expect.stringContaining('append unavailable')) - - disposeBlock() - send(agent, 'resume') - await waitForIdle(ctx, agent) - - expect(events(agent).some(event => event.type === 'user/message' - && JSON.stringify(event.data.content).includes('retained context'))).toBe(true) + expect(agent.inbox.hasPending).toBe(false) + expect(agent.status).toBe('idle') }) - it('adjacent blocked and allowed prompts keep independent turn outcomes', async () => { + it('a blocked prompt cancels adjacent queued prompts', async () => { const adapter = new MockAdapter([textResponse('ran once')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -457,22 +428,18 @@ describe('agent/prompt-submit', () => { const reasons: TurnEndReason[] = [] ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) }) - // The rejected admission is dropped; the allowed prompt owns the only turn. send(agent, 'secret') send(agent, 'safe') await waitForIdle(ctx, agent) const log = events(agent) - // The allowed prompt became a user/message and drove exactly one model call. - const userMsgs = log.filter(e => e.type === 'user/message') - expect(userMsgs).toHaveLength(1) - expect(userMsgs[0]?.type === 'user/message' && userMsgs[0].data.content).toEqual([{ type: 'text', text: 'safe' }]) - expect(adapter.requests.length).toBeGreaterThanOrEqual(1) - expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1) - expect(reasons).toEqual([{ kind: 'completed' }]) + expect(log.filter(e => e.type === 'user/message')).toHaveLength(0) + expect(adapter.requests).toHaveLength(0) + expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0) + expect(reasons).toEqual([]) }) - it('a throwing prompt-submit listener drops that admission while an adjacent message survives', async () => { + it('a throwing prompt-submit listener reports the driver error and retains adjacent work', async () => { const adapter = new MockAdapter([textResponse('after')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -497,14 +464,14 @@ describe('agent/prompt-submit', () => { send(agent, 'first') send(agent, 'second') await idle - expect(errors).toEqual([]) + expect(errors).toEqual([expect.objectContaining({ message: 'prompt hook broke' })]) const log = events(agent) - expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1) - expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1) - expect(reasons).toEqual([{ kind: 'completed' }]) + expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0) + expect(log.filter(e => e.type === 'turn/end')).toHaveLength(0) + expect(reasons).toEqual([]) expect(statuses).toEqual(['running', 'idle']) - expect(adapter.requests).toHaveLength(1) - expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('second') + expect(adapter.requests).toHaveLength(0) + expect(agent.inbox.nextTurn).toHaveLength(2) }) }) diff --git a/packages/core/agent-loop/tests/request-reconstruction.spec.ts b/packages/core/agent-loop/tests/request-reconstruction.spec.ts index d4150c0521..f9153b4de9 100644 --- a/packages/core/agent-loop/tests/request-reconstruction.spec.ts +++ b/packages/core/agent-loop/tests/request-reconstruction.spec.ts @@ -257,10 +257,6 @@ describe('request stability across the loop', () => { } }([]) const ctx = await harness(adapter) - const errors: Error[] = [] - ctx.on('agent/error', (_agent, _turn, _step, error) => { - if (error instanceof Error) errors.push(error) - }) const agent = ctx.agentLoop.create(SessionId(`reasoning-${kind}`), { provider: 'mock', model: 'mock', @@ -269,7 +265,9 @@ describe('request stability across the loop', () => { send(agent, 'go') await waitForIdle(ctx, agent) - expect(errors).toContain(failure) + expect(agent.session.events.findLast(event => event.type === 'turn/end')).toMatchObject({ + data: { reason: { kind: 'error', error: failure.message } }, + }) expect(adapter.requests).toHaveLength(0) }, ) @@ -316,19 +314,13 @@ describe('request stability across the loop', () => { send(agent, 'first') await waitForIdle(ctx, agent) - // A pre-step listener compacts turn 1's history before turn 2's step — - // the sanctioned surface rewrite, landing OUTSIDE the step. - const preStep = ctx.on('agent/step', () => { - preStep() - const session = agent.session - const nodes = session.surface.nodes - session.append('user/message', createUserMessage({ - content: [{ type: 'text', text: '[summary of turn 1]' }], - source: { kind: 'plugin', plugin: 'test-compact' }, - }), { - surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! }, - sourceEventSeqs: [nodes[0]!, nodes[1]!], - }) + const nodes = agent.session.surface.nodes + agent.session.append('user/message', createUserMessage({ + content: [{ type: 'text', text: '[summary of turn 1]' }], + source: { kind: 'plugin', plugin: 'test-compact' }, + }), { + surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! }, + sourceEventSeqs: [nodes[0]!, nodes[1]!], }) send(agent, 'second') @@ -398,10 +390,6 @@ describe('request stability across the loop', () => { const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const errors: Error[] = [] - ctx.on('agent/error', (_agent, _turn, _step, error) => { - if (error instanceof Error) errors.push(error) - }) ctx.on('llm/stream', (options, next) => { // The historical failure mode this design kills: a listener rewriting // request content in place. The freeze turns it into a loud error. @@ -415,8 +403,10 @@ describe('request stability across the loop', () => { send(agent, 'go') await waitForIdle(ctx, agent) - expect(errors).toHaveLength(1) - expect(errors[0]!.message).toMatch(/not extensible|frozen|read only|readonly/i) + const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end') + expect(turnEnd).toMatchObject({ data: { reason: { kind: 'error' } } }) + if (turnEnd?.type !== 'turn/end' || turnEnd.data.reason.kind !== 'error') throw new Error() + expect(turnEnd.data.reason.error).toMatch(/not extensible|frozen|read only|readonly/i) }) it('a fresh loop instance over a seeded log anchors with a resume snapshot and stays cache-aligned', async () => { diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index d03dfe0f59..430d5eb273 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -540,9 +540,10 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await waitForIdle(ctx1, a1) a1.inject(createUserMessage({ content: [{ type: 'text', text: 'background task 42 finished' }], source: { kind: 'plugin', plugin: 'tool-bash' } })) await a1.whenIdle() - await ctx1.fiber.dispose() + await ctx1.sessions.flush(a1.session) - // Lifecycle 2: resume; the injected context is still in the derived history. + // Lifecycle 2: resume; the injected context is still pending and becomes + // model-visible when the next turn admits it. const adapter2 = new MockAdapter([textResponse('next')]) const ctx2 = new Context() await ctx2.plugin(LlmService) @@ -553,10 +554,17 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await ctx2.plugin(AgentLoop, { agents: [] }) await ctx2.plugin(SessionPersistenceJsonl, { root }) ctx2.llm.registerAdapter(['mock'], adapter2) + const loaded = await ctx2.sessionPersistence.load(SessionId('inject-sess')) + expect(loaded.events.some(event => event.type === 'agent/inbox/spliced')).toBe(true) + expect(JSON.stringify(loaded.events)).toContain('background task 42 finished') const a2 = (await ctx2.agents.resume({ resumeSessionId: SessionId('inject-sess') })).agent + expect(JSON.stringify(a2.inbox.nextStep)).toContain('background task 42 finished') + a2.followup(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } })) + await waitForIdle(ctx2, a2) const flat = JSON.stringify(a2.session.deriveMessages()) expect(flat).toContain('background task 42 finished') await ctx2.fiber.dispose() + await ctx1.fiber.dispose() }) it('resume reloads a persisted session: history + turn numbering continue, no duplicate seqs', async () => { diff --git a/packages/core/agent-loop/tests/tool-calls.spec.ts b/packages/core/agent-loop/tests/tool-calls.spec.ts index dd086d8e7f..98ac46270f 100644 --- a/packages/core/agent-loop/tests/tool-calls.spec.ts +++ b/packages/core/agent-loop/tests/tool-calls.spec.ts @@ -648,10 +648,6 @@ describe('tool-call scheduler: failure quiescence', () => { ? new Promise((_resolve, reject) => { rejectFirst = reject }) : dispatch(exec).then(() => { throw drainedError }) const agent = ctx.agentLoop.create(SessionId('scheduler-failure'), { provider: 'mock', model: 'mock' }) - const errors: unknown[] = [] - ctx.on('agent/error', (subject, _turn, _step, error) => { - if (subject === agent) errors.push(error) - }) let idle = false const idlePromise = waitForIdle(ctx, agent).then(() => { idle = true }) @@ -664,15 +660,16 @@ describe('tool-call scheduler: failure quiescence', () => { const startedBeforeDrain = [...gated.started] const idleBeforeDrain = idle - const errorsBeforeDrain = [...errors] + const turnEndBeforeDrain = events(agent).find(event => event.type === 'turn/end') for (const id of gated.pending()) gated.release(id) await idlePromise expect(startedBeforeDrain).toEqual(['2']) expect(idleBeforeDrain).toBe(false) - expect(errorsBeforeDrain).toEqual([]) + expect(turnEndBeforeDrain).toBeUndefined() expect(gated.pending()).toEqual([]) - expect(errors).toEqual([schedulerError]) - expect(errors[0]).toBe(schedulerError) + expect(events(agent).findLast(event => event.type === 'turn/end')).toMatchObject({ + data: { reason: { kind: 'error', error: schedulerError.message } }, + }) }) }) diff --git a/packages/goal/goal-session/src/index.ts b/packages/goal/goal-session/src/index.ts index 3789ec5430..583734a3db 100644 --- a/packages/goal/goal-session/src/index.ts +++ b/packages/goal/goal-session/src/index.ts @@ -32,6 +32,7 @@ interface RoundAttempt extends RoundIdentity { readonly messageId: MessageId readonly content: ContentBlock[] phase: 'queued' | 'admitted' + cancelled: boolean stale: boolean } @@ -124,6 +125,12 @@ export function apply(ctx: Context): void { } } + /** Remove only this driver's still-pending reservation. */ + function cancelReservation(agent: Agent, attempt: RoundAttempt): void { + const index = agent.inbox.nextTurn.findIndex(message => message.id === attempt.messageId) + if (index >= 0) agent.inbox.splice('next-turn', index, 1, [], 'canceled') + } + /** Process admitted work at quiescence, then reserve at most one next round. */ async function drive(state: DriverState): Promise { const { agent } = state @@ -175,6 +182,7 @@ export function apply(ctx: Context): void { messageId: message.id, content, phase: 'queued', + cancelled: false, stale: false, } state.attempt = reservation @@ -252,7 +260,8 @@ export function apply(ctx: Context): void { state.competingQueued = false const attempt = state.attempt const goal = currentGoal(state) - if (attempt?.phase === 'queued' && goal?.phase === 'active' && goal.activation === 'armed') { + if ((attempt?.phase === 'queued' || attempt?.cancelled) + && goal?.phase === 'active' && goal.activation === 'armed') { state.attempt = undefined try { ctx.goals.pause(agent, goalRef(goal)) @@ -292,16 +301,8 @@ export function apply(ctx: Context): void { return case 'turn/end': if (event.data.reason.kind !== 'aborted') return - { - const goal = currentGoal(state) - if (goal?.phase !== 'active' || goal.activation !== 'armed') return - try { - ctx.goals.pause(agent, goalRef(goal)) - } catch (error: unknown) { - ctx.logger.warn(`goal-session: could not pause cancelled goal for agent "${agent.id}": ${renderThrown(error)}`) - disarm(state) - } - } + if (state.attempt?.phase === 'admitted') state.attempt.cancelled = true + else disarm(state) return default: return @@ -324,7 +325,7 @@ export function apply(ctx: Context): void { && source.round === goal.roundsStarted + 1 } - ctx.on('agent/prompt-submit', async (agent, messages, _signal, next): Promise => { + ctx.on('agent/prompt-submit', async (agent, messages, signal, next): Promise => { const submitted = messages.find(message => isGoalRoundSource(message.source)) if (submitted === undefined) return next() const { content, source } = submitted @@ -342,14 +343,16 @@ export function apply(ctx: Context): void { if (attempt !== undefined && sameRound(source, attempt)) { attempt.stale = true state.attempt = undefined + cancelReservation(agent, attempt) } requestDrive(state) - return { kind: 'block', reason: STALE_ROUND_REASON } + return { kind: 'block', reason: STALE_ROUND_REASON, keepInbox: true } } let decision: PromptDecision try { decision = await next() } catch (error: unknown) { + if (signal.aborted) throw error // A throwing downstream hook drops the whole admission: the loop // returns to idle without a turn, so a still-queued reservation would // starve every later drive pass. Clear it and let the driver @@ -357,10 +360,12 @@ export function apply(ctx: Context): void { const attempt = state.attempt if (attempt !== undefined && sameRound(source, attempt) && attempt.phase === 'queued') { state.attempt = undefined + cancelReservation(agent, attempt) requestDrive(state) } throw error } + if (signal.aborted) return decision if (decision.kind === 'block') { const attempt = state.attempt if (attempt !== undefined && sameRound(source, attempt)) state.attempt = undefined @@ -386,9 +391,10 @@ export function apply(ctx: Context): void { if (attempt !== undefined && sameRound(source, attempt)) { attempt.stale = true state.attempt = undefined + cancelReservation(agent, attempt) } requestDrive(state) - return { kind: 'block', reason: STALE_ROUND_REASON } + return { kind: 'block', reason: STALE_ROUND_REASON, keepInbox: true } } return decision }) @@ -410,6 +416,9 @@ export function apply(ctx: Context): void { const attempt = state.attempt if (attempt !== undefined) { attempt.stale = true + if (attempt.phase === 'admitted' && state.agent.status === 'running') { + state.agent.cancel({ kind: 'parent' }) + } } if (state.run !== undefined) waits.push(state.run) } diff --git a/packages/goal/goal-session/tests/goal-session.spec.ts b/packages/goal/goal-session/tests/goal-session.spec.ts index 93d3ef926e..4cff18d80f 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -4,7 +4,7 @@ import type { Agent, PromptDecision } from '@deepseek-ai/dsh-agent' import { agentEvents } from '@deepseek-ai/dsh-agent' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' -import GoalService, { foldGoal, GoalId } from '@deepseek-ai/dsh-goal' +import GoalService, { GoalId } from '@deepseek-ai/dsh-goal' import type { GoalView } from '@deepseek-ai/dsh-goal' import { createUserMessage, LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm' import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm' @@ -221,18 +221,18 @@ describe('same-session goal driving', () => { }) it.each([ - ['rate limit', new LlmError('slow down', 'RATE_LIMIT'), 'usage-limited'], - ['request error', new Error('provider broke'), 'turn-error'], - ['max tokens', maxTokensResponse('unfinished'), 'max-tokens'], - ] as const)('stops after a %s without an automatic retry', async (_label, response, code) => { - const test = await harness([response]) + ['rate limit', new LlmError('slow down', 'RATE_LIMIT')], + ['request error', new Error('provider broke')], + ['max tokens', maxTokensResponse('unfinished')], + ] as const)('does not attribute a %s to one goal follow-up', async (_label, response) => { + const test = await harness(Array.from({ length: 8 }, () => response)) test.ctx.goals.create(test.agent, { objective: 'stop safely', maxGoalRounds: 8 }) const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') - expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' }) - expect(goal?.blockedReason?.code).toBe(code) - expect(test.adapter.requests).toHaveLength(1) + expect(goal).toMatchObject({ roundsStarted: 8, activation: 'disarmed' }) + expect(goal?.blockedReason?.code).toBe('round-limit') + expect(test.adapter.requests).toHaveLength(8) }) it('maps a downstream prompt veto to blocked without admitting the round', async () => { @@ -250,7 +250,7 @@ describe('same-session goal driving', () => { expect(test.agent.session.events.some(event => event.type === 'turn/start')).toBe(false) }) - it('does not reserve again when a stopped-goal observer queues ordinary work', async () => { + it('does not reserve again when a stopped-goal observer queues cancel-scoped work', async () => { const test = await harness([textResponse('human follow-up')]) test.ctx.on('agent/prompt-submit', (_agent, messages, _signal, next) => messages[0]?.source.kind === 'goal' ? Promise.resolve({ kind: 'block', reason: 'stop this round' }) @@ -261,10 +261,10 @@ describe('same-session goal driving', () => { test.ctx.goals.create(test.agent, { objective: 'stop and inspect' }) await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked') - await waitForRequests(test.adapter, 1) await test.agent.whenIdle() - expect(requestText(test.adapter.requests[0]!)).toContain('inspect the blocker') + expect(test.adapter.requests).toHaveLength(0) + expect(test.agent.inbox.nextTurn).toHaveLength(0) }) it('pauses and drops a reserved round when cancellation lands before admission', async () => { @@ -297,10 +297,6 @@ describe('same-session goal driving', () => { const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused') expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' }) - expect(foldGoal(test.agent.session.events)).toMatchObject({ - goal: { phase: 'paused', revision: 2 }, - roundsStarted: 1, - }) expect(test.adapter.requests).toHaveLength(1) }) @@ -486,8 +482,8 @@ describe('same-session goal driving', () => { expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'paused' }) }) - it('reschedules the round when a downstream admission hook throws', async () => { - const test = await harness([textResponse('second admission succeeded')]) + it('fails closed when a downstream admission hook throws', async () => { + const test = await harness([]) // Registered after goal-session's own listener: the throw propagates back // through goal-session's next() await, dropping the whole admission. let threw = false @@ -500,12 +496,10 @@ describe('same-session goal driving', () => { }) test.ctx.goals.create(test.agent, { objective: 'survive a throwing hook', maxGoalRounds: 1 }) - // The cleared reservation lets the driver reschedule; the second - // admission passes and the round completes to its limit. - const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') - expect(goal?.blockedReason?.code).toBe('round-limit') - expect(goal?.roundsStarted).toBe(1) - expect(test.adapter.requests).toHaveLength(1) + const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed') + expect(goal).toMatchObject({ phase: 'active', roundsStarted: 0 }) + expect(test.adapter.requests).toHaveLength(0) + expect(test.agent.inbox.nextTurn).toHaveLength(0) }) it('a retry turn on a non-goal failure leaves the goal reservation untouched', async () => { @@ -619,7 +613,7 @@ describe('same-session goal driving', () => { const test = await harness([textResponse('retry after containment')]) let armed = true onInboxMessage(test.ctx, test.agent, (message) => { - if (message.source.kind !== 'goal' || !armed) return + if (message.source.kind !== 'goal' || message.source.round <= 0 || !armed) return armed = false vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => { throw new Error('admission projection failed') @@ -779,34 +773,8 @@ describe('same-session goal driving', () => { expect(test.adapter.requests).toHaveLength(1) }) - it('leaves a queued reservation pending when the driver runs before its turn settles', async () => { - const test = await harness([textResponse('settled later')]) - let woken = false - test.ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next) => { - if (messages[0]?.source.kind === 'goal' && !woken) { - woken = true - // A concurrent driver pass must observe the still-unsettled attempt - // and yield rather than double-book or clear the reservation. - agentEvents(test.ctx, test.agent).emit('agent/status', 'idle') - await new Promise((resolve) => { setImmediate(resolve) }) - } - return next() - }) - test.ctx.goals.create(test.agent, { objective: 'wake early', maxGoalRounds: 1 }) - - const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') - - expect(goal?.blockedReason?.code).toBe('round-limit') - expect(goal?.roundsStarted).toBe(1) - expect(test.adapter.requests).toHaveLength(1) - }) - - it('yields to a round whose turn/end never committed instead of misreading it as settled', async () => { + it('disarms when a round turn/end cannot commit', async () => { const test = await harness([textResponse('round ran')]) - // A persistent pre-commit turn/end rejection: the loop contains the close - // failure and reaches idle, but the round's attempt holds a turn with no - // terminal reason. The idle drive pass must yield to that unsettled - // attempt rather than classify an absent reason or crash into disarm. test.ctx.on('internal/dispatch', (_mode, name, args) => { if (name !== 'session/event') return const event = args[1] as { type: string } @@ -817,12 +785,10 @@ describe('same-session goal driving', () => { await test.agent.whenIdle() await new Promise((resolve) => { setImmediate(resolve) }) - // One request ran; the unsettled attempt parked the driver without a - // second reservation and without disarming the goal. expect(test.adapter.requests).toHaveLength(1) expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'active', - activation: 'armed', + activation: 'disarmed', }) }) @@ -868,7 +834,9 @@ describe('same-session goal driving', () => { if (session !== test.agent.session || queued) return if (event.type === 'user/message' && event.data.source.kind === 'goal') { queued = true - test.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } })) + queueMicrotask(() => { + test.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } })) + }) } }) test.ctx.goals.create(test.agent, { objective: 'survive a stale failure', maxGoalRounds: 1 })