diff --git a/cloud/server.py b/cloud/server.py index a1d1f5f..e2109b2 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -158,6 +158,49 @@ def _trace_dir_for(server): return os.environ.get("CODEX_TRACE_DIR") or TRACE_DEFAULT_DIR +TENANTS_DEFAULT_ROOT = "/var/lib/codex-app-server/tenants" + + +def _tenants_root_for(server): + """租户根目录：server 属性（单测注入）> 环境变量 > 默认路径。""" + d = getattr(server, "tenants_root", None) + if d: + return str(d) + return os.environ.get("CODEX_TENANTS_ROOT") or TENANTS_DEFAULT_ROOT + + +def _tenant_trace_dir(server, project_id): + """项目对应的租户链路目录（租户 Docker 任务的链路文件落在这里）。 + + 目录名清洗与 worker._tenant_safe_name 同规则；项目 id 映射 + 不出安全名时返回 None（只读共享目录）。 + """ + if not project_id: + return None + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(project_id)).strip(".") + if not safe or safe in (".", ".."): + return None + return os.path.join(_tenants_root_for(server), safe, "work", + "host-traces") + + +def _trace_dirs_for_task(server, task): + """任务的链路目录列表：租户目录优先、共享目录兜底。 + + 任务可能在不同执行器上跑过（重试换执行器），两个目录的 + 文件都读、按 id / (thread, id) 去重合并；通常只有一个目录 + 里有文件，行为与以前完全一致。 + """ + dirs = [] + tenant = _tenant_trace_dir(server, task.get("project_id")) + if tenant: + dirs.append(tenant) + shared = _trace_dir_for(server) + if shared not in dirs: + dirs.append(shared) + return dirs + + def _read_trace_entries(path): """读 host 写的 jsonl 链路文件：跳过坏行；按 id 去重，后到的 完成态覆盖先到的开始态，保首次顺序。文件不存在返回空列表。""" @@ -186,6 +229,21 @@ def _read_trace_entries(path): return entries +def _read_trace_entries_multi(paths): + """多文件合并读（flat trace 用）：按 id 去重，后到的覆盖先到的， + 保首次顺序；文件不存在视为空。""" + entries = [] + index = {} + for path in paths: + for e in _read_trace_entries(path): + if e["id"] in index: + entries[index[e["id"]]] = e + else: + index[e["id"]] = len(entries) + entries.append(e) + return entries + + AGENT_CALL_LIMIT = 1000 @@ -219,9 +277,27 @@ def _read_agent_entries(path): return entries +def _read_agent_entries_multi(paths): + """多文件合并读（agents 聚合用）：按 (thread, id) 去重，后到的 + 覆盖先到的，保首次顺序；文件不存在视为空。""" + entries = [] + index = {} + for path in paths: + for e in _read_agent_entries(path): + key = (e.get("thread") or "", e["id"]) + if key in index: + entries[index[key]] = e + else: + index[key] = len(entries) + entries.append(e) + return entries + + def _latest_ts(values): vals = [v for v in values if isinstance(v, (int, float))] return max(vals) if vals else None + vals = [v for v in values if isinstance(v, (int, float))] + return max(vals) if vals else None def _agent_status_from_task(status): @@ -231,18 +307,20 @@ def _agent_status_from_task(status): "interrupted": "interrupted"}.get(status, "unknown") -def _build_agent_chain(task, trace_dir): +def _build_agent_chain(task, trace_dirs): """任务 → 子代理链：主任务 + 模型内子代理（同文件多线程）+ 委派子任务（父链路里的 agent-start/done 标记 + 子链路文件）。 + trace_dirs 是 _trace_dirs_for_task 给的目录列表（租户优先、 + 共享兜底）；父链路与子链路文件都在这些目录里找、合并读。 状态诚实口径：委派子任务以 agent-done 标记为准，无标记即 进行中——委派进程与父任务相互独立，父任务结束不代表子任务 停；模型内子代理无明确结束信号，父任务运行中即进行中， 否则未知。无 thread 的旧条目一律归主任务。 """ - parent_path = os.path.join( - trace_dir, _trace_file_name(task["id"]) + ".jsonl") - entries = _read_agent_entries(parent_path) + parent_paths = [os.path.join(d, _trace_file_name(task["id"]) + ".jsonl") + for d in trace_dirs] + entries = _read_agent_entries_multi(parent_paths) calls = [e for e in entries if e.get("kind") not in ("agent-start", "agent-done")] main_thread = None @@ -285,9 +363,9 @@ def _build_agent_chain(task, trace_dir): trace_name = s.get("trace") or dones.get(name, {}).get("trace") child_calls = [] if trace_name: - child_path = os.path.join( - trace_dir, os.path.basename(trace_name)) - child_calls = [e for e in _read_agent_entries(child_path) + child_paths = [os.path.join(d, os.path.basename(trace_name)) + for d in trace_dirs] + child_calls = [e for e in _read_agent_entries_multi(child_paths) if e.get("kind") not in ("agent-start", "agent-done")] done = dones.get(name) @@ -2570,15 +2648,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): # -- 任务实时链路 -------------------------------------------------- def _trace_snapshot(self, task): """当前链路快照 (entries, note)：worktree 在别机时读不到 - 文件，返回空链路 + 说明（多机诚实降级，不谎报无过程）。""" + 文件，返回空链路 + 说明（多机诚实降级，不谎报无过程）。 + 租户目录与共享目录都读、合并（租户 Docker 任务的链路 + 文件只在租户目录里，旧版只读共享目录会误报为空）。""" worktree_host = task.get("worktree_host") if worktree_host and worktree_host != socket.gethostname(): return [], ("链路文件在 worktree 主机 %s 上，本机读不到" % worktree_host) - path = os.path.join( - _trace_dir_for(self.server), - _trace_file_name(task["id"]) + ".jsonl") - return _read_trace_entries(path), "" + paths = [os.path.join(d, _trace_file_name(task["id"]) + ".jsonl") + for d in _trace_dirs_for_task(self.server, task)] + return _read_trace_entries_multi(paths), "" def _task_trace(self, user_id, task_id): row = self._fetch_task(task_id) @@ -2611,7 +2690,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "note": ("链路文件在 worktree 主机 %s 上，本机读不到" % worktree_host)}) return - agents = _build_agent_chain(task, _trace_dir_for(self.server)) + agents = _build_agent_chain( + task, _trace_dirs_for_task(self.server, task)) self._send_json(200, {"task_id": task_id, "status": task["status"], "agents": agents, "note": ""}) @@ -2772,7 +2852,14 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): worktree_host = task.get("worktree_host") if worktree_host and worktree_host != socket.gethostname(): return + # 续写已有链路文件所在的目录：租户 Docker 任务的链路 + # 文件在租户目录里，续写共享目录会把一条任务的链路 + # 拆成两处；还没有任何链路文件时默认写共享目录。 d = _trace_dir_for(self.server) + tenant = _tenant_trace_dir(self.server, task.get("project_id")) + if tenant and os.path.exists(os.path.join( + tenant, _trace_file_name(task["id"]) + ".jsonl")): + d = tenant os.makedirs(d, exist_ok=True) entry = {"kind": "message", "id": "fu-" + fu_id, "ts": int(time.time() * 1000), diff --git a/cloud/tests/test_task_trace.py b/cloud/tests/test_task_trace.py index 663ee2a..95b6b85 100644 --- a/cloud/tests/test_task_trace.py +++ b/cloud/tests/test_task_trace.py @@ -18,11 +18,11 @@ from pathlib import Path try: from cloud.auth import AuthStore - from cloud.server import create_server + 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 + from cloud.server import create_server, _tenant_trace_dir class TaskTraceTestCase(unittest.TestCase): @@ -31,11 +31,14 @@ class TaskTraceTestCase(unittest.TestCase): 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) @@ -93,6 +96,15 @@ class TaskTraceTestCase(unittest.TestCase): 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) @@ -246,6 +258,79 @@ class TaskTraceTestCase(unittest.TestCase): 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()