From c4d3ec9bff677730b92b0f879c77a6cf24c69cd1 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 11:51:11 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=91=BD=E4=BB=A4=E5=8F=B0=E5=AD=90?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E9=93=BE=E8=B7=AF=EF=BC=88=E4=B8=BB=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1/=E6=A8=A1=E5=9E=8B=E5=AD=90=E4=BB=A3=E7=90=86/?= =?UTF-8?q?=E5=A7=94=E6=B4=BE=E5=AD=90=E4=BB=BB=E5=8A=A1=20+=20=E8=BF=87?= =?UTF-8?q?=E7=A8=8B=E8=B0=83=E7=94=A8=E5=AE=9E=E6=97=B6=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit host：链路条目带 thread 归属；delegate 向父链路文件写 agent-start/done 标记。服务端：GET /tasks/{id}/agents 聚合父链路 + 子链路文件成代理链（状态口径：委派以结束标记为准、无标记即进行中；模型子代理无结束信号时父任务运行中为进行中、否则未知）。命令台：新增子代理链路区，随 SSE 流防抖刷新；扁平链路也渲染 agent 标记。 --- cloud/console.html | 49 ++++++++++- cloud/server.py | 144 +++++++++++++++++++++++++++++++++ cloud/tests/test_console.py | 2 +- cloud/tests/test_task_trace.py | 73 +++++++++++++++++ host.py | 69 +++++++++++++--- tests/test_host_trace.py | 43 ++++++++++ 6 files changed, 366 insertions(+), 14 deletions(-) diff --git a/cloud/console.html b/cloud/console.html index c45bf88..a5f34c8 100644 --- a/cloud/console.html +++ b/cloud/console.html @@ -248,6 +248,8 @@
+

子代理链路

+
点任务行查看详情后，这里显示子代理链路

实时链路

点任务行查看详情后，这里显示过程链路
@@ -269,7 +271,7 @@ var $ = function (s) { return document.querySelector(s); }; var KEY_STORE = "codex-console-key"; var currentTask = null; var traceAbort = null; -var TRACE_KIND = { exec: "命令", file: "文件", call: "调用", tool: "工具", message: "消息", reasoning: "思考", plan: "计划" }; +var TRACE_KIND = { exec: "命令", file: "文件", call: "调用", tool: "工具", message: "消息", reasoning: "思考", plan: "计划", "agent-start": "子代理", "agent-done": "子代理" }; var TRACE_LIMIT = 1000; var TRACE_TERMINAL = ["completed", "failed", "interrupted", "merged", "merge-failed"]; @@ -485,6 +487,7 @@ async function loadTasks() { function stopTrace() { if (traceAbort) { try { traceAbort.abort(); } catch (e) {} traceAbort = null; } + if (agentTimer) { clearTimeout(agentTimer); agentTimer = null; } } function traceLine(e) { var k = TRACE_KIND[e.kind] || e.kind || "?"; @@ -496,6 +499,8 @@ function traceLine(e) { else if (e.kind === "tool") body = esc(e.server || "") + "/" + esc(e.tool || "") + " " + esc(e.status || ""); else if (e.kind === "call") body = esc(e.name || ""); else if (e.kind === "plan") body = esc(e.text || ""); + else if (e.kind === "agent-start") body = "开始：" + esc(e.name || "") + (e.prompt ? "（" + esc(e.prompt) + "）" : ""); + else if (e.kind === "agent-done") body = "结束：" + esc(e.name || "") + " " + esc(e.status || ""); else body = esc(JSON.stringify(e).slice(0, 300)); return "[" + k + "] " + body; } @@ -512,6 +517,44 @@ function renderTrace(entries, note, doneStatus) { if (doneStatus) html += "
链路已结束（" + esc(doneStatus) + "）"; box.innerHTML = html; } +var AGENT_STATUS = { running: "进行中", completed: "已完成", failed: "失败", interrupted: "已中断", unknown: "未知" }; +var AGENT_KIND = { main: "主任务", delegate: "委派子任务", model: "模型子代理" }; +var AGENT_CALL_SHOWN = 50; +var agentTimer = null; +function renderAgents(data) { + var box = $("#agentBox"); + var html = ""; + if (!data) { box.innerHTML = "（暂无子代理链路）"; return; } + if (data.note) html += "" + esc(data.note) + "
"; + var agents = data.agents || []; + if (!agents.length) html += "（暂无子代理）"; + agents.forEach(function (a) { + var st = AGENT_STATUS[a.status] || a.status || "?"; + html += "
"; + html += "" + esc(AGENT_KIND[a.kind] || a.kind || "") + "：" + esc(a.name || "") + + " " + esc(st) + "" + + " ｜ 过程调用 " + (a.calls ? a.calls.length : 0) + " 条" + + (a.updated_ts ? " ｜ 最近活动 " + fmt(a.updated_ts / 1000) : "") + "
"; + if (a.prompt) html += "任务：" + esc(a.prompt) + "
"; + if (a.calls && a.calls.length) { + var shown = a.calls.slice(-AGENT_CALL_SHOWN); + if (a.calls.length > AGENT_CALL_SHOWN) html += "（仅显示最近 " + AGENT_CALL_SHOWN + " 条，共 " + a.calls.length + " 条）
"; + html += shown.map(traceLine).join("
"); + } else html += "（暂无过程调用）"; + html += "
"; + }); + box.innerHTML = html; +} +async function loadAgents(taskId) { + try { + var d = await api("/tasks/" + encodeURIComponent(taskId) + "/agents", "GET"); + renderAgents(d); + } catch (e) { $("#agentBox").innerHTML = "子代理链路加载失败：" + esc(e.message) + ""; } +} +function scheduleAgents(taskId) { + if (agentTimer) clearTimeout(agentTimer); + agentTimer = setTimeout(function () { loadAgents(taskId); }, 1000); +} async function startTrace(taskId, status) { stopTrace(); var box = $("#traceBox"); @@ -561,10 +604,12 @@ async function startTrace(taskId, status) { entries.forEach(function (e, i) { index[e.id] = i; }); note = m.note || ""; renderTrace(entries, note, ""); + scheduleAgents(taskId); } else if (m.type === "entry") { var e = m.entry; if (e.id in index) entries[index[e.id]] = e; else { index[e.id] = entries.length; entries.push(e); } renderTrace(entries, note, ""); + scheduleAgents(taskId); } else if (m.type === "done") { done = true; doneStatus = m.status; renderTrace(entries, note, doneStatus); @@ -618,6 +663,8 @@ async function showDetail(taskId) { $("#continueMsgSpan").textContent = ""; $("#continueForm").style.display = (TRACE_TERMINAL.indexOf(t.status) >= 0 && t.status !== "merged") ? "block" : "none"; startTrace(taskId, t.status); + $("#agentBox").innerHTML = "子代理链路加载中…"; + loadAgents(taskId); loadFollowups(taskId); } catch (e) { $("#detail").innerHTML = "加载失败：" + esc(e.message) + ""; } } diff --git a/cloud/server.py b/cloud/server.py index 85c0880..a1d1f5f 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -184,6 +184,127 @@ def _read_trace_entries(path): except OSError: return [] return entries + + +AGENT_CALL_LIMIT = 1000 + + +def _read_agent_entries(path): + """读链路文件供子代理聚合：按 (thread, id) 去重，后到的完成态 + 覆盖先到的开始态，保首次顺序；坏行与无 id 行跳过；文件不 + 存在为空。同一 item id 在不同线程出现时不互相覆盖（模型内 + 子代理与主任务各有自己的线程）。""" + entries = [] + index = {} + try: + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except ValueError: + continue + if not isinstance(entry, dict) or "id" not in entry: + continue + key = (entry.get("thread") or "", entry["id"]) + if key in index: + entries[index[key]] = entry + else: + index[key] = len(entries) + entries.append(entry) + except OSError: + return [] + return entries + + +def _latest_ts(values): + vals = [v for v in values if isinstance(v, (int, float))] + return max(vals) if vals else None + + +def _agent_status_from_task(status): + return {"queued": "running", "running": "running", + "completed": "completed", "merged": "completed", + "failed": "failed", "merge-failed": "failed", + "interrupted": "interrupted"}.get(status, "unknown") + + +def _build_agent_chain(task, trace_dir): + """任务 → 子代理链：主任务 + 模型内子代理（同文件多线程）+ + 委派子任务（父链路里的 agent-start/done 标记 + 子链路文件）。 + + 状态诚实口径：委派子任务以 agent-done 标记为准，无标记即 + 进行中——委派进程与父任务相互独立，父任务结束不代表子任务 + 停；模型内子代理无明确结束信号，父任务运行中即进行中， + 否则未知。无 thread 的旧条目一律归主任务。 + """ + parent_path = os.path.join( + trace_dir, _trace_file_name(task["id"]) + ".jsonl") + entries = _read_agent_entries(parent_path) + calls = [e for e in entries + if e.get("kind") not in ("agent-start", "agent-done")] + main_thread = None + for e in calls: + if e.get("thread"): + main_thread = e["thread"] + break + root_calls = [e for e in calls + if not e.get("thread") or e.get("thread") == main_thread] + agents = [{ + "id": "main", "name": "主任务", "kind": "main", "parent": None, + "status": _agent_status_from_task(task.get("status")), + "prompt": None, "trace": None, + "calls": root_calls[:AGENT_CALL_LIMIT], + "updated_ts": _latest_ts([e.get("ts") for e in root_calls]), + }] + task_running = task.get("status") in ("queued", "running") + threads = {} + for e in calls: + t = e.get("thread") + if t and t != main_thread: + threads.setdefault(t, []).append(e) + for t, es in threads.items(): + agents.append({ + "id": t, "name": "模型子代理 " + t[:12], "kind": "model", + "parent": "main", + "status": "running" if task_running else "unknown", + "prompt": None, "trace": None, + "calls": es[:AGENT_CALL_LIMIT], + "updated_ts": _latest_ts([e.get("ts") for e in es]), + }) + starts = {} + dones = {} + for e in entries: + if e.get("kind") == "agent-start" and e.get("name"): + starts[e["name"]] = e + elif e.get("kind") == "agent-done" and e.get("name"): + dones[e["name"]] = e + for name, s in starts.items(): + 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) + if e.get("kind") not in + ("agent-start", "agent-done")] + done = dones.get(name) + status = "running" + if done and done.get("status") in ( + "completed", "failed", "interrupted"): + status = done["status"] + agents.append({ + "id": "delegate:" + name, "name": name, "kind": "delegate", + "parent": "main", "status": status, + "prompt": s.get("prompt"), "trace": trace_name, + "calls": child_calls[:AGENT_CALL_LIMIT], + "updated_ts": _latest_ts( + [e.get("ts") for e in child_calls] + + [s.get("ts"), done.get("ts") if done else None]), + }) + return agents DEFAULT_SWEEP_THRESHOLD_SECONDS = 300 # 限流默认值：每 IP / 每 API key 每分钟 120 次；环境变量 # CLOUD_RATE_LIMIT_PER_MINUTE / CLOUD_RATE_LIMIT_WINDOW_SECONDS 可调。 @@ -1262,6 +1383,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): user_id, rest[: -len("/trace/stream")]) elif rest.endswith("/trace") and method == "GET": self._task_trace(user_id, rest[: -len("/trace")]) + elif rest.endswith("/agents") and method == "GET": + self._task_agents(user_id, rest[: -len("/agents")]) elif rest.endswith("/cancel") and method == "POST": self._cancel_task(user_id, rest[: -len("/cancel")]) elif rest.endswith("/merge") and method == "POST": @@ -2471,6 +2594,27 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(200, {"task_id": task_id, "status": task["status"], "entries": entries, "note": note}) + def _task_agents(self, user_id, task_id): + 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 + worktree_host = task.get("worktree_host") + if worktree_host and worktree_host != socket.gethostname(): + self._send_json(200, {"task_id": task_id, "status": task["status"], + "agents": [], + "note": ("链路文件在 worktree 主机 %s 上，本机读不到" + % worktree_host)}) + return + agents = _build_agent_chain(task, _trace_dir_for(self.server)) + self._send_json(200, {"task_id": task_id, "status": task["status"], + "agents": agents, "note": ""}) + def _task_trace_stream(self, user_id, task_id): row = self._fetch_task(task_id) if row is None: diff --git a/cloud/tests/test_console.py b/cloud/tests/test_console.py index 609dfac..8af4d33 100644 --- a/cloud/tests/test_console.py +++ b/cloud/tests/test_console.py @@ -64,7 +64,7 @@ class ConsoleTestCase(unittest.TestCase): self.assertIn("Codex 命令台", text) # 壳必须引用它实际调的接口，防 JS 与服务端路由漂移。 for marker in ("/health", "/projects", "/tasks", "/workers", - "X-API-Key"): + "X-API-Key", "agentBox", "/agents"): self.assertIn(marker, text) def test_console_trailing_slash_same_page(self): diff --git a/cloud/tests/test_task_trace.py b/cloud/tests/test_task_trace.py index adaf6e1..663ee2a 100644 --- a/cloud/tests/test_task_trace.py +++ b/cloud/tests/test_task_trace.py @@ -173,6 +173,79 @@ class TaskTraceTestCase(unittest.TestCase): 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"], []) + if __name__ == "__main__": unittest.main() diff --git a/host.py b/host.py index 6a475be..b405a43 100644 --- a/host.py +++ b/host.py @@ -214,6 +214,8 @@ def _summarize_trace_item(method, params): 返回 [] 表示该通知不进链路（userMessage 等）。 ts 取通知的 completedAtMs/startedAtMs（毫秒 epoch）， 没有时为 None，服务端展示时按无时间处理。 + thread 取通知的 threadId（模型内子代理与主任务同文件时， + 服务端凭它把条目归属到不同代理；没有时为 None，归主任务）。 """ item = (params or {}).get("item") if not isinstance(item, dict): @@ -221,8 +223,11 @@ def _summarize_trace_item(method, params): item_type = item.get("type") item_id = item.get("id") or "" ts = params.get("completedAtMs") or params.get("startedAtMs") + thread = params.get("threadId") + if not thread and isinstance(params.get("thread"), dict): + thread = params["thread"].get("id") if item_type == "commandExecution": - return [{"kind": "exec", "id": item_id, "ts": ts, + return [{"kind": "exec", "id": item_id, "ts": ts, "thread": thread, "command": (item.get("command") or "")[:500], "cwd": item.get("cwd") or "", "exitCode": item.get("exitCode"), @@ -235,6 +240,7 @@ def _summarize_trace_item(method, params): path = ch.get("path") or "" entries.append({"kind": "file", "id": item_id + ":" + path, "ts": ts, + "thread": thread, "path": path, "change": ch.get("kind") or "", "status": item.get("status") or ""}) return entries @@ -242,32 +248,60 @@ def _summarize_trace_item(method, params): text = item.get("text") or "" if not text: return [] - return [{"kind": "message", "id": item_id, "ts": ts, + return [{"kind": "message", "id": item_id, "ts": ts, "thread": thread, "phase": item.get("phase") or "", "text": text[:300]}] if item_type == "reasoning": summary = item.get("summary") or item.get("content") or [] text = " ".join(s for s in summary if isinstance(s, str)).strip() if not text: return [] - return [{"kind": "reasoning", "id": item_id, "ts": ts, + return [{"kind": "reasoning", "id": item_id, "ts": ts, "thread": thread, "summary": text[:200]}] if item_type == "mcpToolCall": - return [{"kind": "tool", "id": item_id, "ts": ts, + return [{"kind": "tool", "id": item_id, "ts": ts, "thread": thread, "server": item.get("server") or "", "tool": item.get("tool") or "", "status": item.get("status") or ""}] if item_type == "functionCallOutput": - return [{"kind": "call", "id": item_id, "ts": ts, + return [{"kind": "call", "id": item_id, "ts": ts, "thread": thread, "name": item.get("name") or ""}] if item_type == "plan": text = item.get("text") or "" if not text: return [] - return [{"kind": "plan", "id": item_id, "ts": ts, + return [{"kind": "plan", "id": item_id, "ts": ts, "thread": thread, "text": text[:300]}] return [] +def _trace_file_path(state_path, task_name): + """链路文件路径：state 文件同目录下的 host-traces/<清洗后名字>.jsonl。 + + 与 Server.set_trace_file、cloud 的 _trace_file_name 同款清洗， + 三边算出同一名字，delegate 的父链路标记才能落进父任务文件。 + """ + safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name)) + return os.path.join( + os.path.dirname(os.path.abspath(state_path)), + "host-traces", safe + ".jsonl") + + +def _append_trace_marker(state_path, task_name, entry): + """往另一任务的链路文件追加一条标记（delegate 向父链路报到用）。 + + 尽力而为：目录建不出、写失败都不影响回合本身。父链路的 + 行数上限由属主进程自己的计数器管，这里不重复计数（标记 + 一轮至多两条，超限风险可忽略）。 + """ + try: + path = _trace_file_path(state_path, task_name) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + except Exception: + pass + + class Server: """单个 app-server 进程，多 thread 并发。""" @@ -302,12 +336,8 @@ class Server: 时 trace_path 保持 None，不影响回合。 """ try: - safe = re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_name)) - trace_dir = os.path.join( - os.path.dirname(os.path.abspath(state_path)), - "host-traces") - os.makedirs(trace_dir, exist_ok=True) - self.trace_path = os.path.join(trace_dir, safe + ".jsonl") + self.trace_path = _trace_file_path(state_path, task_name) + os.makedirs(os.path.dirname(self.trace_path), exist_ok=True) self._trace_lines = 0 except Exception: self.trace_path = None @@ -2566,11 +2596,26 @@ def main(): "created_at": time.time(), "turns": []} state["tasks"][args.name] = sub save_state(args.state, state) + # 子代理链路：向父任务链路文件报到（开始 + 结束各一条）， + # 服务端凭标记把子链路文件拼进父任务的代理链；子任务自己 + # 的过程条目仍写它自己的链路文件（main 起动时按 --name 定）。 + child_trace = (os.path.basename(srv.trace_path) + if srv.trace_path else None) + _append_trace_marker( + args.state, args.parent_task, + {"kind": "agent-start", "id": "agent:" + args.name, + "name": args.name, "prompt": (args.task or "")[:300], + "trace": child_trace, "ts": int(time.time() * 1000)}) result = run_task_turn(srv, state, sub, args.task, args.model, args.effort, args.sandbox, timeout=args.turn_timeout, interrupt_file=_interrupt_request_path(args.state, args.name)) save_state(args.state, state) + _append_trace_marker( + args.state, args.parent_task, + {"kind": "agent-done", "id": "agent:" + args.name + ":done", + "name": args.name, "status": sub["status"], + "trace": child_trace, "ts": int(time.time() * 1000)}) print(json.dumps({"sub_task": args.name, "status": sub["status"], "text": result.get("text", "")}, indent=2, ensure_ascii=False)) diff --git a/tests/test_host_trace.py b/tests/test_host_trace.py index 79f7474..300157b 100644 --- a/tests/test_host_trace.py +++ b/tests/test_host_trace.py @@ -6,6 +6,7 @@ import json import os +import shutil import sys import tempfile import unittest @@ -130,6 +131,48 @@ class TraceFileTestCase(unittest.TestCase): self.assertEqual(len(lines), 1) self.assertEqual(json.loads(lines[0])["id"], "c1") + def test_summarize_carries_thread(self): + # 模型内子代理与主任务同文件时，服务端凭 thread 归属条目。 + entries = _summarize_trace_item( + "item/completed", + {"item": {"type": "commandExecution", "id": "c1", + "command": "ls"}, + "threadId": "th-sub", "completedAtMs": 1700000000000}) + self.assertEqual(entries[0]["thread"], "th-sub") + entries = _summarize_trace_item( + "item/completed", + {"item": {"type": "commandExecution", "id": "c2", + "command": "ls"}, + "completedAtMs": 1700000000000}) + self.assertIsNone(entries[0]["thread"]) + + def test_append_trace_marker_to_parent_file(self): + # delegate 向父链路报到：路径清洗与 set_trace_file 一致， + # 写失败不抛（尽力而为，不影响回合）。 + from host import _append_trace_marker, _trace_file_path + with tempfile.TemporaryDirectory() as d: + state_path = os.path.join(d, "host-state.json") + self.assertEqual( + _trace_file_path(state_path, "abc/def?id=1"), + os.path.join(d, "host-traces", "abc-def-id-1.jsonl")) + _append_trace_marker( + state_path, "parent task!", + {"kind": "agent-start", "id": "agent:rev", + "name": "rev", "prompt": "review"}) + path = os.path.join(d, "host-traces", "parent-task-.jsonl") + with open(path, encoding="utf-8") as f: + entry = json.loads(f.read().strip()) + self.assertEqual(entry["kind"], "agent-start") + self.assertEqual(entry["name"], "rev") + # 目录被文件占住时也不抛。 + blocker = os.path.join(d, "host-traces") + shutil.rmtree(blocker) + with open(blocker, "w", encoding="utf-8") as f: + f.write("x") + _append_trace_marker(state_path, "parent task!", + {"kind": "agent-done", "id": "agent:rev:done", + "name": "rev", "status": "completed"}) + if __name__ == "__main__": unittest.main()