host
ef30558f7c
同项目任务互发消息、对方做完可回消息：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 跳过）。
193 lines
7.0 KiB
Python
193 lines
7.0 KiB
Python
"""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()
|