From ef30558f7cf7c7647700da9d7801b371afd75f77 Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Tue, 27 Oct 2026 14:12:28 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BB=BB=E5=8A=A1=E9=97=B4=E4=BA=92?=
 =?UTF-8?q?=E5=8F=91=E6=B6=88=E6=81=AF=EF=BC=88API=20+=20agent=20=E6=A0=87?=
 =?UTF-8?q?=E8=AE=B0=E5=9D=97=EF=BC=8C=E4=B8=8D=E4=B8=AD=E6=96=AD=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

同项目任务互发消息、对方做完可回消息：API POST /tasks/<目标>/messages
（from_task_id，目标项目 create_task + 发送项目 read 鉴权）与 agent
回合输出标记块（worker 回合后解析投递，不用联网、不用凭证，本地
与 Docker 都通）。投递不中断任务：对方运行中/排队中排成后续消息
（轮次边界执行，direct 优先）；对方终态则续跑复活（resume_message +
queued），A→B→A 往返不受回合时序影响。拒绝项：自己、跨项目、goal
目标、已合并、复活位被占；每轮最多 5 块，仅 completed 轮次投递；
目标执行时配额门照常生效。控制台后续消息列显示发送方，新增给
其他任务发消息框（落已代理的 /tasks 下，不用改 nginx）。

测试：新增 test_peer_messages（15）、test_peer_messages_server（8）、
test_worker_peer（4）；全量 cloud 654、host 64、deploy 19 全过（cloud 4 跳过）。
---
 cloud/README.md                          |  43 ++++
 cloud/console.html                       |  43 +++-
 cloud/peer_messages.py                   | 215 ++++++++++++++++++++
 cloud/sandbox.py                         |  14 +-
 cloud/server.py                          | 133 ++++++++++--
 cloud/tests/test_console.py              |   3 +-
 cloud/tests/test_peer_messages.py        | 219 ++++++++++++++++++++
 cloud/tests/test_peer_messages_server.py | 248 +++++++++++++++++++++++
 cloud/tests/test_worker_peer.py          | 192 ++++++++++++++++++
 cloud/worker.py                          |  57 +++++-
 examples/peer-message-prompt.txt         |  27 +++
 11 files changed, 1171 insertions(+), 23 deletions(-)
 create mode 100644 cloud/peer_messages.py
 create mode 100644 cloud/tests/test_peer_messages.py
 create mode 100644 cloud/tests/test_peer_messages_server.py
 create mode 100644 cloud/tests/test_worker_peer.py
 create mode 100644 examples/peer-message-prompt.txt

diff --git a/cloud/README.md b/cloud/README.md
index cfe629b..00b4c26 100644
--- a/cloud/README.md
+++ b/cloud/README.md
@@ -512,6 +512,49 @@ goal 模式的继续：server 把任务转成新 goal 语义（goal=新消息、
 合并任务 worktree 已删、请分支（merged 409）；冲突与新建一致
 （plan + approval=full、plan + goal 各 400）。
 
+## 任务间互发消息（peer messages，Muse 式通用互发）
+
+任意两个同项目任务可以互发消息、对方做完可以回消息，不
+用经过用户中转。两个发送方：
+
+- API：`POST /tasks/<目标>/messages`，body
+  `{from_task_id, body, mode}`（mode 缺省 queued）；调用方
+  须对目标项目有 create_task、对发送任务项目有 read（防冒
+  名）。`GET /tasks/<任务>/messages` 列出该任务发出的消息
+  （收到的在 `/followups` 里，带 sender_task_id）。
+- Agent：任务里跑的 agent 在回合输出里写标记块，worker 在
+  回合结束后解析并投递——agent 不用联网、不用凭证，本地
+  与 Docker 执行器都通。块格式见
+  `examples/peer-message-prompt.txt`；每轮最多 5 块。
+
+投递语义（核心约束：互发消息不中断任务）：
+
+- 目标排队中 / 运行中：消息排成一条后续消息（带
+  sender_task_id），worker 排空循环在轮次边界执行——direct
+  优先、队内 FIFO，与用户后续消息同款，当前轮绝不被打断。
+  接收方 agent 看到的 prompt 带发送方前缀
+  `[来自任务 <id> 的消息]`，凭此知道回给哪个任务。
+- 目标已终态（completed / failed / interrupted）：消息复活
+  它——把带前缀的正文写进 resume_message 并把状态置回
+  queued，下次认领按续跑跑这一轮。没有复活，B 回给已结束
+  的 A 的消息会直接丢失，A→B→A 的往返没法成立。
+- 一律拒绝（409 或投递层跳过，只记审计）：发给自己、跨
+  项目、目标是 goal 任务、目标已合并、目标已有待续跑消息
+  （复活位被占）、目标不存在。
+- 只有 completed 轮次的标记块被投递（失败 / 超时 / 取消 /
+  配额中断的轮次不投递）；goal 任务的回合不解析标记块，且
+  目标是 goal 时直接拒绝——所以 goal 任务既不收也不发。
+- 投递不查发送方配额（发送本身不花钱），但目标任务执行时
+  的配额门照常生效：排空前查配额，超限时该条回 pending、
+  任务走配额中断；复活任务认领时认领门同样生效。互发消息
+  不构成配额绕过。
+- 边界：标记块正文不能含围栏（非贪婪匹配会把块截断成非法
+  JSON 而被跳过）；worker 投递的 direct 消息只有排空优先
+  权，没有 API 直接发送那种“立即进链路”的标记（worker 侧
+  链路目录不一定在本机）；复活投递的消息没有 followup 行，
+  发出列表里查不到，查审计 peer_message_sent 与目标的续跑
+  轮次；控制台“给其他任务发消息”框走 /tasks 下的
+  /messages，不用新增 nginx location。
 
 ## 测试
 
diff --git a/cloud/console.html b/cloud/console.html
index a5f34c8..c84942d 100644
--- a/cloud/console.html
+++ b/cloud/console.html
@@ -263,6 +263,17 @@
   </div>
 </div>
 <div id="followupList" class="muted"></div>
+<h3>给其他任务发消息</h3>
+<div id="peerMessageForm">
+  <input id="peerTarget" placeholder="目标任务 id（与本任务同项目；对方运行中排队执行、已终态则续跑复活，绝不打断当前轮）" style="width:100%">
+  <textarea id="peerBody" placeholder="消息正文（以本任务名义发出；对方 agent 会看到发送方任务 id，可照此回消息）"></textarea>
+  <div class="bar">
+    <button id="btnQueuePeerMessage">排队发给对方</button>
+    <button id="btnDirectPeerMessage" class="primary">直接发送</button>
+    <span id="peerMsg" class="muted"></span>
+  </div>
+</div>
+<div id="peerMessageList" class="muted"></div>
 </section>
 
 <script>
@@ -666,6 +677,8 @@ async function showDetail(taskId) {
     $("#agentBox").innerHTML = "子代理链路加载中…";
     loadAgents(taskId);
     loadFollowups(taskId);
+    loadPeerMessages(taskId);
+    $("#peerMsg").textContent = "";
   } catch (e) { $("#detail").innerHTML = "<span class='err'>加载失败：" + esc(e.message) + "</span>"; }
 }
 
@@ -688,7 +701,8 @@ async function loadFollowups(taskId) {
     var list = d.followups || [];
     if (!list.length) { box.innerHTML = "（暂无后续消息）"; return; }
     box.innerHTML = list.map(function (f) {
-      return "[" + esc(f.mode) + "｜" + esc(f.status) + "] " + esc(String(f.body || "").slice(0, 120));
+      var from = f.sender_task_id ? "来自任务 " + esc(String(f.sender_task_id).slice(0, 8)) + "｜" : "";
+      return "[" + esc(f.mode) + "｜" + esc(f.status) + "] " + from + esc(String(f.body || "").slice(0, 120));
     }).join("<br>");
   } catch (e) { box.innerHTML = "<span class='err'>后续消息加载失败：" + esc(e.message) + "</span>"; }
 }
@@ -899,8 +913,35 @@ $("#btnContinueConfirm").onclick = async function () {
     loadTasks();
   } catch (e) { $("#continueMsgSpan").innerHTML = "<span class='err'>继续失败：" + esc(e.message) + "</span>"; }
 };
+async function loadPeerMessages(taskId) {
+  var box = $("#peerMessageList");
+  try {
+    var d = await api("/tasks/" + encodeURIComponent(taskId) + "/messages", "GET");
+    var list = d.messages || [];
+    if (!list.length) { box.innerHTML = "（本任务尚未发出任务间消息）"; return; }
+    box.innerHTML = list.map(function (f) {
+      return "[发给 " + esc(String(f.task_id).slice(0, 8)) + "｜" + esc(f.mode) + "｜" + esc(f.status) + "] " + esc(String(f.body || "").slice(0, 120));
+    }).join("<br>");
+  } catch (e) { box.innerHTML = "<span class='err'>发出消息加载失败：" + esc(e.message) + "</span>"; }
+}
+async function postPeerMessage(mode) {
+  if (!currentTask) return;
+  var target = $("#peerTarget").value.trim();
+  var text = $("#peerBody").value.trim();
+  if (!target) { $("#peerMsg").textContent = "目标任务 id 不能为空"; return; }
+  if (!text) { $("#peerMsg").textContent = "消息正文不能为空"; return; }
+  try {
+    var d = await api("/tasks/" + encodeURIComponent(target) + "/messages", "POST", { from_task_id: currentTask.id, body: text, mode: mode });
+    $("#peerMsg").innerHTML = "<span class='ok'>已发送（" + esc(d.delivery) + "）</span>";
+    $("#peerBody").value = "";
+    loadPeerMessages(currentTask.id);
+  } catch (e) { $("#peerMsg").innerHTML = "<span class='err'>发送失败：" + esc(e.message) + "</span>"; }
+}
+
 $("#btnQueueFollowup").onclick = function () { postFollowup("queued"); };
 $("#btnDirectFollowup").onclick = function () { postFollowup("direct"); };
+$("#btnQueuePeerMessage").onclick = function () { postPeerMessage("queued"); };
+$("#btnDirectPeerMessage").onclick = function () { postPeerMessage("direct"); };
 $("#btnCost").onclick = async function () {
   if (!currentTask) return;
   try {
diff --git a/cloud/peer_messages.py b/cloud/peer_messages.py
new file mode 100644
index 0000000..378b0b9
--- /dev/null
+++ b/cloud/peer_messages.py
@@ -0,0 +1,215 @@
+"""任务间互发消息（peer messages）。
+
+Muse 式通用互发消息在 codex-app-server 里的落地。两个发送方：
+
+1. API：任何持有用户 key 的调用方（控制台、外部编排器）调
+   POST /tasks/<目标>/messages，带 from_task_id，消息以发送
+   任务的名义投递。
+2. Agent：任务里跑的 agent 在回合输出里写标记块，worker 在
+   回合结束后解析并投递——agent 不用联网、不用凭证，本地
+   与 Docker 执行器都通（两者都经 sandbox 的结果解析）。
+
+投递语义（核心约束：互发消息不中断任务）：
+
+- 目标排队中 / 运行中：消息排成一条后续消息（followup），
+  带 sender_task_id；由 worker 排空循环在轮次边界执行——
+  direct 优先、队内 FIFO，与用户后续消息同款，当前轮绝不
+  被打断。执行仍受目标任务的配额门约束（排空前查配额，
+  超限时该条回 pending、任务走配额中断，与后续消息一致）。
+- 目标已终态（completed / failed / interrupted）：消息复活
+  它——把带发送方前缀的正文写进 resume_message 并把状态
+  置回 queued，下一次 worker 认领时按续跑跑这一轮。这样
+  A→B→A 的往返不受双方回合结束时序影响；没有复活，B 回
+  给已结束的 A 的消息会直接丢失。
+- 其它情形一律拒绝，不炸任务、不改状态：发给自己、跨项目、
+  目标是 goal 任务（goal 循环自有轮次）、目标已合并、
+  目标已有待续跑消息（复活位被占）、目标不存在。
+
+发送方约束：
+
+- API 路径由 server 鉴权：调用方须对目标项目有 create_task
+  权限、对发送任务项目有 read 权限（防冒名）。
+- Worker 路径的发送方就是当前任务本身；跨项目在投递层再
+  挡一次（agent 猜到别项目任务 id 也发不出去）。
+- 每轮最多投递 5 个标记块；只有 completed 轮次的标记块被
+  投递（失败 / 超时 / 取消 / 配额中断的轮次不投递）；goal
+  任务的回合不解析标记块（只收不发，见 README）。
+- 标记块正文里不能含围栏——围栏是非贪婪匹配，正文里的
+  围栏会把块截断成非法 JSON 而被跳过。
+
+本模块只依赖标准库，server 与 worker 共用，不 import
+cloud 其它模块，避免循环 import。
+"""
+
+import json
+import re
+import time
+import uuid
+
+MAX_PEER_MESSAGES_PER_TURN = 5
+MAX_PEER_MESSAGE_BODY = 20000
+
+# 复活可接的目标终态。merged / merge-failed 不复活（代码已
+# 合并，续跑语义不明）；其它未知状态一律拒绝。
+REVIVABLE_STATUSES = ("completed", "failed", "interrupted")
+
+_FENCE_RE = re.compile(r"```task-message\s*\n(.*?)\n```", re.DOTALL)
+
+
+def peer_message_prefix(from_task_id):
+    """发送方前缀：排空轮次 prompt 与复活消息都带它，接收方
+    agent 凭此知道回给哪个任务 id。"""
+    return "[来自任务 %s 的消息]\n" % from_task_id
+
+
+def row_get(row, name):
+    """行取值容错：sqlite3.Row 缺列报 IndexError、dict 缺键
+    报 KeyError、空行报 TypeError，一律回 None（worker 直连
+    老库、列还没迁移时排空不能因此停摆）。"""
+    try:
+        return row[name]
+    except (KeyError, IndexError, TypeError):
+        return None
+
+
+def extract_peer_messages(text):
+    """从回合输出文本里提取互发消息标记块。
+
+    块格式::
+
+        ```task-message
+        {"to": "<目标任务 id>", "body": "...", "mode": "queued"}
+        ```
+
+    mode 缺省 queued，非法值也回落 queued；to / body 缺失、
+    非字符串、空串、正文超长、JSON 非法时整块跳过；每轮
+    最多取 5 块，多余的忽略。"""
+    if not text:
+        return []
+    out = []
+    for match in _FENCE_RE.finditer(str(text)):
+        if len(out) >= MAX_PEER_MESSAGES_PER_TURN:
+            break
+        try:
+            payload = json.loads(match.group(1))
+        except (TypeError, ValueError):
+            continue
+        if not isinstance(payload, dict):
+            continue
+        to_task_id = payload.get("to") or payload.get("to_task_id")
+        body = payload.get("body")
+        if not isinstance(to_task_id, str) or not to_task_id.strip():
+            continue
+        if not isinstance(body, str) or not body.strip():
+            continue
+        body = body.strip()
+        if len(body) > MAX_PEER_MESSAGE_BODY:
+            continue
+        mode = payload.get("mode", "queued")
+        if mode not in ("queued", "direct"):
+            mode = "queued"
+        out.append({"to_task_id": to_task_id.strip(),
+                    "body": body, "mode": mode})
+    return out
+
+
+def ensure_followups_table(conn):
+    """幂等建后续消息表并补 sender_task_id 列。
+
+    server 的 _ensure_followups_table 委托到这里，worker 投递
+    路径也调它——两边同一份 DDL，不漂移。列补齐写法与
+    tasks 迁移同款：PRAGMA 查现有列，缺了再 ALTER（db 适配
+    层把 PRAGMA 翻译成 information_schema 查询，Postgres 下
+    同样生效）。"""
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_followups("
+        "id TEXT PRIMARY KEY, "
+        "task_id TEXT, "
+        "project_id TEXT, "
+        "body TEXT, "
+        "mode TEXT, "
+        "status TEXT, "
+        "created_by TEXT, "
+        "sender_task_id TEXT, "
+        "created_at DOUBLE PRECISION, "
+        "updated_at DOUBLE PRECISION)"
+    )
+    try:
+        existing = {row[1] for row in conn.execute(
+            "PRAGMA table_info(task_followups)").fetchall()}
+    except Exception:
+        existing = set()
+    if "sender_task_id" not in existing:
+        try:
+            conn.execute(
+                "ALTER TABLE task_followups"
+                " ADD COLUMN sender_task_id TEXT")
+        except Exception:
+            pass
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_followups_task"
+        " ON task_followups(task_id, status)")
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_followups_sender"
+        " ON task_followups(sender_task_id)")
+
+
+def deliver_peer_message(conn, from_task_id, to_task_id, body, mode="queued"):
+    """投递一条互发消息。调用方自备连接，本函数在投递成功时
+    commit，拒绝时不写任何东西。
+
+    返回 (status, detail)：delivered 时 detail 为 followup id；
+    revived 时 detail 为空；rejected 时 detail 为原因（直接
+    可进 409 / 审计）。mode 非 queued/direct 时回落 queued。"""
+    if not isinstance(body, str) or not body.strip():
+        return "rejected", "body required"
+    body = body.strip()
+    if len(body) > MAX_PEER_MESSAGE_BODY:
+        return "rejected", "body too long"
+    if mode not in ("queued", "direct"):
+        mode = "queued"
+    if not from_task_id or not to_task_id:
+        return "rejected", "task not found"
+    if from_task_id == to_task_id:
+        return "rejected", "cannot message self"
+    sender = conn.execute(
+        "SELECT * FROM tasks WHERE id = ?", (from_task_id,)).fetchone()
+    if sender is None:
+        return "rejected", "sender task not found"
+    target = conn.execute(
+        "SELECT * FROM tasks WHERE id = ?", (to_task_id,)).fetchone()
+    if target is None:
+        return "rejected", "target task not found"
+    if row_get(sender, "project_id") != row_get(target, "project_id"):
+        return "rejected", "cross-project messages not allowed"
+    if row_get(target, "verify_command") or row_get(target, "goal_mode"):
+        return "rejected", "goal tasks do not accept peer messages"
+    status = row_get(target, "status")
+    now = time.time()
+    if status in ("queued", "running"):
+        ensure_followups_table(conn)
+        fu_id = uuid.uuid4().hex
+        conn.execute(
+            "INSERT INTO task_followups(id, task_id, project_id,"
+            " body, mode, status, created_by, sender_task_id,"
+            " created_at, updated_at)"
+            " VALUES(?, ?, ?, ?, ?, 'pending', ?, ?, ?, ?)",
+            (fu_id, to_task_id, row_get(target, "project_id"),
+             body, mode, row_get(sender, "created_by"),
+             from_task_id, now, now),
+        )
+        conn.commit()
+        return "delivered", fu_id
+    if status in REVIVABLE_STATUSES:
+        cur = conn.execute(
+            "UPDATE tasks SET status = 'queued',"
+            " resume_message = ?, updated_at = ?"
+            " WHERE id = ? AND status IN"
+            " ('completed', 'failed', 'interrupted')"
+            " AND resume_message IS NULL",
+            (peer_message_prefix(from_task_id) + body, now, to_task_id))
+        conn.commit()
+        if cur.rowcount == 1:
+            return "revived", ""
+        return "rejected", "target has a pending resume message"
+    return "rejected", "target task is not receivable: %s" % status
diff --git a/cloud/sandbox.py b/cloud/sandbox.py
index 8de55f3..238dd79 100644
--- a/cloud/sandbox.py
+++ b/cloud/sandbox.py
@@ -20,6 +20,8 @@ import shutil
 import subprocess
 import sys
 
+from cloud.peer_messages import extract_peer_messages as _extract_peer_messages
+
 MAX_STDOUT_CHARS = 2000
 
 
@@ -224,6 +226,7 @@ class LocalExecutor(Executor):
         stdout = _truncate_stdout(stdout_full)
         status = "failed"
         usage = None
+        task_messages = []
         if returncode == 0:
             status = "completed"
             # JSON 解析必须用完整 stdout：截头后的前 2000 字在
@@ -242,6 +245,10 @@ class LocalExecutor(Executor):
                 # 原样透传给 worker 记 actual；脏数据由
                 # _clean_usage 挡掉，不进 actual。
                 usage = _clean_usage(parsed.get("usage"))
+                # 互发消息标记块：从完整解析出的 text 里提取（截头
+                # 版 text 尾部的块会丢），worker 在 completed 轮次
+                # 后投递；goal 路径不走这里，goal 任务只收不发。
+                task_messages = _extract_peer_messages(parsed.get("text"))
             else:
                 host_status = None
         else:
@@ -258,6 +265,7 @@ class LocalExecutor(Executor):
             # 的旧结果 dict 不含此字段，保持不变。
             "stderr": _truncate_stdout(getattr(result, "stderr", "")),
             "usage": usage,
+            "task_messages": task_messages,
         }
 
     def _host_name(self, task):
@@ -1231,11 +1239,13 @@ def _parse_host_result(task_id, fake):
     status = "failed"
     host_status = None
     usage = None
+    task_messages = []
     if returncode == 0:
         status = "completed"
         # 与 _result_from_runner 同理：JSON 解析用完整 stdout，
         # 否则长 text 会把 host 的 failed 误判成 completed，
-        # 也会丢掉 JSON 末尾的 usage。
+        # 也会丢掉 JSON 末尾的 usage。标记块同样从完整解析
+        # 出的 text 里提取。
         try:
             parsed = json.loads(stdout_full)
         except (TypeError, ValueError):
@@ -1245,6 +1255,7 @@ def _parse_host_result(task_id, fake):
             status = ("completed" if host_status == "completed"
                       else "failed")
             usage = _clean_usage(parsed.get("usage"))
+            task_messages = _extract_peer_messages(parsed.get("text"))
     return {
         "executor": "docker",
         "task_id": task_id,
@@ -1254,6 +1265,7 @@ def _parse_host_result(task_id, fake):
         "host_status": host_status,
         "stderr": _truncate_stdout(getattr(fake, "stderr", "")),
         "usage": usage,
+        "task_messages": task_messages,
     }
 
 
diff --git a/cloud/server.py b/cloud/server.py
index e2109b2..6ddfada 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -62,6 +62,10 @@ from cloud.db import create_database as _create_database
 from cloud.db import init_postgres_schema as _init_postgres_schema
 from cloud.db import PostgresIntegrityError as _PgIntegrityError
 from cloud.db import PostgresOperationalError as _PgOperationalError
+from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
+from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups
+from cloud.peer_messages import peer_message_prefix as _peer_prefix
+from cloud.peer_messages import row_get as _row_get
 from cloud.kb import KBStore
 from cloud.memory import MemoryStore
 from cloud.pr import PrSkipped, PrStepError
@@ -832,24 +836,12 @@ def _attachments_map(conn, task_ids):
 
 
 def _ensure_followups_table(conn):
-    """幂等建后续消息表：server init_db 与各 followup 接口都调它，
-    老库第一次跑到这里时自动补表，不丢数据。DOUBLE PRECISION 在
-    SQLite 与 Postgres 下都可用（SQLite 按 NUMERIC 亲和处理）。"""
-    conn.execute(
-        "CREATE TABLE IF NOT EXISTS task_followups("
-        "id TEXT PRIMARY KEY, "
-        "task_id TEXT, "
-        "project_id TEXT, "
-        "body TEXT, "
-        "mode TEXT, "
-        "status TEXT, "
-        "created_by TEXT, "
-        "created_at DOUBLE PRECISION, "
-        "updated_at DOUBLE PRECISION)"
-    )
-    conn.execute(
-        "CREATE INDEX IF NOT EXISTS idx_followups_task"
-        " ON task_followups(task_id, status)")
+    """幂等建后续消息表：委托给 peer_messages（两边同一份 DDL，
+    server init_db、各 followup 接口与 worker 投递路径共用）。
+    老库第一次跑到这里时自动补表补列，不丢数据。DOUBLE
+    PRECISION 在 SQLite 与 Postgres 下都可用（SQLite 按 NUMERIC
+    亲和处理）。"""
+    _peer_ensure_followups(conn)
 
 
 def _followup_to_dict(row):
@@ -861,6 +853,7 @@ def _followup_to_dict(row):
         "mode": row["mode"],
         "status": row["status"],
         "created_by": row["created_by"],
+        "sender_task_id": _row_get(row, "sender_task_id"),
         "created_at": row["created_at"],
         "updated_at": row["updated_at"],
     }
@@ -1477,6 +1470,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._create_followup(user_id, rest[: -len("/followups")])
             elif rest.endswith("/followups") and method == "GET":
                 self._list_followups(user_id, rest[: -len("/followups")])
+            elif rest.endswith("/messages") and method == "POST":
+                self._create_peer_message(user_id, rest[: -len("/messages")])
+            elif rest.endswith("/messages") and method == "GET":
+                self._list_peer_messages(user_id, rest[: -len("/messages")])
             elif rest.endswith("/fork") and method == "POST":
                 self._fork_task(user_id, rest[: -len("/fork")])
             elif rest.endswith("/continue") and method == "POST":
@@ -2895,6 +2892,106 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         self._send_json(200, {
             "followups": [_followup_to_dict(r) for r in rows]})
 
+    def _create_peer_message(self, user_id, task_id):
+        """任务间互发消息（API 发送方）：以 from_task_id 任务的名义
+        给目标任务发消息。投递语义与 worker 路径共用 deliver 层，
+        同样不中断目标任务：运行中/排队中排成后续消息，终态则
+        续跑复活。权限：调用方须对目标项目有 create_task、对
+        发送任务项目有 read（防冒名）；from_task_id 必填——用户
+        给任务发消息走 /followups，这里只做任务对任务。"""
+        body, err = self._read_json()
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        target_row = self._fetch_task(task_id)
+        if target_row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        target = _task_to_dict(target_row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, target["project_id"], "create_task"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        from_task_id = body.get("from_task_id")
+        if not from_task_id or not isinstance(from_task_id, str):
+            self._send_json(400, {"error": "from_task_id required"})
+            return
+        sender_row = self._fetch_task(from_task_id)
+        if sender_row is None:
+            self._send_json(404, {"error": "sender task not found"})
+            return
+        sender = _task_to_dict(sender_row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, sender["project_id"], "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        text = body.get("body")
+        if not text or not isinstance(text, str) or not text.strip():
+            self._send_json(400, {"error": "body required"})
+            return
+        text = text.strip()
+        if len(text) > 20000:
+            self._send_json(400, {"error": "body too long"})
+            return
+        mode = body.get("mode", "queued")
+        if mode not in ("queued", "direct"):
+            self._send_json(400, {
+                "error": "mode must be queued or direct"})
+            return
+        conn = self._db()
+        try:
+            _ensure_followups_table(conn)
+            status, detail = _deliver_peer_message(
+                conn, from_task_id, task_id, text, mode)
+        finally:
+            conn.close()
+        if status == "rejected":
+            self._send_json(409, {"error": detail})
+            return
+        if status == "delivered" and mode == "direct":
+            # 直接发送的链路可见性与用户后续消息同款：消息立即
+            # 进链路，但当前轮仍跑完才开下一轮，不中断。
+            self._append_direct_trace(
+                target, detail, _peer_prefix(from_task_id) + text)
+        with _auth_store(self.server.db_path) as store:
+            store.audit(user_id, target["project_id"],
+                        "peer_message_sent",
+                        "%s -> %s (%s)" % (from_task_id, task_id, status))
+        self._send_json(201, {
+            "from_task_id": from_task_id, "to_task_id": task_id,
+            "mode": mode, "delivery": status, "followup_id": (
+                detail if status == "delivered" else None)})
+
+    def _list_peer_messages(self, user_id, task_id):
+        """列出某任务发出的互发消息（收到的消息在 /followups 里，
+        带 sender_task_id）。只含排成后续消息的那部分；复活投递
+        的那部分没有 followup 行，查审计 peer_message_sent 与目标
+        任务的续跑轮次。权限与 followup 列表同款：项目 read。"""
+        row = self._fetch_task(task_id)
+        if row is None:
+            self._send_json(404, {"error": "task not found"})
+            return
+        task = _task_to_dict(row)
+        with _auth_store(self.server.db_path) as store:
+            if not store.check_permission(
+                    user_id, task["project_id"], "read"):
+                self._send_json(403, {"error": "forbidden"})
+                return
+        conn = self._db()
+        try:
+            _ensure_followups_table(conn)
+            rows = conn.execute(
+                "SELECT * FROM task_followups WHERE sender_task_id = ?"
+                " ORDER BY created_at, id",
+                (task_id,),
+            ).fetchall()
+        finally:
+            conn.close()
+        self._send_json(200, {
+            "messages": [_followup_to_dict(r) for r in rows]})
+
     def _fork_task(self, user_id, task_id):
         """会话分支：从已结束任务的 thread 分叉出新任务。
 
diff --git a/cloud/tests/test_console.py b/cloud/tests/test_console.py
index 8af4d33..2049ed8 100644
--- a/cloud/tests/test_console.py
+++ b/cloud/tests/test_console.py
@@ -64,7 +64,8 @@ class ConsoleTestCase(unittest.TestCase):
         self.assertIn("Codex 命令台", text)
         # 壳必须引用它实际调的接口，防 JS 与服务端路由漂移。
         for marker in ("/health", "/projects", "/tasks", "/workers",
-                       "X-API-Key", "agentBox", "/agents"):
+                       "X-API-Key", "agentBox", "/agents",
+                       "peerMessage", "/messages"):
             self.assertIn(marker, text)
 
     def test_console_trailing_slash_same_page(self):
diff --git a/cloud/tests/test_peer_messages.py b/cloud/tests/test_peer_messages.py
new file mode 100644
index 0000000..961c615
--- /dev/null
+++ b/cloud/tests/test_peer_messages.py
@@ -0,0 +1,219 @@
+"""任务间互发消息单测：标记块提取 + 投递分支。
+
+不跑真实模型、不联网：extract 是纯函数；deliver 用内存
+SQLite + 最小表结构（tasks 只含投递读到的列，followups
+表由 peer_messages 自己建）。
+"""
+
+import sqlite3
+import sys
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.peer_messages import (
+        deliver_peer_message,
+        extract_peer_messages,
+        peer_message_prefix,
+    )
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.peer_messages import (
+        deliver_peer_message,
+        extract_peer_messages,
+        peer_message_prefix,
+    )
+
+
+def _conn():
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        "CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
+        " created_by TEXT, status TEXT, verify_command TEXT,"
+        " goal_mode INTEGER, resume_message TEXT, updated_at REAL)")
+    return conn
+
+
+def _task(conn, task_id, project_id="p1", status="queued",
+          created_by="tester", verify_command=None, goal_mode=None,
+          resume_message=None):
+    conn.execute(
+        "INSERT INTO tasks(id, project_id, created_by, status,"
+        " verify_command, goal_mode, resume_message, updated_at)"
+        " VALUES(?, ?, ?, ?, ?, ?, ?, 0)",
+        (task_id, project_id, created_by, status, verify_command,
+         goal_mode, resume_message))
+    conn.commit()
+
+
+def _followups(conn, task_id=None, sender=None):
+    sql = "SELECT * FROM task_followups"
+    conds = []
+    args = []
+    if task_id is not None:
+        conds.append("task_id = ?")
+        args.append(task_id)
+    if sender is not None:
+        conds.append("sender_task_id = ?")
+        args.append(sender)
+    if conds:
+        sql += " WHERE " + " AND ".join(conds)
+    try:
+        return conn.execute(sql, args).fetchall()
+    except Exception:
+        return []
+
+
+def _task_row(conn, task_id):
+    return conn.execute(
+        "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+
+
+class ExtractTestCase(unittest.TestCase):
+    def test_single_block_default_queued(self):
+        text = ("做完了。\n```task-message\n"
+                "{\"to\": \"T2\", \"body\": \"请评审\"}\n```\n")
+        msgs = extract_peer_messages(text)
+        self.assertEqual(len(msgs), 1)
+        self.assertEqual(msgs[0]["to_task_id"], "T2")
+        self.assertEqual(msgs[0]["body"], "请评审")
+        self.assertEqual(msgs[0]["mode"], "queued")
+
+    def test_direct_mode_and_alias_key(self):
+        text = ("```task-message\n{\"to_task_id\": \"T2\","
+                " \"body\": \"急\", \"mode\": \"direct\"}\n```")
+        msgs = extract_peer_messages(text)
+        self.assertEqual(msgs[0]["mode"], "direct")
+
+    def test_bad_mode_falls_back_to_queued(self):
+        text = ("```task-message\n{\"to\": \"T2\", \"body\": \"x\","
+                " \"mode\": \"smoke-signal\"}\n```")
+        self.assertEqual(extract_peer_messages(text)[0]["mode"], "queued")
+
+    def test_invalid_blocks_skipped(self):
+        text = (
+            "```task-message\nnot json\n```\n"
+            "```task-message\n{\"body\": \"no to\"}\n```\n"
+            "```task-message\n{\"to\": \"T2\"}\n```\n"
+            "```task-message\n{\"to\": \"T2\", \"body\": \"  \"}\n```\n"
+            "```task-message\n{\"to\": \"T2\", \"body\": \"ok\"}\n```\n")
+        msgs = extract_peer_messages(text)
+        self.assertEqual(len(msgs), 1)
+        self.assertEqual(msgs[0]["body"], "ok")
+
+    def test_body_too_long_skipped(self):
+        text = ("```task-message\n{\"to\": \"T2\", \"body\": \""
+                + "x" * 20001 + "\"}\n```")
+        self.assertEqual(extract_peer_messages(text), [])
+
+    def test_max_five_per_turn(self):
+        blocks = "".join(
+            "```task-message\n{\"to\": \"T%d\", \"body\": \"m%d\"}\n```\n"
+            % (i, i) for i in range(7))
+        self.assertEqual(len(extract_peer_messages(blocks)), 5)
+
+    def test_no_fence_or_empty(self):
+        self.assertEqual(extract_peer_messages("plain text"), [])
+        self.assertEqual(extract_peer_messages(""), [])
+        self.assertEqual(extract_peer_messages(None), [])
+
+
+class DeliverTestCase(unittest.TestCase):
+    def test_running_target_becomes_followup(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="running")
+        status, detail = deliver_peer_message(conn, "T1", "T2", "请评审")
+        self.assertEqual(status, "delivered")
+        rows = _followups(conn, task_id="T2", sender="T1")
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["body"], "请评审")
+        self.assertEqual(rows[0]["mode"], "queued")
+        self.assertEqual(rows[0]["status"], "pending")
+        self.assertEqual(rows[0]["created_by"], "tester")
+        self.assertEqual(rows[0]["id"], detail)
+
+    def test_queued_target_direct_mode(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="queued")
+        status, _ = deliver_peer_message(conn, "T1", "T2", "急", "direct")
+        self.assertEqual(status, "delivered")
+        self.assertEqual(_followups(conn, task_id="T2")[0]["mode"], "direct")
+
+    def test_completed_target_revives(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="completed")
+        status, detail = deliver_peer_message(conn, "T1", "T2", "回来一下")
+        self.assertEqual(status, "revived")
+        self.assertEqual(detail, "")
+        row = _task_row(conn, "T2")
+        self.assertEqual(row["status"], "queued")
+        self.assertEqual(row["resume_message"],
+                         peer_message_prefix("T1") + "回来一下")
+        self.assertEqual(_followups(conn, task_id="T2"), [])
+
+    def test_revive_blocked_when_resume_occupied(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="completed", resume_message="旧消息")
+        status, detail = deliver_peer_message(conn, "T1", "T2", "新消息")
+        self.assertEqual(status, "rejected")
+        self.assertIn("pending resume", detail)
+        self.assertEqual(_task_row(conn, "T2")["resume_message"], "旧消息")
+
+    def test_failed_and_interrupted_revive(self):
+        for st in ("failed", "interrupted"):
+            conn = _conn()
+            _task(conn, "T1")
+            _task(conn, "T2", status=st)
+            status, _ = deliver_peer_message(conn, "T1", "T2", "回来")
+            self.assertEqual(status, "revived")
+            self.assertEqual(_task_row(conn, "T2")["status"], "queued")
+
+    def test_goal_targets_rejected(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="running", verify_command="true")
+        status, _ = deliver_peer_message(conn, "T1", "T2", "x")
+        self.assertEqual(status, "rejected")
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="running", goal_mode=1)
+        status, _ = deliver_peer_message(conn, "T1", "T2", "x")
+        self.assertEqual(status, "rejected")
+
+    def test_self_cross_project_merged_missing_rejected(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", project_id="p2")
+        _task(conn, "T3", status="merged")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "T1", "x")[0], "rejected")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "T2", "x")[0], "rejected")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "T3", "x")[0], "rejected")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "NOPE", "x")[0], "rejected")
+        self.assertEqual(
+            deliver_peer_message(conn, "NOPE", "T1", "x")[0], "rejected")
+        # 拒绝不写任何东西、不改状态。
+        self.assertEqual(_followups(conn), [])
+        self.assertEqual(_task_row(conn, "T3")["status"], "merged")
+
+    def test_body_validation(self):
+        conn = _conn()
+        _task(conn, "T1")
+        _task(conn, "T2", status="running")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "T2", "  ")[0], "rejected")
+        self.assertEqual(
+            deliver_peer_message(conn, "T1", "T2", "x" * 20001)[0],
+            "rejected")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/tests/test_peer_messages_server.py b/cloud/tests/test_peer_messages_server.py
new file mode 100644
index 0000000..a78044c
--- /dev/null
+++ b/cloud/tests/test_peer_messages_server.py
@@ -0,0 +1,248 @@
+"""任务间互发消息服务端单测：POST /tasks/<id>/messages 与
+GET /tasks/<id>/messages。
+
+不跑真实模型、不联网：本地临时端口 + 临时 SQLite，与
+test_task_links.py 同款骨架。覆盖鉴权、权限、校验、三种
+投递结果（排队 / 复活 / 拒绝）与发出列表。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+class PeerMessageServerTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.outsider_id, self.outsider_key = self.auth.create_user("outsider")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.trace_dir = os.path.join(self.tmp.name, "host-traces")
+        os.makedirs(self.trace_dir, exist_ok=True)
+        self.server.trace_dir = self.trace_dir
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+        status, self.project2 = self._req(
+            "POST", "/projects", {"name": "other"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if body is not None:
+            payload = json.dumps(body).encode("utf-8")
+            hdrs["Content-Type"] = "application/json"
+        if key is not None:
+            hdrs["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=hdrs)
+            resp = conn.getresponse()
+            data = resp.read()
+        finally:
+            conn.close()
+        try:
+            parsed = json.loads(data.decode("utf-8")) if data else None
+        except ValueError:
+            parsed = None
+        return resp.status, parsed
+
+    def _create_task(self, project_id, **extra):
+        body = {"project_id": project_id, "prompt": "做事"}
+        body.update(extra)
+        status, task = self._req("POST", "/tasks", body, self.owner_key)
+        self.assertEqual(status, 201, task)
+        return task
+
+    def _set_status(self, task_id, status):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute("UPDATE tasks SET status = ? WHERE id = ?",
+                         (status, task_id))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _task_row(self, task_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            return conn.execute(
+                "SELECT status, resume_message FROM tasks WHERE id = ?",
+                (task_id,)).fetchone()
+        finally:
+            conn.close()
+
+    def _followups(self, task_id):
+        _status, body = self._req(
+            "GET", "/tasks/%s/followups" % task_id, None, self.owner_key)
+        return body["followups"]
+
+    def test_delivered_to_running_target(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        self._set_status(target["id"], "running")
+        status, body = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "请评审"},
+            self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "delivered")
+        self.assertIsNotNone(body["followup_id"])
+        fus = self._followups(target["id"])
+        self.assertEqual(len(fus), 1)
+        self.assertEqual(fus[0]["body"], "请评审")
+        self.assertEqual(fus[0]["sender_task_id"], sender["id"])
+        self.assertEqual(fus[0]["status"], "pending")
+
+    def test_revived_completed_target(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        self._set_status(target["id"], "completed")
+        status, body = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "回来一下"},
+            self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "revived")
+        self.assertIsNone(body["followup_id"])
+        row = self._task_row(target["id"])
+        self.assertEqual(row[0], "queued")
+        self.assertIn("回来一下", row[1])
+        self.assertIn(sender["id"], row[1])
+
+    def test_direct_mode_writes_trace(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        self._set_status(target["id"], "running")
+        status, body = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "急事",
+             "mode": "direct"},
+            self.owner_key)
+        self.assertEqual(status, 201, body)
+        self.assertEqual(body["delivery"], "delivered")
+        names = os.listdir(self.trace_dir)
+        self.assertEqual(len(names), 1)
+        with open(os.path.join(self.trace_dir, names[0]),
+                  encoding="utf-8") as f:
+            entry = json.loads(f.read().strip())
+        self.assertEqual(entry["phase"], "followup-direct")
+        self.assertIn("急事", entry["text"])
+        self.assertIn(sender["id"], entry["text"])
+
+    def test_validation_and_not_found(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        base = "/tasks/%s/messages" % target["id"]
+        status, _ = self._req("POST", base, {"body": "x"}, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", base, {"from_task_id": sender["id"]}, self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", base,
+            {"from_task_id": sender["id"], "body": "x", "mode": "smoke"},
+            self.owner_key)
+        self.assertEqual(status, 400)
+        status, _ = self._req(
+            "POST", base, {"from_task_id": "nope", "body": "x"},
+            self.owner_key)
+        self.assertEqual(status, 404)
+        status, _ = self._req(
+            "POST", "/tasks/nope/messages",
+            {"from_task_id": sender["id"], "body": "x"}, self.owner_key)
+        self.assertEqual(status, 404)
+        status, _ = self._req(
+            "POST", base, {"from_task_id": sender["id"], "body": "x"},
+            None)
+        self.assertEqual(status, 401)
+
+    def test_forbidden_for_outsider(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        status, _ = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "x"}, self.outsider_key)
+        self.assertEqual(status, 403)
+
+    def test_cross_project_rejected(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project2["id"])
+        self._set_status(target["id"], "running")
+        status, body = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "x"}, self.owner_key)
+        self.assertEqual(status, 409, body)
+        self.assertIn("cross-project", body["error"])
+
+    def test_self_and_goal_rejected(self):
+        task = self._create_task(self.project["id"])
+        self._set_status(task["id"], "running")
+        status, body = self._req(
+            "POST", "/tasks/%s/messages" % task["id"],
+            {"from_task_id": task["id"], "body": "x"}, self.owner_key)
+        self.assertEqual(status, 409)
+        goal = self._create_task(self.project["id"],
+                                 verify_command="true")
+        self._set_status(goal["id"], "running")
+        sender = self._create_task(self.project["id"])
+        status, _ = self._req(
+            "POST", "/tasks/%s/messages" % goal["id"],
+            {"from_task_id": sender["id"], "body": "x"}, self.owner_key)
+        self.assertEqual(status, 409)
+
+    def test_list_outgoing_messages(self):
+        sender = self._create_task(self.project["id"])
+        target = self._create_task(self.project["id"])
+        self._set_status(target["id"], "running")
+        status, _ = self._req(
+            "POST", "/tasks/%s/messages" % target["id"],
+            {"from_task_id": sender["id"], "body": "请评审"},
+            self.owner_key)
+        self.assertEqual(status, 201)
+        status, body = self._req(
+            "GET", "/tasks/%s/messages" % sender["id"], None, self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(body["messages"]), 1)
+        self.assertEqual(body["messages"][0]["task_id"], target["id"])
+        self.assertEqual(body["messages"][0]["sender_task_id"], sender["id"])
+        status, body = self._req(
+            "GET", "/tasks/%s/messages" % sender["id"], None,
+            self.outsider_key)
+        self.assertEqual(status, 403)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/tests/test_worker_peer.py b/cloud/tests/test_worker_peer.py
new file mode 100644
index 0000000..6a9fb96
--- /dev/null
+++ b/cloud/tests/test_worker_peer.py
@@ -0,0 +1,192 @@
+"""cloud worker 任务间互发消息单测：agent 标记块 → 投递 →
+接收方排空前缀 / 终态复活。
+
+不跑真实模型、不联网：注入假 runner（假 CLI），任务与
+目标都在临时 SQLite 里；env 里的值只是占位用的假名，绝
+不是真实密钥。
+"""
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+MARKER_TO_T2 = ("结果如下。\n```task-message\n"
+                "{\"to\": \"T2\", \"body\": \"请评审这份结论\"}\n```\n")
+PREFIX = "[来自任务 T1 的消息]\n"
+
+
+def _insert_task(db_path, task_id, created_by="tester", status="queued",
+                 project_id="p1"):
+    now = time.time()
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " idempotency_key, depends_on, priority, created_by,"
+            " created_at, updated_at, last_heartbeat)"
+            " VALUES(?, ?, ?, ?, NULL, '[]', 0, ?, ?, ?, NULL)",
+            (task_id, project_id, "do work", status, created_by, now, now),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _status(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT status FROM tasks WHERE id = ?", (task_id,)).fetchone()
+    finally:
+        conn.close()
+    return row[0] if row else None
+
+
+def _resume_message(db_path, task_id):
+    conn = sqlite3.connect(db_path)
+    try:
+        row = conn.execute(
+            "SELECT resume_message FROM tasks WHERE id = ?",
+            (task_id,)).fetchone()
+    finally:
+        conn.close()
+    return row[0] if row else None
+
+
+def _followups(db_path, task_id=None):
+    conn = sqlite3.connect(db_path)
+    try:
+        if task_id is None:
+            return conn.execute("SELECT * FROM task_followups").fetchall()
+        return conn.execute(
+            "SELECT * FROM task_followups WHERE task_id = ?",
+            (task_id,)).fetchall()
+    finally:
+        conn.close()
+
+
+def _prompt_values(calls):
+    """假 runner 收到的全部 prompt：dispatch 在 --task 后，
+    run 在 --prompt 后，两个都扫。"""
+    values = []
+    for cmd in calls:
+        for flag in ("--task", "--prompt"):
+            if flag in cmd:
+                values.append(cmd[cmd.index(flag) + 1])
+    return values
+
+
+def _marker_runner(calls, marker_text):
+    """第一轮输出含标记块，之后一律 plain done。"""
+    def fake(cmd):
+        calls.append(list(cmd))
+        text = marker_text if len(calls) == 1 else "done"
+        return SimpleNamespace(
+            returncode=0,
+            stdout=json.dumps({"status": "completed", "text": text}),
+            stderr="")
+    return fake
+
+
+class WorkerPeerTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+        self.calls = []
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _worker(self, runner):
+        return Worker(self.db_path, "/fake/host.py", runner=runner,
+                      env=self.env, model="gpt-6.1-sol",
+                      retry_backoff_seconds=0)
+
+    def test_marker_delivered_and_drain_prefixed(self):
+        _insert_task(self.db_path, "T1")
+        _insert_task(self.db_path, "T2")
+        worker = self._worker(_marker_runner(self.calls, MARKER_TO_T2))
+        result = worker.run_one(task=worker.get_task("T1"))
+        self.assertEqual(result["status"], "completed")
+        # T1 的标记块投递成 T2 的后续消息（发送方 id 在列里）。
+        fus = _followups(self.db_path, "T2")
+        self.assertEqual(len(fus), 1)
+        row = fus[0]
+        # 列序与 DDL 一致：id, task_id, project_id, body, mode,
+        # status, created_by, sender_task_id, created_at, updated_at。
+        self.assertEqual(row[3], "请评审这份结论")
+        self.assertEqual(row[5], "pending")
+        self.assertEqual(row[7], "T1")
+        # T2 跑起来：首轮原 prompt，排空轮次 prompt 带发送方前缀。
+        result = worker.run_one(task=worker.get_task("T2"))
+        self.assertEqual(result["status"], "completed")
+        prompts = _prompt_values(self.calls)
+        self.assertTrue(
+            any(PREFIX in p and "请评审这份结论" in p for p in prompts),
+            prompts)
+        self.assertEqual(_followups(self.db_path, "T2")[0][5], "done")
+
+    def test_marker_revives_completed_target(self):
+        _insert_task(self.db_path, "T1")
+        _insert_task(self.db_path, "T2", status="completed")
+        worker = self._worker(_marker_runner(self.calls, MARKER_TO_T2))
+        result = worker.run_one(task=worker.get_task("T1"))
+        self.assertEqual(result["status"], "completed")
+        # T2 被复活：状态回 queued，续跑消息带发送方前缀。
+        self.assertEqual(_status(self.db_path, "T2"), "queued")
+        self.assertEqual(_resume_message(self.db_path, "T2"),
+                         PREFIX + "请评审这份结论")
+        self.assertEqual(_followups(self.db_path, "T2"), [])
+        # 复活后的 T2 续跑：prompt 就是续跑消息，跑完清掉。
+        result = worker.run_one(task=worker.get_task("T2"))
+        self.assertEqual(result["status"], "completed")
+        prompts = _prompt_values(self.calls)
+        self.assertTrue(any(PREFIX in p for p in prompts), prompts)
+        self.assertIsNone(_resume_message(self.db_path, "T2"))
+
+    def test_failed_round_markers_not_delivered(self):
+        _insert_task(self.db_path, "T1")
+        _insert_task(self.db_path, "T2")
+
+        def failing(cmd):
+            self.calls.append(list(cmd))
+            return SimpleNamespace(
+                returncode=1,
+                stdout=json.dumps({"status": "completed",
+                                   "text": MARKER_TO_T2}),
+                stderr="boom")
+
+        worker = self._worker(failing)
+        result = worker.run_one(task=worker.get_task("T1"))
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(_followups(self.db_path), [])
+
+    def test_self_and_cross_project_markers_rejected(self):
+        _insert_task(self.db_path, "T1")
+        _insert_task(self.db_path, "T3", project_id="p2")
+        text = ("```task-message\n{\"to\": \"T1\", \"body\": \"自己\"}\n```\n"
+                "```task-message\n{\"to\": \"T3\", \"body\": \"别项目\"}\n```\n")
+        worker = self._worker(_marker_runner(self.calls, text))
+        result = worker.run_one(task=worker.get_task("T1"))
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(_followups(self.db_path), [])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/cloud/worker.py b/cloud/worker.py
index 6da3021..9e5ce27 100644
--- a/cloud/worker.py
+++ b/cloud/worker.py
@@ -137,6 +137,9 @@ from cloud.cost import CostStore, estimate_tokens, price_usd_micros, sum_usage
 from cloud.db import connect as _db_connect
 from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, is_fallbackable, record_fallback
+from cloud.peer_messages import deliver_peer_message as _deliver_peer_message
+from cloud.peer_messages import peer_message_prefix as _peer_prefix
+from cloud.peer_messages import row_get as _row_get
 from cloud.pr import PrSkipped, PrStepError
 from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job)
 from cloud.queue import budget_exceeded, ready_tasks
@@ -1672,7 +1675,8 @@ class Worker:
             if cur.rowcount != 1:
                 return None
             return {"id": row["id"], "body": row["body"],
-                    "mode": row["mode"]}
+                    "mode": row["mode"],
+                    "sender_task_id": _row_get(row, "sender_task_id")}
         except Exception:
             return None
         finally:
@@ -1707,6 +1711,46 @@ class Worker:
         except Exception:
             pass
 
+    def _deliver_peer_messages(self, task, result):
+        """投递本轮 agent 输出里的互发消息标记块（见 peer_messages）。
+
+        只投递 completed 轮次的块：失败 / 超时 / 取消 / 配额中断
+        的轮次不投递；任务执行期间被取消时也不投递。投递本身
+        不中断任何任务：对方运行中/排队中排成后续消息（轮次边
+        界执行），对方终态则续跑复活；拒绝（自己、跨项目、
+        goal 目标、已合并、复活位被占等）只记审计，不炸 run_one。
+        投递不查发送方配额（发送本身不花钱），目标任务执行时
+        的配额门照常生效——互发消息不构成配额绕过。"""
+        if not result or result.get("status") != "completed":
+            return
+        if self._was_cancelled(task["id"]):
+            return
+        messages = result.get("task_messages") or []
+        if not messages:
+            return
+        delivered = 0
+        rejected = []
+        for msg in messages:
+            try:
+                conn = _db_connect(self.db_path)
+                try:
+                    status, detail = _deliver_peer_message(
+                        conn, task["id"], msg["to_task_id"],
+                        msg["body"], msg["mode"])
+                finally:
+                    conn.close()
+            except Exception as exc:
+                status, detail = "rejected", "delivery error: %s" % exc
+            if status in ("delivered", "revived"):
+                delivered += 1
+            else:
+                rejected.append("%s: %s" % (msg["to_task_id"], detail))
+        detail = "%s: delivered=%d rejected=%d" % (
+            task["id"], delivered, len(rejected))
+        if rejected:
+            detail += " (" + "; ".join(rejected) + ")"
+        self._audit(task["project_id"], "worker_peer_messages", detail)
+
     def _run_attempt(self, task, quota_check=None):
         """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。
 
@@ -2121,6 +2165,7 @@ class Worker:
             result, error = self._run_fork(task, quota_check=_quota_hit)
             usage_total = sum_usage(usage_total,
                                     (result or {}).get("usage"))
+            self._deliver_peer_messages(task, result)
             if (result or {}).get("status") == "quota_exceeded":
                 quota_error = self._quota_exceeded(task, usage_total) or \
                     "quota_exceeded: 本月配额已用完"
@@ -2137,6 +2182,7 @@ class Worker:
                     break
             result, error = self._run_attempt(task, quota_check=_quota_hit)
             usage_total = sum_usage(usage_total, (result or {}).get("usage"))
+            self._deliver_peer_messages(task, result)
             if (result or {}).get("status") == "quota_exceeded":
                 # 轮询掐断：进程已被杀，配额原因以现查为准。
                 quota_error = self._quota_exceeded(task, usage_total) or \
@@ -2188,6 +2234,7 @@ class Worker:
                 result, error = self._run_attempt(task, quota_check=_quota_hit)
                 usage_total = sum_usage(usage_total,
                                         (result or {}).get("usage"))
+                self._deliver_peer_messages(task, result)
                 if (result or {}).get("status") == "quota_exceeded":
                     quota_error = self._quota_exceeded(task, usage_total) or \
                         "quota_exceeded: 本月配额已用完"
@@ -2239,11 +2286,17 @@ class Worker:
                     quota_stop = True
                     break
                 fu_task = dict(task)
-                fu_task["prompt"] = fu["body"]
+                # 任务间消息的接收方提示：带发送方 id，接收方
+                # agent 凭此知道回给哪个任务；用户后续消息无前缀。
+                fu_prompt = fu["body"]
+                if fu.get("sender_task_id"):
+                    fu_prompt = _peer_prefix(fu["sender_task_id"]) + fu_prompt
+                fu_task["prompt"] = fu_prompt
                 fu_result, fu_error = self._run_attempt(
                     fu_task, quota_check=_quota_hit)
                 usage_total = sum_usage(
                     usage_total, (fu_result or {}).get("usage"))
+                self._deliver_peer_messages(task, fu_result)
                 if (fu_result or {}).get("status") == "quota_exceeded":
                     self._restore_followup(fu["id"])
                     quota_error = self._quota_exceeded(task, usage_total) or \
diff --git a/examples/peer-message-prompt.txt b/examples/peer-message-prompt.txt
new file mode 100644
index 0000000..780dbdb
--- /dev/null
+++ b/examples/peer-message-prompt.txt
@@ -0,0 +1,27 @@
+给另一个任务发消息时，不要等用户中转，直接在你的回合输出
+里写一个标记块（平台 worker 会在回合结束后解析并投递，
+你不需要联网、不需要凭证）：
+
+```task-message
+{"to": "<目标任务 id>", "body": "你要说的话", "mode": "queued"}
+```
+
+规则（违反的块会被跳过，不报错、不重发，你要自己检查）：
+
+- to 只能是与你同项目的任务 id；跨项目、发给自己、发给
+  goal 任务、发给已合并任务都会被拒绝。
+- 收到别的任务的消息时，你的 prompt 开头会带发送方前缀，
+  形如“[来自任务 <id> 的消息]”——回消息时 to 就填那个 id，
+  不要猜别的 id。
+- mode 缺省 queued：对方当前轮跑完后按序执行你的消息，
+  绝不打断对方当前轮。mode 为 direct 时对方跑完当前轮后
+  立刻以你的消息开下一轮，且优先于排队中的消息；同样不
+  打断当前轮。
+- 对方已经结束时，你的消息会复活它（续跑一轮），不用担
+  心对方先结束就收不到；对方正在跑时，你的消息排队等它
+  的轮次边界。
+- 每轮最多发 5 块，多余的忽略；只有你成功完成的轮次里的
+  块才会被投递，失败轮次里的块不投递。
+- body 里不要含围栏（三个反引号），否则块会被截断成非法
+  JSON 而被跳过；body 最长 20000 字。
+- goal 任务收不到也发不出任务间消息，不要给 goal 任务发。