feat(web): steer queued messages into active turns

2026-07-30 03:17:54 +08:00
parent fff0172208
commit 955a12cca4
49 changed files with 508 additions and 147 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/host/apiproxy/README.md
-README.md: 933b5f6167263545b3bef5ca9fb8f8b945e86ef9
-README.zh.md: f1f1106dbd0c50889eca6f6ae52fbb29d1c4c03f
+README.md: 87c477362dc159621a38a1ff923bebf07cbd3c6f
+README.zh.md: 7316f3a6d93d30e101537c40608037fedfa77321
@@ -22,7 +22,7 @@ Session model routing is a session-domain contract. `session.models` returns the

 Generic Agent-bound session, command, and goal operations serve ordinary sessions only. They return `agent-busy` for a session-backed subagent instead of resuming or driving it; explicit-id `session.create` adoption and the attached-only queue controls enforce the same ownership boundary. Subagent conversation reads and continuation use the dedicated `subagent.*` domain, which retains catalog-mode and direct-parent authorization.

-Pending queued input is a live control-plane contract, not session history. The gateway mirrors queued `InboxItem` occurrences from `agent/inbox/*` and broadcasts authoritative `session/queue` snapshots on every queued change and reconnect; pending steering stays outside this Web projection. `session.updateQueue` addresses one `InboxItemId`: edit replaces pending content and remove discards it. `session.cancel` aborts only the active turn and preserves pending inbox work; after cancellation reaches quiescence and the closing turn flushes, AgentLoop claims the next waking occurrence in FIFO order. The browser never resends or promotes that occurrence. A driver claim wins races by retiring the address before admission; a later operation returns `queue-item-not-found`. Queue operations query only an attached Agent and never resume a cold session because process-local inbox identities do not survive restart or disposal. The client never infers retirement from turn or status events.
+Pending queued input is a live control-plane contract, not session history. The gateway mirrors queued `InboxItem` occurrences from `agent/inbox/*` and broadcasts authoritative `session/queue` snapshots on every queued change and reconnect; pending steering stays outside this Web projection. `session.updateQueue` addresses one `InboxItemId`: edit replaces pending content, remove discards it, and strict steer transfers its complete message into the current next-step window. A closed window returns `steer-unavailable` without changing the row. `session.cancel` aborts only the active turn and preserves pending inbox work; after cancellation reaches quiescence and the closing turn flushes, AgentLoop claims the next waking occurrence in FIFO order. The browser never resends or promotes that occurrence. A driver claim wins races by retiring the address before admission; a later operation returns `queue-item-not-found`. Queue operations query only an attached ordinary-session Agent and never resume a cold session because process-local inbox identities do not survive restart or disposal. The client never infers retirement from turn or status events.

 Workspace and Session lists are separate reconnect baselines. `workspace.create({ name })` creates a uniquely titled directory under the configured root, while `workspace.create({ path })` adopts an existing canonical directory and permits basename-derived titles to repeat. `workspace.delete` removes only the Workspace registration, `session.create` accepts an optional preallocated Session id, and `host/workspace-changed`, `host/workspace-removed`, plus `host/session-added` carry committed increments in either arrival order. `workspace.archiveSession` adds one session to the registry-global archive set and answers the full updated set; `workspace.list` carries that set as the reconnect baseline and `host/archived-sessions-changed` pushes the full snapshot after every durable change. Archiving hides the session from grouping surfaces without touching its log or its workspace account; a session neither live nor persisted fails with `session-not-found`. Registration deletion preserves the directory and session logs; its Sessions remain in `session.list` and become Ungrouped. `SessionSummary.blank` and the `host/session-added` frame carry the derived zero-events bit: clients hide blank sessions and reuse them per workspace, flip blank on the first `host/session-status(running:true)`, and treat `session.list` as the reconnect authority; cold summaries are never blank because lazy persistence keeps never-appended sessions out of `list()`. The `session.list` summaries and `host/session-added` frames also carry the optional durable `origin: 'subagent'` classification so navigation can suppress duplicate child rows immediately and after reconnect; that bit is never continuation authority.

@@ -22,7 +22,7 @@

 绑定到 Agent 的通用会话、命令与目标操作只服务普通会话。对于由会话支撑的 subagent，它们会返回 `agent-busy`，而不是恢复或驱动它；显式 id 的 `session.create` 接纳与仅针对已附加会话的队列控件也会执行同一所有权边界。subagent 对话读取与继续执行使用专用的 `subagent.*` 领域，该领域保留目录 mode 与直接 parent 授权。

-待处理的 queued 输入属于实时控制平面契约，而非会话历史。网关镜像来自 `agent/inbox/*` 的 queued `InboxItem` 入队项，并在每次 queued 变更和重连时广播权威的 `session/queue` 快照；待处理 steering（中途引导）不进入此 Web 投影。`session.updateQueue` 通过 `InboxItemId` 寻址单个项：编辑会替换待处理内容，移除会将其丢弃。`session.cancel` 仅中止活动轮次，并保留待处理 inbox 工作；取消达到完全停稳且结束中的轮次完成 flush 后，AgentLoop 按 FIFO 顺序认领下一个可唤醒入队项。浏览器绝不重发或提升该入队项。驱动器在接纳前退役寻址标识，因此认领会赢得竞态；之后的操作返回 `queue-item-not-found`。队列操作只查询当前已挂载的 Agent，绝不恢复冷会话，因为进程本地 inbox 标识无法在重启或资源释放后存活。客户端绝不根据轮次或状态事件推断项已退役。
+待处理的 queued 输入属于实时控制平面契约，而非会话历史。网关镜像来自 `agent/inbox/*` 的 queued `InboxItem` 入队项，并在每次 queued 变更和重连时广播权威的 `session/queue` 快照；待处理 steering（中途引导）不进入此 Web 投影。`session.updateQueue` 通过 `InboxItemId` 寻址单个项：编辑会替换待处理内容，移除会将其丢弃，严格 steering 会把其完整消息转移到当前 next-step 窗口。窗口关闭时返回 `steer-unavailable`，且不改变该行。`session.cancel` 仅中止活动轮次，并保留待处理 inbox 工作；取消达到完全停稳且结束中的轮次完成 flush 后，AgentLoop 按 FIFO 顺序认领下一个可唤醒入队项。浏览器绝不重发或提升该入队项。驱动器在接纳前退役寻址标识，因此认领会赢得竞态；之后的操作返回 `queue-item-not-found`。队列操作只查询当前已挂载的普通会话 Agent，绝不恢复冷会话，因为进程本地 inbox 标识无法在重启或资源释放后存活。客户端绝不根据轮次或状态事件推断项已退役。

 Workspace 列表与 Session 列表是相互独立的重连基线。`workspace.create({ name })` 会在配置根目录下创建显示标题唯一的目录，而 `workspace.create({ path })` 会接纳已有的规范目录，并允许由 basename 派生的标题重复。`workspace.delete` 只移除 Workspace 注册记录，`session.create` 接受可选的预分配 Session id，`host/workspace-changed`、`host/workspace-removed` 与 `host/session-added` 则以任意到达顺序携带已提交的增量。`workspace.archiveSession` 向注册表级全局归档集合添加一个会话，并应答完整的更新后集合；`workspace.list` 携带该集合作为重连基线，`host/archived-sessions-changed` 在每次持久变更后推送完整快照。归档只把会话从各分组视图中隐藏，不触碰其日志和 workspace 记账；既非实时也未持久化的会话以 `session-not-found` 失败。删除注册记录会保留目录和会话日志；相关 Session 仍留在 `session.list` 中，并进入 Ungrouped。`SessionSummary.blank` 与 `host/session-added` 帧携带派生的零事件位：客户端隐藏空白会话并按 workspace 复用它们，在首个 `host/session-status(running:true)` 时翻转 blank，并以 `session.list` 作为重连权威；冷会话摘要永远不是空白：惰性持久化让从未追加过事件的会话根本不出现在 `list()` 中。`session.list` 摘要与 `host/session-added` 帧还会携带可选的持久化分类 `origin: 'subagent'`，使导航在实时创建与重连后都能隐藏重复的 child 行；该标记绝不是继续执行的权威依据。

@@ -1847,13 +1847,28 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
        if (agent !== undefined && hasSubagentOwner(agent.session, agent)) {
          return Promise.resolve(err(request, subagentOwnershipError(sessionId)))
        }
-        if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') {
+        if (agent === undefined) {
          return Promise.resolve(err(request, {
            code: 'queue-item-not-found',
            message: 'queued item is no longer pending',
            details: { itemId },
          }))
        }
+        const result = agent.updateInbox(itemId, action)
+        if (result === 'not-found') {
+          return Promise.resolve(err(request, {
+            code: 'queue-item-not-found',
+            message: 'queued item is no longer pending',
+            details: { itemId },
+          }))
+        }
+        if (result === 'steer-unavailable') {
+          return Promise.resolve(err(request, {
+            code: 'steer-unavailable',
+            message: 'current turn no longer accepts steering',
+            details: { itemId },
+          }))
+        }
        return Promise.resolve(ok(request, { accepted: true as const }))
      },

@@ -48,6 +48,7 @@ export const rpcErrorSchema: z.ZodType<RpcError> = z.discriminatedUnion('code',
  z.object({ code: z.literal('directory-picker-unavailable'), message: z.string(), details: z.object({ capability: z.string() }) }),
  z.object({ code: z.literal('agent-busy'), message: z.string(), details: z.object({ reason: z.string() }) }),
  z.object({ code: z.literal('queue-item-not-found'), message: z.string(), details: z.object({ itemId: z.string() }) }),
+  z.object({ code: z.literal('steer-unavailable'), message: z.string(), details: z.object({ itemId: z.string() }) }),
  z.object({ code: z.literal('command-error'), message: z.string(), details: z.object({}) }),
  z.object({ code: z.literal('unknown-command'), message: z.string(), details: z.object({}) }),
  z.object({ code: z.literal('settings-rejected'), message: z.string(), details: z.object({ ns: z.string() }) }),
@@ -46,6 +46,7 @@ export interface RpcErrorDetailsMap {
  'directory-picker-unavailable': { capability: string }
  'agent-busy': { reason: string }
  'queue-item-not-found': { itemId: InboxItemId }
+  'steer-unavailable': { itemId: InboxItemId }
  /** A known slash command reported a usage/state error; the message is the command's own text. */
  'command-error': {}
  /** A leading-/ prompt named no registered command; the message names the token. */
@@ -269,6 +269,7 @@ 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('steer') }),
  ]),
 }) as unknown as z.ZodType<RequestPayload<'session.updateQueue'>>

@@ -129,6 +129,7 @@ export interface SessionModels {
 export type QueueAction =
  | { kind: 'edit'; content: ContentBlock[] }
  | { kind: 'remove' }
+  | { kind: 'steer' }

 /** Session list entry (v1 builds no index: list does readdir+stat). */
 export interface SessionSummary {
@@ -285,7 +286,7 @@ export interface SessionsApi {
  Promise<RpcResponse<{ accepted: true; command?: { kind: 'success'; text?: string } }>>

  /**
-   * Edits or removes one pending queued occurrence on an ordinary session.
+   * Edits, removes, or strictly steers one pending queued occurrence on an ordinary session.
   * Session-backed subagents reject with `agent-busy`.
   */
  updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>):
@@ -291,13 +291,14 @@ function inboxItem(id: string, message: UserMessage, placement: InboxPlacement):
 }

 describe('session.updateQueue', () => {
-  it('routes an addressable action and reports a lost claim race', async () => {
+  it('routes addressable actions and reports strict steer races', async () => {
    const ctx = await harness()
    const agent = stubAgent(ctx)
    const seen: unknown[] = []
    agent.updateInbox = (id, action) => {
      seen.push({ id, action })
-      return id === InboxItemId('present') ? 'applied' : 'not-found'
+      if (id === InboxItemId('present')) return 'applied'
+      return id === InboxItemId('closed') ? 'steer-unavailable' : 'not-found'
    }
    const api = createApiProxy(ctx, DEFAULTS)

@@ -319,9 +320,22 @@ describe('session.updateQueue', () => {
      },
    })
    expect(expectErr(missing)).toMatchObject({ code: 'queue-item-not-found' })
+    const closed = await api.sessions.updateQueue({
+      rpcId: RpcId('q-closed'),
+      payload: {
+        sessionId: agent.id,
+        itemId: InboxItemId('closed'),
+        action: { kind: 'steer' },
+      },
+    })
+    expect(expectErr(closed)).toMatchObject({
+      code: 'steer-unavailable',
+      details: { itemId: 'closed' },
+    })
    expect(seen).toEqual([
      { id: 'present', action: { kind: 'edit', content: [{ type: 'text', text: 'edited' }] } },
      { id: 'claimed', action: { kind: 'remove' } },
+      { id: 'closed', action: { kind: 'steer' } },
    ])
  })

@@ -77,6 +77,7 @@ describe('rpcErrorSchema', () => {
    }).code).toBe('model-unavailable')
    expect(rpcErrorSchema.parse({ code: 'agent-busy', message: 'm', details: { reason: 'r' } }).code).toBe('agent-busy')
    expect(rpcErrorSchema.parse({ code: 'queue-item-not-found', message: 'm', details: { itemId: 'i' } }).code).toBe('queue-item-not-found')
+    expect(rpcErrorSchema.parse({ code: 'steer-unavailable', message: 'm', details: { itemId: 'i' } }).code).toBe('steer-unavailable')
    expect(rpcErrorSchema.parse({ code: 'command-error', message: 'm', details: {} }).code).toBe('command-error')
    expect(rpcErrorSchema.parse({ code: 'unknown-command', message: 'm', details: {} }).code).toBe('unknown-command')
    expect(rpcErrorSchema.parse({ code: 'title-invalid', message: 'm', details: { sessionId: 's' } }).code).toBe('title-invalid')
@@ -280,6 +281,9 @@ describe('sessions domain schemas', () => {
    expect(sessionUpdateQueueRequestSchema.parse({
      sessionId: 's1', itemId: 'i1', action: { kind: 'remove' },
    }).action.kind).toBe('remove')
+    expect(sessionUpdateQueueRequestSchema.parse({
+      sessionId: 's1', itemId: 'i1', action: { kind: 'steer' },
+    }).action.kind).toBe('steer')
    expect(() => sessionUpdateQueueRequestSchema.parse({
      sessionId: 's1', itemId: 'i1', action: { kind: 'promote' },
    })).toThrow()