fix(agent-loop): scope blocked admission cleanup

2026-07-31 14:35:58 +08:00
parent 12a48558f2
commit c2ff9ddec8
18 changed files with 162 additions and 51 deletions
--- a/packages/core/agent-loop/README.md
+++ b/packages/core/agent-loop/README.md
@@ -55,7 +55,7 @@ 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.
+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 block's mandatory `discardClaimed` controls only its submitted batch; later next-step input and queued prompts remain pending for 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 queued-item boundary: edit freezes replacement content without changing message identity or position, while remove publishes discard. Edit publishes `agent/inbox/update`; steering and claimed occurrences return `not-found`. Claim publishes `agent/inbox/dequeue` and irrevocably removes the live address before prompt admission, so a racing update cannot rewrite durable history; `cancel()` without `keepInbox` publishes `agent/inbox/discard`.

--- a/packages/core/agent-loop/README.zh.md
+++ b/packages/core/agent-loop/README.zh.md
@@ -55,7 +55,7 @@ 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`，不开启轮次也不运行模型。
+统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。block 必须通过 `discardClaimed` 选择是否丢弃本次提交的批次；之后到达的 next-step 输入和排队提示词会继续保持待处理，等待后续获准的提示词。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。

 每次 FIFO 接受项时都会铸造一个 `InboxItemId`，并通过 `agent/inbox/enqueue` 发布完整的单次入队项。`updateInbox()` 持有同步 queued 项边界：编辑会冻结替换内容，但不改变消息标识或位置；移除会发布 discard。编辑会发布 `agent/inbox/update`；steering 项和已被认领的项会返回 `not-found`。认领操作会发布 `agent/inbox/dequeue`，并在提示词接纳前不可逆地移除实时寻址标识，因此竞态中的更新无法改写持久历史；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。

--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -176,7 +176,11 @@ export class ReactLoopAgent implements Agent {
      if (queued !== undefined) this.inbox.splice('next-turn', 0, 1, [], 'admitted')
      return { kind: 'admitted', messages: decision.messages }
    }
-    this.cancel({ kind: 'hook', reason: decision.reason }, { keepInbox: decision.keepInbox })
+    if (decision.discardClaimed) {
+      this.inbox.splice('next-step', 0, outboxLength, [], 'canceled')
+      if (queued !== undefined) this.inbox.splice('next-turn', 0, 1, [], 'canceled')
+    }
+    this.cancel({ kind: 'hook', reason: decision.reason }, { keepInbox: true })
    return { kind: 'blocked' }
  }

--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -212,7 +212,7 @@ describe('agent/prompt-submit', () => {
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })

    ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
-      ({ kind: 'block', reason: 'blocked by policy' }))
+      ({ kind: 'block', reason: 'blocked by policy', discardClaimed: true }))

    const reasons: TurnEndReason[] = []
    ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
@@ -230,6 +230,26 @@ describe('agent/prompt-submit', () => {
    expect(reasons).toEqual([])
  })

+  it('block can retain the claimed prompt without opening a turn', async () => {
+    const adapter = new MockAdapter([])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('retained-claim'), { provider: 'mock', model: 'mock' })
+
+    ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> => ({
+      kind: 'block',
+      reason: 'try later',
+      discardClaimed: false,
+    }))
+
+    send(agent, 'retained')
+    await agent.whenIdle()
+
+    expect(agent.inbox.nextTurn.map(message => message.content[0]))
+      .toEqual([{ type: 'text', text: 'retained' }])
+    expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
+    expect(adapter.requests).toEqual([])
+  })
+
  it('stages inject and steer during admission for the admitted turn', async () => {
    const adapter = new MockAdapter([textResponse('ok')])
    const ctx = await harness(adapter)
@@ -291,7 +311,7 @@ describe('agent/prompt-submit', () => {
    expect(nextRequest).toContain('admission steering')
  })

-  it('cancels admission-time input when admission is blocked', async () => {
+  it('preserves input staged after the blocked batch was claimed', async () => {
    const adapter = new MockAdapter([textResponse('retried')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('blocked-admission-outbox'), { provider: 'mock', model: 'mock' })
@@ -310,10 +330,14 @@ describe('agent/prompt-submit', () => {
      source: { kind: 'plugin', plugin: 'test' },
    }))
    agent.steer(createUserMessage({ content: [{ type: 'text', text: 'staged steering' }], source: { kind: 'user' } }))
-    decision.resolve({ kind: 'block', reason: 'policy' })
+    decision.resolve({ kind: 'block', reason: 'policy', discardClaimed: true })
    await blockedIdle

-    expect(agent.inbox.nextStep).toHaveLength(0)
+    expect(agent.inbox.nextStep.map(message => message.content[0]))
+      .toEqual([
+        { type: 'text', text: 'staged context' },
+        { type: 'text', text: 'staged steering' },
+      ])
    expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
    expect(adapter.requests).toEqual([])

@@ -323,14 +347,21 @@ describe('agent/prompt-submit', () => {

    const staged = events(agent).filter(event =>
      event.type === 'user/message' || event.type === 'steering/message')
-    expect(staged.map(event => event.type)).toEqual(['user/message'])
+    expect(staged.map(event => event.type)).toEqual([
+      'user/message',
+      'user/message',
+      'user/message',
+    ])
    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt')
-    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged context')
-    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged steering')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged context')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering')
  })

-  it('cancels later queued work when an admission is blocked', async () => {
-    const adapter = new MockAdapter([textResponse('continued')])
+  it('preserves later queued work when an admission is blocked', async () => {
+    const adapter = new MockAdapter([
+      textResponse('continued'),
+      textResponse('wake reply'),
+    ])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('rejected-admission-order'), {
      provider: 'mock',
@@ -340,7 +371,7 @@ describe('agent/prompt-submit', () => {
      const decision = await next()
      return messages.some(message =>
        message.content.some(block => block.type === 'text' && block.text === 'blocked prompt'))
-        ? { kind: 'block', reason: 'policy' }
+        ? { kind: 'block', reason: 'policy', discardClaimed: true }
        : decision
    })
    ctx.on('agent/prompt-submit', async (subject, messages, _signal, next) => {
@@ -364,17 +395,32 @@ describe('agent/prompt-submit', () => {
    await idle

    expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
-    expect(agent.inbox.hasPending).toBe(false)
+    expect(agent.inbox.nextStep.map(message => message.content[0]))
+      .toEqual([
+        { type: 'text', text: 'earlier state change' },
+        { type: 'text', text: 'earlier steering' },
+      ])
+    expect(agent.inbox.nextTurn.map(message => message.content[0]))
+      .toEqual([{ type: 'text', text: 'later prompt' }])
    expect(adapter.requests).toEqual([])
+
+    const resumed = waitForIdle(ctx, agent)
+    send(agent, 'wake')
+    await resumed
+    const request = JSON.stringify(adapter.requests[0]?.messages)
+    expect(request).toContain('earlier state change')
+    expect(request).toContain('earlier steering')
+    expect(request).toContain('later prompt')
+    expect(request).not.toContain('blocked prompt')
  })

-  it('cancels context-only injection when admission closes without a turn', async () => {
-    const adapter = new MockAdapter([])
+  it('preserves context-only injection staged after admission began', async () => {
+    const adapter = new MockAdapter([textResponse('continued')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('blocked-admission-context'), { provider: 'mock', model: 'mock' })
    const entered = Promise.withResolvers<undefined>()
    const decision = Promise.withResolvers<PromptDecision>()
-    ctx.on('agent/prompt-submit', async () => {
+    const disposeBlock = ctx.on('agent/prompt-submit', async () => {
      entered.resolve(undefined)
      return decision.promise
    })
@@ -386,13 +432,21 @@ describe('agent/prompt-submit', () => {
      content: [{ type: 'text', text: 'independent context' }],
      source: { kind: 'plugin', plugin: 'test' },
    }))
-    decision.resolve({ kind: 'block', reason: 'policy' })
+    decision.resolve({ kind: 'block', reason: 'policy', discardClaimed: true })
    await idle

    const log = events(agent)
    expect(log.some(event => event.type === 'user/message')).toBe(false)
-    expect(agent.inbox.hasPending).toBe(false)
+    expect(agent.inbox.nextStep.map(message => message.content[0]))
+      .toEqual([{ type: 'text', text: 'independent context' }])
    expect(adapter.requests).toEqual([])
+
+    disposeBlock()
+    const resumed = waitForIdle(ctx, agent)
+    send(agent, 'wake')
+    await resumed
+    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('independent context')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt')
  })

  it('leaves inbox state unchanged when its durable append fails', async () => {
@@ -414,15 +468,20 @@ describe('agent/prompt-submit', () => {
    expect(agent.status).toBe('idle')
  })

-  it('a blocked prompt cancels adjacent queued prompts', async () => {
-    const adapter = new MockAdapter([textResponse('ran once')])
+  it('a blocked prompt preserves adjacent queued prompts', async () => {
+    const adapter = new MockAdapter([
+      textResponse('safe reply'),
+      textResponse('wake reply'),
+    ])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })

    ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next): Promise<PromptDecision> => {
      const text = messages.flatMap(message => message.content)
        .map(b => (b.type === 'text' ? b.text : '')).join('')
-      return text === 'secret' ? { kind: 'block', reason: 'policy: no secrets' } : next()
+      return text === 'secret'
+        ? { kind: 'block', reason: 'policy: no secrets', discardClaimed: true }
+        : next()
    })

    const reasons: TurnEndReason[] = []
@@ -437,6 +496,14 @@ describe('agent/prompt-submit', () => {
    expect(adapter.requests).toHaveLength(0)
    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0)
    expect(reasons).toEqual([])
+    expect(agent.inbox.nextTurn.map(message => message.content[0]))
+      .toEqual([{ type: 'text', text: 'safe' }])
+
+    const resumed = waitForIdle(ctx, agent)
+    send(agent, 'wake')
+    await resumed
+    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('safe')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('secret')
  })

  it('a throwing prompt-submit listener reports the driver error and retains adjacent work', async () => {
@@ -653,7 +720,13 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
      ctx.on('agent/prompt-submit', async (_agent, messages, _signal, next): Promise<PromptDecision> => {
        const text = messages.flatMap(message => message.content)
          .map(b => (b.type === 'text' ? b.text : '')).join('')
-        if (text.includes('rm -rf')) return { kind: 'block', reason: 'destructive prompt blocked' }
+        if (text.includes('rm -rf')) {
+          return {
+            kind: 'block',
+            reason: 'destructive prompt blocked',
+            discardClaimed: true,
+          }
+        }
        return next()
      })
      // 3. PreToolUse: deny a dangerous tool by name.