refactor(agent): scope queue actions to edit and remove
This commit is contained in:
@@ -508,9 +508,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
})
|
||||
|
||||
/**
|
||||
* Per-session inbox occurrence mirror serving the mux-open queue snapshot
|
||||
* Per-session queued-occurrence mirror serving the mux-open queue snapshot
|
||||
* (the same refresh-recovery baseline as pending questions). Each terminal
|
||||
* inbox event retires one matching occurrence, so repeated sends of the same
|
||||
* 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<SessionId, InboxItem[]>()
|
||||
@@ -522,7 +522,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
items: items.map(item => ({
|
||||
id: item.id,
|
||||
message: item.message,
|
||||
placement: item.placement,
|
||||
})),
|
||||
})
|
||||
}
|
||||
@@ -531,12 +530,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
const entries = queuedMirror.get(agent.id)
|
||||
if (entries === undefined) return
|
||||
const index = entries.findIndex(entry => entry.id === item.id)
|
||||
if (index !== -1) entries.splice(index, 1)
|
||||
if (index === -1) return
|
||||
entries.splice(index, 1)
|
||||
if (entries.length === 0) queuedMirror.delete(agent.id)
|
||||
publishQueue(agent.id)
|
||||
}
|
||||
const disposers = [
|
||||
ctx.on('agent/inbox/enqueue', (agent: Agent, item: InboxItem) => {
|
||||
if (item.placement !== 'queued') return
|
||||
let entries = queuedMirror.get(agent.id)
|
||||
if (entries === undefined) {
|
||||
entries = []
|
||||
@@ -545,18 +546,12 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
entries.push(item)
|
||||
publishQueue(agent.id)
|
||||
}),
|
||||
ctx.on('agent/inbox/update', (agent: Agent, item: InboxItem, action) => {
|
||||
ctx.on('agent/inbox/update', (agent: Agent, item: InboxItem) => {
|
||||
const entries = queuedMirror.get(agent.id)
|
||||
if (entries === undefined) return
|
||||
const index = entries.findIndex(entry => entry.id === item.id)
|
||||
if (index === -1) return
|
||||
entries.splice(index, 1)
|
||||
if (action === 'promote') {
|
||||
const first = entries.findIndex(entry => entry.placement === item.placement)
|
||||
entries.splice(first === -1 ? entries.length : first, 0, item)
|
||||
} else {
|
||||
entries.splice(index, 0, item)
|
||||
}
|
||||
entries.splice(index, 1, item)
|
||||
publishQueue(agent.id)
|
||||
}),
|
||||
ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => {
|
||||
@@ -567,6 +562,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
if (entries === undefined) return
|
||||
const ids = new Set(items.map(item => item.id))
|
||||
const kept = entries.filter(entry => !ids.has(entry.id))
|
||||
if (kept.length === entries.length) return
|
||||
if (kept.length === 0) queuedMirror.delete(agent.id)
|
||||
else queuedMirror.set(agent.id, kept)
|
||||
publishQueue(agent.id)
|
||||
@@ -1540,7 +1536,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
||||
items: items.map(item => ({
|
||||
id: item.id,
|
||||
message: item.message,
|
||||
placement: item.placement,
|
||||
})),
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -50,7 +50,6 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
|
||||
items: z.array(z.object({
|
||||
id: inboxItemIdSchema,
|
||||
message: messageSchema,
|
||||
placement: z.union([z.literal('queued'), z.literal('steering')]),
|
||||
})),
|
||||
}),
|
||||
// value stays wide: it already passed its unit's own schema on the host,
|
||||
|
||||
@@ -32,14 +32,12 @@ export type ToolEventView =
|
||||
| { for: 'call'; view: ToolCallView }
|
||||
| { for: 'result'; view: ToolResultView }
|
||||
|
||||
/** One pending inbox occurrence in an authoritative queue snapshot. */
|
||||
/** 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
|
||||
/** Acceptance-time FIFO classification. */
|
||||
placement: 'queued' | 'steering'
|
||||
}
|
||||
|
||||
/** Streaming face of the contract: the two SSE stream openers (mux + host). */
|
||||
@@ -73,10 +71,11 @@ export type MuxFrame =
|
||||
| { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] }
|
||||
| { type: 'question/resolved'; sessionId: SessionId; questionRpcId: RpcId; outcome: 'answered' | 'cancelled' }
|
||||
/**
|
||||
* Complete transient inbox state after every enqueue, mutation, claim, or
|
||||
* 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, reorder, deletion, cancel,
|
||||
* and reconnect converge through one authoritative signal.
|
||||
* session event; the whole snapshot makes edit, deletion, cancel, and
|
||||
* reconnect converge through one authoritative signal. Pending steering is
|
||||
* outside this Web queue projection.
|
||||
*/
|
||||
| { type: 'session/queue'; sessionId: SessionId; items: QueuedInboxItem[] }
|
||||
/**
|
||||
|
||||
@@ -225,7 +225,6 @@ export const sessionUpdateQueueRequestSchema = z.object({
|
||||
action: z.discriminatedUnion('kind', [
|
||||
z.object({ kind: z.literal('edit'), content: z.array(contentBlockSchema) }),
|
||||
z.object({ kind: z.literal('remove') }),
|
||||
z.object({ kind: z.literal('promote') }),
|
||||
]),
|
||||
}) as unknown as z.ZodType<RequestPayload<'session.updateQueue'>>
|
||||
|
||||
|
||||
@@ -129,7 +129,6 @@ export interface SessionModels {
|
||||
export type QueueAction =
|
||||
| { kind: 'edit'; content: ContentBlock[] }
|
||||
| { kind: 'remove' }
|
||||
| { kind: 'promote' }
|
||||
|
||||
/** Session list entry (v1 builds no index: list does readdir+stat). */
|
||||
export interface SessionSummary {
|
||||
@@ -237,9 +236,7 @@ export interface SessionsApi {
|
||||
Promise<RpcResponse<{ accepted: true; command?: { kind: 'success'; text?: string } }>>
|
||||
|
||||
/**
|
||||
* Edits, removes, or promotes one pending inbox occurrence. Promotion means
|
||||
* first in its current FIFO; a queued item is also made waking, so an idle
|
||||
* agent starts it and a running agent takes it as the next independent turn.
|
||||
* Edits or removes one pending queued occurrence.
|
||||
*/
|
||||
updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>):
|
||||
Promise<RpcResponse<{ accepted: true }>>
|
||||
|
||||
Reference in New Issue
Block a user