diff --git a/packages/acp/acp/tests/turns.spec.ts b/packages/acp/acp/tests/turns.spec.ts index c42ef7e5c0..a4ae22080e 100644 --- a/packages/acp/acp/tests/turns.spec.ts +++ b/packages/acp/acp/tests/turns.spec.ts @@ -70,10 +70,13 @@ describe('ACP prompt lifecycle', () => { const sessionId = await newSession(harness) const agent = harness.ctx.agents.get(SessionId(sessionId))! let injected = false - harness.ctx.on('agent/inbox/enqueue', (subject) => { - if (subject === agent && !injected) { + harness.ctx.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'agent/inbox/spliced' + && event.data.inserted.some(message => message.source.kind === 'user') && !injected) { injected = true - agent.inject(createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'test' } })) + queueMicrotask(() => { + agent.inject(createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'test' } })) + }) } }) @@ -86,30 +89,44 @@ describe('ACP prompt lifecycle', () => { harness = await makeBridgeHarness({ script: ['hang'] }) const sessionId = await newSession(harness) const agent = harness.ctx.agents.get(SessionId(sessionId))! - let inserted = false - harness.ctx.on('agent/inbox/enqueue', (subject, item) => { - if (subject !== agent || item.message.source.kind !== 'user' || inserted) return - inserted = true - const source = { kind: 'plugin', plugin: 'test' } as const - agent.session.append('turn/start', { turn: 1 }) - agent.session.append('user/message', createUserMessage({ - content: [{ type: 'text', text: 'autonomous work' }], - source, - }), { surfaceOp: 'append' }) - agent.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + let autonomousStarted!: () => void + const started = new Promise((resolve) => { autonomousStarted = resolve }) + harness.ctx.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'assistant/chunk') autonomousStarted() }) + agent.followup(createUserMessage({ + content: [{ type: 'text', text: 'autonomous work' }], + source: { kind: 'plugin', plugin: 'test' }, + })) + await started let settled = false const prompt = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }) .finally(() => { settled = true }) await vi.waitFor(() => { - expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2) + expect(agent.session.events.filter(event => event.type === 'agent/inbox/spliced' + && event.data.inserted.length > 0)).toHaveLength(2) }) expect(settled).toBe(false) await harness.client.cancel({ sessionId }) await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }) }) + it('correlates a prompt whose admitted history is replaced', async () => { + harness = await makeBridgeHarness({ script: [textResponse('rewritten answer')] }) + harness.ctx.on('agent/prompt-submit', async () => ({ + kind: 'allow', + messages: [createUserMessage({ + content: [{ type: 'text', text: 'rewritten prompt' }], + source: { kind: 'plugin', plugin: 'test' }, + })], + })) + const sessionId = await newSession(harness) + + await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'original' }] })) + .resolves.toEqual({ stopReason: 'end_turn' }) + }) + it('frees the prompt slot when the agent rejects the send synchronously', async () => { harness = await makeBridgeHarness({ script: [] }) const sessionId = await newSession(harness) @@ -143,7 +160,8 @@ describe('ACP prompt lifecycle', () => { await harness.client.cancel({ sessionId }) await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' }) await agent.whenIdle() - expect(agent.session.events.findLast(event => event.type === 'turn/end')?.data.reason).toEqual({ kind: 'aborted' }) + expect(agent.session.events.findLast(event => event.type === 'turn/end')?.data.reason) + .toEqual({ kind: 'aborted', reason: { kind: 'user' } }) }) it('an idle cancel does not affect the following prompt', async () => { @@ -203,4 +221,29 @@ describe('ACP prompt lifecycle', () => { // The blocked prompt opened no turn and streamed nothing. expect(messageText(harness)).toBe('') }) + + it('discards and settles a turnless prompt retained by its admission policy', async () => { + harness = await makeBridgeHarness({ script: [] }) + harness.ctx.on('agent/prompt-submit', async () => ({ + kind: 'block' as const, + reason: 'defer forever', + keepInbox: true, + })) + const sessionId = await newSession(harness) + const agent = harness.ctx.agents.get(SessionId(sessionId))! + + await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })) + .resolves.toEqual({ stopReason: 'cancelled' }) + expect(agent.status).toBe('idle') + expect(agent.session.events.some(event => event.type === 'turn/start')).toBe(false) + }) + + it('settles a prompt when admission fails before opening a turn', async () => { + harness = await makeBridgeHarness({ script: [] }) + harness.ctx.on('agent/prompt-submit', async () => { throw new Error('admission exploded') }) + const sessionId = await newSession(harness) + + await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })) + .resolves.toEqual({ stopReason: 'cancelled' }) + }) }) diff --git a/packages/client/connection/src/client/api.ts b/packages/client/connection/src/client/api.ts index b58718134d..23aa0d20c0 100644 --- a/packages/client/connection/src/client/api.ts +++ b/packages/client/connection/src/client/api.ts @@ -12,7 +12,7 @@ export type { WorkspaceApi, WorkspaceId, WorkspaceView, CommandsApi, CommandDescriptor, SkillsApi, SkillEntry, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning, - InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels, + ModelReasoningEffort, ModelTarget, QueueAction, SessionModels, GoalsApi, GoalRef, } from '@deepseek-ai/dsh-host-apiproxy/api' export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation' @@ -27,6 +27,7 @@ export { RpcId, transportError } from '@deepseek-ai/dsh-host-apiproxy/api' export { AbstractApiClient } from '@deepseek-ai/dsh-host-apiproxy/client' export type { IApiClient } from '@deepseek-ai/dsh-host-apiproxy/client' export type { SessionId, SessionEvent } from '@deepseek-ai/dsh-session/types' +export type { MessageId } from '@deepseek-ai/dsh-llm/brand' export type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm/types' import type { RpcResponse, RpcResult } from '@deepseek-ai/dsh-host-apiproxy/api' diff --git a/packages/client/connection/src/client/index.ts b/packages/client/connection/src/client/index.ts index 7d26b1c526..1599084c1c 100644 --- a/packages/client/connection/src/client/index.ts +++ b/packages/client/connection/src/client/index.ts @@ -17,7 +17,7 @@ export type { ToolCallView, ToolResultView, WorkspaceApi, WorkspaceId, WorkspaceView, CommandsApi, CommandDescriptor, SkillsApi, SkillEntry, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning, - InboxItemId, ModelReasoningEffort, ModelTarget, QueueAction, QueuedInboxItem, SessionModels, + MessageId, ModelReasoningEffort, ModelTarget, QueueAction, SessionModels, RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode, ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt, IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk, diff --git a/packages/client/runtime/src/client/contract/session.ts b/packages/client/runtime/src/client/contract/session.ts index 82bde108b4..0e75648ea1 100644 --- a/packages/client/runtime/src/client/contract/session.ts +++ b/packages/client/runtime/src/client/contract/session.ts @@ -9,7 +9,7 @@ */ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { - InboxItemId, QueueAction, RpcResult, SessionId, + MessageId, QueueAction, RpcResult, SessionId, } from '@deepseek-ai/dsh-client-connection/client' import type { ConversationSnapshot } from '../sessions/conversation.ts' import type { ObservableSnapshot } from './store.ts' @@ -44,7 +44,7 @@ export interface ISession { * @param action - edit or remove operation. * @returns acceptance, or a business/transport error. */ - updateQueue(itemId: InboxItemId, action: QueueAction): Promise> + updateQueue(itemId: MessageId, action: QueueAction): Promise> /** * Cancel the running turn. * @returns acceptance, or the business error. diff --git a/packages/client/runtime/src/client/sessions/conversation.ts b/packages/client/runtime/src/client/sessions/conversation.ts index f68a271c63..55cc3fc73e 100644 --- a/packages/client/runtime/src/client/sessions/conversation.ts +++ b/packages/client/runtime/src/client/sessions/conversation.ts @@ -7,7 +7,7 @@ import type { CommandId } from '@deepseek-ai/dsh-commands/brand' import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { TodoItem } from '@deepseek-ai/dsh-session/types' import type { - InboxItemId, RpcError, SessionId, ToolCallView, ToolResultView, + MessageId, RpcError, SessionId, ToolCallView, ToolResultView, } from '@deepseek-ai/dsh-client-connection/client' import type { PendingInteraction } from './pending.ts' export type { TodoItem } @@ -218,7 +218,7 @@ export interface RunningToolCall { /** One independently addressable row from the transient queue snapshot. */ export interface QueuedMessage { - readonly id: InboxItemId + readonly id: MessageId readonly preview: string /** Complete editable text; null when the message contains non-text blocks. */ readonly text: string | null diff --git a/packages/client/runtime/tests/queue-store.spec.ts b/packages/client/runtime/tests/queue-store.spec.ts index 192885b66b..ce684694f5 100644 --- a/packages/client/runtime/tests/queue-store.spec.ts +++ b/packages/client/runtime/tests/queue-store.spec.ts @@ -4,10 +4,10 @@ * projection, and snapshot reference stability. */ import { describe, expect, it } from 'vitest' -import { createUserMessage } from '@deepseek-ai/dsh-llm' +import { createUserMessage, freezeMessage } from '@deepseek-ai/dsh-llm' import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { - InboxItemId, MuxFrame, RpcId, SessionId, + MessageId, MuxFrame, RpcId, SessionId, } from '@deepseek-ai/dsh-client-connection/client' import { Session } from '../src/client/sessions/session.ts' import { SessionManager } from '../src/client/sessions/manager.ts' @@ -16,7 +16,7 @@ import { FakeApiClient } from './fake-api.ts' const SID = 'fk-q1' as SessionId const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }] const rid = (id: string): RpcId => id as RpcId -const iid = (id: string): InboxItemId => id as InboxItemId +const mid = (id: string): MessageId => id as MessageId interface QueueFixture { id: string @@ -29,12 +29,12 @@ function queueFrame(items: QueueFixture[]): MuxFrame { return { type: 'session/queue', sessionId: SID, - items: items.map(item => ({ - id: iid(item.id), - message: createUserMessage({ + items: items.map(item => freezeMessage({ + ...createUserMessage({ content: item.content ?? text(item.body), source: { kind: 'user', rpcId: rid(`rpc-${item.id}`) } as never, }), + id: mid(item.id), })), } } diff --git a/packages/client/ui-conversation/src/client/input/contract.ts b/packages/client/ui-conversation/src/client/input/contract.ts index 317adc0ec5..0c1d72dbc2 100644 --- a/packages/client/ui-conversation/src/client/input/contract.ts +++ b/packages/client/ui-conversation/src/client/input/contract.ts @@ -191,7 +191,7 @@ export interface InputState { readonly occurrences: readonly Occurrence[] /** Live paste-match attempt (absent when no paste is matchable). */ readonly paste?: PasteAttemptState - /** Read-only queue projection (session/queued frames + connect snapshot). */ + /** Read-only queue projection from the reconnect baseline and durable inbox events. */ readonly queue: readonly QueuedMessage[] } diff --git a/packages/context/time-context/tests/time-context.spec.ts b/packages/context/time-context/tests/time-context.spec.ts index ea13a36728..377e11abd7 100644 --- a/packages/context/time-context/tests/time-context.spec.ts +++ b/packages/context/time-context/tests/time-context.spec.ts @@ -41,10 +41,7 @@ function sessionAgent(session: Session, id = 'agent'): Agent { options: {}, session, status: 'running', - acceptsNextStep: true, ctx: new Context(), - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { diff --git a/packages/context/tmux-context/tests/tmux-context.spec.ts b/packages/context/tmux-context/tests/tmux-context.spec.ts index 4c96fe0a78..a3a78c0153 100644 --- a/packages/context/tmux-context/tests/tmux-context.spec.ts +++ b/packages/context/tmux-context/tests/tmux-context.spec.ts @@ -97,15 +97,12 @@ function sessionAgent(session: Session, id = 'agent'): Agent { options: {}, session, status: 'running', - acceptsNextStep: true, ctx: new Context(), followup: () => {}, steer: () => {}, - updateInbox: () => 'not-found', inject(input) { session.append('user/message', input, { surfaceOp: 'append' }) }, - send: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } diff --git a/packages/context/workspace-context/tests/workspace-context.spec.ts b/packages/context/workspace-context/tests/workspace-context.spec.ts index ad6eb500a7..cb4ca3533f 100644 --- a/packages/context/workspace-context/tests/workspace-context.spec.ts +++ b/packages/context/workspace-context/tests/workspace-context.spec.ts @@ -177,9 +177,6 @@ function stubAgent(cwd?: string, seed: SessionEvent[] = []): Agent { options: {}, session, status: 'idle', - acceptsNextStep: false, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index d42c2790d3..f33e36a3e3 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -1,9 +1,6 @@ /** - * Concrete Agent loop over two pending-input lists: queued prompts each open a - * turn that logs its admitted input after `turn/start` commits, while steering - * and injected context enter through the outbox at step boundaries. Every - * request is derived from the session log. - * + * Default Agent driver over queued turns and step-boundary input. Every request + * is derived from the session log. * @module dsh-agent-loop/agent */ @@ -13,9 +10,10 @@ import type { AgentOptions, AgentStatus, CancelOptions, + InboxTarget, RequestErrorAction, } from '@deepseek-ai/dsh-agent' -import { agentCarrier, agentEvents, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent' +import { Inbox, agentCarrier, agentEvents, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent' import type { GenerateOptions, LlmCallConfig, Message, PreparedLlmCall } from '@deepseek-ai/dsh-llm' import { BlockAssembler, @@ -27,7 +25,7 @@ import { } from '@deepseek-ai/dsh-llm' import type { Scope } from '@deepseek-ai/dsh-scope' import { createScope } from '@deepseek-ai/dsh-scope' -import type { AssistantMessage, Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session' +import type { Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session' import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session' import { renderPrompt } from '@deepseek-ai/dsh-system-prompt' import type { Context } from 'cordis' @@ -40,25 +38,17 @@ type Phase = type Admission = | { kind: 'empty' } - | { kind: 'admitted'; claimed: UserMessage[]; messages: UserMessage[] } + | { kind: 'admitted'; messages: UserMessage[] } | { kind: 'blocked' } -/** - * The concrete {@link Agent}: each `run()` owns one turn and repeats model - * steps while tools or steering require another request. - */ +/** Drives one session through turn and step boundaries. */ export class ReactLoopAgent implements Agent { - /** Prompts awaiting individual turns. */ - private queued: UserMessage[] = [] - /** Input taken into the session log at step boundaries. */ - private outbox: UserMessage[] = [] - + readonly inbox: Inbox private phase: Phase private driverDone: Promise = Promise.resolve() /** The agent-scoped registration boundary; the lifecycle owner unwinds it after the driver exits. */ readonly scope: Scope - /** The agent's scoped composition context ({@link Agent.ctx}). */ readonly ctx: Context /** Whether this loop instance has appended its initial/resume request anchor. */ @@ -70,13 +60,13 @@ export class ReactLoopAgent implements Agent { public readonly options: AgentOptions, public readonly session: Session, ) { + this.inbox = new Inbox(session) const lastTurn = session.events.findLast(event => event.type === 'turn/start')?.data.turn ?? 0 this.phase = { kind: 'idle', lastTurn } this.scope = createScope(loopCtx, this) this.ctx = this.scope.ctx.extend({ agent: this }) } - /** Last activity state published to observers. */ get status(): AgentStatus { return this.phase.kind === 'idle' ? 'idle' : 'running' } @@ -91,49 +81,32 @@ export class ReactLoopAgent implements Agent { } } - /** Accept and route one unified send item. */ - private send(message: UserMessage, target: 'next-turn' | 'next-step', wakeup: boolean): void { - this.session.append('agent/inbox/added', message) + private send(message: UserMessage, target: InboxTarget, wakeup: boolean): void { // Waking input cannot join an aborted admission or turn, so it starts the next turn. const wakingAfterAbort = wakeup && this.phase.kind !== 'idle' && this.phase.abort.signal.aborted - const inbox = target === 'next-turn' || wakingAfterAbort ? this.queued : this.outbox - inbox.push(message) - if (wakeup) { - this.scheduleKick() - } + const resolvedTarget = wakingAfterAbort ? 'next-turn' : target + this.inbox.splice(resolvedTarget, Infinity, 0, [message]) + if (wakeup) this.scheduleKick() } - /** Queue one ordinary prompt turn and wake the driver. */ followup(input: UserMessage): void { this.send(input, 'next-turn', true) } - /** Steer the open turn, falling back to a waking prompt while idle. */ steer(input: UserMessage): void { this.send(input, 'next-step', true) } - /** Append model-facing context without waking the driver. */ inject(input: UserMessage): void { this.send(input, 'next-step', false) } - /** - * Clear all pending work and abort the active turn; the first cause wins. - * The cause is signal payload for observers and the durable turn/end - * classification — it selects no machine behavior. Teardown is just - * `cancel({kind:'disposed'})` + driver join + {@link scope} dispose, all - * owned by the factory. - */ cancel(cause: AgentCancelCause, options: CancelOptions = {}): void { if (!options.keepInbox) { - for (const message of [...this.outbox.splice(0), ...this.queued.splice(0)]) { - emitAgentEvent(this.loopCtx, this, 'agent/inbox/canceled', message) - } - } - if (this.phase.kind !== 'idle') { - this.phase.abort.abort(cause) + this.inbox.splice('next-step', 0, this.inbox.nextStep.length, [], 'canceled') + this.inbox.splice('next-turn', 0, this.inbox.nextTurn.length, [], 'canceled') } + if (this.phase.kind !== 'idle') this.phase.abort.abort(cause) } /** Reserve a driver before deferring idle admission. */ @@ -147,7 +120,6 @@ export class ReactLoopAgent implements Agent { }) } - /** Resolve after the current driver and synchronous replacement chain exits. */ async whenIdle(): Promise { let driver: Promise do { @@ -171,13 +143,12 @@ export class ReactLoopAgent implements Agent { } } - /** Claim and admit the next queued prompt, then start its turn. */ private async admit(onTurnBoundary: boolean): Promise { if (this.phase.kind !== 'running') throw new Error() const signal = this.phase.abort.signal - const claimed = this.outbox.slice() - const outboxLength = this.outbox.length - const queued = onTurnBoundary ? this.queued[0] : undefined + const claimed = [...this.inbox.nextStep] + const outboxLength = this.inbox.nextStep.length + const queued = onTurnBoundary ? this.inbox.nextTurn[0] : undefined if (queued !== undefined) claimed.push(queued) if (claimed.length === 0) return { kind: 'empty' } const decision = await agentEvents(this.loopCtx, this).waterfall( @@ -186,34 +157,31 @@ export class ReactLoopAgent implements Agent { ) signal.throwIfAborted() if (decision.kind === 'allow') { - this.outbox.splice(0, outboxLength) - if (queued !== undefined) this.queued.shift() - return { kind: 'admitted', claimed, messages: decision.messages } - } else { - this.cancel({ kind: 'hook', reason: decision.reason }, { keepInbox: decision.keepInbox }) - return { kind: 'blocked' } + this.inbox.splice('next-step', 0, outboxLength, [], 'admitted') + if (queued !== undefined) this.inbox.splice('next-turn', 0, 1, [], 'admitted') + return { kind: 'admitted', messages: decision.messages } } + this.cancel({ kind: 'hook', reason: decision.reason }, { keepInbox: decision.keepInbox }) + return { kind: 'blocked' } } - /** - * Run one turn and any request-error retry. `admitted` input enters the log - * only after `turn/start` commits; until then it has no owner state to unwind. - */ + /** Admitted input stays unowned until `turn/start` commits. */ private async turn(): Promise { if (this.phase.kind === 'idle') throw new Error() 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 const phase = { kind: 'running' as const, abort, turn: lastTurn, step: 0 } this.setPhase(phase) - if (abort.signal.aborted) return this.outbox.length > 0 || this.queued.length > 0 + if (signal.aborted) return this.inbox.hasPending let admission: Admission try { admission = await this.admit(true) if (admission.kind !== 'admitted') return false - abort.signal.throwIfAborted() + signal.throwIfAborted() } catch (error: unknown) { // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- cancel may abort while admission awaits - if (abort.signal.aborted) return this.outbox.length > 0 || this.queued.length > 0 + if (signal.aborted) return this.inbox.hasPending throw error } const turn = ++phase.turn @@ -222,14 +190,11 @@ export class ReactLoopAgent implements Agent { try { while (true) { if (admission.kind === 'admitted') { - for (const message of admission.claimed) { - emitAgentEvent(this.loopCtx, this, 'agent/inbox/admitted', message) - } for (const message of admission.messages) { this.session.append('user/message', message, { surfaceOp: 'append' }) } } - abort.signal.throwIfAborted() + signal.throwIfAborted() const step = ++phase.step this.session.append('step/start', { turn, step }) try { @@ -237,34 +202,30 @@ export class ReactLoopAgent implements Agent { } finally { this.session.append('step/end', { turn, step }) } - abort.signal.throwIfAborted() - if (turnEnds && this.outbox.length === 0) { - await this.loopCtx.serial(agentCarrier(this), 'agent/turn-stopping', this, turn, abort.signal) - abort.signal.throwIfAborted() + signal.throwIfAborted() + if (turnEnds && this.inbox.nextStep.length === 0) { + await this.loopCtx.serial(agentCarrier(this), 'agent/turn-stopping', this, turn, signal) + signal.throwIfAborted() } admission = await this.admit(false) if (admission.kind === 'blocked') { - turnEnds = { kind: 'aborted', reason: abort.signal.reason as AgentCancelCause } + turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause } return false } - abort.signal.throwIfAborted() + signal.throwIfAborted() if (admission.kind === 'empty' && turnEnds) break } } catch (error: unknown) { // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- cancel may abort during any awaited turn operation - if (abort.signal.aborted) turnEnds = { kind: 'aborted', reason: abort.signal.reason as AgentCancelCause } + if (signal.aborted) turnEnds = { kind: 'aborted', reason: signal.reason as AgentCancelCause } else turnEnds = { kind: 'error', error: errorChain(error) } } finally { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- the turn is always ended in this block this.session.append('turn/end', { turn, reason: turnEnds! }) } - return this.outbox.length > 0 || this.queued.length > 0 + return this.inbox.hasPending } - /** - * Run the `agent/step` extension point, commit pending input, derive one - * request, and execute its tool calls inside one durable step boundary. - */ private async step(): Promise { if (this.phase.kind !== 'running') throw new Error() const { turn, step, abort: { signal } } = this.phase @@ -275,11 +236,9 @@ export class ReactLoopAgent implements Agent { signal.throwIfAborted() const system = renderPrompt(assembly) - let message: AssistantMessage while (true) { - const boundaryMessages = this.session.deriveMessages() const { request, preparedCall } = await this.buildRequest( - turn, step, assembly.tools, system, boundaryMessages, signal, + turn, step, assembly.tools, system, this.session.deriveMessages(), signal, ) const assembler = new BlockAssembler() const chunkSeqs: number[] = [] @@ -287,8 +246,7 @@ export class ReactLoopAgent implements Agent { signal.throwIfAborted() for await (const chunk of stream) { signal.throwIfAborted() - const chunkEvent = this.session.append('assistant/chunk', { turn, step, chunk }) - chunkSeqs.push(chunkEvent.seq) + chunkSeqs.push(this.session.append('assistant/chunk', { turn, step, chunk }).seq) assembler.push(chunk) } signal.throwIfAborted() @@ -305,47 +263,38 @@ export class ReactLoopAgent implements Agent { () => Promise.resolve(undefined), ) signal.throwIfAborted() - if (action?.kind !== 'retry') { - return { kind: 'error', error: finish.failure } - } - } else { - message = createAssistantMessage({ - content: assembler.blocks(), - source: { - provider: request.provider, - model: request.model, - ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {}, - }, - }) - this.session.append( - 'assistant/message', - { - turn, - step, - message, - ...assembler.usage === undefined ? {} : { usage: assembler.usage }, - }, - { surfaceOp: 'append', sourceEventSeqs: chunkSeqs }, - ) - if (finish.kind === 'max-tokens') { - return { kind: 'max-tokens' } - } - break + if (action?.kind !== 'retry') return { kind: 'error', error: finish.failure } + continue } - } - const toolCalls = message.content.filter(block => block.type === 'tool-call') - let result: TurnEndReason | null - if (toolCalls.length > 0) { + const message = createAssistantMessage({ + content: assembler.blocks(), + source: { + provider: request.provider, + model: request.model, + ...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {}, + }, + }) + this.session.append( + 'assistant/message', + { + turn, + step, + message, + ...assembler.usage === undefined ? {} : { usage: assembler.usage }, + }, + { surfaceOp: 'append', sourceEventSeqs: chunkSeqs }, + ) + if (finish.kind === 'max-tokens') return { kind: 'max-tokens' } + + const toolCalls = message.content.filter(block => block.type === 'tool-call') + if (toolCalls.length === 0) return { kind: 'completed' } const { concluded } = await executeToolCalls( this.loopCtx, turn, step, toolCalls, signal, - context => this.outbox.push(context), + context => this.inbox.splice('next-step', this.inbox.nextStep.length, 0, [context]), ) - result = concluded ? { kind: 'completed' } : null - } else { - result = { kind: 'completed' } + return concluded ? { kind: 'completed' } : null } - return result } /** @@ -360,8 +309,6 @@ export class ReactLoopAgent implements Agent { boundaryMessages: Message[], signal: AbortSignal, ): Promise<{ request: GenerateOptions; preparedCall?: PreparedLlmCall }> { - // A loop instance starts from its declared route, restoring only an opaque - // effort owned by that exact model. Later steps fold the config it logged. const persistedConfig = this.session.requestHeader()?.config const route = { provider: this.options.provider ?? '', model: this.options.model ?? '' } const reasoningEffort = persistedConfig?.provider === route.provider @@ -369,16 +316,14 @@ export class ReactLoopAgent implements Agent { ? persistedConfig.reasoningEffort : undefined const maxTokens = this.options.maxTokens - const seedConfig = deepFreeze(structuredClone( - this.requestHeaderLogged - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- the instance logged the header it now folds - ? persistedConfig! - : { - ...route, - ...reasoningEffort === undefined ? {} : { reasoningEffort }, - ...maxTokens === undefined ? {} : { maxTokens }, - }, - )) + const seedConfig = this.requestHeaderLogged + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- the instance logged the frozen header it now folds + ? persistedConfig! + : deepFreeze({ + ...route, + ...reasoningEffort === undefined ? {} : { reasoningEffort }, + ...maxTokens === undefined ? {} : { maxTokens }, + }) const proposedConfig = await this.loopCtx.waterfall( agentCarrier(this), 'agent/request', this, turn, step, signal, () => Promise.resolve(seedConfig), @@ -393,8 +338,7 @@ export class ReactLoopAgent implements Agent { preparedCall = await this.loopCtx.llm.prepareCall(proposedConfig, signal) config = preparedCall.config } catch (error: unknown) { - // A llm/stream listener may own and short-circuit a route with no - // adapter. Terminal dispatch still raises NO_ADAPTER when none does. + // Middleware may serve an unregistered route; terminal dispatch still requires an adapter. if (!(error instanceof LlmError) || error.code !== 'NO_ADAPTER') throw error config = proposedConfig } diff --git a/packages/core/agent-loop/tests/cancel.spec.ts b/packages/core/agent-loop/tests/cancel.spec.ts index 1bc9d58406..87796c97b6 100644 --- a/packages/core/agent-loop/tests/cancel.spec.ts +++ b/packages/core/agent-loop/tests/cancel.spec.ts @@ -76,8 +76,6 @@ describe('Agent.cancel()', () => { const adapter = new MockAdapter([textResponse('reply')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) - const canceled: unknown[] = [] - ctx.on('agent/inbox/canceled', (subject, message) => { if (subject === agent) canceled.push(message) }) agent.followup(createUserMessage({ content: [{ type: 'text', text: 'preserved' }], @@ -85,7 +83,8 @@ describe('Agent.cancel()', () => { })) // Abort the collecting activity while preserving its queued item. agent.cancel({ kind: 'user' }, { keepInbox: true }) - expect(canceled).toEqual([]) + expect(agent.session.events.some(event => + event.type === 'agent/inbox/spliced' && event.data.outcome === 'canceled')).toBe(false) // The preserved item still runs once a later follow-up wakes the driver. send(agent, 'wake it') diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index 31e6d7c2b1..a87afb578b 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -1,10 +1,10 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import LlmService, { createUserMessage, CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm' +import LlmService, { createUserMessage, freezeMessage, CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm' import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools' -import AgentRegistry, { type Agent, type InboxItem, type InboxPlacement } from '@deepseek-ai/dsh-agent' +import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import { ReactLoopAgent } from '../src/agent.ts' import InvariantService from '@deepseek-ai/dsh-invariants' @@ -53,8 +53,8 @@ function send(agent: Agent, text: string) { agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } })) } -function inboxText(item: InboxItem): string { - return item.message.content +function inboxText(message: UserMessage): string { + return message.content .flatMap(block => block.type === 'text' ? [block.text] : []) .join('') } @@ -69,42 +69,28 @@ describe('addressable inbox operations', () => { const agent = ctx.agentLoop.create(SessionId('inbox-actions'), { provider: 'mock', model: 'mock' }) const admission = Promise.withResolvers() const release = Promise.withResolvers() - ctx.on('agent/prompt-submit', async (_subject, message, _signal, next) => { - if (message.content[0]?.type === 'text' && message.content[0].text === 'first') { + ctx.on('agent/prompt-submit', async (_subject, messages, _signal, next) => { + if (messages[0]?.content[0]?.type === 'text' && messages[0].content[0].text === 'first') { admission.resolve(undefined) await release.promise } return next() }) - const pending: InboxItem[] = [] - const updates: { id: string; text: string }[] = [] - const discards: string[][] = [] - ctx.on('agent/inbox/enqueue', (subject, item) => { - if (subject === agent && inboxText(item) !== 'first') pending.push(item) - }) - ctx.on('agent/inbox/update', (subject, item) => { - if (subject === agent) updates.push({ id: item.id, text: inboxText(item) }) - }) - ctx.on('agent/inbox/discard', (subject, items) => { - if (subject === agent) discards.push(items.map(item => item.id)) - }) - send(agent, 'first') await admission.promise send(agent, 'remove me') send(agent, 'edit me') + const pending = agent.inbox.nextTurn expect(pending.map(inboxText)).toEqual(['remove me', 'edit me']) const remove = pending[0]! const edit = pending[1]! - expect(agent.updateInbox(edit.id, { - kind: 'edit', + expect(agent.inbox.splice('next-turn', 1, 1, [freezeMessage({ + ...edit, content: [{ type: 'text', text: 'edited' }], - })).toBe('applied') - expect(agent.updateInbox(remove.id, { kind: 'remove' })).toBe('applied') - expect(updates).toEqual([{ id: edit.id, text: 'edited' }]) - expect(discards).toEqual([[remove.id]]) + })])).toEqual([edit]) + expect(agent.inbox.splice('next-turn', 0, 1, [])).toEqual([remove]) const idle = waitForIdle(ctx, agent) release.resolve(undefined) @@ -115,46 +101,7 @@ describe('addressable inbox operations', () => { ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('') : '')) .toEqual(['first', 'edited']) - expect(agent.updateInbox(edit.id, { kind: 'remove' })).toBe('not-found') - }) - - it('does not mutate steering occurrences', async () => { - const adapter = new MockAdapter([textResponse('done')]) - const ctx = await harness(adapter) - const agent = ctx.agentLoop.create(SessionId('steering-inbox-actions'), { provider: 'mock', model: 'mock' }) - const entered = Promise.withResolvers() - const decision = Promise.withResolvers<{ kind: 'allow' }>() - ctx.on('agent/prompt-submit', async () => { - entered.resolve(undefined) - return decision.promise - }) - - const pending: InboxItem[] = [] - ctx.on('agent/inbox/enqueue', (subject, item) => { - if (subject === agent && item.placement === 'steering') pending.push(item) - }) - - const idle = waitForIdle(ctx, agent) - send(agent, 'admitted prompt') - await entered.promise - agent.steer(createUserMessage({ content: [{ type: 'text', text: 'keep me' }], source: { kind: 'user' } })) - expect(pending.map(inboxText)).toEqual(['keep me']) - - const steering = pending[0]! - expect(agent.updateInbox(steering.id, { - kind: 'edit', - content: [{ type: 'text', text: 'edited' }], - })).toBe('not-found') - expect(agent.updateInbox(steering.id, { kind: 'remove' })).toBe('not-found') - - decision.resolve({ kind: 'allow' }) - await idle - expect(agent.session.events - .filter(event => event.type === 'steering/message') - .map(event => event.type === 'steering/message' - ? event.data.message.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('') - : '')) - .toEqual(['keep me']) + expect(agent.inbox.splice('next-turn', 0, 1, [])).toEqual([]) }) }) @@ -590,7 +537,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => { expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }]) }) - it('agent/inbox/enqueue carries the exact message; steering/message records its source', async () => { + it('durable inbox splices carry exact messages and steering/message preserves its source', async () => { const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) @@ -604,27 +551,30 @@ describe('adapter registration, routing, and accepted-input ownership', () => { }, })) - const queuedSources: MessageSource[] = [] - const queuedShapes: string[][] = [] - const placements: InboxPlacement[] = [] - ctx.on('agent/inbox/enqueue', (_agent, item) => { - queuedSources.push(item.message.source) - queuedShapes.push(Object.keys(item.message).sort()) - placements.push(item.placement) + const insertedSources: MessageSource[] = [] + const insertedShapes: string[][] = [] + const targets: string[] = [] + ctx.on('session/event', (session, event) => { + if (session !== agent.session || event.type !== 'agent/inbox/spliced') return + for (const message of event.data.inserted) { + insertedSources.push(message.source) + insertedShapes.push(Object.keys(message).sort()) + targets.push(event.data.target) + } }) send(agent, 'go') // no explicit source → default {kind:'user'} must be visible await waitForIdle(ctx, agent) - expect(queuedSources).toEqual([ + expect(insertedSources).toEqual([ { kind: 'user' }, { kind: 'plugin', plugin: 'goal' }, ]) - expect(queuedShapes).toEqual([ + expect(insertedShapes).toEqual([ ['content', 'id', 'role', 'source'], ['content', 'id', 'role', 'source'], ]) - expect(placements).toEqual(['queued', 'steering']) + expect(targets).toEqual(['next-turn', 'next-step']) // The drain appends the durable steering/message with the caller's source // intact — the log, not a transient emit, is where consumers read it. const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.message.source] : []) diff --git a/packages/core/agent/package.json b/packages/core/agent/package.json index 9113030eef..db0637ecc2 100644 --- a/packages/core/agent/package.json +++ b/packages/core/agent/package.json @@ -15,10 +15,6 @@ "types": "./lib/types/invariant.d.ts", "default": "./lib/invariant.js" }, - "./brand": { - "types": "./lib/types/brand.d.ts", - "default": "./lib/types/brand.js" - }, "./src/*": "./src/*", "./package.json": "./package.json" }, @@ -32,7 +28,6 @@ ], "license": "BSD-3-Clause", "peerDependencies": { - "@deepseek-ai/dsh-brand": "^0.0.1", "@deepseek-ai/dsh-invariants": "^0.0.1", "@deepseek-ai/dsh-llm": "^0.0.1", "@deepseek-ai/dsh-scope": "^0.0.1", @@ -41,7 +36,6 @@ "cordis": "^4.0.0-rc.7" }, "devDependencies": { - "@deepseek-ai/dsh-brand": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-scope": "workspace:^", diff --git a/packages/core/agent/src/brand.ts b/packages/core/agent/src/brand.ts deleted file mode 100644 index 58d50259c1..0000000000 --- a/packages/core/agent/src/brand.ts +++ /dev/null @@ -1,23 +0,0 @@ -/** - * dsh-agent's owned branded ids for live inbox occurrences. - * - * @module @deepseek-ai/dsh-agent/brand - */ - -import type { Branded } from '@deepseek-ai/dsh-brand' - -/** - * Identifies one accepted occurrence in an agent inbox. Re-sending the same - * message creates a distinct item id, so pending work remains independently - * addressable. - */ -export type InboxItemId = Branded<'InboxItemId'> - -/** - * Brand a string as an {@link InboxItemId}. - * @param id - the agent-loop-minted occurrence identifier. - * @returns the same string, branded; no validation is performed. - */ -export function InboxItemId(id: string): InboxItemId { - return id as InboxItemId -} diff --git a/packages/core/agent/src/inbox.ts b/packages/core/agent/src/inbox.ts new file mode 100644 index 0000000000..d6dafa7e4e --- /dev/null +++ b/packages/core/agent/src/inbox.ts @@ -0,0 +1,109 @@ +/** + * Incremental projection of durable agent inbox events. + * + * @module @deepseek-ai/dsh-agent/inbox + */ + +import type { Session, SessionEventMap, UserMessage } from '@deepseek-ai/dsh-session' + +/** One of the two ordered pending-message lists owned by an agent. */ +export type InboxTarget = 'next-turn' | 'next-step' + +/** Mutable state privately owned by an {@link Inbox}. */ +type InboxState = Record + +/** A replay-once projection that incrementally consumes later inbox splices. */ +export class Inbox { + private readonly state: InboxState = { 'next-turn': [], 'next-step': [] } + + constructor(private readonly session: Session) { + for (const event of session.events.slice(session.header.seedLength ?? 0)) { + if (event.type !== 'agent/inbox/spliced') continue + try { + this.apply(event.data) + } catch (error: unknown) { + throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error }) + } + } + } + + /** Prompts awaiting individual turns. */ + get nextTurn(): readonly UserMessage[] { + return this.state['next-turn'] + } + + /** Input awaiting admission at a step boundary. */ + get nextStep(): readonly UserMessage[] { + return this.state['next-step'] + } + + /** Whether either pending-message list contains work. */ + get hasPending(): boolean { + return this.nextTurn.length > 0 || this.nextStep.length > 0 + } + + /** + * Apply standard splice semantics and durably record the normalized result. + * @param target - pending list to mutate. + * @param start - splice position. + * @param deleteCount - maximum number of messages to remove. + * @param inserted - messages to insert at the resolved position. + * @param outcome - terminal disposition of removed messages. + * @returns messages removed by the splice. + */ + splice( + target: InboxTarget, + start: number, + deleteCount: number, + inserted: UserMessage[], + outcome?: 'admitted' | 'canceled', + ): UserMessage[] { + const inbox = this.state[target] + const offset = Math.trunc(start) || 0 + const actualStart = offset < 0 + ? Math.max(inbox.length + offset, 0) + : Math.min(offset, inbox.length) + const actualDeleteCount = Math.min( + Math.max(Math.trunc(deleteCount) || 0, 0), + inbox.length - actualStart, + ) + if (actualDeleteCount === 0 && inserted.length === 0) return [] + const resolvedOutcome = outcome ?? (actualDeleteCount > 0 ? 'canceled' : undefined) + const splice = { + target, + start: actualStart, + ...(actualDeleteCount === 0 ? {} : { removedCount: actualDeleteCount }), + inserted, + ...(resolvedOutcome === undefined ? {} : { outcome: resolvedOutcome }), + } + this.validate(splice) + const event = this.session.append('agent/inbox/spliced', splice) + return inbox.splice(actualStart, actualDeleteCount, ...event.data.inserted) + } + + /** Apply one normalized durable splice to the projection. */ + private apply(splice: SessionEventMap['agent/inbox/spliced']): UserMessage[] { + this.validate(splice) + const inbox = this.state[splice.target] + return inbox.splice(splice.start, splice.removedCount ?? 0, ...splice.inserted) + } + + /** Validate one normalized splice against the current projection. */ + private validate(splice: SessionEventMap['agent/inbox/spliced']): void { + const inbox = this.state[splice.target] + const removedCount = splice.removedCount ?? 0 + if (!Number.isSafeInteger(splice.start) || splice.start < 0 || splice.start > inbox.length + || !Number.isSafeInteger(removedCount) || removedCount < 0 + || splice.start + removedCount > inbox.length) { + throw new Error('invalid inbox splice') + } + const candidate = inbox.toSpliced(splice.start, removedCount, ...splice.inserted) + const ids = new Set() + for (const message of splice.target === 'next-turn' + ? [...candidate, ...this.nextStep] + : [...this.nextTurn, ...candidate]) { + if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`) + ids.add(message.id) + } + } +} diff --git a/packages/core/agent/src/index.ts b/packages/core/agent/src/index.ts index 0d87778135..d46d8833c5 100644 --- a/packages/core/agent/src/index.ts +++ b/packages/core/agent/src/index.ts @@ -15,7 +15,7 @@ import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import type { Agent, AgentOptions } from './types.ts' export * from './types.ts' -export * from './brand.ts' +export * from './inbox.ts' export * from './llm-target.ts' export { agentCarrier, agentEvents, assembleContextFor, emitAgentEvent } from './dispatch.ts' export type { AgentEventDispatch, AgentSubjectEvent } from './dispatch.ts' diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 804ad4a963..b1451768ca 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -7,10 +7,10 @@ import type { Context } from 'cordis' import type { Scoped } from '@deepseek-ai/dsh-scope' -import type { ContentBlock, LlmCallConfig, LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' +import type { LlmCallConfig, LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm' import type { AgentCancelCause, Session, SessionId, UserMessage } from '@deepseek-ai/dsh-session' export type { AgentCancelCause } from '@deepseek-ai/dsh-session' -import type { InboxItemId } from './brand.ts' +import type { Inbox, InboxTarget } from './inbox.ts' import type {} from '@deepseek-ai/dsh-system-prompt' declare module '@deepseek-ai/dsh-system-prompt' { interface AssembleContext { @@ -29,63 +29,12 @@ export interface AgentOptions { maxTokens?: number } -/** - * Which inbox queue a {@link Agent.send} item joins: - * - `next-turn` — the item becomes its own turn, claimed at a turn boundary. - * - `next-step` — during prompt admission or an open turn, the item stages for - * the next safe step boundary; otherwise it is promoted per its `wakeup` - * flag. - */ -export type SendTarget = 'next-turn' | 'next-step' - -/** Resolved inbox placement reported when an accepted message is enqueued. */ -export type InboxPlacement = 'queued' | 'steering' - -/** One independently addressable accepted occurrence in an agent inbox. */ -export interface InboxItem { - /** Agent-loop-minted occurrence identity. */ - readonly id: InboxItemId - /** Identified message delivered by the caller. */ - readonly message: UserMessage - /** Acceptance-time FIFO classification. */ - readonly placement: InboxPlacement -} - -/** A user-requested mutation of one still-pending queued occurrence. */ -export type InboxAction = - | { readonly kind: 'edit'; readonly content: ContentBlock[] } - | { readonly kind: 'remove' } - -/** Result of applying an inbox action at the synchronous ownership boundary. */ -export type InboxActionResult = 'applied' | 'not-found' - -/** - * Options for the unified {@link Agent.send} primitive over the - * (`target` × `wakeup`) matrix. Named presets: {@link Agent.followup} - * (`next-turn`/wakeup), {@link Agent.steer} (`next-step`/wakeup), and - * {@link Agent.inject} (`next-step`/no-wakeup). - * - * The object is complete so routing policy is explicit. - */ -export interface SendOptions { - /** Queue the item joins. */ - target: SendTarget - /** - * Whether this item makes the model run: wake a parked driver (`next-turn`) - * or force a continuation step (`next-step` while running). A `false` - * `next-turn` item queues without waking; a `false` - * `next-step` item attaches durable context without forcing another step - * (the injection preset). - */ - wakeup: boolean -} - /** Options for {@link Agent.cancel}. */ export interface CancelOptions { /** * Preserve queued and steering inbox items instead of discarding them. The * active turn is still aborted, but un-started and pending work survives for a - * later turn and no `agent/inbox/canceled` fires. + * later turn and no canceled inbox splice is logged. */ keepInbox?: boolean | undefined } @@ -136,30 +85,13 @@ export interface Agent { readonly options: AgentOptions /** The live session this agent drives; its log is the durable source of truth. */ readonly session: Session + /** The agent-owned projection of durable pending work. */ + readonly inbox: Inbox /** The current lifecycle state, mirrored on every `agent/status` transition. */ readonly status: AgentStatus - /** Whether a next-step send currently remains in the open turn. */ - readonly acceptsNextStep: boolean /** Agent-scoped context; its contributions are agent-local, unwind on disposal, and reject registration afterward. */ readonly ctx: Context - /** - * The unified delivery primitive over the (`target` × `wakeup`) matrix. - * @param message - identified model-facing content and its producer provenance. - * @param options - target queue and wakeup decision. - */ - send(message: UserMessage, options: SendOptions): void - - /** - * Mutate one still-pending queued occurrence synchronously. Editing preserves - * the message identity and queue position; removal publishes its terminal - * discard. Steering occurrences and driver-claimed items return `not-found`. - * @param id - independently addressable queued occurrence. - * @param action - edit or remove operation. - * @returns whether the pending occurrence was found and updated. - */ - updateInbox(id: InboxItemId, action: InboxAction): InboxActionResult - /** * Clear queued and steering work — unless `keepInbox` — and abort the active * turn. The first cause wins for the active turn. Idle cancellation is a @@ -236,48 +168,6 @@ declare module 'cordis' { * @mode emit */ 'agent/status'(this: Scoped, agent: Agent, status: AgentStatus): void - /** - * An item entered the queued or steering inbox. `placement` is the - * acceptance-time routing result. - * @param agent - the owning agent. - * @param item - accepted occurrence, message, and resolved placement. - * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. - * @mode emit - */ - 'agent/inbox/enqueue'(this: Scoped, agent: Agent, item: InboxItem): void - /** - * A still-pending queued item changed content. - * @param agent - the owning agent. - * @param item - the complete post-update occurrence. - * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. - * @mode emit - */ - 'agent/inbox/update'(this: Scoped, agent: Agent, item: InboxItem): void - /** - * The driver claimed one item out of the inbox. - * @param agent - the agent whose inbox item was claimed. - * @param item - the exact claimed occurrence. - * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. - * @mode emit - */ - 'agent/inbox/dequeue'(this: Scoped, agent: Agent, item: InboxItem): void - /** - * Pending inbox items were dropped without delivery. - * @param agent - the agent whose inbox items were dropped. - * @param items - the discarded occurrences in FIFO order. - * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. - * @mode emit - */ - 'agent/inbox/discard'(this: Scoped, agent: Agent, items: InboxItem[]): void - /** - * Effective broad cancellation was requested before pending work clears or - * the active turn aborts. - * @param agent - the agent whose current work is being cancelled. - * @param cause - the explicit typed cancellation cause. - * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. - * @mode emit - */ - 'agent/cancel-requested'(this: Scoped, agent: Agent, cause: AgentCancelCause): void // ---- session lifecycle (emit) ---- /** * The session lifecycle began, once before the first turn. Use @@ -374,7 +264,13 @@ declare module 'cordis' { declare module '@deepseek-ai/dsh-session' { interface SessionEventMap { - /** One message was accepted into the agent inbox. */ - 'agent/inbox/added': UserMessage + /** One normalized mutation of an agent's durable pending-message lists. */ + 'agent/inbox/spliced': { + target: InboxTarget + start: number + removedCount?: number + inserted: UserMessage[] + outcome?: 'admitted' | 'canceled' + } } } diff --git a/packages/core/agent/tests/agent.spec.ts b/packages/core/agent/tests/agent.spec.ts index ffd1bd91c4..b2178aea7e 100644 --- a/packages/core/agent/tests/agent.spec.ts +++ b/packages/core/agent/tests/agent.spec.ts @@ -21,8 +21,6 @@ function stubAgent(rawId: string, overrides: Partial = {}): Agent { session: new Session(id), status: 'idle', ctx: new Context(), - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject: () => {}, diff --git a/packages/core/agent/tests/invariant.spec.ts b/packages/core/agent/tests/invariant.spec.ts index 7744726465..158376a3d7 100644 --- a/packages/core/agent/tests/invariant.spec.ts +++ b/packages/core/agent/tests/invariant.spec.ts @@ -1,7 +1,6 @@ -import { freezeMessage, MessageId } from '@deepseek-ai/dsh-llm' import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import { InboxItemId, type Agent, type InboxItem, type InboxPlacement } from '@deepseek-ai/dsh-agent' +import type { Agent } from '@deepseek-ai/dsh-agent' import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant' import { scopeTarget } from '@deepseek-ai/dsh-scope' import InvariantService from '@deepseek-ai/dsh-invariants' @@ -44,51 +43,3 @@ describe('agent status invariants', () => { expect(() => { ctx.emit(scopeTarget(b, b), 'agent/status', b, 'running') }).not.toThrow() }) }) - -describe('agent inbox invariants', () => { - let nextItem = 0 - const info = (placement: InboxPlacement = 'queued'): InboxItem => ({ - id: InboxItemId(`i-${nextItem++}`), - message: freezeMessage({ - id: MessageId('m'), - role: 'user' as const, - content: [], - source: { kind: 'user' as const }, - }), - placement, - }) - - it('accepts a dequeue and a discard covered by prior enqueues', async () => { - const ctx = await setup() - const agent = mockAgent('i1') - const at = scopeTarget(agent, agent) - expect(() => { - ctx.emit(at, 'agent/inbox/enqueue', agent, info()) - ctx.emit(at, 'agent/inbox/enqueue', agent, info('steering')) - ctx.emit(at, 'agent/inbox/dequeue', agent, info()) - ctx.emit(at, 'agent/inbox/discard', agent, [info()]) - }).not.toThrow() - }) - - it('rejects a dequeue with no outstanding item', async () => { - const ctx = await setup() - const agent = mockAgent('i2') - expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/dequeue', agent, info()) }) - .toThrow(/without a matching prior enqueue/) - }) - - it('rejects a discard larger than the outstanding count', async () => { - const ctx = await setup() - const agent = mockAgent('i3') - const at = scopeTarget(agent, agent) - ctx.emit(at, 'agent/inbox/enqueue', agent, info()) - expect(() => { ctx.emit(at, 'agent/inbox/discard', agent, [info(), info()]) }) - .toThrow(/dropped 2 items but only 1 were outstanding/) - }) - - it('accepts an empty discard against a fresh agent', async () => { - const ctx = await setup() - const agent = mockAgent('i4') - expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/discard', agent, []) }).not.toThrow() - }) -}) diff --git a/packages/core/agent/tsconfig.json b/packages/core/agent/tsconfig.json index b6d6c9e6bf..1561175ed9 100644 --- a/packages/core/agent/tsconfig.json +++ b/packages/core/agent/tsconfig.json @@ -14,9 +14,6 @@ { "path": "../../../vendor/cordis" }, - { - "path": "../../util/brand" - }, { "path": "../../core/scope" }, diff --git a/packages/core/scope/src/scoped-events.generated.ts b/packages/core/scope/src/scoped-events.generated.ts index fff3107f48..e08fd2aabd 100644 --- a/packages/core/scope/src/scoped-events.generated.ts +++ b/packages/core/scope/src/scoped-events.generated.ts @@ -8,14 +8,9 @@ type ScopedSubjectResolver = (args: readonly unknown[]) => unknown const scopedSubjectResolvers: Readonly> = Object.freeze({ - 'agent/cancel-requested': args => args[0], 'agent/created': args => args[0], 'agent/disposed': args => args[0], 'agent/error': args => args[0], - 'agent/inbox/dequeue': args => args[0], - 'agent/inbox/discard': args => args[0], - 'agent/inbox/enqueue': args => args[0], - 'agent/inbox/update': args => args[0], 'agent/prompt-submit': args => args[0], 'agent/request': args => args[0], 'agent/request-error': args => args[0], diff --git a/packages/core/scope/tests/invariant.spec.ts b/packages/core/scope/tests/invariant.spec.ts index f773f3a9aa..ccc90e7843 100644 --- a/packages/core/scope/tests/invariant.spec.ts +++ b/packages/core/scope/tests/invariant.spec.ts @@ -2,7 +2,7 @@ import { freezeMessage, MessageId } from '@deepseek-ai/dsh-llm' import { describe, expect, it } from 'vitest' import { Context } from 'cordis' import type { Events } from 'cordis' -import { InboxItemId, type Agent } from '@deepseek-ai/dsh-agent' +import type { Agent } from '@deepseek-ai/dsh-agent' import { scopeTarget } from '@deepseek-ai/dsh-scope' import * as ScopeInvariant from '@deepseek-ai/dsh-scope/invariant' import InvariantService from '@deepseek-ai/dsh-invariants' @@ -44,27 +44,22 @@ describe('scoped-dispatch invariants', () => { content: [], source: { kind: 'user' }, }) - const item = { id: InboxItemId('i'), message, placement: 'queued' as const } const agentRows = { 'agent/created': [agent], 'agent/disposed': [agent], 'agent/status': [agent, 'idle'], - 'agent/inbox/enqueue': [agent, item], - 'agent/inbox/update': [agent, item], - 'agent/inbox/dequeue': [agent, item], - 'agent/inbox/discard': [agent, []], 'agent/session-start': [agent, 'startup'], 'agent/step': [agent, 1, 1, signal], - 'agent/prompt-submit': [agent, message, signal, () => Promise.resolve({ kind: 'allow' })], + 'agent/prompt-submit': [agent, [message], signal, () => Promise.resolve({ kind: 'allow', messages: [message] })], 'agent/request': [agent, 1, 1, signal, () => Promise.resolve(config)], 'agent/request-error': [ agent, - 1, - 1, - new Error('request'), - { message: 'request', code: 'UNKNOWN' }, - [], - undefined, + { + turn: 1, + step: 1, + provider: 'p', + failure: { message: 'request', code: 'UNKNOWN' }, + }, signal, () => Promise.resolve(undefined), ], diff --git a/packages/examples/cli-demo/tests/cli.spec.ts b/packages/examples/cli-demo/tests/cli.spec.ts index fc77b11331..1414a0691a 100644 --- a/packages/examples/cli-demo/tests/cli.spec.ts +++ b/packages/examples/cli-demo/tests/cli.spec.ts @@ -114,14 +114,13 @@ const liveContexts: Context[] = [] async function harness(script: readonly ScriptEntry[]): Promise { const root = await mkdtemp(join(tmpdir(), 'dsh-cli-runner-')) - const skillHome = await mkdtemp(join(tmpdir(), 'dsh-cli-runner-skills-')) const ctx = new Context() liveContexts.push(ctx) await ctx.plugin(cliDemo, { provider: 'mock', model: 'mock', persistenceRoot: root, - skills: { local: { dshHome: join(skillHome, '.dsh'), agentsHome: join(skillHome, '.agents') } }, + skills: { enabled: false }, workspaceContext: false, }) await new Promise(resolve => setTimeout(resolve, 80)) @@ -380,29 +379,94 @@ describe('runOneShot and executeCli', () => { expect(result.result).toBe('working') }) - it('streams only the correlated main message turn and then the result envelope', async () => { - const { ctx, agent } = await harness([textResponse('streamed')]) + it('observes only the correlated main message turn', async () => { + const { ctx, agent } = await harness([ + textResponse('startup'), + textResponse('autonomous'), + textResponse('streamed'), + ]) const other = ctx.sessions.create(SessionId('unrelated')) - let injected = false - ctx.on('agent/inbox/enqueue', (subject) => { - if (subject !== agent || injected) return - injected = true - agent.inject(createUserMessage({ content: [{ type: 'text', text: 'startup injection' }], source: { kind: 'plugin', plugin: 'test' } })) + let startupStarted!: () => void + const started = new Promise((resolve) => { startupStarted = resolve }) + const releaseStartup = Promise.withResolvers() + ctx.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'assistant/message' + && event.data.turn === 1) startupStarted() + }) + ctx.on('agent/turn-stopping', async (subject, turn) => { + if (subject === agent && turn === 1) await releaseStartup.promise + }) + agent.followup(createUserMessage({ + content: [{ type: 'text', text: 'startup' }], + source: { kind: 'plugin', plugin: 'startup' }, + })) + await started + + let replacementQueued = false + ctx.on('agent/status', (subject, status) => { + if (subject !== agent || status !== 'idle' || replacementQueued) return + replacementQueued = true + agent.followup(createUserMessage({ + content: [{ type: 'text', text: 'autonomous' }], + source: { kind: 'plugin', plugin: 'test' }, + })) other.append('turn/start', { turn: 1 }) other.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) }) - const output = await invoke(ctx, ['--output-format', 'stream-json', 'task']) - const lines = output.stdout.trimEnd().split('\n').map(line => JSON.parse(line) as Record) - const events = lines.slice(0, -1).map(line => line['event'] as SessionEvent) - expect(lines.at(-1)).toMatchObject({ type: 'result', success: true, turn: 1, result: 'streamed' }) - expect(events[0]).toMatchObject({ type: 'turn/start', data: { turn: 1 } }) - expect(events.at(-1)).toMatchObject({ type: 'turn/end', data: { turn: 1 } }) - expect(lines.slice(0, -1).every(line => line['sessionId'] === agent.session.id)).toBe(true) + const streamed: { sessionId: string; event: SessionEvent }[] = [] + const result = runOneShot(ctx, { + task: 'task', + onEvent: (sessionId, event) => { streamed.push({ sessionId, event }) }, + }) + releaseStartup.resolve(undefined) + + const outcome = await result + expect(outcome.reason).toEqual({ kind: 'completed' }) + expect(outcome).toMatchObject({ success: true, turn: 3, result: 'streamed' }) + const events = streamed.map(item => item.event) + expect(events[0]).toMatchObject({ type: 'turn/start', data: { turn: 3 } }) + expect(events.at(-1)).toMatchObject({ type: 'turn/end', data: { turn: 3 } }) + expect(streamed.every(item => item.sessionId === agent.session.id)).toBe(true) expect(events.some(event => event.type === 'user/message' && event.data.source.kind === 'plugin' && event.data.source.plugin === 'test')).toBe(false) }) + it('correlates a task whose admitted history is replaced', async () => { + const { ctx } = await harness([textResponse('rewritten answer')]) + ctx.on('agent/prompt-submit', async () => ({ + kind: 'allow', + messages: [createUserMessage({ + content: [{ type: 'text', text: 'rewritten task' }], + source: { kind: 'plugin', plugin: 'test' }, + })], + })) + + await expect(runOneShot(ctx, { task: 'original task' })).resolves.toMatchObject({ + success: true, + result: 'rewritten answer', + }) + }) + + it('rejects tasks blocked before admission, including retained tasks', async () => { + const blocked = await harness([]) + blocked.ctx.on('agent/prompt-submit', async () => ({ kind: 'block' as const, reason: 'denied' })) + await expect(runOneShot(blocked.ctx, { task: 'task' })).rejects.toThrow('canceled before admission') + + const retained = await harness([]) + retained.ctx.on('agent/prompt-submit', async () => ({ + kind: 'block' as const, + reason: 'deferred', + keepInbox: true, + })) + await expect(runOneShot(retained.ctx, { task: 'task' })).rejects.toThrow('not admitted') + expect(retained.agent.status).toBe('idle') + + const failed = await harness([]) + failed.ctx.on('agent/prompt-submit', async () => { throw new Error('admission exploded') }) + await expect(runOneShot(failed.ctx, { task: 'task' })).rejects.toThrow('not admitted') + }) + it('emits partial data and a diagnostic for non-completed turns', async () => { const { ctx } = await harness([textResponse('partial', { inputTokens: 2, outputTokens: 3 }, 'max-tokens')]) const output = await invoke(ctx, ['--output-format', 'json', 'task']) @@ -498,8 +562,11 @@ describe('runOneShot and executeCli', () => { const queued = await harness([textResponse('unused')]) const queuedAbort = new AbortController() - queued.ctx.on('agent/inbox/enqueue', (agent) => { - if (agent === queued.agent) queuedAbort.abort('cancel queued') + queued.ctx.on('session/event', (session, event) => { + if (session === queued.agent.session && event.type === 'agent/inbox/spliced' + && event.data.inserted.some(message => message.source.kind === 'user')) { + queueMicrotask(() => { queuedAbort.abort('cancel queued') }) + } }) await expect(runOneShot(queued.ctx, { task: 'task', signal: queuedAbort.signal })).rejects.toThrow('cancel queued') await queued.agent.whenIdle() diff --git a/packages/goal/command-goal/tests/command-goal.spec.ts b/packages/goal/command-goal/tests/command-goal.spec.ts index 89f115df4d..ef7204e7c0 100644 --- a/packages/goal/command-goal/tests/command-goal.spec.ts +++ b/packages/goal/command-goal/tests/command-goal.spec.ts @@ -36,9 +36,6 @@ function stubAgent(ctx: Context, id: string): { agent: Agent; session: Session } session, ctx: new Context(), get status() { return status }, - get acceptsNextStep() { return status === 'running' }, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { appendInjection(session, input) }, diff --git a/packages/goal/goal-session/src/index.ts b/packages/goal/goal-session/src/index.ts index 3260016d5f..5694da1dae 100644 --- a/packages/goal/goal-session/src/index.ts +++ b/packages/goal/goal-session/src/index.ts @@ -324,37 +324,6 @@ export function apply(ctx: Context): void { requestDrive(state) } }) - ctx.on('agent/inbox/enqueue', (agent, item) => { - const state = stateFor(agent) - const attempt = state.attempt - if (attempt !== undefined && sameQueued(item.message.content, item.message.source, attempt)) return - state.competingQueued = true - if (attempt?.phase === 'queued') attempt.stale = true - }) - ctx.on('agent/cancel-requested', (agent, cause) => { - const state = stateFor(agent) - const attempt = state.attempt - state.competingQueued = false - const goal = currentGoal(state) - if (goal?.phase === 'active' && goal.activation === 'armed') { - if (attempt === undefined) { - disarm(state) - return - } - // An admitted round closes durably as aborted; retain it so the normal - // turn outcome path appends pause after cancellation reaches idle. - // Pausing here would stage context into the active outbox only for this - // same cancel() call to discard it. - if (attempt.turn !== undefined || attempt.phase === 'admitted') return - state.attempt = undefined - try { - applyOutcome(state, goal, { kind: 'pause', reason: cause.kind }) - } catch (error: unknown) { - ctx.logger.warn(`goal-session: could not pause cancelled goal for agent "${agent.id}": ${renderThrown(error)}`) - disarm(state) - } - } - }) ctx.on('goal/changed', (agent) => { const state = stateFor(agent) state.needsCheckpoint = true @@ -366,12 +335,14 @@ export function apply(ctx: Context): void { if (agent === undefined || agent.session !== session) return const state = stateFor(agent) switch (event.type) { - case 'agent/inbox/added': { + case 'agent/inbox/spliced': { + if (event.data.target !== 'next-turn') return const attempt = state.attempt - const { content, source } = event.data - if (attempt !== undefined && sameQueued(content, source, attempt)) return - state.competingQueued = true - if (attempt?.phase === 'queued') attempt.stale = true + for (const message of event.data.inserted) { + if (attempt !== undefined && sameQueued(message.content, message.source, attempt)) continue + state.competingQueued = true + if (attempt?.phase === 'queued') attempt.stale = true + } return } case 'turn/start': { diff --git a/packages/goal/goal-session/tests/goal-session.spec.ts b/packages/goal/goal-session/tests/goal-session.spec.ts index 20c74b9f15..d1144bea1d 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -9,7 +9,7 @@ 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' import { SessionId } from '@deepseek-ai/dsh-session' -import type { TurnEndReason } from '@deepseek-ai/dsh-session' +import type { TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session' import * as goalSession from '../src/index.ts' type ScriptEntry = StreamChunk[] | Error | 'hang' | ((options: GenerateOptions) => StreamChunk[]) @@ -101,6 +101,18 @@ async function harness(script: ScriptEntry[]): Promise { return { ctx, adapter, agent, driver } } +/** Observe inserted inbox messages after the session append boundary closes. */ +function onInboxMessage( + ctx: Context, + agent: Agent, + listener: (message: UserMessage) => void, +): () => void { + return ctx.on('session/event', (session, event) => { + if (session !== agent.session || event.type !== 'agent/inbox/spliced') return + for (const message of event.data.inserted) queueMicrotask(() => { listener(message) }) + }) +} + /** Await a stable goal projection selected by the caller. */ async function waitForGoal( ctx: Context, @@ -279,10 +291,10 @@ describe('same-session goal driving', () => { it('pauses and drops a reserved round when cancellation lands before admission', async () => { const test = await harness([]) - const cancel = test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent === test.agent && info.message.source.kind === 'goal') { + const cancel = onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind === 'goal') { cancel() - agent.cancel({ kind: 'user' }) + test.agent.cancel({ kind: 'user' }) } }) test.ctx.goals.create(test.agent, { objective: 'do not start yet' }) @@ -330,10 +342,10 @@ describe('same-session goal driving', () => { it('makes a reserved round stale when a listener queues human work behind it', async () => { const test = await harness([textResponse('human batch'), textResponse('later goal')]) let inserted = false - test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent !== test.agent || info.message.source.kind !== 'goal' || inserted) return + onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind !== 'goal' || inserted) return inserted = true - agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human joined the pending batch' }], source: { kind: 'user' } })) + test.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human joined the pending batch' }], source: { kind: 'user' } })) }) test.ctx.goals.create(test.agent, { objective: 'yield to nested human input', maxGoalRounds: 1 }) @@ -348,12 +360,12 @@ describe('same-session goal driving', () => { it('blocks a queued reservation made stale by a goal edit and continues the new revision', async () => { const test = await harness([textResponse('new revision')]) let edited = false - test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent !== test.agent || info.message.source.kind !== 'goal' || edited) return + onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind !== 'goal' || edited) return edited = true - const current = test.ctx.goals.get(agent) + const current = test.ctx.goals.get(test.agent) if (current === undefined) throw new Error('missing goal during queued edit') - test.ctx.goals.edit(agent, current, { objective: 'new objective' }) + test.ctx.goals.edit(test.agent, current, { objective: 'new objective' }) }) test.ctx.goals.create(test.agent, { objective: 'old objective', maxGoalRounds: 1 }) @@ -628,8 +640,8 @@ describe('same-session goal driving', () => { it('fails a pre-admission read closed even when the first disarm attempt throws', async () => { const test = await harness([textResponse('retry after containment')]) let armed = true - test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent !== test.agent || info.message.source.kind !== 'goal' || !armed) return + onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind !== 'goal' || !armed) return armed = false vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => { throw new Error('admission projection failed') @@ -704,13 +716,13 @@ describe('same-session goal driving', () => { it('falls back to disarming when a cancelled reservation cannot be paused', async () => { const test = await harness([]) - const cancel = test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent !== test.agent || info.message.source.kind !== 'goal') return + const cancel = onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind !== 'goal') return cancel() vi.spyOn(test.ctx.goals, 'pause').mockImplementationOnce(() => { throw new Error('pause failed') }) - agent.cancel({ kind: 'user' }) + test.agent.cancel({ kind: 'user' }) }) test.ctx.goals.create(test.agent, { objective: 'fail closed after cancellation' }) @@ -758,8 +770,8 @@ describe('same-session goal driving', () => { it('cancels an accepted queued round and awaits its driver task during teardown', async () => { const test = await harness([]) let unloading: Promise | undefined - test.ctx.on('agent/inbox/enqueue', (agent, info) => { - if (agent === test.agent && info.message.source.kind === 'goal' && unloading === undefined) { + onInboxMessage(test.ctx, test.agent, (message) => { + if (message.source.kind === 'goal' && unloading === undefined) { unloading = Promise.resolve(test.driver.dispose()) } }) diff --git a/packages/goal/goal/tests/goal.spec.ts b/packages/goal/goal/tests/goal.spec.ts index e01356f271..f395996011 100644 --- a/packages/goal/goal/tests/goal.spec.ts +++ b/packages/goal/goal/tests/goal.spec.ts @@ -46,9 +46,6 @@ function stubAgentForSession(session: Session): StubAgent { session, ctx: new Context(), get status() { return status }, - get acceptsNextStep() { return status === 'running' }, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { diff --git a/packages/goal/goal/tests/projection.spec.ts b/packages/goal/goal/tests/projection.spec.ts index f787a881f8..70fcbc0d5a 100644 --- a/packages/goal/goal/tests/projection.spec.ts +++ b/packages/goal/goal/tests/projection.spec.ts @@ -37,9 +37,6 @@ function liveAgent(ctx: Context, session: Session): Agent { session, ctx, get status() { return status }, - get acceptsNextStep() { return false }, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input: UserMessage) { diff --git a/packages/goal/tool-goal/tests/tool-goal.spec.ts b/packages/goal/tool-goal/tests/tool-goal.spec.ts index 7bcee0c0e7..d411e0c9b2 100644 --- a/packages/goal/tool-goal/tests/tool-goal.spec.ts +++ b/packages/goal/tool-goal/tests/tool-goal.spec.ts @@ -30,10 +30,7 @@ function stubAgent(rawId: string, supplied?: Session): StubAgent { options: {}, session, get status() { return status }, - get acceptsNextStep() { return status === 'running' }, ctx: new Context(), - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index e7c1d96acf..c68f81745d 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -9,9 +9,9 @@ import { join } from 'node:path' import type { Context } from 'cordis' import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent' import type { - Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxItem, InboxItemId, + Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, } from '@deepseek-ai/dsh-agent' -import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm' +import { createUserMessage, freezeMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm' import { errorChain } from '@deepseek-ai/dsh-llm' import type { MessageSource } from '@deepseek-ai/dsh-llm' import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session' @@ -507,112 +507,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro }) }) - /** - * Per-session queued-occurrence mirror serving the mux-open queue snapshot - * (the same refresh-recovery baseline as pending questions). Each terminal - * queue event retires one matching occurrence, so repeated sends of the same - * identified message remain visible until every occurrence is claimed. - */ - const queuedMirror = new Map() - type UnseenQueueEvent = - | { readonly kind: 'update'; readonly item: InboxItem } - | { readonly kind: 'terminal' } - const unseenQueueEvents = new Map>() - const rememberUnseen = (sessionId: SessionId, itemId: InboxItemId, event: UnseenQueueEvent): void => { - let events = unseenQueueEvents.get(sessionId) - if (events === undefined) { - events = new Map() - unseenQueueEvents.set(sessionId, events) - } - events.set(itemId, event) - // Only synchronous re-entrancy may deliver a mutation before its outer - // enqueue observer. Drop unmatched protocol-invalid observations instead - // of retaining process-local ids indefinitely. - queueMicrotask(() => { - const current = unseenQueueEvents.get(sessionId) - if (current?.get(itemId) !== event) return - current.delete(itemId) - if (current.size === 0) unseenQueueEvents.delete(sessionId) - }) - } - const takeUnseen = (sessionId: SessionId, itemId: InboxItemId): UnseenQueueEvent | undefined => { - const events = unseenQueueEvents.get(sessionId) - const event = events?.get(itemId) - if (event === undefined) return undefined - events?.delete(itemId) - if (events?.size === 0) unseenQueueEvents.delete(sessionId) - return event - } - const publishQueue = (sessionId: SessionId): void => { - const items = queuedMirror.get(sessionId) ?? [] - broadcast({ - type: 'session/queue', - sessionId, - items: items.map(item => ({ - id: item.id, - message: item.message, - })), - }) - } - ctx.effect(() => { - const retire = (agent: Agent, item: InboxItem): boolean => { - const entries = queuedMirror.get(agent.id) - if (entries === undefined) { - rememberUnseen(agent.id, item.id, { kind: 'terminal' }) - return false - } - const index = entries.findIndex(entry => entry.id === item.id) - if (index === -1) { - rememberUnseen(agent.id, item.id, { kind: 'terminal' }) - return false - } - entries.splice(index, 1) - if (entries.length === 0) queuedMirror.delete(agent.id) - return true - } - const disposers = [ - ctx.on('agent/inbox/enqueue', (agent: Agent, item: InboxItem) => { - if (item.placement !== 'queued') return - const unseen = takeUnseen(agent.id, item.id) - if (unseen?.kind === 'terminal') return - let entries = queuedMirror.get(agent.id) - if (entries === undefined) { - entries = [] - queuedMirror.set(agent.id, entries) - } - entries.push(unseen?.kind === 'update' ? unseen.item : item) - publishQueue(agent.id) - }), - ctx.on('agent/inbox/update', (agent: Agent, item: InboxItem) => { - const entries = queuedMirror.get(agent.id) - if (entries === undefined) { - rememberUnseen(agent.id, item.id, { kind: 'update', item }) - return - } - const index = entries.findIndex(entry => entry.id === item.id) - if (index === -1) { - rememberUnseen(agent.id, item.id, { kind: 'update', item }) - return - } - entries.splice(index, 1, item) - publishQueue(agent.id) - }), - ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => { - if (retire(agent, item)) publishQueue(agent.id) - }), - ctx.on('agent/inbox/discard', (agent: Agent, items: InboxItem[]) => { - let changed = false - for (const item of items) changed = retire(agent, item) || changed - if (changed) publishQueue(agent.id) - }), - ctx.on('session/disposed', (session: Session) => { - queuedMirror.delete(session.id) - unseenQueueEvents.delete(session.id) - }), - ] - return () => { for (const dispose of disposers) dispose() } - }, 'api-proxy: queued mirror') - /** Remove a wait before settling it: synchronous deletion makes the first claimant win. */ function claimQuestion(pending: PendingQuestion, outcome: 'answered' | 'cancelled'): void { pendingQuestions.delete(pending.rpcId) @@ -1162,13 +1056,21 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro updateQueue(request) { const { sessionId, itemId, action } = request.payload const agent = ctx.agents.get(sessionId) - if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') { + const queued = agent?.inbox.nextTurn + const index = queued?.findIndex(message => message.id === itemId) ?? -1 + const message = queued?.[index] + if (agent === undefined || message === undefined) { return Promise.resolve(err(request, { code: 'queue-item-not-found', message: 'queued item is no longer pending', details: { itemId }, })) } + if (action.kind === 'edit') { + agent.inbox.splice('next-turn', index, 1, [freezeMessage({ ...message, content: action.content })]) + } else { + agent.inbox.splice('next-turn', index, 1, []) + } return Promise.resolve(ok(request, { accepted: true as const })) }, @@ -1567,14 +1469,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro // Queue snapshot baseline (pendingQuestions precedent): frames replayed // in arrival order per session; a reconnecting client rebuilds its // queue view from these alone. - for (const [sessionId, items] of queuedMirror) { + for (const agent of ctx.agents.list()) { + const items = agent.inbox.nextTurn + if (items.length === 0) continue queue.push(frame({ type: 'session/queue', - sessionId, - items: items.map(item => ({ - id: item.id, - message: item.message, - })), + sessionId: agent.id, + items: [...items], })) } // Per-session open-call table for result-view pairing. Bounded by the diff --git a/packages/host/apiproxy/src/api/events.schema.ts b/packages/host/apiproxy/src/api/events.schema.ts index 4e17b1f403..040540d830 100644 --- a/packages/host/apiproxy/src/api/events.schema.ts +++ b/packages/host/apiproxy/src/api/events.schema.ts @@ -11,7 +11,7 @@ import type { Wire } from './rpc.schema.ts' import { rpcErrorSchema, rpcIdSchema } from './rpc.schema.ts' import { approvalRequestIdSchema } from './approvals.schema.ts' import { - contentBlockSchema, inboxItemIdSchema, sessionEventSchema, sessionIdSchema, toolEventViewSchema, + contentBlockSchema, messageIdSchema, sessionEventSchema, sessionIdSchema, toolEventViewSchema, } from './sessions.schema.ts' import { workspaceIdSchema, workspaceViewSchema } from './workspace.schema.ts' @@ -25,10 +25,10 @@ export const askUserQuestionItemSchema = z.object({ multiSelect: z.boolean().optional(), }) satisfies z.ZodType> -/** Unified message envelope carried by transient queue frames. */ -const messageSchema = z.object({ - id: z.string().min(1), - role: z.union([z.literal('system'), z.literal('user'), z.literal('assistant')]), +/** User-message envelope carried by queue baselines. */ +const userMessageSchema = z.object({ + id: messageIdSchema, + role: z.literal('user'), content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }), }) @@ -47,10 +47,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [ z.object({ type: z.literal('session/queue'), sessionId: sessionIdSchema, - items: z.array(z.object({ - id: inboxItemIdSchema, - message: messageSchema, - })), + items: z.array(userMessageSchema), }), // value stays wide: it already passed its unit's own schema on the host, // and deep-validating here would import every domain's schema into the carrier. diff --git a/packages/host/apiproxy/src/api/events.ts b/packages/host/apiproxy/src/api/events.ts index ad541b4e4d..859f4e1bc2 100644 --- a/packages/host/apiproxy/src/api/events.ts +++ b/packages/host/apiproxy/src/api/events.ts @@ -8,9 +8,8 @@ import type { AskUserQuestionItem } from '@deepseek-ai/dsh-user-interaction/types' import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval/types' -import type { Message } from '@deepseek-ai/dsh-llm/types' -import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand' import type { CallId } from '@deepseek-ai/dsh-llm/brand' +import type { UserMessage } from '@deepseek-ai/dsh-llm/message' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation' import type { RpcError, RpcId, RpcRequest } from './rpc.ts' @@ -32,14 +31,6 @@ export type ToolEventView = | { for: 'call'; view: ToolCallView } | { for: 'result'; view: ToolResultView } -/** One pending queued occurrence in an authoritative queue snapshot. */ -export interface QueuedInboxItem { - /** Agent-owned occurrence identity used by queue mutations. */ - id: InboxItemId - /** Complete pending message; it is not durable until the Agent claims it. */ - message: Message -} - /** Streaming face of the contract: the two SSE stream openers (mux + host). */ export interface EventsApi { /** @@ -71,13 +62,11 @@ export type MuxFrame = | { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] } | { type: 'question/resolved'; sessionId: SessionId; questionRpcId: RpcId; outcome: 'answered' | 'cancelled' } /** - * Complete transient queue state after every enqueue, mutation, claim, or - * discard. Pending work is not model-visible and therefore has no durable - * session event; the whole snapshot makes edit, deletion, cancel, and - * reconnect converge through one authoritative signal. Pending steering is - * outside this Web queue projection. + * 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. */ - | { type: 'session/queue'; sessionId: SessionId; items: QueuedInboxItem[] } + | { type: 'session/queue'; sessionId: SessionId; items: UserMessage[] } /** * One projection unit's finished value changed (session-projection RFC). * Live push state, never logged — replay recomputes on the host (the diff --git a/packages/host/apiproxy/src/api/index.ts b/packages/host/apiproxy/src/api/index.ts index 0372c8cc30..a89f6db2ea 100644 --- a/packages/host/apiproxy/src/api/index.ts +++ b/packages/host/apiproxy/src/api/index.ts @@ -35,7 +35,7 @@ export type { DirectoryEntry, DirectoryListing, HostApi } from './host.ts' export type { WorkspaceApi, WorkspaceId, WorkspaceView } from './workspace.ts' export type { CommandsApi, CommandDescriptor } from './commands.ts' export type { SkillsApi, SkillEntry } from './skills.ts' -export type { EventsApi, MuxFrame, HostFrame, QueuedInboxItem, ToolCallView, ToolEventView, ToolResultView } from './events.ts' +export type { EventsApi, MuxFrame, HostFrame, ToolCallView, ToolEventView, ToolResultView } from './events.ts' export type { GoalsApi, GoalId, GoalRef } from './goals.ts' export type { ApprovalResponsePayload } from './approvals.ts' export type { QuestionResponsePayload } from './questions.ts' @@ -56,7 +56,5 @@ export type { // ---- Errors and ids ---- export { RpcId, transportError } from './rpc.ts' export type { RpcError, RpcErrorCode, RpcErrorDetailsMap, RpcResult } from './rpc.ts' -export type { InboxItemId } from '@deepseek-ai/dsh-agent/brand' - // ---- Method registry and derived generics ---- export type { RequestPayload, ResponseValue, RpcMethodMap } from './rpc-map.ts' diff --git a/packages/host/apiproxy/src/api/rpc.ts b/packages/host/apiproxy/src/api/rpc.ts index c1fa4e1611..2c27ee16f2 100644 --- a/packages/host/apiproxy/src/api/rpc.ts +++ b/packages/host/apiproxy/src/api/rpc.ts @@ -8,8 +8,8 @@ import type { z as zCore } from 'zod' type ZodIssue = zCore.core.$ZodIssue import type { Branded } from '@deepseek-ai/dsh-brand' +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' import type { SessionId } from '@deepseek-ai/dsh-session/types' -import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand' /** * Message correlation id: the initiator mints it on a request; a response @@ -45,7 +45,7 @@ export interface RpcErrorDetailsMap { 'directory-create-failed': { path: string } 'directory-picker-unavailable': { capability: string } 'agent-busy': { reason: string } - 'queue-item-not-found': { itemId: InboxItemId } + 'queue-item-not-found': { itemId: MessageId } /** A known slash command reported a usage/state error; the message is the command's own text. */ 'command-error': {} /** A leading-/ prompt named no registered command; the message names the token. */ diff --git a/packages/host/apiproxy/src/api/sessions.schema.ts b/packages/host/apiproxy/src/api/sessions.schema.ts index e3744ec37a..6b64b80d81 100644 --- a/packages/host/apiproxy/src/api/sessions.schema.ts +++ b/packages/host/apiproxy/src/api/sessions.schema.ts @@ -7,7 +7,7 @@ import { z } from 'zod' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' -import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand' +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' import type { RequestPayload, ResponseValue } from './rpc-map.ts' import type { Wire } from './rpc.schema.ts' import type { @@ -20,8 +20,8 @@ import type { WorkspaceId } from './workspace.ts' /** SessionId: one brand cast after shape validation (the only cast point in this domain). */ export const sessionIdSchema = z.string().min(1) as unknown as z.ZodType -/** InboxItemId: one brand cast after non-empty string validation. */ -export const inboxItemIdSchema = z.string().min(1) as unknown as z.ZodType +/** MessageId: one brand cast after non-empty string validation. */ +export const messageIdSchema = z.string().min(1) as unknown as z.ZodType /** * WorkspaceId: the workspace domain's one brand cast. Hosted here rather @@ -221,7 +221,7 @@ export const sessionPromptValueSchema = z.object({ /** session.updateQueue request payload. */ export const sessionUpdateQueueRequestSchema = z.object({ sessionId: sessionIdSchema, - itemId: inboxItemIdSchema, + itemId: messageIdSchema, action: z.discriminatedUnion('kind', [ z.object({ kind: z.literal('edit'), content: z.array(contentBlockSchema) }), z.object({ kind: z.literal('remove') }), diff --git a/packages/host/apiproxy/src/api/sessions.ts b/packages/host/apiproxy/src/api/sessions.ts index a57c3f4b73..eac40ffbf3 100644 --- a/packages/host/apiproxy/src/api/sessions.ts +++ b/packages/host/apiproxy/src/api/sessions.ts @@ -4,8 +4,8 @@ * else references RequestPayload<'session.*'> / ResponseValue<'session.*'>. */ +import type { MessageId } from '@deepseek-ai/dsh-llm/brand' import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' -import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types' // The pure-type outlet: api/ is browser-importable, and the package root's // cordis Context merge (via dsh-agent) must not enter client aggregates. @@ -238,7 +238,7 @@ export interface SessionsApi { /** * Edits or removes one pending queued occurrence. */ - updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>): + updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: MessageId; action: QueueAction }>): Promise> /** Stops: clears both FIFOs + aborts the current step (1:1 with agent.cancel). */ diff --git a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts index 6df71258be..73a3a793e6 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -11,8 +11,8 @@ import { MessageId, freezeMessage } from '@deepseek-ai/dsh-llm' import { describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' -import AgentRegistry, { InboxItemId } from '@deepseek-ai/dsh-agent' -import type { Agent, InboxItem, InboxPlacement } from '@deepseek-ai/dsh-agent' +import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent' +import type { Agent } from '@deepseek-ai/dsh-agent' import SessionStore from '@deepseek-ai/dsh-session' import type { SessionId, UserMessage } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' @@ -63,7 +63,14 @@ async function harness(options: { commands?: boolean; skills?: boolean } = {}): /** Register a live structural agent stub (api-proxy-view precedent: only id/session/status/ctx are read). */ function stubAgent(ctx: Context, sessionId?: SessionId): Agent { const session = ctx.sessions.create(sessionId) - const agent = { id: session.id, session, status: 'idle', ctx } as Agent + const inbox = new Inbox(session) + const agent = { + id: session.id, + session, + inbox, + status: 'idle', + ctx, + } as Agent ctx.agents.register(agent) return agent } @@ -264,7 +271,7 @@ describe('host/commands-changed frame', () => { }) }) -/** Build one frozen inbox message for the live `agent/inbox/*` events. */ +/** Build one frozen inbox message. */ function inboxMessage(id: string, text: string, rpcId?: string): UserMessage { return freezeMessage({ id: MessageId(id), @@ -274,27 +281,19 @@ function inboxMessage(id: string, text: string, rpcId?: string): UserMessage { }) } -/** Build one addressable inbox occurrence around a frozen message. */ -function inboxItem(id: string, message: UserMessage, placement: InboxPlacement): InboxItem { - return { id: InboxItemId(id), message, placement } -} - describe('session.updateQueue', () => { - it('routes an addressable action and reports a lost claim race', async () => { + it('splices a queued message and reports a lost claim race', async () => { const ctx = await harness() const agent = stubAgent(ctx) - const seen: unknown[] = [] - agent.updateInbox = (id, action) => { - seen.push({ id, action }) - return id === InboxItemId('present') ? 'applied' : 'not-found' - } + const present = inboxMessage('present', 'before') + agent.inbox.splice('next-turn', 0, 0, [present]) const api = createApiProxy(ctx, DEFAULTS) const applied = await api.sessions.updateQueue({ rpcId: RpcId('q-apply'), payload: { sessionId: agent.id, - itemId: InboxItemId('present'), + itemId: MessageId('present'), action: { kind: 'edit', content: [{ type: 'text', text: 'edited' }] }, }, }) @@ -303,15 +302,15 @@ describe('session.updateQueue', () => { rpcId: RpcId('q-missing'), payload: { sessionId: agent.id, - itemId: InboxItemId('claimed'), + itemId: MessageId('claimed'), action: { kind: 'remove' }, }, }) expect(expectErr(missing)).toMatchObject({ code: 'queue-item-not-found' }) - expect(seen).toEqual([ - { id: 'present', action: { kind: 'edit', content: [{ type: 'text', text: 'edited' }] } }, - { id: 'claimed', action: { kind: 'remove' } }, - ]) + expect(agent.inbox.nextTurn[0]).toMatchObject({ + id: 'present', + content: [{ type: 'text', text: 'edited' }], + }) }) it('rejects a stale occurrence without resuming a cold agent', async () => { @@ -322,7 +321,7 @@ describe('session.updateQueue', () => { rpcId: RpcId('q-cold'), payload: { sessionId: 'cold-session' as SessionId, - itemId: InboxItemId('stale-item'), + itemId: MessageId('stale-item'), action: { kind: 'remove' }, }, }) @@ -333,100 +332,24 @@ describe('session.updateQueue', () => { }) describe('session/queue frames', () => { - it('folds nested mutations observed before their outer enqueue', async () => { - const ctx = await harness() - const agent = stubAgent(ctx) - const original = inboxItem('i-edit', inboxMessage('m-edit', 'before'), 'queued') - const edited = inboxItem('i-edit', inboxMessage('m-edit', 'after'), 'queued') - const removed = inboxItem('i-remove', inboxMessage('m-remove', 'remove me'), 'queued') - ctx.on('agent/inbox/enqueue', (subject, item) => { - if (subject !== agent) return - if (item.id === original.id) ctx.emit('agent/inbox/update', agent, edited) - if (item.id === removed.id) ctx.emit('agent/inbox/discard', agent, [removed]) - }) - const api = createApiProxy(ctx, DEFAULTS) - const live = new AbortController() - const collected = collect( - api.events.mux({ rpcId: RpcId('t-mux-reentrant'), payload: {} }, live.signal), 2, live) - - ctx.emit('agent/inbox/enqueue', agent, original) - ctx.emit('agent/inbox/enqueue', agent, removed) - - const liveFrames = (await collected).filter(frame => frame.type === 'session/queue') - expect(liveFrames.map(frame => frame.items)).toEqual([ - [{ id: edited.id, message: edited.message }], - ]) - const replay = new AbortController() - const replayFrames = await collect( - api.events.mux({ rpcId: RpcId('t-mux-reentrant-replay'), payload: {} }, replay.signal), 2, replay) - expect(replayFrames.filter(frame => frame.type === 'session/queue')).toEqual(liveFrames) - }) - - it('publishes complete live snapshots and replays the latest snapshot on reconnect', async () => { + it('publishes the durable next-turn baseline without duplicating message identity', async () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS) const agent = stubAgent(ctx) - const live = new AbortController() - const liveStream = api.events.mux({ rpcId: RpcId('t-mux-live'), payload: {} }, live.signal) - // subscribed baseline + one queued snapshot; pending steering stays off this wire. - const liveCollected = collect(liveStream, 2, live) - - const queued = inboxItem('i-1', inboxMessage('m-1', 'queued prompt'), 'queued') - const steering = inboxItem('i-2', inboxMessage('m-2', 'steering prompt'), 'steering') - ctx.emit('agent/inbox/enqueue', agent, queued) - ctx.emit('agent/inbox/enqueue', agent, steering) - - const liveFrames = (await liveCollected).filter(f => f.type === 'session/queue') - expect(liveFrames).toEqual([ - { - type: 'session/queue', - sessionId: agent.id, - items: [{ id: queued.id, message: queued.message }], - }, - ]) - - // A fresh mux connection replays only the current authoritative snapshot. - const replay = new AbortController() - const replayFrames = await collect( - api.events.mux({ rpcId: RpcId('t-mux-replay'), payload: {} }, replay.signal), 2, replay) - expect(replayFrames.filter(f => f.type === 'session/queue')).toEqual([liveFrames[0]]) - }) - - it('publishes edits in place in the authoritative order', async () => { - const ctx = await harness() - const api = createApiProxy(ctx, DEFAULTS) - const agent = stubAgent(ctx) - const abort = new AbortController() - const collected = collect( - api.events.mux({ rpcId: RpcId('t-mux-updates'), payload: {} }, abort.signal), 5, abort) - const first = inboxItem('i-a', inboxMessage('m-a', 'a'), 'queued') - const second = inboxItem('i-b', inboxMessage('m-b', 'b'), 'queued') - const edited = inboxItem('i-b', inboxMessage('m-b', 'b edited'), 'queued') - ctx.emit('agent/inbox/enqueue', agent, first) - ctx.emit('agent/inbox/enqueue', agent, second) - ctx.emit('agent/inbox/update', agent, edited) - ctx.emit('agent/inbox/dequeue', agent, edited) - - const frames = (await collected).filter(frame => frame.type === 'session/queue') - expect(frames.map(frame => frame.items)).toEqual([ - [{ id: first.id, message: first.message }], - [{ id: first.id, message: first.message }, { id: second.id, message: second.message }], - [{ id: first.id, message: first.message }, { id: edited.id, message: edited.message }], - [{ id: first.id, message: first.message }], - ]) - }) - - it('publishes an empty snapshot after terminal discard', async () => { - const ctx = await harness() - const api = createApiProxy(ctx, DEFAULTS) - const agent = stubAgent(ctx) - const doomed = inboxItem('i-doomed', inboxMessage('m-5', 'doomed'), 'queued') - ctx.emit('agent/inbox/enqueue', agent, doomed) - ctx.emit('agent/inbox/discard', agent, [doomed]) + const queued = inboxMessage('m-1', 'queued 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-swept'), payload: {} }, abort.signal), 1, abort) - expect(frames.filter(frame => frame.type === 'session/queue')).toHaveLength(0) + api.events.mux({ rpcId: RpcId('t-mux-baseline'), payload: {} }, abort.signal), 2, abort) + expect(frames.filter(frame => frame.type === 'session/queue')).toEqual([ + { + type: 'session/queue', + sessionId: agent.id, + items: [queued], + }, + ]) }) }) diff --git a/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts b/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts index 103ea25bc1..52e40decdc 100644 --- a/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts @@ -45,10 +45,7 @@ function stubAgent(session: Session): Agent { options: {}, session, status: 'idle', - acceptsNextStep: false, ctx: new Context(), - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject: () => {}, diff --git a/packages/host/apiproxy/tests/rpc-schemas.spec.ts b/packages/host/apiproxy/tests/rpc-schemas.spec.ts index 80cd0baf4c..9d97501e74 100644 --- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts +++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts @@ -380,7 +380,7 @@ describe('events frame schemas', () => { { type: 'question/requested', sessionId: 's', questions: [{ id: 'q', question: 'Q?', options: [{ label: 'L' }], multiSelect: true }] }, { type: 'question/resolved', sessionId: 's', questionRpcId: 'r', outcome: 'answered' }, { type: 'session/queue', sessionId: 's', items: [ - { id: 'i1', message: { id: 'm1', role: 'user', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' } } }, + { id: 'm1', role: 'user', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' } }, ] }, { type: 'session/projection', sessionId: 's', key: 'todos', value: [{ content: 'x', status: 'pending' }], seq: 7 }, { type: 'stream/error', error: { code: 'internal', message: 'm', details: {} } }, @@ -401,8 +401,8 @@ describe('events frame schemas', () => { it('rejects a queue snapshot with malformed items', () => { expect(() => muxFrameSchema.parse({ type: 'session/queue', sessionId: 's', items: 'x' })).toThrow() - expect(() => muxFrameSchema.parse({ type: 'session/queue', sessionId: 's', items: [{ id: '', message: {} }] })).toThrow() - expect(() => muxFrameSchema.parse({ type: 'session/queue', sessionId: 's', items: [{ id: 'i', message: { id: 'm', role: 'user', content: [], source: {} } }] })).toThrow() + expect(() => muxFrameSchema.parse({ type: 'session/queue', sessionId: 's', items: [{ id: '', role: 'user', content: [], source: { kind: 'user' } }] })).toThrow() + expect(() => muxFrameSchema.parse({ type: 'session/queue', sessionId: 's', items: [{ id: 'm', role: 'assistant', content: [], source: { kind: 'user' } }] })).toThrow() }) it('accepts every host frame branch', () => { diff --git a/packages/pty/pty-local/tests/index.spec.ts b/packages/pty/pty-local/tests/index.spec.ts index 7b87fe6b10..3976e586cb 100644 --- a/packages/pty/pty-local/tests/index.spec.ts +++ b/packages/pty/pty-local/tests/index.spec.ts @@ -41,8 +41,8 @@ function config(): ResolvedConfig { function agent(ctx: Context): Agent { const id = SessionId('agent') return { - id, options: {}, session: new Session(id), status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id, options: {}, session: new Session(id), status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } } @@ -248,8 +248,8 @@ describe('pty-local plugin shape', () => { const session = ctx.sessions.create(SessionId('mode-owner')) const ownerFiber = await ctx.plugin(() => {}) const owner: Agent = { - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx: ownerFiber.ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx: ownerFiber.ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } ctx.agents.register(owner) const providerFiber = await registerStubLocalBackend(ctx, () => stubLocalSession()) @@ -291,8 +291,8 @@ describe('pty-local plugin shape', () => { const session = ctx.sessions.create(SessionId('pending-mode-owner')) const ownerFiber = await ctx.plugin(() => {}) const owner: Agent = { - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx: ownerFiber.ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx: ownerFiber.ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } ctx.agents.register(owner) const gate = Promise.withResolvers() diff --git a/packages/pty/pty-local/tests/local.spec.ts b/packages/pty/pty-local/tests/local.spec.ts index c3fb33c75c..c2cb23cb59 100644 --- a/packages/pty/pty-local/tests/local.spec.ts +++ b/packages/pty/pty-local/tests/local.spec.ts @@ -34,8 +34,8 @@ function stubAgent(ctx: Context, rawId: string): Agent { const id = SessionId(rawId) const scope = ctx.plugin(() => {}) return { - id, options: {}, session: new Session(id), status: 'idle', acceptsNextStep: false, ctx: scope.ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id, options: {}, session: new Session(id), status: 'idle', ctx: scope.ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } } diff --git a/packages/pty/pty/tests/service.spec.ts b/packages/pty/pty/tests/service.spec.ts index ba0996841c..cf5c4f0bab 100644 --- a/packages/pty/pty/tests/service.spec.ts +++ b/packages/pty/pty/tests/service.spec.ts @@ -26,10 +26,7 @@ function stubAgent(ctx: Context, rawId: string): Agent { options: {}, session: new Session(id), status: 'idle', - acceptsNextStep: false, ctx: scopeFiber.ctx, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject: () => {}, diff --git a/packages/pty/tool-pty/tests/loader-composition.spec.ts b/packages/pty/tool-pty/tests/loader-composition.spec.ts index f6ad1084c6..da8555e4c8 100644 --- a/packages/pty/tool-pty/tests/loader-composition.spec.ts +++ b/packages/pty/tool-pty/tests/loader-composition.spec.ts @@ -39,8 +39,8 @@ function agent(ctx: Context): Agent { const scope = ctx.plugin(() => {}) const id = SessionId('pty-loader-agent') const value: Agent = { - id, options: {}, session: new Session(id), status: 'idle', acceptsNextStep: false, ctx: scope.ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id, options: {}, session: new Session(id), status: 'idle', ctx: scope.ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } ctx.agents.register(value) return value diff --git a/packages/pty/tool-pty/tests/tools.spec.ts b/packages/pty/tool-pty/tests/tools.spec.ts index a09fcac714..ed9a41d370 100644 --- a/packages/pty/tool-pty/tests/tools.spec.ts +++ b/packages/pty/tool-pty/tests/tools.spec.ts @@ -17,8 +17,8 @@ function fakeAgent(ctx: Context, rawId: string): Agent { const scope = ctx.plugin(() => {}) const id = SessionId(rawId) const agent: Agent = { - id, options: {}, session: new Session(id), status: 'idle', acceptsNextStep: false, ctx: scope.ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id, options: {}, session: new Session(id), status: 'idle', ctx: scope.ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } ctx.agents.register(agent) return agent diff --git a/packages/skill/tool-skill/tests/tool-skill.spec.ts b/packages/skill/tool-skill/tests/tool-skill.spec.ts index e866805a2d..98ae51e38a 100644 --- a/packages/skill/tool-skill/tests/tool-skill.spec.ts +++ b/packages/skill/tool-skill/tests/tool-skill.spec.ts @@ -45,9 +45,6 @@ function agentForCwd(cwd: string): Agent { options: {}, session, status: 'idle', - acceptsNextStep: false, - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { @@ -64,10 +61,7 @@ function sessionAgent(session: Session, id = 'tool-skill-agent'): Agent { options: {}, session, status: 'running', - acceptsNextStep: false, ctx: new Context(), - send: () => {}, - updateInbox: () => 'not-found', followup: () => {}, steer: () => {}, inject(input) { diff --git a/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts b/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts index 2d6338701d..f13c68cfa1 100644 --- a/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts +++ b/packages/subagent/subagent-spawn/tests/subagent-spawn.spec.ts @@ -200,10 +200,14 @@ describe('dsh-subagent-spawn', () => { expect(published).toEqual([]) }) - it('a cancel from agent/inbox/enqueue maps a no-turn child log to aborted', async () => { + it('a cancel after the child prompt is queued maps a no-turn child log to aborted', async () => { const { ctx, parent } = await setup([]) const controller = new AbortController() - ctx.on('agent/inbox/enqueue', () => { controller.abort('queued-window') }) + ctx.on('session/event', (_session, event) => { + if (event.type === 'agent/inbox/spliced' && event.data.inserted.length > 0) { + queueMicrotask(() => { controller.abort('queued-window') }) + } + }) const run = await start(ctx, 'spawn', { prompt: [{ type: 'text', text: 'p' }], parent, signal: controller.signal }) const result = await run.result expect(result).toMatchObject({ stopReason: 'aborted', output: [] }) diff --git a/packages/tasks/tasks-local/tests/tasks.spec.ts b/packages/tasks/tasks-local/tests/tasks.spec.ts index 555eb6146a..5cf11732cf 100644 --- a/packages/tasks/tasks-local/tests/tasks.spec.ts +++ b/packages/tasks/tasks-local/tests/tasks.spec.ts @@ -23,10 +23,7 @@ function stubAgent(ctx: Context, rawId: string): Agent { options: {}, session: new Session(id), status: 'idle' as const, - acceptsNextStep: false, ctx: scopeFiber.ctx, - send: () => {}, - updateInbox: (): 'not-found' => 'not-found', followup: () => {}, steer: () => {}, inject: () => {}, diff --git a/packages/ui/tui/src/index.ts b/packages/ui/tui/src/index.ts index 8c5322f8e6..de8288d71d 100644 --- a/packages/ui/tui/src/index.ts +++ b/packages/ui/tui/src/index.ts @@ -1198,7 +1198,7 @@ export function createTuiChat( appendNotice(`Agent "${agent.id}" is disposed.`, 'error') return } - if (agent.acceptsNextStep) { + if (agent.status === 'running') { // Steering is never subject to prompt admission; an attached snapshot // drains beside it at the same step boundary through the outbox. if (attachedContext !== undefined) { diff --git a/packages/ui/tui/tests/harness.ts b/packages/ui/tui/tests/harness.ts index 6be1183d78..a6a22ec2d9 100644 --- a/packages/ui/tui/tests/harness.ts +++ b/packages/ui/tui/tests/harness.ts @@ -36,8 +36,6 @@ interface FakeAgent extends Agent { export interface TuiHarnessOptions { status?: AgentStatus - /** Override the fake agent's next-step capability independently of status. */ - acceptsNextStep?: boolean config?: Config /** Leave the session event log empty instead of seeding one turn and step. */ omitInitialLifecycle?: boolean @@ -193,9 +191,6 @@ export async function createTuiTestHarness 'not-found', followup(input) { sent.push(input.content) sentMessages.push(input) diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index e18c96c4db..0979714468 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -2636,45 +2636,6 @@ describe('pi-tui chat lifecycle and transcript', () => { await dispose(result) }) - it('keeps a referenced prompt on admission when running no longer accepts next-step input', async () => { - const result = await setup({ - status: 'running', - acceptsNextStep: false, - omitInitialLifecycle: true, - async configureContext(ctx) { - ctx.provide('tools', { get: () => undefined } as never) - await ctx.plugin(TestSessionQueryService) - await ctx.plugin(SessionReferenceService) - const source = ctx.sessions.create(SessionId('admission-src'), { - meta: { cwd: process.cwd(), createdAt: 1 }, - }) - appendUser(source, 'source background') - source.append('session/title', { - title: 'Admission source', - messageSeqs: [0], - source: { kind: 'fallback' }, - }) - }, - }) - - result.terminal.send(formatSessionReferenceMention({ - sessionId: SessionId('admission-src'), - label: 'Admission source', - })) - result.terminal.send('\r') - await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(1) }) - - expect(result.agent.steered).toHaveLength(0) - expect(result.agent.injected).toHaveLength(0) - const decision = await agentEvents(result.ctx, result.agent).waterfall( - 'agent/prompt-submit', result.agent.sentMessages[0]!, - new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }), - ) - expect(decision.kind === 'allow' && decision.additionalContexts?.[0]?.source) - .toMatchObject({ kind: 'session-reference', references: [{ sessionId: 'admission-src' }] }) - await dispose(result) - }) - it('releases the reference-admission wrapper on the ordinary allowed path', async () => { const result = await setup({ async configureContext(ctx) { @@ -4974,8 +4935,8 @@ describe('terminal mounting', () => { ctx.provide('tools', { get: () => undefined } as never) const session = ctx.sessions.create(SessionId('main')) ctx.agents.register({ - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), }) const terminal = new FakeTerminal() mountTui(ctx, { theme: { color: false } }, { terminal, exit: vi.fn() }) @@ -4999,8 +4960,8 @@ describe('terminal mounting', () => { ctx.provide('tools', { get: () => undefined } as never) const session = ctx.sessions.create(SessionId('main')) ctx.agents.register({ - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), }) const terminal = new FakeTerminal() // Mirror dsh-tui's own inject (minus loader, the absence under test). @@ -5034,15 +4995,15 @@ describe('terminal mounting', () => { const otherSession = ctx.sessions.create(SessionId('other-session')) ctx.agents.register({ - id: otherSession.id, options: {}, session: otherSession, status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: otherSession.id, options: {}, session: otherSession, status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), }) expect(terminal.started).toBe(0) const session = ctx.sessions.create(SessionId('late-session')) const agent = { - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), } as Agent ctx.agents.register(agent) await tick() @@ -5072,8 +5033,8 @@ describe('terminal mounting', () => { const session = ctx.sessions.create(SessionId('main-session')) ctx.agents.register({ - id: session.id, options: {}, session, status: 'idle', acceptsNextStep: false, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'idle', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), }) await tick() expect(terminal.started).toBe(0) @@ -5116,8 +5077,8 @@ describe('terminal mounting', () => { session.append('turn/start', { turn: 1 }) session.append('step/start', { turn: 1, step: 1 }) ctx.agents.register({ - id: session.id, options: {}, session, status: 'running', acceptsNextStep: true, ctx, - followup: () => {}, steer: () => {}, inject: () => {}, send: () => {}, updateInbox: () => 'not-found', cancel() {}, whenIdle: () => Promise.resolve(), + id: session.id, options: {}, session, status: 'running', ctx, + followup: () => {}, steer: () => {}, inject: () => {}, cancel() {}, whenIdle: () => Promise.resolve(), }) const terminal = new FakeTerminal() terminal.start = () => { throw new Error('terminal startup failed') } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 61930c815d..58228736c6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2048,9 +2048,6 @@ importers: packages/core/agent: devDependencies: - '@deepseek-ai/dsh-brand': - specifier: workspace:^ - version: link:../../util/brand '@deepseek-ai/dsh-invariants': specifier: workspace:^ version: link:../../support/invariants