host
f39ed3654e
基底：#251 头 174b631（堆叠于 #242 feat-ci-expand 之上）， 基线 origin/main aebc6238；本轨不改 #251 已定语义、不动 workflow 门槛、不删检查、不加 skip。 lint（ruff E,F@120）：109 -> 10 - F401 43 -> 0：删未用导入；try/except 双绑定的名字出现 “打地鼠”（删 except 分支后 try 分支又被标出），两分支 均已核对正文无引用后清掉；MCP 三模块的 Error 名不删， 改走 importlib 属性绑定（见下），两路径绑定都保留。 - E702 40 -> 0：全在 host.py main() 的 argparse 段，纯分行， tokenize 按分号 token 拆（含 2377 行多行语句只断首个分号）； 一次 5 空格缩进失误已回滚重做并以 ast + 全量 host 测试验证。 - E501 13 -> 0：host.py 5 项随 E702 分行连带清掉；mcp_server 3 项把描述拆成相邻字面量（AST 比对 24 条工具描述与原值 完全一致，注意 ruff 按东亚宽度计长，CJK 算 2）；test_host 5 项断开 git commit 参数表。 - E402 8 -> 10：生产侧 2 项清掉（db.py _re、github_app _base64 移到顶部，之前无引用、无重名）；剩余 10 项全是 tests/ 的 sys.path 前置 import（6 项旧 + 4 个新测试文件 沿用同一既定模式），移上去会破坏单文件直跑，有意保留。 - F841 4 -> 0：只去绑定、不删调用（run_task、DB 入队、 tarfile 上下文、git 提交），各有注释与原测试/新测试覆盖； host dispatch_from_branch 新增真实 git 仓库回归测试。 - F811 1 -> 0：server 本地 _row_get 与 peer_messages.row_get 逐行相同，删本地定义，全部调用点解析到同一 import； 新增 test_server_row_get（同一对象 + 三路行为）。 mypy（生产范围）：24 -> 0，零新增 type: ignore - no-redef 9 -> 0：MCP 三模块 try/except 两次 from-import 改 importlib.import_module + 属性绑定（cloud 优先、flat copy 回落、内层 ImportError 同样回落，语义与原来一致； flat-copy 子进程测试证实回落后 __module__ 为顶层名）； server _row_get 随 F811 消除。 - arg-type 10 -> 0：harness2_tasks shared 改 TypedDict 逐键精确类型；CodexProbeAdapter 补 no-op validate_tree （_adapter_validate 本就跳过缺失方法，行为一致）。 - var-annotated 3 -> 0：sub2api _SHARED_ADMIN/_login_slots、 verify-multiuser RESULTS 加精确注解。 - assignment/misc 2 -> 0：marker_tools ZoneInfo 先声明 Any 再 try import（草稿先验 mypy 通过）。 bandit：总数 286 不变，High 仍 4 项，全在 host.py，本轨不改 （属第 3 项白名单轨范围），只做 triage： - 1319 run_verify：无插值，整串验测命令，管道/&& 须 shell 语义（新增 test_host_run_verify 断言管道与 && 真实生效）。 - 1497 notify：模板来自算子配置，三处插值全部 shlex.quote， 证据为既有 test_notify_failure_quotes_shell_metacharacters （; touch 与 $(touch ) 不执行）。 - 2099/2103 goal verify：无插值，任务配置验测串，同 run_verify。 - 无新增 High；pip-audit 0 漏洞。 覆盖率（同口径三套件 coverage 计量）：75.96% -> 77.87% （22392 行/缺 5385 -> 22437 行/缺 4963，显示 76% -> 78%） - client.py 0% -> 90%（pipe 假子进程 + main 分支） - run_parallel.py 0% -> 97%、migrate 包装 0% -> 95% - playwright_mcp 37% -> 62%、imagegen_mcp 47% -> 79%、 websearch_mcp 51% -> 86%、deploy.py 56% -> 77% - verify-multiuser 0% -> 27%（仅 check/req helper；main() 是主机侧真实多用户实弹验证，按门禁不建生产任务、 不以 mock 冒充，仍为缺口） - host.py 55% -> 57%（文件 2020 -> 2060 行系 E702 分行 的计量口径变化，非代码膨胀） - 未到 85%：门槛保持 75% 不动，不删低覆盖模块出统计口径； 缺口主力仍是 server.py / worker.py / host.py 大文件， 下一阶段须针对性深测，非本轨快赢可及。 验证：三套件 1576 -> 1629 项全过（tests 127、cloud/tests 1449 skipped=4、deploy/tests 53）；ruff E9 通过；mypy 0； bandit High 4（同上）；pip-audit 0 漏洞；试跑产物已删， 提交后工作树干净。
336 lines
14 KiB
Python
336 lines
14 KiB
Python
"""任务实时链路（服务端）单测：GET /tasks/{id}/trace 与 /trace/stream。
|
|
|
|
覆盖：鉴权（401/403/404）、快照去重与坏行跳过、worktree 在别机
|
|
时的空链路 + 说明、已结束任务的流（snapshot 后立刻 done 关流）。
|
|
不跑真实模型，不启动 worker；链路目录经 server.trace_dir 注入。
|
|
"""
|
|
|
|
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, _tenant_trace_dir
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.auth import AuthStore
|
|
from cloud.server import create_server, _tenant_trace_dir
|
|
|
|
|
|
class TaskTraceTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
self.trace_dir = os.path.join(self.tmp.name, "host-traces")
|
|
os.makedirs(self.trace_dir, exist_ok=True)
|
|
self.tenants_root = os.path.join(self.tmp.name, "tenants")
|
|
os.makedirs(self.tenants_root, exist_ok=True)
|
|
self.auth = AuthStore(self.db_path)
|
|
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
|
self.other_id, self.other_key = self.auth.create_user("other")
|
|
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
|
self.server.trace_dir = self.trace_dir
|
|
self.server.tenants_root = self.tenants_root
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(
|
|
target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
status, self.project = self._req(
|
|
"POST", "/projects", {"name": "demo"}, self.owner_key)
|
|
self.assertEqual(status, 201)
|
|
status, self.task = self._req(
|
|
"POST", "/tasks",
|
|
{"project_id": self.project["id"], "prompt": "hi"},
|
|
self.owner_key)
|
|
self.assertEqual(status, 201)
|
|
self.task_id = self.task["id"]
|
|
|
|
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)
|
|
headers = {"Content-Type": "application/json"}
|
|
if key:
|
|
headers["X-API-Key"] = key
|
|
conn.request(method, path,
|
|
json.dumps(body) if body is not None else None, headers)
|
|
resp = conn.getresponse()
|
|
raw = resp.read()
|
|
conn.close()
|
|
try:
|
|
data = json.loads(raw)
|
|
except ValueError:
|
|
data = None
|
|
return resp.status, data
|
|
|
|
def _set_task(self, **fields):
|
|
conn = sqlite3.connect(self.db_path)
|
|
try:
|
|
for k, v in fields.items():
|
|
conn.execute(
|
|
"UPDATE tasks SET %s = ? WHERE id = ?" % k,
|
|
(v, self.task_id))
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
def _write_trace(self, lines):
|
|
path = os.path.join(self.trace_dir, self.task_id + ".jsonl")
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
for line in lines:
|
|
f.write(line + "\n")
|
|
|
|
def _write_tenant_trace(self, lines, filename=None):
|
|
d = os.path.join(self.tenants_root, self.project["id"],
|
|
"work", "host-traces")
|
|
os.makedirs(d, exist_ok=True)
|
|
path = os.path.join(d, filename or (self.task_id + ".jsonl"))
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
for line in lines:
|
|
f.write(line + "\n")
|
|
|
|
def test_trace_auth_and_not_found(self):
|
|
status, _ = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id)
|
|
self.assertEqual(status, 401)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.other_key)
|
|
self.assertEqual(status, 403)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/nope/trace", key=self.owner_key)
|
|
self.assertEqual(status, 404)
|
|
|
|
def test_trace_snapshot_dedup_and_bad_lines(self):
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "ls",
|
|
"exitCode": None}),
|
|
"this is not json",
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "ls",
|
|
"exitCode": 0, "durationMs": 5}),
|
|
json.dumps({"kind": "file", "id": "f1:a.py", "path": "a.py"}),
|
|
])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["status"], "queued")
|
|
# 同 id 后到的完成态覆盖先到的开始态，保首次顺序。
|
|
self.assertEqual([e["id"] for e in data["entries"]],
|
|
["c1", "f1:a.py"])
|
|
self.assertEqual(data["entries"][0]["exitCode"], 0)
|
|
self.assertEqual(data["note"], "")
|
|
|
|
def test_trace_missing_file_is_empty(self):
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["entries"], [])
|
|
|
|
def test_trace_remote_worktree_host_is_empty_with_note(self):
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "ls"})])
|
|
self._set_task(worktree_host="some-other-host")
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["entries"], [])
|
|
self.assertIn("some-other-host", data["note"])
|
|
|
|
def _read_sse(self, path, key):
|
|
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
|
headers = {}
|
|
if key:
|
|
headers["X-API-Key"] = key
|
|
conn.request("GET", path, headers=headers)
|
|
resp = conn.getresponse()
|
|
raw = resp.read()
|
|
conn.close()
|
|
events = []
|
|
for block in raw.decode("utf-8").split("\n\n"):
|
|
for line in block.split("\n"):
|
|
if line.startswith("data: "):
|
|
events.append(json.loads(line[len("data: "):]))
|
|
return resp.status, events
|
|
|
|
def test_stream_finished_task_snapshot_then_done(self):
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "ls",
|
|
"exitCode": 0})])
|
|
self._set_task(status="completed")
|
|
status, events = self._read_sse(
|
|
"/tasks/%s/trace/stream" % self.task_id, self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual([e["type"] for e in events], ["snapshot", "done"])
|
|
self.assertEqual(events[0]["entries"][0]["id"], "c1")
|
|
self.assertEqual(events[1]["status"], "completed")
|
|
|
|
def test_stream_auth(self):
|
|
status, events = self._read_sse(
|
|
"/tasks/%s/trace/stream" % self.task_id, None)
|
|
self.assertEqual(status, 401)
|
|
self.assertEqual(events, [])
|
|
|
|
def test_agents_auth_and_not_found(self):
|
|
status, _ = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id)
|
|
self.assertEqual(status, 401)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.other_key)
|
|
self.assertEqual(status, 403)
|
|
status, _ = self._req(
|
|
"GET", "/tasks/nope/agents", key=self.owner_key)
|
|
self.assertEqual(status, 404)
|
|
|
|
def test_agents_chain_groups_calls_by_agent(self):
|
|
self._set_task(status="running")
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "thread": "th-main",
|
|
"command": "make", "ts": 1000}),
|
|
json.dumps({"kind": "tool", "id": "t1", "thread": "th-sub",
|
|
"server": "s", "tool": "read", "ts": 1001}),
|
|
json.dumps({"kind": "agent-start", "id": "agent:rev",
|
|
"name": "rev", "prompt": "review code",
|
|
"trace": "rev.jsonl", "ts": 1002}),
|
|
json.dumps({"kind": "agent-done", "id": "agent:rev:done",
|
|
"name": "rev", "status": "completed",
|
|
"trace": "rev.jsonl", "ts": 1004}),
|
|
])
|
|
with open(os.path.join(self.trace_dir, "rev.jsonl"),
|
|
"w", encoding="utf-8") as f:
|
|
f.write(json.dumps({"kind": "exec", "id": "c9",
|
|
"command": "grep", "ts": 1003}) + "\n")
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data["status"], "running")
|
|
by_kind = {a["kind"]: a for a in data["agents"]}
|
|
main = by_kind["main"]
|
|
self.assertEqual(main["status"], "running")
|
|
self.assertEqual([e["id"] for e in main["calls"]], ["c1"])
|
|
model = by_kind["model"]
|
|
self.assertEqual(model["status"], "running")
|
|
self.assertEqual([e["id"] for e in model["calls"]], ["t1"])
|
|
delegate = [a for a in data["agents"]
|
|
if a["kind"] == "delegate"][0]
|
|
self.assertEqual(delegate["name"], "rev")
|
|
self.assertEqual(delegate["status"], "completed")
|
|
self.assertEqual(delegate["prompt"], "review code")
|
|
self.assertEqual([e["id"] for e in delegate["calls"]], ["c9"])
|
|
self.assertEqual(delegate["updated_ts"], 1004)
|
|
|
|
def test_agents_delegate_without_done_stays_running(self):
|
|
# 委派进程与父任务相互独立：父任务已结束、无结束标记时，
|
|
# 子任务仍算进行中（不能谎报它已中断）。
|
|
self._set_task(status="completed")
|
|
self._write_trace([
|
|
json.dumps({"kind": "agent-start", "id": "agent:rev",
|
|
"name": "rev", "prompt": "review",
|
|
"trace": "rev.jsonl", "ts": 1002}),
|
|
])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
delegate = [a for a in data["agents"]
|
|
if a["kind"] == "delegate"][0]
|
|
self.assertEqual(delegate["status"], "running")
|
|
self.assertEqual(data["agents"][0]["status"], "completed")
|
|
|
|
def test_agents_missing_file_is_main_only(self):
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(len(data["agents"]), 1)
|
|
self.assertEqual(data["agents"][0]["kind"], "main")
|
|
self.assertEqual(data["agents"][0]["calls"], [])
|
|
|
|
def test_tenant_trace_dir_mapping(self):
|
|
class S:
|
|
tenants_root = "/tenants"
|
|
self.assertEqual(
|
|
_tenant_trace_dir(S(), "abc-123"),
|
|
"/tenants/abc-123/work/host-traces")
|
|
self.assertIsNone(_tenant_trace_dir(S(), None))
|
|
self.assertIsNone(_tenant_trace_dir(S(), ""))
|
|
self.assertIsNone(_tenant_trace_dir(S(), ".."))
|
|
self.assertIsNone(_tenant_trace_dir(S(), "..."))
|
|
|
|
def test_trace_reads_tenant_dir_only(self):
|
|
# 租户 Docker 任务的链路文件只在租户目录里：旧版只读
|
|
# 共享目录会误报为空，修后 /trace 与 /agents 都看得见。
|
|
self._write_tenant_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "make"}),
|
|
])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual([e["id"] for e in data["entries"]], ["c1"])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual([e["id"] for e in data["agents"][0]["calls"]],
|
|
["c1"])
|
|
|
|
def test_trace_merges_tenant_and_shared(self):
|
|
self._write_tenant_trace([
|
|
json.dumps({"kind": "exec", "id": "c1", "command": "tenant"}),
|
|
])
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c2", "command": "shared"}),
|
|
])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual([e["id"] for e in data["entries"]], ["c1", "c2"])
|
|
# 同 id 冲突时后读的共享目录覆盖（顺序有文档、可复现）。
|
|
self._write_tenant_trace([
|
|
json.dumps({"kind": "exec", "id": "c1",
|
|
"command": "tenant-cmd"}),
|
|
])
|
|
self._write_trace([
|
|
json.dumps({"kind": "exec", "id": "c1",
|
|
"command": "shared-cmd"}),
|
|
])
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/trace" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(data["entries"][0]["command"], "shared-cmd")
|
|
|
|
def test_agents_tenant_child_file(self):
|
|
# 委派子任务的子链路文件也在租户目录里：父标记的
|
|
# trace 名要在租户目录里找得到，子任务调用才进链。
|
|
self._write_tenant_trace([
|
|
json.dumps({"kind": "agent-start", "id": "agent:rev",
|
|
"name": "rev", "prompt": "review",
|
|
"trace": "rev.jsonl", "ts": 1002}),
|
|
json.dumps({"kind": "agent-done", "id": "agent:rev:done",
|
|
"name": "rev", "status": "completed",
|
|
"trace": "rev.jsonl", "ts": 1004}),
|
|
])
|
|
self._write_tenant_trace([
|
|
json.dumps({"kind": "exec", "id": "c9", "command": "grep"}),
|
|
], filename="rev.jsonl")
|
|
status, data = self._req(
|
|
"GET", "/tasks/%s/agents" % self.task_id, key=self.owner_key)
|
|
self.assertEqual(status, 200)
|
|
delegate = [a for a in data["agents"]
|
|
if a["kind"] == "delegate"][0]
|
|
self.assertEqual(delegate["status"], "completed")
|
|
self.assertEqual([e["id"] for e in delegate["calls"]], ["c9"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|