feat: 任务间互发消息(API + agent 标记块,不中断任务)

同项目任务互发消息、对方做完可回消息: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 跳过)。
This commit is contained in:
host committed 2026-10-27 14:12:28 +08:00
1 parent 4901a1090b
commit ef30558f7c
11 files changed
+1171 -23

No files matched your search

+115 -18
View File
@@ -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 分叉出新任务。