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