fix(apiproxy): derive queued steering on the client, drop it from the wire
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).
This commit is contained in:
@@ -105,6 +105,11 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
||||
private dispatchesRev = 0
|
||||
private dispatchesCache: { rev: number; value: ReadonlyMap<string, readonly CodeSubCall[]> } | 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<ConversationSnapshot> {
|
||||
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<ConversationSnapshot> {
|
||||
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++
|
||||
|
||||
@@ -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([])
|
||||
})
|
||||
|
||||
@@ -36,7 +36,7 @@ function effectAt<T extends InputEffect['type']>(
|
||||
): Extract<InputEffect, { type: T }> {
|
||||
const e = effects[index]
|
||||
expect(e?.type).toBe(type)
|
||||
return e
|
||||
return e as Extract<InputEffect, { type: T }>
|
||||
}
|
||||
|
||||
/** Drive plain → adjudicating and hand back the minted attempt. */
|
||||
|
||||
@@ -22,7 +22,7 @@ function summary(partial: Partial<SessionSummary> & { id: SessionId }): SessionS
|
||||
running: false,
|
||||
updatedAt: 0,
|
||||
...partial,
|
||||
}
|
||||
} as SessionSummary
|
||||
}
|
||||
|
||||
const sid = (id: string) => id as SessionId
|
||||
|
||||
Reference in New Issue
Block a user