Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web): surface and configure composer steering

Browse Source
This commit is contained in:
imccyu
2026-08-02 15:36:17 +08:00
parent 5b17d3bab5
commit dffe955ed2
48 changed files with 1006 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -818,29 +818,28 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
sessionId,
items: items.map(item => ({
id: item.id,
placement: item.placement,
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
}
const retireKnown = (sessionId: SessionId, itemId: InboxItemId): boolean => {
const entries = queuedMirror.get(sessionId)
if (entries === undefined) return false
const index = entries.findIndex(entry => entry.id === itemId)
if (index === -1) return false
entries.splice(index, 1)
if (entries.length === 0) queuedMirror.delete(agent.id)
if (entries.length === 0) queuedMirror.delete(sessionId)
return true
}
const retire = (agent: Agent, item: InboxItem): boolean => {
if (retireKnown(agent.id, item.id)) return true
rememberUnseen(agent.id, item.id, { kind: 'terminal' })
return false
}
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)
@@ -866,7 +865,24 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
publishQueue(agent.id)
}),
ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => {
if (retire(agent, item)) publishQueue(agent.id)
if (item.placement === 'steering') {
// AgentLoop appends the durable steering/message synchronously after
// this claim. Retain and retire the mirror row in the following
// microtask so any re-entrant snapshot and the Host's linear mux
// stream keep it visible until the durable event exists.
const present = queuedMirror.get(agent.id)?.some(entry => entry.id === item.id) === true
if (!present) {
retire(agent, item)
return
}
queueMicrotask(() => {
if (retireKnown(agent.id, item.id)) publishQueue(agent.id)
})
} else if (retire(agent, item)) {
// Queued claims have no durable same-message handoff to order.
// Publish retirement synchronously as before.
publishQueue(agent.id)
}
}),
ctx.on('agent/inbox/discard', (agent: Agent, items: InboxItem[]) => {
let changed = false
@@ -2509,6 +2525,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
sessionId,
items: items.map(item => ({
id: item.id,
placement: item.placement,
message: item.message,
})),
}))

1
packages/host/apiproxy/src/api/events.schema.ts
View File

@@ -54,6 +54,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
sessionId: sessionIdSchema,
items: z.array(z.object({
id: inboxItemIdSchema,
placement: z.union([z.literal('queued'), z.literal('steering')]),
message: messageSchema,
})),
}),

13
packages/host/apiproxy/src/api/events.ts
View File

@@ -32,10 +32,12 @@ export type ToolEventView =
| { for: 'call'; view: ToolCallView }
| { for: 'result'; view: ToolResultView }
/** One pending queued occurrence in an authoritative queue snapshot. */
/** One pending inbox occurrence in the authoritative `session/queue` snapshot. */
export interface QueuedInboxItem {
/** Agent-owned occurrence identity used by queue mutations. */
/** Agent-owned occurrence identity; queue mutations address only `queued` items. */
id: InboxItemId
/** Agent-resolved FIFO placement; clients render queued and steering items on different surfaces. */
placement: 'queued' | 'steering'
/** Complete pending message; it is not durable until the Agent claims it. */
message: Message
}
@@ -71,11 +73,12 @@ 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
* Complete transient inbox 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.
* reconnect converge through one authoritative signal. `session/queue`
* covers both resolved placements: queued items render
* in QueueDock, while pending steering renders at the conversation tail.
*/
| { type: 'session/queue'; sessionId: SessionId; items: QueuedInboxItem[] }
/**

113
packages/host/apiproxy/tests/api-proxy-commands.spec.ts
View File

@@ -379,7 +379,7 @@ describe('session/queue frames', () => {
const liveFrames = (await collected).filter(frame => frame.type === 'session/queue')
expect(liveFrames.map(frame => frame.items)).toEqual([
[{ id: edited.id, message: edited.message }],
[{ id: edited.id, placement: edited.placement, message: edited.message }],
])
const replay = new AbortController()
const replayFrames = await collect<MuxFrame>(
@@ -393,8 +393,8 @@ describe('session/queue frames', () => {
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<MuxFrame>(liveStream, 2, live)
// subscribed baseline + one snapshot per accepted inbox occurrence.
const liveCollected = collect<MuxFrame>(liveStream, 3, live)
const queued = inboxItem('i-1', inboxMessage('m-1', 'queued prompt'), 'queued')
const steering = inboxItem('i-2', inboxMessage('m-2', 'steering prompt'), 'steering')
@@ -406,7 +406,15 @@ describe('session/queue frames', () => {
{
type: 'session/queue',
sessionId: agent.id,
items: [{ id: queued.id, message: queued.message }],
items: [{ id: queued.id, placement: 'queued', message: queued.message }],
},
{
type: 'session/queue',
sessionId: agent.id,
items: [
{ id: queued.id, placement: 'queued', message: queued.message },
{ id: steering.id, placement: 'steering', message: steering.message },
],
},
])
@@ -414,7 +422,88 @@ describe('session/queue frames', () => {
const replay = new AbortController()
const replayFrames = await collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-mux-replay'), payload: {} }, replay.signal), 2, replay)
expect(replayFrames.filter(f => f.type === 'session/queue')).toEqual([liveFrames[0]])
expect(replayFrames.filter(f => f.type === 'session/queue')).toEqual([liveFrames[1]])
})
it('publishes the durable steering event before retiring its transient row', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const abort = new AbortController()
const collected = collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-steering-order'), payload: {} }, abort.signal), 5, abort)
const steering = inboxItem('i-steering', inboxMessage('m-steering', 'interrupt now'), 'steering')
agent.session.append('turn/start', {
turn: 1,
trigger: { kind: 'message', source: { kind: 'user' } },
})
ctx.emit('agent/inbox/enqueue', agent, steering)
ctx.emit('agent/inbox/dequeue', agent, steering)
agent.session.append('steering/message', {
turn: 1,
message: steering.message,
}, { surfaceOp: 'append' })
const frames = await collected
expect(frames.map(frame => frame.type)).toEqual([
'session/subscribed',
'session/event',
'session/queue',
'session/event',
'session/queue',
])
expect(frames[2]).toMatchObject({
type: 'session/queue',
items: [{ id: steering.id, placement: 'steering' }],
})
expect(frames[3]).toMatchObject({
type: 'session/event',
event: { type: 'steering/message', data: { message: { id: steering.message.id } } },
})
expect(frames[4]).toMatchObject({ type: 'session/queue', items: [] })
})
it('retains claimed steering in re-entrant snapshots until its durable event', async () => {
const ctx = await harness()
const api = createApiProxy(ctx, DEFAULTS)
const agent = stubAgent(ctx)
const steering = inboxItem('i-steering', inboxMessage('m-steering', 'interrupt now'), 'steering')
const queued = inboxItem('i-reentrant', inboxMessage('m-reentrant', 'later'), 'queued')
ctx.on('agent/inbox/dequeue', (subject, item) => {
if (subject === agent && item.id === steering.id) ctx.emit('agent/inbox/enqueue', agent, queued)
})
const abort = new AbortController()
const collected = collect<MuxFrame>(
api.events.mux({ rpcId: RpcId('t-steering-reentrant-order'), payload: {} }, abort.signal), 6, abort)
agent.session.append('turn/start', {
turn: 1,
trigger: { kind: 'message', source: { kind: 'user' } },
})
ctx.emit('agent/inbox/enqueue', agent, steering)
ctx.emit('agent/inbox/dequeue', agent, steering)
agent.session.append('steering/message', {
turn: 1,
message: steering.message,
}, { surfaceOp: 'append' })
const frames = await collected
expect(frames[3]).toMatchObject({
type: 'session/queue',
items: [
{ id: steering.id, placement: 'steering' },
{ id: queued.id, placement: 'queued' },
],
})
expect(frames[4]).toMatchObject({
type: 'session/event',
event: { type: 'steering/message', data: { message: { id: steering.message.id } } },
})
expect(frames[5]).toMatchObject({
type: 'session/queue',
items: [{ id: queued.id, placement: 'queued' }],
})
})
it('publishes edits in place in the authoritative order', async () => {
@@ -434,10 +523,16 @@ describe('session/queue frames', () => {
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 }],
[{ id: first.id, placement: first.placement, message: first.message }],
[
{ id: first.id, placement: first.placement, message: first.message },
{ id: second.id, placement: second.placement, message: second.message },
],
[
{ id: first.id, placement: first.placement, message: first.message },
{ id: edited.id, placement: edited.placement, message: edited.message },
],
[{ id: first.id, placement: first.placement, message: first.message }],
])
})

2
packages/host/apiproxy/tests/rpc-schemas.spec.ts
View File

@@ -481,7 +481,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: 'i1', placement: 'steering', message: { 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: {} } },