From 1aefcbf4d744995d6552ff53a71eae35cae5fd56 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Mon, 27 Jul 2026 10:22:28 +0800 Subject: [PATCH] fix(apiproxy): derive queued steering on the client, drop it from the wire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The session/queued frame no longer carries steering — AgentMessage no longer has the field. The client derives it from the same ordered turn boundaries the host saw (a frame arriving while a turn is open joined the steering FIFO). --- .../runtime/src/client/sessions/session.ts | 13 ++++++++++- .../client/runtime/tests/queue-store.spec.ts | 23 ++++++++++++------- .../tests/input-machine.spec.ts | 2 +- .../ui-subagent/tests/browser-plugin.spec.ts | 2 +- packages/host/apiproxy/src/api-proxy.ts | 4 ++-- .../host/apiproxy/src/api/events.schema.ts | 2 +- packages/host/apiproxy/src/api/events.ts | 2 +- .../apiproxy/tests/api-proxy-commands.spec.ts | 21 ++++++++--------- .../host/apiproxy/tests/rpc-schemas.spec.ts | 9 ++++---- 9 files changed, 46 insertions(+), 32 deletions(-) diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index 75c55bc4bd..dc56ee584a 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -105,6 +105,11 @@ 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 → @@ -334,6 +339,12 @@ 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 } @@ -343,7 +354,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: frame.steering, + steering: this.turnOpen, 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 360f7c1a9d..8481243cb7 100644 --- a/packages/client/runtime/tests/queue-store.spec.ts +++ b/packages/client/runtime/tests/queue-store.spec.ts @@ -17,10 +17,10 @@ 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, steering = false): MuxFrame { +function queuedFrame(body: string, rpcId: string): MuxFrame { return { type: 'session/queued', sessionId: SID, content: text(body), - source: { kind: 'user', rpcId: rid(rpcId) } as never, steering, + source: { kind: 'user', rpcId: rid(rpcId) } as never, } } @@ -41,7 +41,7 @@ describe('queue intake', () => { session.handleMuxEnvelope(rid('env-2'), { type: 'session/queued', sessionId: SID, content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never], - source: { kind: 'plugin', plugin: 'loop' }, steering: false, + source: { kind: 'plugin', plugin: 'loop' }, }) expect(session.getSnapshot().queue).toEqual([{ key: 'f:env-2', preview: 'hi [image]' }]) }) @@ -85,22 +85,29 @@ 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')) - session.handleMuxEnvelope(rid('e2'), queuedFrame('插话', 'p-2', true)) + 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 // Loop-authored steering (different source) must not consume the user entry. const foreignSteering = { seq: 0, time: 1, type: 'steering/message', surfaceOp: 'append', data: { turn: 0, content: text('loop'), source: { kind: 'plugin', plugin: 'loop' } }, } as never - session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: foreignSteering }) + session.handleMuxEnvelope(rid('e4'), { type: 'session/event', sessionId: SID, event: foreignSteering }) expect(session.getSnapshot().queue).toHaveLength(2) const matchedSteering = { seq: 1, time: 2, type: 'steering/message', surfaceOp: 'append', data: { turn: 0, content: text('插话'), source: { kind: 'user', rpcId: rid('p-2') } }, } as never - session.handleMuxEnvelope(rid('e4'), { type: 'session/event', sessionId: SID, event: matchedSteering }) + session.handleMuxEnvelope(rid('e5'), { type: 'session/event', sessionId: SID, event: matchedSteering }) expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-1']) }) @@ -108,7 +115,7 @@ describe('queue retirement (host queuedMirror rules)', () => { const session = makeSession() session.handleRunning(true) session.handleMuxEnvelope(rid('e1'), queuedFrame('一', 'p-1')) - session.handleMuxEnvelope(rid('e2'), queuedFrame('二', 'p-2', true)) + session.handleMuxEnvelope(rid('e2'), queuedFrame('二', 'p-2')) session.handleRunning(false) expect(session.getSnapshot().queue).toEqual([]) }) diff --git a/packages/client/ui-conversation/tests/input-machine.spec.ts b/packages/client/ui-conversation/tests/input-machine.spec.ts index a470044481..206a66e4c6 100644 --- a/packages/client/ui-conversation/tests/input-machine.spec.ts +++ b/packages/client/ui-conversation/tests/input-machine.spec.ts @@ -36,7 +36,7 @@ function effectAt( ): Extract { const e = effects[index] expect(e?.type).toBe(type) - return e + return e as Extract } /** Drive plain → adjudicating and hand back the minted attempt. */ diff --git a/packages/client/ui-subagent/tests/browser-plugin.spec.ts b/packages/client/ui-subagent/tests/browser-plugin.spec.ts index 828ef38837..fc74470406 100644 --- a/packages/client/ui-subagent/tests/browser-plugin.spec.ts +++ b/packages/client/ui-subagent/tests/browser-plugin.spec.ts @@ -22,7 +22,7 @@ function summary(partial: Partial & { id: SessionId }): SessionS running: false, updatedAt: 0, ...partial, - } + } as SessionSummary } const sid = (id: string) => id as SessionId diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index ca422f945f..2b0fcea120 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -384,7 +384,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro 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, steering: message.steering }) + broadcast({ type: 'session/queued', sessionId: agent.id, content: message.content, source: message.source }) }), ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => { retire(agent, message.id) @@ -907,7 +907,7 @@ 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, steering: entry.steering })) + queue.push(frame({ type: 'session/queued', sessionId, content: entry.content, source: entry.source })) } } // 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 e95b371c54..4798cc4e86 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() }), steering: z.boolean() }), + z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }) }), 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 db572215cb..5f70558e84 100644 --- a/packages/host/apiproxy/src/api/events.ts +++ b/packages/host/apiproxy/src/api/events.ts @@ -72,7 +72,7 @@ export type MuxFrame = * 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; steering: boolean } + | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource } | { 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 9861d19c27..0483f3837e 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -238,14 +238,11 @@ describe('host/commands-changed frame', () => { }) /** Build one frozen inbox message for the live `agent/inbox/*` events. */ -function inboxMessage(id: string, text: string, steering: boolean, rpcId?: string): AgentMessage { +function inboxMessage(id: string, text: string, rpcId?: string): AgentMessage { return Object.freeze({ id: AgentMessageId(id), content: [{ type: 'text' as const, text }], source: rpcId === undefined ? { kind: 'user' as const } : { kind: 'user' as const, rpcId: RpcId(rpcId) }, - contexts: [], - steering, - wakeup: true, }) } @@ -259,15 +256,15 @@ describe('session/queued frames', () => { // subscribed baseline + 2 queued frames const liveCollected = collect(liveStream, 3, live) - const queued = inboxMessage('m-1', 'queued prompt', false) - const steering = inboxMessage('m-2', 'queued prompt', true) + 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) 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' }, steering: false }, - { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' }, steering: true }, + { type: 'session/queued', sessionId: agent.id, content: queued.content, source: { kind: 'user' } }, + { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' } }, ]) // A fresh mux connection replays the still-pending entries as its baseline. @@ -281,8 +278,8 @@ describe('session/queued frames', () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS) const agent = stubAgent(ctx) - const queued = inboxMessage('m-3', 'x', false) - const steering = inboxMessage('m-4', 'x', true, 'r-1') + 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/dequeue', agent, queued) @@ -298,8 +295,8 @@ describe('session/queued frames', () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS) const agent = stubAgent(ctx) - const doomed = inboxMessage('m-5', 'doomed', false) - const survivor = inboxMessage('m-6', 'survivor', false) + 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/discard', agent, [doomed]) diff --git a/packages/host/apiproxy/tests/rpc-schemas.spec.ts b/packages/host/apiproxy/tests/rpc-schemas.spec.ts index 02ca8dec22..322b005e20 100644 --- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts +++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts @@ -240,8 +240,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' }, steering: false }, - { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' }, steering: true }, + { 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: 'stream/error', error: { code: 'internal', message: 'm', details: {} } }, ] for (const frame of frames) expect(muxFrameSchema.parse(frame)).toMatchObject({ type: frame.type }) @@ -261,9 +261,8 @@ describe('events frame schemas', () => { }) it('rejects a queued frame missing its members', () => { - expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [{ type: 'text' }], source: { kind: 'user' } })).toThrow() - expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: 'x', source: { kind: 'user' }, steering: false })).toThrow() - expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: {}, steering: false })).toThrow() + 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() }) it('accepts every host frame branch', () => {