diff --git a/packages/context/time-context/src/index.ts b/packages/context/time-context/src/index.ts index 01d1393f3c..6762498bc4 100644 --- a/packages/context/time-context/src/index.ts +++ b/packages/context/time-context/src/index.ts @@ -174,6 +174,6 @@ export function apply(ctx: Context, config: Config): void { const previous = step === 1 ? precedingMessageTime(agent) : precedingStepContextTime(agent, turn) - agent.inject(createUserMessage({ content: [{ type: 'text', text: renderText(now, turn, step, previous, formatter, resolvedTimeZone) }], source: { kind: 'plugin', plugin: name } })) + agent.session.append('user/message', createUserMessage({ content: [{ type: 'text', text: renderText(now, turn, step, previous, formatter, resolvedTimeZone) }], source: { kind: 'plugin', plugin: name } }), { surfaceOp: 'append' }) }, { prepend: true }) } diff --git a/packages/context/time-context/src/invariant.ts b/packages/context/time-context/src/invariant.ts index aa8f0418dd..37b5a08610 100644 --- a/packages/context/time-context/src/invariant.ts +++ b/packages/context/time-context/src/invariant.ts @@ -18,31 +18,22 @@ export const name = 'time-context-invariant' /** Service required before the companion can reserve package ownership. */ export const inject = ['invariants'] -/** Derive the pre-step position at which a time-context reading may append. */ +/** Derive the open step in which a time-context reading may append. */ function preparationPosition(history: readonly SessionEvent[], fail: InvariantFailure): { turn: number; step: number } { - const currentTurnEvents: SessionEvent[] = [] - let openTurn: number | undefined for (const event of history.slice().reverse()) { - if (event.type === 'turn/end') { - fail('time-context reading must be appended inside an open turn') - } - if (event.type === 'turn/start') { - openTurn = event.data.turn - break - } - currentTurnEvents.push(event) - } - if (openTurn === undefined) fail('time-context reading must be appended inside an open turn') - - for (const event of currentTurnEvents) { - if (event.type === 'step/start') { - fail(`time-context reading must precede step/start, but step ${event.data.step} is already open`) - } - if (event.type === 'step/end') { - return { turn: openTurn, step: event.data.step + 1 } + switch (event.type) { + case 'step/start': + return event.data + case 'step/end': + case 'turn/start': + case 'turn/end': + fail('time-context reading must be appended inside an open step') + break + default: + break } } - return { turn: openTurn, step: 1 } + fail('time-context reading must be appended inside an open step') } /** Validate one plugin-attributed time reading against its session position and timestamp. */ diff --git a/packages/context/time-context/tests/invariant.spec.ts b/packages/context/time-context/tests/invariant.spec.ts index f117df5081..db64595938 100644 --- a/packages/context/time-context/tests/invariant.spec.ts +++ b/packages/context/time-context/tests/invariant.spec.ts @@ -58,6 +58,7 @@ function preparing(turn: number, step: number): Session { session.append('step/start', { turn, step: priorStep }) session.append('step/end', { turn, step: priorStep }) } + session.append('step/start', { turn, step }) return session } @@ -92,8 +93,8 @@ describe('time-context invariants', () => { content: [{ type: 'text', text: 'prepare' }], source: { kind: 'user' }, }), { surfaceOp: 'append' }) - appendReading(session, reading()) session.append('step/start', { turn: 1, step: 1 }) + appendReading(session, reading()) await ctx.plugin(InvariantService, { enabled: true }) await expect(ctx.plugin(TimeInvariant)).resolves.toBeDefined() @@ -108,6 +109,7 @@ describe('time-context invariants', () => { content: [{ type: 'text', text: 'prepare' }], source: { kind: 'user' }, }), { surfaceOp: 'append' }) + session.append('step/start', { turn: 1, step: 1 }) appendReading(session, reading('1', '2', 'step context')) await ctx.plugin(InvariantService, { enabled: true }) @@ -127,17 +129,17 @@ describe('time-context invariants', () => { const session = preparing(1, 2) session.append('turn/end', { turn: 1, reason: { kind: 'aborted', reason: { kind: 'user' } } }) expect(() => { ctx.emit('session/event', session, event(reading('1', '2', 'step context'))) }) - .toThrow(/inside an open turn/) + .toThrow(/inside an open step/) }) - it('rejects a reading after step/start or without any open turn', async () => { + it('rejects a reading outside an open step', async () => { const ctx = await setup() - const started = preparing(1, 1) - started.append('step/start', { turn: 1, step: 1 }) - expect(() => { ctx.emit('session/event', started, event(reading())) }).toThrow(/must precede step\/start/) + const ended = preparing(1, 1) + ended.append('step/end', { turn: 1, step: 1 }) + expect(() => { ctx.emit('session/event', ended, event(reading())) }).toThrow(/inside an open step/) expect(() => { ctx.emit('session/event', new Session(SessionId('time-invariant-empty')), event(reading())) - }).toThrow(/inside an open turn/) + }).toThrow(/inside an open step/) }) it.each([ diff --git a/packages/context/time-context/tests/time-context.e2e.ts b/packages/context/time-context/tests/time-context.e2e.ts index 02704d3eba..f56d369db2 100644 --- a/packages/context/time-context/tests/time-context.e2e.ts +++ b/packages/context/time-context/tests/time-context.e2e.ts @@ -54,7 +54,7 @@ describe('time-context through a real headless cordis.yml', () => { expect(contexts).toHaveLength(2) expect(starts).toHaveLength(2) for (let index = 0; index < contexts.length; index += 1) { - expect(contexts[index]!.seq).toBeLessThan(starts[index]!.seq) + expect(contexts[index]!.seq).toBeGreaterThan(starts[index]!.seq) expect(contexts[index]!.surfaceOp).toBe('append') expect(contexts[index]!.data.source).toEqual({ kind: 'plugin', plugin: 'time-context' }) } diff --git a/packages/context/time-context/tests/time-context.spec.ts b/packages/context/time-context/tests/time-context.spec.ts index 884c9430ad..17e86c3562 100644 --- a/packages/context/time-context/tests/time-context.spec.ts +++ b/packages/context/time-context/tests/time-context.spec.ts @@ -45,9 +45,7 @@ function sessionAgent(session: Session, id = 'agent'): Agent { ctx: new Context(), followup: () => {}, steer: () => {}, - inject(input) { - session.append('user/message', input, { surfaceOp: 'append' }) - }, + inject: () => { throw new Error('time-context must append directly to the open step') }, cancel() {}, whenIdle: () => Promise.resolve(), } @@ -358,7 +356,7 @@ describe('real agent-loop request history', () => { it.each([ ['throws', 'error'], ['cancels', 'aborted'], - ] as const)('discards the pending preparation reading when a later step listener %s', async (mode, reasonKind) => { + ] as const)('retains the durable preparation reading when a later step listener %s', async (mode, reasonKind) => { const adapter = new ScriptedAdapter([textResponse('unused')]) const ctx = await loopHarness(adapter) let laterSawReading = false @@ -372,10 +370,10 @@ describe('real agent-loop request history', () => { agent.followup(createUserMessage({ content: [{ type: 'text', text: 'start' }], source: { kind: 'user' } })) await agent.whenIdle() - expect(laterSawReading).toBe(false) - expect(contextTexts(agent.session)).toHaveLength(0) + expect(laterSawReading).toBe(true) + expect(contextTexts(agent.session)).toHaveLength(1) expect(adapter.requests).toHaveLength(0) - expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false) + expect(agent.session.events.some(event => event.type === 'step/start')).toBe(true) const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end') expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe(reasonKind) await ctx.fiber.dispose() @@ -405,7 +403,7 @@ describe('real agent-loop request history', () => { expect(contexts).toHaveLength(adapter.requests.length) expect(starts).toHaveLength(adapter.requests.length) for (let index = 0; index < contexts.length; index += 1) { - expect(contexts[index]!.seq).toBeLessThan(starts[index]!.seq) + expect(contexts[index]!.seq).toBeGreaterThan(starts[index]!.seq) } expect(contexts.every(event => event.data.source.kind === 'plugin' && event.data.source.plugin === 'time-context' diff --git a/packages/context/tmux-context/src/index.ts b/packages/context/tmux-context/src/index.ts index 35f4c4a22f..4e1694ca1a 100644 --- a/packages/context/tmux-context/src/index.ts +++ b/packages/context/tmux-context/src/index.ts @@ -233,9 +233,9 @@ export function apply(ctx: Context, config: Config): void { if (location === undefined) return const state = renderState(location) if (previous !== undefined && previous.state === state) return - agent.inject(createUserMessage({ + agent.session.append('user/message', createUserMessage({ content: [{ type: 'text', text: renderReading(location, turn) }], source: { kind: 'plugin', plugin: name }, - })) + }), { surfaceOp: 'append' }) }, { prepend: true }) } diff --git a/packages/context/tmux-context/tests/tmux-context.spec.ts b/packages/context/tmux-context/tests/tmux-context.spec.ts index 4262383b7f..097cc66802 100644 --- a/packages/context/tmux-context/tests/tmux-context.spec.ts +++ b/packages/context/tmux-context/tests/tmux-context.spec.ts @@ -101,9 +101,7 @@ function sessionAgent(session: Session, id = 'agent'): Agent { ctx: new Context(), followup: () => {}, steer: () => {}, - inject(input) { - session.append('user/message', input, { surfaceOp: 'append' }) - }, + inject: () => { throw new Error('tmux-context must append directly to the open step') }, cancel() {}, whenIdle: () => Promise.resolve(), } diff --git a/packages/context/workspace-context/src/index.ts b/packages/context/workspace-context/src/index.ts index 52dc76070c..10fb2b0c33 100644 --- a/packages/context/workspace-context/src/index.ts +++ b/packages/context/workspace-context/src/index.ts @@ -116,20 +116,20 @@ export function apply(ctx: Context, config: Config): void { { includeBaselineScopes: false, signal }, ) if (update !== undefined) { - agent.inject(update.context) + agent.session.append('user/message', update.context, { surfaceOp: 'append' }) applyInstructionVersionUpdates(agent.session, update.versionUpdates, instructionVersions) } const keepVisibleBaseline = !lifecycleWitnessed.has(agent.session) && hasVisibleBaseline(agent) if (!keepVisibleBaseline && instructions !== undefined && instructions.rendered.text.length > 0) { const baselineMessage = workspaceContextMessage(instructions.rendered.text) - agent.inject(createUserMessage({ + agent.session.append('user/message', createUserMessage({ content: baselineMessage.content, source: { kind: 'workspace-instructions', baseline: true, changes: [...baseline.changes.values()], }, - })) + }), { surfaceOp: 'append' }) } baselineLoaded.add(agent.session) }) diff --git a/packages/context/workspace-context/tests/workspace-context.spec.ts b/packages/context/workspace-context/tests/workspace-context.spec.ts index e23ea3205c..04e7fa4fdc 100644 --- a/packages/context/workspace-context/tests/workspace-context.spec.ts +++ b/packages/context/workspace-context/tests/workspace-context.spec.ts @@ -180,9 +180,7 @@ function stubAgent(cwd?: string, seed: SessionEvent[] = []): Agent { status: 'idle', followup: () => {}, steer: () => {}, - inject(input) { - session.append('user/message', input, { surfaceOp: 'append' }) - }, + inject: () => { throw new Error('workspace-context must append directly to the open step') }, cancel() {}, whenIdle: () => Promise.resolve(), } @@ -1164,7 +1162,7 @@ describe('workspace context request injection', () => { const ctx = new Context() await mountWorkspaceContext(ctx, { dshHome: home, maxBytes: 65536 }) ctx.on('agent/step', (agent) => { - agent.inject(createUserMessage({ content: [{ type: 'text', text: 'Available skills' }], source: { kind: 'plugin', plugin: 'test-skills' } })) + agent.session.append('user/message', createUserMessage({ content: [{ type: 'text', text: 'Available skills' }], source: { kind: 'plugin', plugin: 'test-skills' } }), { surfaceOp: 'append' }) }) const prefix = await composeBaselinePrefix(ctx, stubAgent(root)) diff --git a/packages/core/agent-loop/src/invariant.ts b/packages/core/agent-loop/src/invariant.ts index 5d96efc70c..d87655d1fc 100644 --- a/packages/core/agent-loop/src/invariant.ts +++ b/packages/core/agent-loop/src/invariant.ts @@ -6,7 +6,7 @@ import type { Context } from 'cordis' import { isAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm' import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants' -import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session' +import { foldRequestHeader } from '@deepseek-ai/dsh-session' const PACKAGE_NAME = '@deepseek-ai/dsh-agent-loop' @@ -17,8 +17,7 @@ export const inject = ['invariants'] /** Install the request-reconstruction contribution into its child registration fiber. */ const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => { - // Prepend prevents a short-circuiting replay listener from silencing the - // check; correctness itself comes from the sequence-bounded reconstruction. + // Prepend prevents a short-circuiting replay listener from silencing the check. ctx.on('llm/stream', (options: GenerateOptions, next) => { if (!isAgentLoopRequest(options)) return next() if (!Object.isFrozen(options)) fail('a loop-built request must be frozen') @@ -30,27 +29,16 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant } const events = session.events - let boundary = -1 - for (let index = events.length - 1; index >= 0; index -= 1) { - if (events[index]?.type === 'step/start') { - boundary = index - break - } - } - if (boundary === -1) { + if (!events.some(event => event.type === 'step/start')) { return fail('a loop-built request with no step/start in its session log') } const header = foldRequestHeader(events) if (header === undefined) { return fail('a loop-built request with no request/header event in its session log') } - const rebuilt = new Session( - SessionId(`${String(session.id)}-invariant-rebuild`), - structuredClone(events.slice(0, boundary)), - ) - const expected = rebuilt.deriveMessages() + const expected = session.deriveMessages() if (JSON.stringify(options.messages) !== JSON.stringify(expected)) { - fail(`llm request for session "${String(session.id)}" diverges from the boundary derivation (log-reconstruction desync)`) + fail(`llm request for session "${String(session.id)}" diverges from the dispatch-time durable derivation (log-reconstruction desync)`) } const headerMatches = options.model === header.config.model diff --git a/packages/core/agent-loop/tests/invariant.spec.ts b/packages/core/agent-loop/tests/invariant.spec.ts index 3abdf04aa0..d77ad3a7d6 100644 --- a/packages/core/agent-loop/tests/invariant.spec.ts +++ b/packages/core/agent-loop/tests/invariant.spec.ts @@ -42,12 +42,16 @@ describe('request-reconstruction invariant', () => { expect(() => { dispatch(ctx, options) }).not.toThrow() }) - it('uses the step boundary rather than content appended afterward', async () => { - const { ctx, session, boundary } = await requestSetup() + it('includes context appended inside the open step before dispatch', async () => { + const { ctx, session } = await requestSetup() session.append('user/message', createUserMessage({ - content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' }, + content: [{ type: 'text', text: '[step context]' }], source: { kind: 'plugin', plugin: 'x' }, }), { surfaceOp: 'append' }) - const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id }) + const options = loopRequest({ + model: 'm', + messages: Object.freeze(session.deriveMessages()), + sessionId: session.id, + }) expect(() => { dispatch(ctx, options) }).not.toThrow() }) @@ -57,16 +61,16 @@ describe('request-reconstruction invariant', () => { expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary]), sessionId: session.id })) }) .not.toThrow() expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([extra, ...boundary]), sessionId: session.id })) }) - .toThrow(/diverges from the boundary derivation/) + .toThrow(/diverges from the dispatch-time durable derivation/) expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary, extra]), sessionId: session.id })) }) - .toThrow(/diverges from the boundary derivation/) + .toThrow(/diverges from the dispatch-time durable derivation/) }) it('rejects message and header divergence', async () => { const { ctx, session, boundary } = await requestSetup() const divergent = [...boundary, { role: 'user', content: [{ type: 'text', text: 'phantom' }] }] expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze(divergent), sessionId: session.id })) }) - .toThrow(/diverges from the boundary derivation/) + .toThrow(/diverges from the dispatch-time durable derivation/) expect(() => { dispatch(ctx, loopRequest({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })) }) .toThrow(/diverges from the folded request header/) }) @@ -133,6 +137,6 @@ describe('request-reconstruction invariant', () => { messages: Object.freeze([{ role: 'user', content: [{ type: 'text', text: 'phantom' }] }]), sessionId: session.id, }) - expect(() => { dispatch(ctx, divergent) }).toThrow(/diverges from the boundary derivation/) + expect(() => { dispatch(ctx, divergent) }).toThrow(/diverges from the dispatch-time durable derivation/) }) }) diff --git a/packages/skill/tool-skill/src/index.ts b/packages/skill/tool-skill/src/index.ts index efc352fd3f..82741420c1 100644 --- a/packages/skill/tool-skill/src/index.ts +++ b/packages/skill/tool-skill/src/index.ts @@ -149,7 +149,7 @@ export function apply(ctx: Context, config: Config = {}): void { const catalog = history.published ? renderCatalogUpdate(skills, catalogDescriptionMaxLength) : renderCatalogMessage(skills, catalogDescriptionMaxLength) - agent.inject(catalog) + agent.session.append('user/message', catalog, { surfaceOp: 'append' }) }) } diff --git a/packages/skill/tool-skill/tests/tool-skill.spec.ts b/packages/skill/tool-skill/tests/tool-skill.spec.ts index f49d05e55a..c633e8faca 100644 --- a/packages/skill/tool-skill/tests/tool-skill.spec.ts +++ b/packages/skill/tool-skill/tests/tool-skill.spec.ts @@ -48,9 +48,7 @@ function agentForCwd(cwd: string): Agent { status: 'idle', followup: () => {}, steer: () => {}, - inject(input) { - session.append('user/message', input, { surfaceOp: 'append' }) - }, + inject: () => { throw new Error('step-boundary catalog must not use agent.inject()') }, cancel() {}, whenIdle: () => Promise.resolve(), } @@ -66,9 +64,7 @@ function sessionAgent(session: Session, id = 'tool-skill-agent'): Agent { ctx: new Context(), followup: () => {}, steer: () => {}, - inject(input) { - session.append('user/message', input, { surfaceOp: 'append' }) - }, + inject: () => { throw new Error('step-boundary catalog must not use agent.inject()') }, cancel() {}, whenIdle: () => Promise.resolve(), } @@ -200,7 +196,10 @@ describe('dsh-tool-skill', () => { content: 'User-only body.', }) ctx.on('agent/step', (agent) => { - agent.inject(createUserMessage({ content: [{ type: 'text', text: 'later contribution' }], source: { kind: 'plugin', plugin: 'later-contribution' } })) + agent.session.append('user/message', createUserMessage({ + content: [{ type: 'text', text: 'later contribution' }], + source: { kind: 'plugin', plugin: 'later-contribution' }, + }), { surfaceOp: 'append' }) }) const prefix = await composePrefix(ctx, '/workspace') diff --git a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts index 0eef295911..816f60d522 100644 --- a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts +++ b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts @@ -185,7 +185,12 @@ describe('dsh-jsonrpc plugin apply', () => { params: { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }] }, }) const response = await harness.waitForFrame(frame => frame.id === 2, 'prompt response') - expect(response.result).toEqual({ accepted: true }) + expect((response.result as { messageId?: unknown }).messageId).toBeTypeOf('string') + await harness.waitForFrame( + frame => frame.method === 'session.status' + && (frame.params as { status?: string } | undefined)?.status === 'idle', + 'idle session status', + ) expect(llmServer.requests).toHaveLength(1) const body = llmServer.requests[0] as { model: string; messages: { role: string }[] } @@ -195,9 +200,9 @@ describe('dsh-jsonrpc plugin apply', () => { // Notifications use the same transport and arrive as id-less frames. const notifications = harness.frames().filter(frame => frame.id === undefined) expect(notifications.some(frame => frame.method === 'session.event')).toBe(true) - expect(notifications.find(frame => frame.method === 'session.finished')).toMatchObject({ + expect(notifications.findLast(frame => frame.method === 'session.status')).toMatchObject({ jsonrpc: '2.0', - params: { sessionId: 'main', status: 'ok' }, + params: { sessionId: 'main', status: 'idle' }, }) } finally { await harness.dispose() diff --git a/packages/ui/jsonrpc/tests/server.spec.ts b/packages/ui/jsonrpc/tests/server.spec.ts index 7f634474da..373d0769da 100644 --- a/packages/ui/jsonrpc/tests/server.spec.ts +++ b/packages/ui/jsonrpc/tests/server.spec.ts @@ -8,7 +8,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' import AgentRegistry, { type Agent, type AgentHandle } from '@deepseek-ai/dsh-agent' -import SessionStore, { SessionId, type UserMessage } from '@deepseek-ai/dsh-session' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo' import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' @@ -127,12 +127,13 @@ describe('HarnessSdkServer', () => { }) as { serverInfo: { name: string } } expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime') - await server.handleRequest('session/prompt', { + const receipt = await server.handleRequest('session/prompt', { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }], }) + expect((receipt as { messageId?: unknown }).messageId).toBeTypeOf('string') - expect(llmServer.requests).toHaveLength(1) + await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) }) const body = llmServer.requests[0] as { model: string; messages: { role: string }[]; max_tokens?: number } expect(body.model).toBe('dsagent-model') expect(body.max_tokens).toBe(321) @@ -140,16 +141,18 @@ describe('HarnessSdkServer', () => { expect(body.messages.at(-1)?.role).toBe('user') expect(llmServer.headers[0]?.authorization).toBe('Bearer test-key') expect(transport.notifications.some(n => n.method === 'session.event')).toBe(true) - expect(transport.notifications.at(-1)).toMatchObject({ - method: 'session.finished', - params: { sessionId: 'main', status: 'ok' }, + await vi.waitFor(() => { + expect(transport.notifications.findLast(n => n.method === 'session.status')).toEqual({ + method: 'session.status', + params: { sessionId: 'main', status: 'idle' }, + }) }) await server.handleRequest('session/prompt', { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'again' }], }) - expect(llmServer.requests).toHaveLength(2) + await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(2) }) const orphanHandle = await ctx.agents.create({ sessionId: SessionId('orphan-session'), @@ -168,24 +171,17 @@ describe('HarnessSdkServer', () => { } }) - it('rejects overlapping prompts for one session without serializing other sessions', async () => { - let releaseMain: (() => void) | undefined - const firstMainIdle = new Promise((resolve) => { releaseMain = resolve }) - const mainWhenIdle = vi.fn<() => Promise>() - .mockReturnValueOnce(firstMainIdle) - .mockResolvedValue(undefined) + it('queues overlapping prompts for one session without blocking other sessions', async () => { const mainFollowup = vi.fn() const mainAgent = ({ id: SessionId('main'), followup: mainFollowup, - whenIdle: mainWhenIdle, - } satisfies Pick) as unknown as Agent + } satisfies Pick) as unknown as Agent const otherFollowup = vi.fn() const otherAgent = ({ id: SessionId('other'), followup: otherFollowup, - whenIdle: vi.fn(() => Promise.resolve()), - } satisfies Pick) as unknown as Agent + } satisfies Pick) as unknown as Agent const mainHandle = { agent: mainAgent, dispose: vi.fn(() => Promise.resolve()) } const otherHandle = { agent: otherAgent, dispose: vi.fn(() => Promise.resolve()) } const create = vi.fn(async (options: { sessionId: SessionId }) => @@ -202,20 +198,11 @@ describe('HarnessSdkServer', () => { contentBlocks: [{ type: 'text', text }], }) - const first = prompt('main', 'first') - await vi.waitFor(() => { expect(mainFollowup).toHaveBeenCalledOnce() }) + expect((await prompt('main', 'first')).messageId).toBeTypeOf('string') + expect((await prompt('main', 'overlap')).messageId).toBeTypeOf('string') + expect((await prompt('other', 'independent')).messageId).toBeTypeOf('string') - await expect(prompt('main', 'overlap')).rejects.toThrow('session already has an active prompt: main') - await expect(prompt('other', 'independent')).resolves.toEqual({ accepted: true }) - releaseMain?.() - await expect(first).resolves.toEqual({ accepted: true }) - await expect(prompt('main', 'sequential')).resolves.toEqual({ accepted: true }) - - mainWhenIdle.mockRejectedValueOnce(new Error('turn wait failed')) - await expect(prompt('main', 'failing')).rejects.toThrow('turn wait failed') - await expect(prompt('main', 'after failure')).resolves.toEqual({ accepted: true }) - - expect(mainFollowup).toHaveBeenCalledTimes(4) + expect(mainFollowup).toHaveBeenCalledTimes(2) expect(otherFollowup).toHaveBeenCalledOnce() await server.shutdown() expect(mainHandle.dispose).toHaveBeenCalledOnce() @@ -247,7 +234,7 @@ describe('HarnessSdkServer', () => { contentBlocks: [{ type: 'text', text }], }) - await expect(prompt('while live')).resolves.toEqual({ accepted: true }) + expect((await prompt('while live')).messageId).toBeTypeOf('string') live = false await expect(prompt('after detach')).rejects.toThrow('session agent was disposed outside the server: zombie') // The detached agent was never driven by the rejected prompt. @@ -255,59 +242,26 @@ describe('HarnessSdkServer', () => { await server.shutdown() }) - it('reports the final whole-agent outcome after later activity settles', async () => { + it('forwards whole-agent status without attributing a turn outcome', async () => { const ctx = new Context() await ctx.plugin(SessionStore) await ctx.plugin(AgentRegistry) const transport = new FakeTransport() - const server = new HarnessSdkServer(ctx, transport) as unknown as { - prompt(params: { sessionId: string; contentBlocks: { type: 'text'; text: string }[] }): Promise - sessions: Map - shutdown(): Promise> - } + const server = new HarnessSdkServer(ctx, transport) const session = ctx.sessions.create(SessionId('message-outcome')) const agent = ({ id: SessionId('message-outcome'), session, - followup(input: UserMessage) { - session.append('turn/start', { - turn: 1, - }) - session.append('user/message', input, { surfaceOp: 'append' }) - session.append('turn/end', { turn: 1, reason: { kind: 'max-tokens' } }) - session.append('turn/start', { - turn: 2, - }) - session.append('user/message', createUserMessage({ - content: [{ type: 'text', text: 'late metadata' }], - source: { kind: 'plugin', plugin: 'late-metadata' }, - }), { surfaceOp: 'append' }) - session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) - return input.id - }, - whenIdle: () => Promise.resolve(), - } satisfies Pick) as unknown as Agent - ctx.agents.register(agent) - server.sessions.set('message-outcome', { - handle: { agent, dispose: () => Promise.resolve() }, - lastTurnEnd: undefined, - activePrompt: false, - }) + } satisfies Pick) as Agent - await server.prompt({ - sessionId: 'message-outcome', - contentBlocks: [{ type: 'text', text: 'bounded prompt' }], - }) + ctx.emit('agent/status', agent, 'running') + ctx.emit('agent/status', agent, 'idle') - expect(transport.notifications.findLast(notification => notification.method === 'session.finished')) - .toEqual({ - method: 'session.finished', - params: { - sessionId: 'message-outcome', - status: 'ok', - reason: { kind: 'completed' }, - }, - }) + expect(transport.notifications.filter(notification => notification.method === 'session.status')) + .toEqual([ + { method: 'session.status', params: { sessionId: 'message-outcome', status: 'running' } }, + { method: 'session.status', params: { sessionId: 'message-outcome', status: 'idle' } }, + ]) await server.shutdown() await ctx.fiber.dispose() }) @@ -356,7 +310,7 @@ describe('HarnessSdkServer', () => { contentBlocks: [{ type: 'text', text: 'hello' }], }) - expect(llmServer.requests).toHaveLength(1) + await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) }) await server.shutdown() } finally { await ctx.fiber.dispose() @@ -881,43 +835,6 @@ describe('HarnessSdkServer', () => { }, ) - it('classifies defensive finish states', async () => { - const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-finish-states-')) - const ctx = await makeHarness(storageDir) - try { - const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as { - finishedStatus(reason: unknown): 'ok' | 'error' - shutdown(): Promise> - } - - expect(server.finishedStatus(undefined)).toBe('error') - expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('error') - expect(server.finishedStatus({ kind: 'error' })).toBe('error') - await server.shutdown() - } finally { - await ctx.fiber.dispose() - await rm(storageDir, { recursive: true, force: true }) - } - }) - - it('can report max-token turn termination as an accepted evaluation result', async () => { - const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-max-tokens-success-')) - const ctx = await makeHarness(storageDir) - try { - const server = new HarnessSdkServer(ctx, new FakeTransport(), { maxTokensAsSuccess: true }) as unknown as { - finishedStatus(reason: unknown): 'ok' | 'error' - shutdown(): Promise> - } - - expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('ok') - expect(server.finishedStatus({ kind: 'error' })).toBe('error') - await server.shutdown() - } finally { - await ctx.fiber.dispose() - await rm(storageDir, { recursive: true, force: true }) - } - }) - it('reports no adapter when the LLM service is absent', async () => { const ctx = new Context() try { @@ -1045,6 +962,6 @@ describe('HarnessSdkServer', () => { const server = new HarnessSdkServer(ctx, new FakeTransport()) await expect(server.shutdown()).rejects.toBe(listenerFailure) - expect(on).toHaveBeenCalledTimes(3) + expect(on).toHaveBeenCalledTimes(4) }) }) diff --git a/packages/ui/user-approval/src/index.ts b/packages/ui/user-approval/src/index.ts index e4f60da037..d037f7edb2 100644 --- a/packages/ui/user-approval/src/index.ts +++ b/packages/ui/user-approval/src/index.ts @@ -277,10 +277,10 @@ export class ApprovalService extends Service { const cause = overrideSource === 'delegation' ? 'inherited from the delegating session' : overrideIndex > headerIndex ? 'changed by the user' : 'changed by the operator/config' - agent.inject(createUserMessage({ + session.append('user/message', createUserMessage({ content: [{ type: 'text', text: `The approval policy changed from "${told}" to "${current}" (${cause}).` }], source: { kind: 'plugin', plugin: 'user-approval' }, - })) + }), { surfaceOp: 'append' }) }) } diff --git a/packages/ui/user-approval/tests/approval.spec.ts b/packages/ui/user-approval/tests/approval.spec.ts index f779b1e250..cdc8be1d7a 100644 --- a/packages/ui/user-approval/tests/approval.spec.ts +++ b/packages/ui/user-approval/tests/approval.spec.ts @@ -358,23 +358,27 @@ describe('approval policy (the approval/policy fold)', () => { * An agent stand-in over a REAL Session — gate, section, and narrator fold * real events; the opened turn satisfies request()'s enclosure precondition. */ - function sessionAgent(id: string): { agent: Agent; session: Session; injected: string[] } { + function sessionAgent(id: string): { agent: Agent; session: Session } { const session = new Session(SessionId(id)) session.append('turn/start', { turn: 1 }) - const injected: string[] = [] const agent = { id, session, - inject: (input: { content: Array<{ type: string; text: string }> }) => { - injected.push(input.content[0]?.text ?? '') - }, + inject: () => { throw new Error('step-boundary narration must not use agent.inject()') }, } as unknown as Agent - return { agent, session, injected } + return { agent, session } } const preStep = (ctx: Context, agent: Agent): Promise => agentEvents(ctx, agent).serial('agent/step', 1, 1, new AbortController().signal) + const narrations = (session: Session): string[] => session.events.flatMap(event => + event.type === 'user/message' + && event.data.source.kind === 'plugin' + && event.data.source.plugin === 'user-approval' + ? [event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')] + : []) + /** Append a `request/header` snapshot whose system text is exactly `system`. */ function appendHeader(session: Session, system: string): void { session.append('request/header', { header: { config: { provider: 'mock', model: 'mock' }, system }, reason: 'initial' }) @@ -482,20 +486,20 @@ describe('approval policy (the approval/policy fold)', () => { it('narrates nothing cold, once per coalesced switch (user wording), and idempotently', async () => { const ctx = new Context() await ctx.plugin(ApprovalService) - const { agent, session, injected } = sessionAgent('sess-narr-1') + const { agent, session } = sessionAgent('sess-narr-1') await preStep(ctx, agent) - expect(injected).toEqual([]) + expect(narrations(session)).toEqual([]) setApprovalPolicy(session, 'never') setApprovalPolicy(session, 'ask') setApprovalPolicy(session, 'never') await preStep(ctx, agent) - expect(injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).']) + expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).']) await preStep(ctx, agent) - expect(injected).toHaveLength(1) + expect(narrations(session)).toHaveLength(1) setApprovalPolicy(session, 'ask') setApprovalPolicy(session, 'never') await preStep(ctx, agent) - expect(injected).toHaveLength(1) + expect(narrations(session)).toHaveLength(1) }) it('reads what the model was told back from the folded header text after a restart', async () => { @@ -503,69 +507,69 @@ describe('approval policy (the approval/policy fold)', () => { // an ask default: the narrator attributes the change to the operator. const ctx = new Context() await ctx.plugin(ApprovalService) - const { agent, session, injected } = sessionAgent('sess-narr-2') + const { agent, session } = sessionAgent('sess-narr-2') appendHeader(session, `persona\n\n${NEVER_SENTENCE}\n${NEVER_MARKER}`) await preStep(ctx, agent) - expect(injected).toEqual(['The approval policy changed from "never" to "ask" (changed by the operator/config).']) + expect(narrations(session)).toEqual(['The approval policy changed from "never" to "ask" (changed by the operator/config).']) }) it('attributes a constructor-seeded policy event to delegation', async () => { const ctx = new Context() await ctx.plugin(ApprovalService) - const { agent, session, injected } = sessionAgent('sess-narr-inherited') + const { agent, session } = sessionAgent('sess-narr-inherited') appendHeader(session, ASK_MARKER) session.append('approval/policy', { policy: 'never', source: 'delegation' }) await preStep(ctx, agent) - expect(injected).toEqual(['The approval policy changed from "ask" to "never" (inherited from the delegating session).']) + expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (inherited from the delegating session).']) }) it('narrates a config default drift from the logged ask marker', async () => { const ctx = new Context() await ctx.plugin(ApprovalService, { policy: 'never' }) - const { agent, session, injected } = sessionAgent('sess-narr-3') + const { agent, session } = sessionAgent('sess-narr-3') appendHeader(session, `persona only\n${ASK_MARKER}`) await preStep(ctx, agent) - expect(injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the operator/config).']) + expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the operator/config).']) }) it('a pinned override survives a default change silently', async () => { const ctx = new Context() await ctx.plugin(ApprovalService, { policy: 'never' }) - const { agent, session, injected } = sessionAgent('sess-narr-4') + const { agent, session } = sessionAgent('sess-narr-4') appendHeader(session, `persona only\n${ASK_MARKER}`) setApprovalPolicy(session, 'ask') appendHeader(session, `persona only\n${ASK_MARKER}`) await preStep(ctx, agent) - expect(injected).toEqual([]) + expect(narrations(session)).toEqual([]) }) it('does not infer never from deployment prose that quotes the never sentence', async () => { const ctx = new Context() await ctx.plugin(ApprovalService) - const { agent, session, injected } = sessionAgent('sess-narr-spoof-prose') + const { agent, session } = sessionAgent('sess-narr-spoof-prose') appendHeader(session, `persona quotes this warning: ${NEVER_SENTENCE}\n${ASK_MARKER}`) await preStep(ctx, agent) - expect(injected).toEqual([]) + expect(narrations(session)).toEqual([]) }) it('treats a legacy header with no source-owned marker as untold', async () => { const ctx = new Context() await ctx.plugin(ApprovalService, { policy: 'never' }) - const { agent, session, injected } = sessionAgent('sess-narr-unmarked-header') + const { agent, session } = sessionAgent('sess-narr-unmarked-header') appendHeader(session, 'legacy persona-only header') await preStep(ctx, agent) - expect(injected).toEqual([]) + expect(narrations(session)).toEqual([]) }) it('uses the service marker after an earlier persona marker', async () => { const ctx = new Context() await ctx.plugin(ApprovalService) - const { agent, session, injected } = sessionAgent('sess-narr-spoof-marker') + const { agent, session } = sessionAgent('sess-narr-spoof-marker') appendHeader(session, `persona quotes ${NEVER_MARKER}\n${ASK_MARKER}`) await preStep(ctx, agent) - expect(injected).toEqual([]) + expect(narrations(session)).toEqual([]) }) it('disposes the service prompt section and pre-step narrator together (HMR safety)', async () => { @@ -581,7 +585,7 @@ describe('approval policy (the approval/policy fold)', () => { appendHeader(live.session, `persona\n${ASK_MARKER}`) setApprovalPolicy(live.session, 'never') await preStep(ctx, live.agent) - expect(live.injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).']) + expect(narrations(live.session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).']) appendHeader(afterDispose.session, `persona\n${ASK_MARKER}`) setApprovalPolicy(afterDispose.session, 'never') @@ -589,6 +593,6 @@ describe('approval policy (the approval/policy fold)', () => { expect(await sectionFor()).toBeUndefined() await preStep(ctx, afterDispose.agent) - expect(afterDispose.injected).toEqual([]) + expect(narrations(afterDispose.session)).toEqual([]) }) })