diff --git a/.agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.i18n.yaml b/.agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.i18n.yaml similarity index 100% rename from .agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.i18n.yaml rename to .agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.i18n.yaml diff --git a/.agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.md b/.agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.md similarity index 94% rename from .agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.md rename to .agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.md index ada6d222f2..27da3f1c27 100644 --- a/.agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.md +++ b/.agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.md @@ -1,6 +1,6 @@ # Agent Note: Separate context injection from turn execution -Status: proposed +Status: implemented English | [中文](2026-07-24-separate-context-injection-from-turn-execution.zh.md) @@ -14,7 +14,7 @@ Idle `inject()` exposes a second mismatch. Injection does not request model exec `HookContext` also names its producer rather than its role. The value may come from a native plugin, a hook bridge, prompt admission, or tool post-processing. Its stable meaning is simply additional model-facing context with provenance. -## Proposal +## Decision Make `inject()` the only caller-facing operation for adding supplementary model-facing input, and define a turn exclusively as one execution of the model loop. @@ -42,7 +42,7 @@ Caller-driven injection and hook-produced additional context deliberately have d Cross-session references follow the ordinary composition: the host prepares the snapshot, injects it with session-reference provenance, then sends or steers the readable direct prompt. The target log contains two simple messages, so later source mutation cannot change replay and transcript consumers do not need a prompt envelope. This supersedes the attachment mechanism in the [cross-session reference decision](../../implemented/feature/2026-07-21-cross-session-references.md) while retaining its snapshot and trust-boundary rules. -This proposal preserves the caller-owned framing decision from [unwrapped injected content](../../implemented/simplification/2026-07-20-unwrap-injected-content-envelopes.md), the one-item turn rule from [one send, one turn](../../implemented/simplification/2026-07-17-one-send-one-turn.md), and narrows the [turn-enclosure decision](../../implemented/architecture/2026-06-15-turn-enclosure-invariant.md) so turns enclose execution rather than every session event. +This decision preserves the caller-owned framing decision from [unwrapped injected content](../simplification/2026-07-20-unwrap-injected-content-envelopes.md), the one-item turn rule from [one send, one turn](../simplification/2026-07-17-one-send-one-turn.md), and narrows the [turn-enclosure decision](2026-06-15-turn-enclosure-invariant.md) so turns enclose execution rather than every session event. ## Alternatives considered @@ -56,7 +56,7 @@ This proposal preserves the caller-owned framing decision from [unwrapped inject **Let hooks call `inject()` directly instead of returning additional contexts.** Direct injection would erase the extension point's admission ownership: a listener could append context before a downstream listener blocks the operation. Returning `additionalContexts` keeps the waterfall result authoritative while sharing the same post-admission outbox path. -## Acceptance criteria +## Verification - `SendOptions` and steering inbox records contain no attached contexts; `agent/queued` reports only the retained message and steering facts. - `UserMessageData` replaces `HookContext` across prompt interception, tool execution, hook bridges, guards, and context producers. @@ -66,7 +66,7 @@ This proposal preserves the caller-owned framing decision from [unwrapped inject - Blocked prompt admission opens no turn and appends neither the prompt nor hook-produced additional contexts; independently injected caller context remains. - Unit, persistence/resume, invariant, ACP/TUI replay, and keyless assembled-application snapshots cover the new event order and durability semantics. -## Risks +## Consequences - Allowing one surface event outside turns weakens a simple invariant and may expose hidden assumptions in persistence scanning, crash repair, forking, compaction, and session queries. - Consecutive user-role messages replace one baked prompt message; provider adapters and cache behavior must accept and preserve that ordering. diff --git a/.agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md b/.agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md similarity index 99% rename from .agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md rename to .agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md index 85cad4bef7..7c3946043b 100644 --- a/.agents/notes/proposed/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md +++ b/.agents/notes/implemented/architecture/2026-07-24-separate-context-injection-from-turn-execution.zh.md @@ -1,6 +1,6 @@ # Agent Note: 将上下文注入与轮次执行分离 -Status: proposed +Status: implemented [English](2026-07-24-separate-context-injection-from-turn-execution.md) | 中文 diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index d0333d97d6..1dd0283429 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -108,11 +108,6 @@ export class Session implements ObservableSnapshot { private dispatchesRev = 0 private dispatchesCache: { rev: number; value: ReadonlyMap } | null = null private running = false - /** Whether a turn is open on the live stream — set on turn/start, cleared on - * turn/end. A session/queued frame arriving while this is true joined the - * steering FIFO; the host no longer stamps steering on the frame, so the - * client derives it from the same ordered turn boundaries the host saw. */ - private turnOpen = false /** * Sticky send marker, private input of the composerPhase derivation: set * synchronously before prompt()'s first await, never reset — the blank → @@ -342,12 +337,6 @@ export class Session implements ObservableSnapshot { switch (frame.type) { case 'session/event': { this.retireQueued(frame.event) - // Track turn-open state AFTER retirement (a message turn/start first - // claims its queued entry, then opens the turn), so a later queued - // frame is stamped steering iff a turn is open — the host no longer - // stamps it on the frame. - if (frame.event.type === 'turn/start') this.turnOpen = true - else if (frame.event.type === 'turn/end') this.turnOpen = false this.acceptLiveEvent(frame.event, frame.view) return } @@ -357,7 +346,7 @@ export class Session implements ObservableSnapshot { const key = 'rpcId' in frame.source ? String(frame.source.rpcId) : `f:${rpcId}` this.queued.push({ row: { key, preview: queuePreviewOf(frame.content) }, - steering: this.turnOpen, + steering: frame.steering, sourceJson: JSON.stringify(frame.source), }) this.queueRev++ diff --git a/packages/client/runtime/tests/queue-store.spec.ts b/packages/client/runtime/tests/queue-store.spec.ts index 8481243cb7..6fe9f33c80 100644 --- a/packages/client/runtime/tests/queue-store.spec.ts +++ b/packages/client/runtime/tests/queue-store.spec.ts @@ -17,10 +17,11 @@ const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }] const rid = (id: string): RpcId => id as RpcId /** session/queued frame with the wire-sourced rpcId key (the host prompt path). */ -function queuedFrame(body: string, rpcId: string): MuxFrame { +function queuedFrame(body: string, rpcId: string, steering = false): MuxFrame { return { type: 'session/queued', sessionId: SID, content: text(body), source: { kind: 'user', rpcId: rid(rpcId) } as never, + steering, } } @@ -42,6 +43,7 @@ describe('queue intake', () => { type: 'session/queued', sessionId: SID, content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never], source: { kind: 'plugin', plugin: 'loop' }, + steering: false, }) expect(session.getSnapshot().queue).toEqual([{ key: 'f:env-2', preview: 'hi [image]' }]) }) @@ -86,14 +88,7 @@ describe('queue retirement (host queuedMirror rules)', () => { it('steering/message drains the source-matched steering row only', () => { const session = makeSession() session.handleMuxEnvelope(rid('e1'), queuedFrame('普通', 'p-1')) // idle → non-steering - // Injection-triggered turn/start opens the turn without claiming a row, so - // the next queued frame is derived steering (arrived mid-turn). - const injection = { - ...ev.turnStart(0, 0), - data: { turn: 0, trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'x' } } }, - } as never - session.handleMuxEnvelope(rid('e2'), { type: 'session/event', sessionId: SID, event: injection }) - session.handleMuxEnvelope(rid('e3'), queuedFrame('插话', 'p-2')) // turn open → steering + session.handleMuxEnvelope(rid('e3'), queuedFrame('插话', 'p-2', true)) // Loop-authored steering (different source) must not consume the user entry. const foreignSteering = { seq: 0, time: 1, @@ -151,6 +146,19 @@ describe('queue reconnect semantics', () => { await session.resync() expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-fresh']) }) + + it('replayed steering retires without a replayed turn/start', () => { + const session = makeSession() + session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 }) + session.handleMuxEnvelope(rid('e2'), queuedFrame('重连插话', 'p-steer', true)) + const committed = { + seq: 6, time: 2, + type: 'steering/message', surfaceOp: 'append', + data: { turn: 1, content: text('重连插话'), source: { kind: 'user', rpcId: rid('p-steer') } }, + } as never + session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: committed }) + expect(session.getSnapshot().queue).toEqual([]) + }) }) describe('manager buffering of queued frames', () => { diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index ba5ec3dae9..6a3c68e959 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -17,6 +17,7 @@ import type { Agent, CancelOptions, AgentInterruptReason, + InboxPlacement, AgentOptions, AgentStatus, SettleReason, @@ -51,6 +52,8 @@ export class ReactLoopAgent implements Agent { /** Whether observers see a running interval; consecutive turns share it. */ private busy = false + /** Whether next-step input belongs to the current admission or open turn. */ + private acceptingNextStep = false /** Abort owner for the current admission or turn. */ private abort: AbortController | undefined /** Coalesced retry capability scoped to the active request-error waterfall. */ @@ -96,7 +99,7 @@ export class ReactLoopAgent implements Agent { const { target, wakeup } = options const id = AgentMessageId(randomUUID()) if (target === 'next-step' && !wakeup) { - if (this.turnOpen) { + if (this.acceptingNextStep) { this.outbox.push({ content, source }) return id } @@ -104,19 +107,19 @@ export class ReactLoopAgent implements Agent { return id } - const steering = target === 'next-step' && this.turnOpen + const placement: InboxPlacement = target === 'next-step' && this.acceptingNextStep ? 'steering' : 'queued' const message: AgentMessage = { id, content, source, } - if (steering) { + if (placement === 'steering') { this.outbox.push(message) } else { this.queued.push({ message, wakeup }) } - emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message) - if (!steering && wakeup) this.kick() + emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message, placement) + if (placement === 'queued' && wakeup) this.kick() return id } @@ -212,6 +215,7 @@ export class ReactLoopAgent implements Agent { const admission = new AbortController() this.abort = admission + this.acceptingNextStep = true // Claimed admission is part of the running interval: it is cancellable // activity, so observers (and their cancel routing) must see it. if (!this.busy) { @@ -253,6 +257,7 @@ export class ReactLoopAgent implements Agent { // still owns the slot here and releasing it unconditionally is exact. this.abort = undefined if (admitted === undefined) { + this.acceptingNextStep = false // A synchronously aborted admission would otherwise publish idle // inside send()'s own synchronous extent, before any post-send // subscriber could observe the transition. @@ -279,6 +284,7 @@ export class ReactLoopAgent implements Agent { if (this.abort !== undefined) throw new Error(`agent "${this.id}" is already running`) const controller = new AbortController() this.abort = controller + this.acceptingNextStep = true if (!this.busy) { this.busy = true emitAgentEvent(this.loopCtx, this, 'agent/status', 'running') @@ -384,6 +390,7 @@ export class ReactLoopAgent implements Agent { // Every step-close happens before this point on both success and // failure paths (step(), the request-failed branch, the catch), so the // finally owes only the turn boundary. + this.acceptingNextStep = false try { if (this.turnOpen) { // Re-entrant turn/end listeners must route new input to a later turn. diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index a31ac3369c..1cd6fa9bc5 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -4,7 +4,7 @@ import LlmService, { CallId, MessageSource, ProviderRequestId, StreamChunk } fro 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 } from '@deepseek-ai/dsh-agent' +import AgentRegistry, { type Agent, type InboxPlacement } 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' @@ -500,9 +500,11 @@ describe('adapter registration, routing, and accepted-input ownership', () => { const queuedSources: MessageSource[] = [] const queuedShapes: string[][] = [] - ctx.on('agent/inbox/enqueue', (_agent, message) => { + const placements: InboxPlacement[] = [] + ctx.on('agent/inbox/enqueue', (_agent, message, placement) => { queuedSources.push(message.source) queuedShapes.push(Object.keys(message).sort()) + placements.push(placement) }) send(agent, 'go') // no explicit source → default {kind:'user'} must be visible @@ -516,6 +518,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => { ['content', 'id', 'source'], ['content', 'id', 'source'], ]) + expect(placements).toEqual(['queued', 'steering']) // 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.source] : []) diff --git a/packages/core/agent-loop/tests/interception.spec.ts b/packages/core/agent-loop/tests/interception.spec.ts index abf512ab3c..0e690ab8ae 100644 --- a/packages/core/agent-loop/tests/interception.spec.ts +++ b/packages/core/agent-loop/tests/interception.spec.ts @@ -4,7 +4,7 @@ import LlmService, { CallId } from '@deepseek-ai/dsh-llm' import SessionStore, { SessionId, type SessionEvent, type TurnEndReason } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineContentToolFixture, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools' -import AgentRegistry, { type Agent, type PromptDecision, type SessionStartSource } from '@deepseek-ai/dsh-agent' +import AgentRegistry, { type Agent, type InboxPlacement, type PromptDecision, type SessionStartSource } from '@deepseek-ai/dsh-agent' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts' @@ -164,6 +164,95 @@ describe('agent/prompt-submit', () => { expect(reasons).toEqual([]) }) + it('stages inject and steer during admission for the admitted turn', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('admission-outbox'), { provider: 'mock', model: 'mock' }) + const entered = Promise.withResolvers() + const decision = Promise.withResolvers() + const placements: InboxPlacement[] = [] + ctx.on('agent/prompt-submit', async () => { + entered.resolve(undefined) + return decision.promise + }) + ctx.on('agent/inbox/enqueue', (subject, _message, placement) => { + if (subject === agent) placements.push(placement) + }) + + const idle = waitForIdle(ctx, agent) + send(agent, 'admitted prompt') + await entered.promise + expect(agent.status).toBe('running') + expect(events(agent).some(event => event.type === 'turn/start')).toBe(false) + + agent.inject({ + content: [{ type: 'text', text: 'attached context' }], + source: { kind: 'plugin', plugin: 'test' }, + }) + agent.steer({ content: [{ type: 'text', text: 'admission steering' }], source: { kind: 'user' } }) + expect(events(agent).some(event => event.type === 'user/message')).toBe(false) + expect(placements).toEqual(['queued', 'steering']) + + decision.resolve({ kind: 'allow' }) + await idle + + const staged = events(agent).filter(event => + event.type === 'turn/start' || event.type === 'user/message' || event.type === 'steering/message') + expect(staged.map(event => event.type)).toEqual([ + 'turn/start', + 'user/message', + 'user/message', + 'steering/message', + ]) + expect(staged[1]?.type === 'user/message' && staged[1].data.content) + .toEqual([{ type: 'text', text: 'admitted prompt' }]) + expect(staged[2]?.type === 'user/message' && staged[2].data.content) + .toEqual([{ type: 'text', text: 'attached context' }]) + expect(staged[3]?.type === 'steering/message' && staged[3].data.content) + .toEqual([{ type: 'text', text: 'admission steering' }]) + const request = JSON.stringify(adapter.requests[0]?.messages) + expect(request).toContain('admitted prompt') + expect(request).toContain('attached context') + expect(request).toContain('admission steering') + }) + + it('keeps admission-time outbox input staged when admission is blocked', async () => { + const adapter = new MockAdapter([textResponse('retried')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('blocked-admission-outbox'), { provider: 'mock', model: 'mock' }) + const entered = Promise.withResolvers() + const decision = Promise.withResolvers() + ctx.on('agent/prompt-submit', async () => { + entered.resolve(undefined) + return decision.promise + }) + + const blockedIdle = waitForIdle(ctx, agent) + send(agent, 'blocked prompt') + await entered.promise + agent.inject({ + content: [{ type: 'text', text: 'staged context' }], + source: { kind: 'plugin', plugin: 'test' }, + }) + agent.steer({ content: [{ type: 'text', text: 'staged steering' }], source: { kind: 'user' } }) + decision.resolve({ kind: 'block', reason: 'policy' }) + await blockedIdle + + expect(events(agent)).toEqual([]) + expect(adapter.requests).toEqual([]) + + const retryIdle = waitForIdle(ctx, agent) + agent.retry() + await retryIdle + + const staged = events(agent).filter(event => + event.type === 'user/message' || event.type === 'steering/message') + expect(staged.map(event => event.type)).toEqual(['user/message', 'steering/message']) + expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt') + expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged context') + expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering') + }) + it('adjacent blocked and allowed prompts keep independent turn outcomes', async () => { const adapter = new MockAdapter([textResponse('ran once')]) const ctx = await harness(adapter) diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index ac2350dafe..f59f0d92fb 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -29,11 +29,15 @@ export interface AgentOptions { /** * Which inbox queue a {@link Agent.send} item joins: * - `next-turn` — the item becomes its own turn, claimed at a turn boundary. - * - `next-step` — the item joins the active turn between steps as steering, - * or, when no turn is active, is promoted per its `wakeup` flag. + * - `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' + /** * Options for the unified {@link Agent.send} primitive over the * (`target` × `wakeup`) matrix. Named presets: {@link Agent.followup} @@ -153,12 +157,13 @@ export interface Agent { * - `next-turn` queues an item that becomes the sole ordinary message of its * own FIFO-ordered turn; `wakeup:true` wakes a * parked driver, while `wakeup:false` queues without waking. - * - `next-step` with `wakeup:true` submits steering into the active turn - * (idle falls back to a woken `next-turn`). + * - `next-step` with `wakeup:true` stages steering during prompt admission + * or an open turn; outside that window it falls back to a woken + * `next-turn`. * - `next-step` with `wakeup:false` injects durable model-facing context - * without running the model: an open turn stages it for the next safe log - * position, while an idle injection appends it immediately without opening - * a turn. + * without running the model: admission or an open turn stages it for the + * next safe log position, while an injection outside that window appends + * immediately without opening a turn. * @param input - model-facing content and its producer provenance. * @param options - target queue and wakeup decision. * @returns the accepted message's {@link AgentMessageId}, stable across its `agent/inbox/*` events. @@ -189,12 +194,13 @@ export interface Agent { followup(input: UserMessageData): AgentMessageId /** - * Submit steering into the running turn — the `next-step`/wakeup preset of - * {@link send}. An open turn records it at the next steering checkpoint before - * a request or stop decision. If the turn fails before that boundary, the - * remainder stays staged without waking the agent; retry or a later prompt - * takes it. Idle steering falls back to a woken follow-up turn, while - * cancellation or disposal may discard pending steering. + * Submit steering during prompt admission or an open turn — the + * `next-step`/wakeup preset of {@link send}. It stages for the next steering + * checkpoint before a request or stop decision. If the activity fails before + * that boundary, the remainder stays staged without waking the agent; retry + * or a later prompt takes it. Outside that window steering falls back to a + * woken follow-up turn, while cancellation or disposal may discard pending + * steering. * @param input - steering content and its producer provenance. * @returns the accepted message's {@link AgentMessageId}. */ @@ -202,9 +208,9 @@ export interface Agent { /** * Append model-facing context without running the model — the - * `next-step`/no-wakeup preset of {@link send}. An open-turn injection stages - * at the next safe log position; an idle injection appends immediately - * without opening a turn. + * `next-step`/no-wakeup preset of {@link send}. Admission or an open turn + * stages it at the next safe log position; outside that window it appends + * immediately without opening a turn. * @param input - injected context and its producer provenance. * @returns the accepted message's {@link AgentMessageId}. */ @@ -253,13 +259,16 @@ declare module 'cordis' { */ 'agent/status'(this: Scoped, agent: Agent, status: AgentStatus): void /** - * An item entered the queued or steering inbox. + * An item entered the queued or steering inbox. `placement` is the + * acceptance-time routing result; listeners must not reconstruct it from + * later agent or session state. * @param agent - the owning agent. * @param message - accepted content, source, and correlation identity. + * @param placement - resolved queued or steering placement. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode emit */ - 'agent/inbox/enqueue'(this: Scoped, agent: Agent, message: AgentMessage): void + 'agent/inbox/enqueue'(this: Scoped, agent: Agent, message: AgentMessage, placement: InboxPlacement): void /** * The driver claimed one item out of the inbox: a queued item at a turn * boundary, or steering drained between steps. Fires after the item leaves diff --git a/packages/core/agent/tests/invariant.spec.ts b/packages/core/agent/tests/invariant.spec.ts index 8b5e6ae7d7..d95d97ea32 100644 --- a/packages/core/agent/tests/invariant.spec.ts +++ b/packages/core/agent/tests/invariant.spec.ts @@ -52,8 +52,8 @@ describe('agent inbox invariants', () => { 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()) + ctx.emit(at, 'agent/inbox/enqueue', agent, info(), 'queued') + 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() @@ -70,7 +70,7 @@ describe('agent inbox invariants', () => { const ctx = await setup() const agent = mockAgent('i3') const at = scopeTarget(agent, agent) - ctx.emit(at, 'agent/inbox/enqueue', agent, info()) + ctx.emit(at, 'agent/inbox/enqueue', agent, info(), 'queued') expect(() => { ctx.emit(at, 'agent/inbox/discard', agent, [info(), info()]) }) .toThrow(/dropped 2 items but only 1 were outstanding/) }) diff --git a/packages/core/scope/tests/invariant.spec.ts b/packages/core/scope/tests/invariant.spec.ts index e0acb2809c..e94a061cf6 100644 --- a/packages/core/scope/tests/invariant.spec.ts +++ b/packages/core/scope/tests/invariant.spec.ts @@ -41,7 +41,7 @@ describe('scoped-dispatch invariants', () => { 'agent/created': [agent], 'agent/disposed': [agent], 'agent/status': [agent, 'idle'], - 'agent/inbox/enqueue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' } }], + 'agent/inbox/enqueue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' } }, 'queued'], 'agent/inbox/dequeue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' } }], 'agent/inbox/discard': [agent, []], 'agent/cancel-requested': [agent, { kind: 'user' }], diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index bb962ab84c..0ac45fa588 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -380,7 +380,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro * `agent/inbox/dequeue` OR `agent/inbox/discard` (the inbox contract), so * the mirror needs no consumption heuristics or sweeps beyond disposal. */ - const queuedMirror = new Map>() + const queuedMirror = new Map>() ctx.effect(() => { const retire = (agent: Agent, id: AgentMessageId): void => { const entries = queuedMirror.get(agent.id) @@ -389,11 +389,21 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro if (entries.size === 0) queuedMirror.delete(agent.id) } const disposers = [ - ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage) => { + ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage, placement) => { let entries = queuedMirror.get(agent.id) - if (entries === undefined) queuedMirror.set(agent.id, entries = new Map()) - entries.set(message.id, message) - broadcast({ type: 'session/queued', sessionId: agent.id, content: message.content, source: message.source }) + if (entries === undefined) { + entries = new Map() + queuedMirror.set(agent.id, entries) + } + const steering = placement === 'steering' + entries.set(message.id, { message, steering }) + broadcast({ + type: 'session/queued', + sessionId: agent.id, + content: message.content, + source: message.source, + steering, + }) }), ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => { retire(agent, message.id) @@ -929,7 +939,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro // queue view from these alone. for (const [sessionId, entries] of queuedMirror) { for (const entry of entries.values()) { - queue.push(frame({ type: 'session/queued', sessionId, content: entry.content, source: entry.source })) + queue.push(frame({ + type: 'session/queued', + sessionId, + content: entry.message.content, + source: entry.message.source, + steering: entry.steering, + })) } } // 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 4a0d88fa38..973db5a91e 100644 --- a/packages/host/apiproxy/src/api/events.schema.ts +++ b/packages/host/apiproxy/src/api/events.schema.ts @@ -36,7 +36,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [ z.object({ type: z.literal('question/requested'), sessionId: sessionIdSchema, questions: z.array(askUserQuestionItemSchema).min(1) }), z.object({ type: z.literal('question/resolved'), sessionId: sessionIdSchema, questionRpcId: rpcIdSchema, outcome: z.union([z.literal('answered'), z.literal('cancelled')]) }), // content/source reuse the wide passthroughs (both are merge-extensible in core). - z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }) }), + z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }), steering: z.boolean() }), z.object({ type: z.literal('stream/error'), error: rpcErrorSchema }), ]) as unknown as z.ZodType diff --git a/packages/host/apiproxy/src/api/events.ts b/packages/host/apiproxy/src/api/events.ts index debb1750c3..804f689f97 100644 --- a/packages/host/apiproxy/src/api/events.ts +++ b/packages/host/apiproxy/src/api/events.ts @@ -69,10 +69,12 @@ export type MuxFrame = * host replays the current queue snapshot for every attached session (same * refresh-recovery baseline as pending questions); queue clearing on cancel * has no dedicated frame — clients fold it from the status flip. - * source carries the prompt's rpcId when the message came over this wire - * (the client's provisional-echo reconciliation key). + * `steering` is the host's acceptance-time queue classification and remains + * authoritative in reconnect snapshots. `source` carries the prompt's rpcId + * when the message came over this wire (the client's provisional-echo + * reconciliation key). */ - | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource } + | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource; steering: boolean } | { type: 'stream/error'; error: RpcError } /** diff --git a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts index 0483f3837e..480f821b95 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -258,20 +258,20 @@ describe('session/queued frames', () => { const queued = inboxMessage('m-1', 'queued prompt') const steering = inboxMessage('m-2', 'queued prompt') - ctx.emit('agent/inbox/enqueue', agent, queued) - ctx.emit('agent/inbox/enqueue', agent, steering) + ctx.emit('agent/inbox/enqueue', agent, queued, 'queued') + ctx.emit('agent/inbox/enqueue', agent, steering, 'steering') const liveFrames = (await liveCollected).filter(f => f.type === 'session/queued') expect(liveFrames).toEqual([ - { type: 'session/queued', sessionId: agent.id, content: queued.content, source: { kind: 'user' } }, - { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' } }, + { type: 'session/queued', sessionId: agent.id, content: queued.content, source: { kind: 'user' }, steering: false }, + { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' }, steering: true }, ]) // A fresh mux connection replays the still-pending entries as its baseline. const replay = new AbortController() const replayFrames = await collect( api.events.mux({ rpcId: RpcId('t-mux-replay'), payload: {} }, replay.signal), 3, replay) - expect(replayFrames.filter(f => f.type === 'session/queued')).toHaveLength(2) + expect(replayFrames.filter(f => f.type === 'session/queued')).toEqual(liveFrames) }) it('retires mirror entries on their terminal dequeue', async () => { @@ -280,8 +280,8 @@ describe('session/queued frames', () => { const agent = stubAgent(ctx) const queued = inboxMessage('m-3', 'x') const steering = inboxMessage('m-4', 'x', 'r-1') - ctx.emit('agent/inbox/enqueue', agent, queued) - ctx.emit('agent/inbox/enqueue', agent, steering) + ctx.emit('agent/inbox/enqueue', agent, queued, 'queued') + ctx.emit('agent/inbox/enqueue', agent, steering, 'steering') ctx.emit('agent/inbox/dequeue', agent, queued) ctx.emit('agent/inbox/dequeue', agent, steering) @@ -297,8 +297,8 @@ describe('session/queued frames', () => { const agent = stubAgent(ctx) const doomed = inboxMessage('m-5', 'doomed') const survivor = inboxMessage('m-6', 'survivor') - ctx.emit('agent/inbox/enqueue', agent, doomed) - ctx.emit('agent/inbox/enqueue', agent, survivor) + ctx.emit('agent/inbox/enqueue', agent, doomed, 'queued') + ctx.emit('agent/inbox/enqueue', agent, survivor, 'queued') ctx.emit('agent/inbox/discard', agent, [doomed]) const abort = new AbortController() diff --git a/packages/host/apiproxy/tests/rpc-schemas.spec.ts b/packages/host/apiproxy/tests/rpc-schemas.spec.ts index 032be04e44..5002f8b857 100644 --- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts +++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts @@ -248,8 +248,8 @@ describe('events frame schemas', () => { { type: 'approval/resolved', sessionId: 's', approvalId: 'a', outcome: 'allowed-once' }, { 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/queued', sessionId: 's', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' } }, - { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } }, + { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' }, steering: false }, + { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' }, steering: true }, { type: 'stream/error', error: { code: 'internal', message: 'm', details: {} } }, ] for (const frame of frames) expect(muxFrameSchema.parse(frame)).toMatchObject({ type: frame.type }) @@ -270,7 +270,8 @@ describe('events frame schemas', () => { it('rejects a queued frame missing its members', () => { expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: 'x', source: { kind: 'user' } })).toThrow() - expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: {} })).toThrow() + expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: { kind: 'user' } })).toThrow() + expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: {}, steering: false })).toThrow() }) it('accepts every host frame branch', () => { diff --git a/packages/ui/tui/src/index.ts b/packages/ui/tui/src/index.ts index 35044b668f..ba31c7ca24 100644 --- a/packages/ui/tui/src/index.ts +++ b/packages/ui/tui/src/index.ts @@ -2907,10 +2907,8 @@ export function createTuiChat( agent.followup({ content, source: { kind: 'user' } }) return } - // Idle: the snapshot rides the prompt's own admission transaction - // (PromptDecision.additionalContexts), so a blocking hook discards the - // prompt and its attached context together instead of stranding the - // snapshot in history for the next unrelated prompt. + // Idle: the snapshot rides the prompt's admission transaction so a + // blocking hook discards both together. let cleanedUp = false const cleanup = (): void => { // Each trigger detaches both listeners, so a second call needs a diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index bf84a5ebbf..0929f3fa99 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -1384,7 +1384,7 @@ describe('pi-tui chat lifecycle and transcript', () => { // A steering queue for a different agent never touches this status line. const other = { ...result.agent, id: SessionId('other') } as unknown as Agent result.terminal.output = '' - result.ctx.emit('agent/inbox/enqueue', other, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'elsewhere' }], source: { kind: 'user' } }) + result.ctx.emit('agent/inbox/enqueue', other, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'elsewhere' }], source: { kind: 'user' } }, 'queued') await tick() expect(result.terminal.output).not.toContain('queued') @@ -1474,7 +1474,7 @@ describe('pi-tui chat lifecycle and transcript', () => { // A live event before the turn runs has no status controller to move. const idle = await setup() // Inbox notifications do not affect the status phase while idle. - idle.ctx.emit('agent/inbox/enqueue', idle.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'early' }], source: { kind: 'user' } }) + idle.ctx.emit('agent/inbox/enqueue', idle.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'early' }], source: { kind: 'user' } }, 'queued') idle.session.append('tool/call', { turn: 1, step: 0, callId: 'pre' as never, name: 'bash', arguments: '{}' }) await tick() expect(idle.terminal.output).not.toContain('Executing tools')