agentao

12 Explicit Routing And Push Delegation

Parent doc: ACP Client And Project-Local Servers

Issue index: ACP Client Project-Local Servers Issues

Status

Part B 依赖一个私有 sessionUpdate: "task_complete" 扩展，不属于 ACP 标准枚举（标准集合仅含 user_message_chunk / agent_message_chunk / agent_thought_chunk / tool_call / tool_call_update / plan）。为了不让 Agentao 作为 ACP client 的行为偏离标准，并避免对 server 端引入非标扩展义务，决定彻底去掉该能力，而不是保留一个默认关闭但私有的路径。以下关于 Part B 的章节作为历史设计保留，不再代表当前实现。

Goal

在现有 project-local ACP client 基础上，补两个增强能力：

  1. 显式指定目标服务器 当用户消息中明确点名某个 ACP server 时，直接把该消息路由给对应 server，而不是继续走主 Agent 的普通 chat()。

  2. 后台 ACP server 主动 push 通知并触发后续委派 (已取消，见 Status) 当特定 server 开启实验性开关后，如果它通过 session/update 主动推送 task_complete 类型通知，Agentao 会在安全空闲点把结果注入主对话上下文，并自动触发一轮后续处理。

Scope

本 issue 只覆盖：

本 issue 不覆盖：

Part A: Explicit Server Routing

User Stories

Trigger Rules

v1 建议只做确定性匹配，不做模糊猜测：

  1. @server-name <task>
  2. server-name: <task>

可选扩展规则（v1.1 或 v2）：

  1. 让 server-name ...
  2. 请 server-name ...
  3. server-name 帮我 ...

Behavior

命中显式路由时：

  1. 不进入主 Agent 的普通 agent.chat(user_input) 路径
  2. 直接复用 /acp send <server> <task> 的执行逻辑
  3. CLI 显示明确的路由提示，如：
    • ACP Delegation → code-reviewer
  4. 任务文本中应移除路由前缀，仅保留真正的任务内容

Failure Handling

Proposed Data Model

新增一个轻量解析结果类型：

@dataclass(frozen=True)
class AcpExplicitRoute:
    server: str
    task: str
    syntax: str   # at_mention | colon | zh_explicit
    raw_prefix: str

Proposed Files

Proposed Flow

user input
  ↓
detect_explicit_route()
  ├─ no match → normal agent.chat(user_input)
  └─ match    → run_acp_prompt_inline(cli, route.server, route.task)

Why This Shape

Part B: Experimental Push Delegation (Dropped)

⚠️ 本节仅作为历史设计保留，不在代码库中实现。原因见顶部 Status：task_complete 不属于 ACP 标准 sessionUpdate 枚举，继续保留会迫使 server 侧引入私有扩展。

Goal

允许特定 ACP server 在后台主动推送“任务完成”结果，并把结果交给主 Agent 做后续决策。

Config Shape

在 .agentao/acp.json 的具体 server 下增加实验性开关：

{
  "servers": {
    "security-auditor": {
      "command": "python",
      "args": ["-m", "security_server"],
      "env": {},
      "cwd": ".",
      "experimental": {
        "pushTaskCompleteToAgent": true
      }
    }
  }
}

Why Under experimental

Wire Contract

复用已有 session/update，约定私有 sessionUpdate 类型：

{
  "method": "session/update",
  "params": {
    "sessionId": "sess_xxx",
    "update": {
      "sessionUpdate": "task_complete",
      "taskId": "audit-20250412-001",
      "title": "Dependency audit finished",
      "content": {
        "type": "text",
        "text": "Found 3 vulnerable packages..."
      },
      "metadata": {
        "severity": "high"
      }
    }
  }
}

Detection Rules

仅当同时满足以下条件时，才触发 push delegation：

  1. method == "session/update"
  2. params.update.sessionUpdate == "task_complete"
  3. 对应 server 开启 experimental.pushTaskCompleteToAgent

否则：

Threading Rule

禁止在 ACP reader 线程或 notification callback 线程内直接调用 agent.chat()。

正确做法：

  1. notification callback 识别符合条件的 task_complete
  2. 放入一个单独的 delegate queue
  3. 仅在 CLI 主线程的 safe idle point 消费该 queue
  4. 由主线程执行 synthetic message 注入和后续 agent.chat()

Proposed Queue Model

@dataclass(frozen=True)
class ACPDelegateEvent:
    server: str
    session_id: str
    task_id: str
    title: str
    text: str
    metadata: dict[str, Any]
    timestamp: float

新增：

Proposed Main-Thread Injection

在 safe idle point 把事件转成 synthetic user message：

<system-reminder>
ACP background task complete

server: security-auditor
task_id: audit-20250412-001
title: Dependency audit finished

result:
Found 3 vulnerable packages...
</system-reminder>

An ACP background server finished a task. Incorporate this update and decide whether any follow-up action is needed.

然后由主线程触发一轮：

self.agent.chat(synthetic_message)

Why Use Synthetic User Message

Safety Constraints

1. Dedup

task_complete 应强制要求 taskId。

去重键：

若缺失 taskId：

2. Safe Idle Only

只允许在这些空闲点消费 delegate queue：

不得在：

3. Batch Merge

同一空闲点如有多个 task_complete 事件：

4. No Recursive ACP Routing

synthetic message 不参与 explicit route 检测。

即：

5. Rate Limit

建议增加简单限流：

v1 可先不实现复杂配额，只保留 hook 点。

Proposed Code Changes

Config / Models

建议新增字段：

experimental_push_task_complete_to_agent: bool = False

或：

experimental: dict[str, Any] = field(default_factory=dict)

更推荐前者在内部落成稳定布尔字段，避免下游处处读裸 dict。

Routing

Push Delegation

Testing Plan

Explicit Routing

  1. @server task 命中并走 ACP runner
  2. server: task 命中并走 ACP runner
  3. 未命中 server 时回退普通 chat
  4. 空任务内容时报错
  5. 多 server 冲突时报错

Push Delegation

  1. 未开 experimental 开关时，task_complete 只进 inbox
  2. 开启开关时，task_complete 进入 delegate queue
  3. safe idle 点消费后，触发一轮主 Agent follow-up
  4. 相同 (server, task_id) 不重复注入
  5. 缺失 taskId 时只渲染 inbox，不自动注入
  6. 多个事件同轮合并为一次 follow-up

Rollout Plan

Phase 1 (shipped)

Phase 2 — dropped

Documentation Impact

需要同步更新：

其中用户文档应明确：

This site is open source. Improve this page.