feat(agent): add addressable queue operations

2026-07-30 00:05:00 +08:00
parent 87c9ab06b3
commit f893e2281d
108 changed files with 1785 additions and 608 deletions
--- a/packages/core/agent-loop/README.i18n.yaml
+++ b/packages/core/agent-loop/README.i18n.yaml
@@ -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/core/agent-loop/README.md
-README.md: 6bb8b12af69f54c2a75cd672e4d3802887808c76
-README.zh.md: f9eb8aa3cdead427a88492e35c00eab80ba12f91
+README.md: f38fc5232cab49cdd6311cb24de9b83e62bc19a6
+README.zh.md: 4f81bfd182e739ba7645e364a873cc461b2aefe5
--- a/packages/core/agent-loop/README.md
+++ b/packages/core/agent-loop/README.md
@@ -55,7 +55,9 @@ Configured agents start automatically. A model call requires both `provider` and

 The concrete `ReactLoopAgent`, its queued input, outbox, and run controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.

-The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue` with the same placement, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
+The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model.
+
+Every FIFO acceptance mints an `InboxItemId` and publishes `agent/inbox/enqueue` with the complete occurrence. `updateInbox()` owns the synchronous pending-item boundary: edit freezes replacement content without changing message identity or position, remove publishes discard, and promote moves the occurrence to the head of its queued or steering FIFO; promoting queued work also makes it waking. Edit and promote publish `agent/inbox/update`. Claim publishes `agent/inbox/dequeue` and irrevocably removes the live address before prompt admission, so a racing update returns `not-found`; `cancel()` without `keepInbox` publishes `agent/inbox/discard`.

 ### Loop lifecycle (`agent.ts`)

--- a/packages/core/agent-loop/README.zh.md
+++ b/packages/core/agent-loop/README.zh.md
@@ -55,7 +55,9 @@ interface Config {

 实体 `ReactLoopAgent`、其排队输入、outbox 与运行控制均为包内部实现。包根只导出插件／服务／配置契约，包导出映射不提供 `./src/*` 逃逸路径；生命周期拥有方通过 `ctx.agents` 创建 agent，而不是点名、构造或启动驱动器内部组件。一个准备完成的会话只能由一个实体驱动器认领；所有可观测行为都通过会话事件和 `agent/*` 事件分类体系发生。

-统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`，并携带相同的路由归类；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
+统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。
+
+每次 FIFO 接受项时都会铸造一个 `InboxItemId`，并通过 `agent/inbox/enqueue` 发布完整的单次入队项。`updateInbox()` 持有同步待处理项边界：编辑会冻结替换内容，但不改变消息标识或位置；移除会发布 discard；前移会把该项移至其 queued 或 steering FIFO 的队首，其中 queued 工作还会变为可唤醒。编辑和前移会发布 `agent/inbox/update`。认领操作会发布 `agent/inbox/dequeue`，并在提示词接纳前不可逆地移除实时寻址标识，因此竞态中的更新会返回 `not-found`；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。

 ### 循环生命周期（`agent.ts`）

--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -8,13 +8,18 @@
 */

 import type { Context } from 'cordis'
-import { agentCarrier, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
+import { randomUUID } from 'node:crypto'
+import { agentCarrier, assembleContextFor, emitAgentEvent, InboxItemId } from '@deepseek-ai/dsh-agent'
 import { createScope } from '@deepseek-ai/dsh-scope'
 import type { Scope } from '@deepseek-ai/dsh-scope'
 import type {
  Agent,
  CancelOptions,
  AgentInterruptReason,
+  InboxAction,
+  InboxActionResult,
+  InboxItem,
+  InboxItemId as InboxItemIdType,
  InboxPlacement,
  AgentOptions,
  AgentStatus,
@@ -55,9 +60,9 @@ type StepOutcome =
 */
 export class ReactLoopAgent implements Agent {
  /** Prompts awaiting individual turns. */
-  private queued: { message: UserMessage; wakeup: boolean }[] = []
+  private queued: { item: InboxItem; wakeup: boolean }[] = []
  /** Input taken into the session log at step boundaries. */
-  private outbox: { message: UserMessage; steering: boolean }[] = []
+  private outbox: { message: UserMessage; steering: boolean; item?: InboxItem }[] = []

  /** Whether observers see a running interval; consecutive turns share it. */
  private busy = false
@@ -115,16 +120,77 @@ export class ReactLoopAgent implements Agent {
    }

    const placement: InboxPlacement = target === 'next-step' && this.acceptsNextStep ? 'steering' : 'queued'
+    const item: InboxItem = Object.freeze({
+      id: InboxItemId(randomUUID()),
+      message,
+      placement,
+    })
    if (placement === 'steering') {
-      this.outbox.push({ message, steering: true })
+      this.outbox.push({ message, steering: true, item })
    } else {
-      this.queued.push({ message, wakeup })
+      this.queued.push({ item, wakeup })
    }
    // Preserve the routing decision for every send in this synchronous caller
    // stack, while installing quiescence ownership before enqueue observers
    // can cancel or dispose.
    if (placement === 'queued' && wakeup) this.scheduleKick()
-    emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message, placement)
+    emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', item)
+  }
+
+  /** Apply one synchronous mutation to a still-pending inbox occurrence. */
+  updateInbox(id: InboxItemIdType, action: InboxAction): InboxActionResult {
+    const queuedIndex = this.queued.findIndex(candidate => candidate.item.id === id)
+    const outboxIndex = queuedIndex === -1
+      ? this.outbox.findIndex(candidate => candidate.item?.id === id)
+      : -1
+    if (queuedIndex === -1 && outboxIndex === -1) return 'not-found'
+
+    const pending = queuedIndex === -1 ? this.outbox[outboxIndex] : this.queued[queuedIndex]
+    if (pending === undefined || pending.item === undefined) {
+      throw new Error(`agent "${this.id}" inbox index changed during synchronous update`)
+    }
+
+    switch (action.kind) {
+      case 'edit': {
+        const item: InboxItem = Object.freeze({
+          ...pending.item,
+          message: freezeMessage({ ...pending.item.message, content: action.content }),
+        })
+        if (queuedIndex !== -1) {
+          const queued = this.queued[queuedIndex]
+          if (queued === undefined) throw new Error(`agent "${this.id}" queued item disappeared during edit`)
+          this.queued[queuedIndex] = { ...queued, item }
+        } else {
+          const outbox = this.outbox[outboxIndex]
+          if (outbox === undefined) throw new Error(`agent "${this.id}" steering item disappeared during edit`)
+          this.outbox[outboxIndex] = { ...outbox, message: item.message, item }
+        }
+        emitAgentEvent(this.loopCtx, this, 'agent/inbox/update', item, 'edit')
+        return 'applied'
+      }
+      case 'remove': {
+        if (queuedIndex !== -1) this.queued.splice(queuedIndex, 1)
+        else this.outbox.splice(outboxIndex, 1)
+        emitAgentEvent(this.loopCtx, this, 'agent/inbox/discard', [pending.item])
+        return 'applied'
+      }
+      case 'promote': {
+        if (queuedIndex !== -1) {
+          const queued = this.queued.splice(queuedIndex, 1)[0]
+          if (queued === undefined) throw new Error(`agent "${this.id}" queued item disappeared during promotion`)
+          this.queued.unshift({ item: queued.item, wakeup: true })
+          this.scheduleKick()
+        } else {
+          const outbox = this.outbox.splice(outboxIndex, 1)[0]
+          if (outbox === undefined) throw new Error(`agent "${this.id}" steering item disappeared during promotion`)
+          this.outbox.unshift(outbox)
+        }
+        emitAgentEvent(this.loopCtx, this, 'agent/inbox/update', pending.item, 'promote')
+        return 'applied'
+      }
+      default:
+        return assertNever(action)
+    }
  }

  /** Queue one ordinary prompt turn and wake the driver. */
@@ -169,9 +235,9 @@ export class ReactLoopAgent implements Agent {
      if (cause.kind !== 'disposed') emitAgentEvent(this.loopCtx, this, 'agent/cancel-requested', cause)
    }
    if (!options.keepInbox) {
-      const discarded = this.queued.map(item => item.message)
+      const discarded = this.queued.map(item => item.item)
      for (const item of this.outbox) {
-        if (item.steering) discarded.push(item.message)
+        if (item.steering && item.item !== undefined) discarded.push(item.item)
      }
      // Clear before abort observers run: replacement work belongs to the next turn.
      this.queued.length = 0
@@ -222,7 +288,8 @@ export class ReactLoopAgent implements Agent {
    // The some() guard above proves the queue is non-empty; the non-null
    // assertion expresses that invariant.
    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    const { message } = this.queued.shift()!
+    const { item } = this.queued.shift()!
+    const { message } = item
    const inheritedOutboxLength = this.outbox.length

    const admission = new AbortController()
@@ -293,7 +360,7 @@ export class ReactLoopAgent implements Agent {
    // Published only after the abort owner and pending done are installed: a
    // dequeue listener that cancels or disposes must find live cancellation
    // and quiescence ownership, not the previous activity's settled state.
-    emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message, 'queued')
+    emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item)
  }

  /**
@@ -643,7 +710,8 @@ export class ReactLoopAgent implements Agent {
    for (const item of this.outbox.splice(0, limit)) {
      if (item.steering) {
        steered = true
-        emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.message, 'steering')
+        if (item.item === undefined) throw new Error(`agent "${this.id}" steering outbox item has no inbox identity`)
+        emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.item)
        this.session.append(
          'steering/message',
          { turn, message: item.message },
--- a/packages/core/agent-loop/tests/contract-regressions.spec.ts
+++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts
@@ -4,7 +4,7 @@ import LlmService, { createUserMessage, CallId, MessageSource, ProviderRequestId
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
-import AgentRegistry, { type Agent, type InboxPlacement } from '@deepseek-ai/dsh-agent'
+import AgentRegistry, { type Agent, type InboxItem, type InboxPlacement } from '@deepseek-ai/dsh-agent'
 import AgentLoop from '@deepseek-ai/dsh-agent-loop'
 import { ReactLoopAgent } from '../src/agent.ts'
 import InvariantService from '@deepseek-ai/dsh-invariants'
@@ -53,6 +53,79 @@ function send(agent: Agent, text: string) {
  agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
 }

+function inboxText(item: InboxItem): string {
+  return item.message.content
+    .flatMap(block => block.type === 'text' ? [block.text] : [])
+    .join('')
+}
+
+describe('addressable inbox operations', () => {
+  it('edits in place, removes exactly one item, and promotes the next independent turn', async () => {
+    const adapter = new MockAdapter([
+      textResponse('first reply'),
+      textResponse('promoted reply'),
+      textResponse('edited reply'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('inbox-actions'), { provider: 'mock', model: 'mock' })
+    const admission = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    ctx.on('agent/prompt-submit', async (_subject, message, _signal, next) => {
+      if (message.content[0]?.type === 'text' && message.content[0].text === 'first') {
+        admission.resolve(undefined)
+        await release.promise
+      }
+      return next()
+    })
+
+    const pending: InboxItem[] = []
+    const updates: { id: string; action: string; text: string }[] = []
+    const discards: string[][] = []
+    ctx.on('agent/inbox/enqueue', (subject, item) => {
+      if (subject === agent && inboxText(item) !== 'first') pending.push(item)
+    })
+    ctx.on('agent/inbox/update', (subject, item, action) => {
+      if (subject === agent) updates.push({ id: item.id, action, text: inboxText(item) })
+    })
+    ctx.on('agent/inbox/discard', (subject, items) => {
+      if (subject === agent) discards.push(items.map(item => item.id))
+    })
+
+    send(agent, 'first')
+    await admission.promise
+    send(agent, 'remove me')
+    send(agent, 'edit me')
+    send(agent, 'promote me')
+    expect(pending.map(inboxText)).toEqual(['remove me', 'edit me', 'promote me'])
+
+    const remove = pending[0]!
+    const edit = pending[1]!
+    const promote = pending[2]!
+    expect(agent.updateInbox(edit.id, {
+      kind: 'edit',
+      content: [{ type: 'text', text: 'edited' }],
+    })).toBe('applied')
+    expect(agent.updateInbox(remove.id, { kind: 'remove' })).toBe('applied')
+    expect(agent.updateInbox(promote.id, { kind: 'promote' })).toBe('applied')
+    expect(updates).toEqual([
+      { id: edit.id, action: 'edit', text: 'edited' },
+      { id: promote.id, action: 'promote', text: 'promote me' },
+    ])
+    expect(discards).toEqual([[remove.id]])
+
+    const idle = waitForIdle(ctx, agent)
+    release.resolve(undefined)
+    await idle
+    expect(agent.session.events
+      .filter(event => event.type === 'user/message')
+      .map(event => event.type === 'user/message'
+        ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')
+        : ''))
+      .toEqual(['first', 'promote me', 'edited'])
+    expect(agent.updateInbox(promote.id, { kind: 'remove' })).toBe('not-found')
+  })
+})
+
 describe('assistant replay provenance', () => {
  it('records adapter replay state with the assembled assistant content', async () => {
    const response = textResponse('unchanged')
@@ -502,10 +575,10 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    const queuedSources: MessageSource[] = []
    const queuedShapes: string[][] = []
    const placements: InboxPlacement[] = []
-    ctx.on('agent/inbox/enqueue', (_agent, message, placement) => {
-      queuedSources.push(message.source)
-      queuedShapes.push(Object.keys(message).sort())
-      placements.push(placement)
+    ctx.on('agent/inbox/enqueue', (_agent, item) => {
+      queuedSources.push(item.message.source)
+      queuedShapes.push(Object.keys(item.message).sort())
+      placements.push(item.placement)
    })

    send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -86,8 +86,9 @@ describe('agent/prompt-submit', () => {
    const entered = Promise.withResolvers<undefined>()
    const decision = Promise.withResolvers<PromptDecision>()
    const observed: UserMessage[] = []
-    ctx.on('agent/inbox/enqueue', (subject, message) => {
+    ctx.on('agent/inbox/enqueue', (subject, item) => {
      if (subject !== agent) return
+      const message = item.message
      expect(Object.isFrozen(message)).toBe(true)
      expect(Object.isFrozen(message.content)).toBe(true)
      expect(Object.isFrozen(message.content[0])).toBe(true)
@@ -97,8 +98,8 @@ describe('agent/prompt-submit', () => {
        if (block?.type === 'text') block.text = 'listener mutation'
      }).toThrow()
    })
-    ctx.on('agent/inbox/enqueue', (subject, message) => {
-      if (subject === agent) observed.push(message)
+    ctx.on('agent/inbox/enqueue', (subject, item) => {
+      if (subject === agent) observed.push(item.message)
    })
    ctx.on('agent/prompt-submit', async () => {
      entered.resolve(undefined)
@@ -240,8 +241,8 @@ describe('agent/prompt-submit', () => {
      entered.resolve(undefined)
      return decision.promise
    })
-    ctx.on('agent/inbox/enqueue', (subject, _message, placement) => {
-      if (subject === agent) placements.push(placement)
+    ctx.on('agent/inbox/enqueue', (subject, item) => {
+      if (subject === agent) placements.push(item.placement)
    })

    const idle = waitForIdle(ctx, agent)