Merge pull request #83 from openqj/feat-tenant-trace

fix: 租户 Docker 任务链路可见（租户目录优先、共享目录兜底）

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),

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()