diff --git a/docs/persistence-catalog.md b/docs/persistence-catalog.md index 8878521ba8..9ac1f466f1 100644 --- a/docs/persistence-catalog.md +++ b/docs/persistence-catalog.md @@ -87,7 +87,11 @@ Sources: [`packages/core/session/src/types.ts:258`](../packages/core/session/src #### `agent/inbox/spliced` — log-only ```ts persistence-catalog -/** One normalized mutation of an agent's durable pending-message lists. */ +/** + * One normalized mutation of an agent's durable pending-message lists. + * Live dispatch precedes projection mutation, so synchronous observers may + * read the pre-splice inbox to recover the removed messages. + */ 'agent/inbox/spliced': { target: InboxTarget start: number @@ -97,7 +101,7 @@ Sources: [`packages/core/session/src/types.ts:258`](../packages/core/session/src } ``` -Source: [`packages/core/agent/src/types.ts:268`](../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:272`](../packages/core/agent/src/types.ts) ### `approval/*` diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index f33e36a3e3..144a12634c 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -144,10 +144,10 @@ export class ReactLoopAgent implements Agent { } private async admit(onTurnBoundary: boolean): Promise { - if (this.phase.kind !== 'running') throw new Error() + if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": admit outside running phase`) const signal = this.phase.abort.signal const claimed = [...this.inbox.nextStep] - const outboxLength = this.inbox.nextStep.length + const outboxLength = claimed.length const queued = onTurnBoundary ? this.inbox.nextTurn[0] : undefined if (queued !== undefined) claimed.push(queued) if (claimed.length === 0) return { kind: 'empty' } @@ -167,7 +167,7 @@ export class ReactLoopAgent implements Agent { /** Admitted input stays unowned until `turn/start` commits. */ private async turn(): Promise { - if (this.phase.kind === 'idle') throw new Error() + if (this.phase.kind === 'idle') throw new Error(`agent "${this.id}": turn without driver reservation`) const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController() const { signal } = abort const lastTurn = this.phase.kind === 'collecting' ? this.phase.lastTurn : this.phase.turn @@ -227,7 +227,7 @@ export class ReactLoopAgent implements Agent { } private async step(): Promise { - if (this.phase.kind !== 'running') throw new Error() + if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": step outside running phase`) const { turn, step, abort: { signal } } = this.phase signal.throwIfAborted() await this.loopCtx.serial(agentCarrier(this), 'agent/step', this, turn, step, signal) diff --git a/packages/core/agent/src/inbox.ts b/packages/core/agent/src/inbox.ts index d6dafa7e4e..2cdacfddfd 100644 --- a/packages/core/agent/src/inbox.ts +++ b/packages/core/agent/src/inbox.ts @@ -44,6 +44,9 @@ export class Inbox { /** * Apply standard splice semantics and durably record the normalized result. + * The durable event commits before the live projection mutates, so synchronous + * `session/event` observers see the pre-splice lists and can reconstruct the + * removed messages from the normalized coordinates. * @param target - pending list to mutate. * @param start - splice position. * @param deleteCount - maximum number of messages to remove. @@ -59,12 +62,14 @@ export class Inbox { outcome?: 'admitted' | 'canceled', ): UserMessage[] { const inbox = this.state[target] - const offset = Math.trunc(start) || 0 + const truncatedStart = Math.trunc(start) + const offset = Number.isNaN(truncatedStart) ? 0 : truncatedStart const actualStart = offset < 0 ? Math.max(inbox.length + offset, 0) : Math.min(offset, inbox.length) + const truncatedDeleteCount = Math.trunc(deleteCount) const actualDeleteCount = Math.min( - Math.max(Math.trunc(deleteCount) || 0, 0), + Math.max(Number.isNaN(truncatedDeleteCount) ? 0 : truncatedDeleteCount, 0), inbox.length - actualStart, ) if (actualDeleteCount === 0 && inserted.length === 0) return [] diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index b1451768ca..29f536e2c0 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -264,7 +264,11 @@ declare module 'cordis' { declare module '@deepseek-ai/dsh-session' { interface SessionEventMap { - /** One normalized mutation of an agent's durable pending-message lists. */ + /** + * One normalized mutation of an agent's durable pending-message lists. + * Live dispatch precedes projection mutation, so synchronous observers may + * read the pre-splice inbox to recover the removed messages. + */ 'agent/inbox/spliced': { target: InboxTarget start: number diff --git a/packages/goal/goal-session/src/index.ts b/packages/goal/goal-session/src/index.ts index 583734a3db..e65fb41aa6 100644 --- a/packages/goal/goal-session/src/index.ts +++ b/packages/goal/goal-session/src/index.ts @@ -418,6 +418,7 @@ export function apply(ctx: Context): void { attempt.stale = true if (attempt.phase === 'admitted' && state.agent.status === 'running') { state.agent.cancel({ kind: 'parent' }) + waits.push(state.agent.whenIdle()) } } 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 4cff18d80f..7cd8090e07 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -735,7 +735,7 @@ describe('same-session goal driving', () => { activation: 'disarmed', roundsStarted: 1, }) - await test.agent.whenIdle() + expect(test.agent.status).toBe('idle') expect(test.adapter.requests).toHaveLength(1) }) diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 10c3102122..6f600f114a 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -1502,6 +1502,20 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro const view = viewFor(ctx, event, callId => openCalls.get(session.id)?.get(callId) ?? backscanArgs(session.events, callId)) queue.push(frame({ type: 'session/event', sessionId: session.id, event, ...view === undefined ? {} : { view } })) + if (event.type === 'agent/inbox/spliced' && event.data.target === 'next-turn') { + const agent = ctx.agents.get(session.id) + if (agent?.session === session) { + queue.push(frame({ + type: 'session/queue', + sessionId: session.id, + items: agent.inbox.nextTurn.toSpliced( + event.data.start, + event.data.removedCount ?? 0, + ...event.data.inserted, + ), + })) + } + } }), ctx.on('session/created', (session: Session) => { subscribeSession(queue, session) diff --git a/packages/host/apiproxy/src/api/events.ts b/packages/host/apiproxy/src/api/events.ts index 859f4e1bc2..c6742e5352 100644 --- a/packages/host/apiproxy/src/api/events.ts +++ b/packages/host/apiproxy/src/api/events.ts @@ -62,9 +62,9 @@ export type MuxFrame = | { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] } | { type: 'question/resolved'; sessionId: SessionId; questionRpcId: RpcId; outcome: 'answered' | 'cancelled' } /** - * Complete next-turn queue baseline emitted when a mux stream opens. Live - * mutations arrive through durable `agent/inbox/spliced` session events. - * Pending next-step input is outside this Web queue projection. + * Complete next-turn queue snapshot emitted when a mux stream opens and + * after every live next-turn mutation. Pending next-step input is outside + * this Web queue projection. */ | { type: 'session/queue'; sessionId: SessionId; items: UserMessage[] } /** diff --git a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts index 73a3a793e6..6008b16922 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -20,7 +20,7 @@ import ToolRegistry from '@deepseek-ai/dsh-tools' import UserInteractionService from '@deepseek-ai/dsh-user-interaction' import CommandService from '@deepseek-ai/dsh-commands' import SkillService from '@deepseek-ai/dsh-skill' -import type { HostFrame, MuxFrame } from '../src/api/index.ts' +import type { HostFrame } from '../src/api/index.ts' import type { RpcRequest, RpcResponse } from '../src/api/rpc.ts' import { RpcId } from '../src/api/rpc.ts' import { createApiProxy } from '../src/api-proxy.ts' @@ -85,6 +85,13 @@ async function collect(iterable: AsyncIterable>, count: number, return frames } +/** Read the next payload from an open stream. */ +async function nextFrame(iterator: AsyncIterator>): Promise { + const result = await iterator.next() + if (result.done) throw new Error('stream ended') + return result.value.payload +} + describe('command.list', () => { it('serves the addressed agent\'s name-sorted catalog', async () => { const ctx = await harness() @@ -332,24 +339,41 @@ describe('session.updateQueue', () => { }) describe('session/queue frames', () => { - it('publishes the durable next-turn baseline without duplicating message identity', async () => { + it('publishes authoritative next-turn snapshots without duplicating message identity', async () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS) const agent = stubAgent(ctx) const queued = inboxMessage('m-1', 'queued prompt') + const edited = inboxMessage('m-1', 'edited prompt') const steering = inboxMessage('m-2', 'steering prompt') agent.inbox.splice('next-turn', 0, 0, [queued]) agent.inbox.splice('next-step', 0, 0, [steering]) const abort = new AbortController() - const frames = await collect( - api.events.mux({ rpcId: RpcId('t-mux-baseline'), payload: {} }, abort.signal), 2, abort) + const iterator = api.events.mux({ + rpcId: RpcId('t-mux-baseline'), + payload: {}, + }, abort.signal)[Symbol.asyncIterator]() + const frames = [ + await nextFrame(iterator), + await nextFrame(iterator), + ] + agent.inbox.splice('next-turn', 0, 1, [edited]) + frames.push(await nextFrame(iterator), await nextFrame(iterator)) + abort.abort() + await iterator.return?.() + expect(frames.filter(frame => frame.type === 'session/queue')).toEqual([ { type: 'session/queue', sessionId: agent.id, items: [queued], }, + { + type: 'session/queue', + sessionId: agent.id, + items: [edited], + }, ]) }) }) diff --git a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts index 4916f93e32..f6acfe81d4 100644 --- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts @@ -10,7 +10,7 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' import { z } from 'zod' -import AgentRegistry from '@deepseek-ai/dsh-agent' +import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent' import type { Agent } from '@deepseek-ai/dsh-agent' import { createUserMessage } from '@deepseek-ai/dsh-llm' import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' @@ -53,9 +53,8 @@ async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: await ctx.plugin(AgentRegistry) if (withRegistry) await ctx.plugin(SessionProjectionRegistry) const session = ctx.sessions.create() - // history resolves the agent first; a live structural stub is enough (only - // .session is read on this path). - ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent) + // The gateway reads both the session and durable inbox baseline. + ctx.agents.register({ id: session.id, session, inbox: new Inbox(session), status: 'idle', ctx } as Agent) return { ctx, session } }