From 799f7c3edbe336f977b4a34d3b7239a3dc5dd7b4 Mon Sep 17 00:00:00 2001
From: openqj <openqj@gmail.com>
Date: Tue, 27 Oct 2026 02:27:15 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20=E4=BB=BB=E5=8A=A1=E5=AE=9E=E6=97=B6?=
 =?UTF-8?q?=E9=93=BE=E8=B7=AF=20SSE=20(#61)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

host.py：回合通知摘要成链路条目，增量追加到 <state>/host-traces/<task>.jsonl（上限 5000 行）；dispatch/run/goal/delegate 设链路文件
cloud/server.py：GET /tasks/{id}/trace 快照 + /trace/stream SSE（snapshot → entry → done；worktree 在别机时空链路 + 说明）
console：详情链路区，运行中 fetch 流式渲染（EventSource 不能带 X-API-Key 头），终态走快照；DOM 上限 1000 条
单测：test_host_trace 7 + test_task_trace 6；host 49、cloud 525、deploy 14 全过

Co-authored-by: host <host@test.local>
---
 README.md                      |   4 +
 cloud/console.html             | 106 ++++++++++++++++++++
 cloud/server.py                | 152 ++++++++++++++++++++++++++++
 cloud/tests/test_task_trace.py | 178 +++++++++++++++++++++++++++++++++
 host.py                        | 121 ++++++++++++++++++++++
 tests/test_host_trace.py       | 135 +++++++++++++++++++++++++
 6 files changed, 696 insertions(+)
 create mode 100644 cloud/tests/test_task_trace.py
 create mode 100644 tests/test_host_trace.py

diff --git a/README.md b/README.md
index e3c4a68..1e3efd7 100644
--- a/README.md
+++ b/README.md
@@ -234,6 +234,10 @@ python3 host.py daemon-stop
 - 云端控制面演示另起一端（不跑模型回合）：
   `python3 -m cloud.server`（默认 127.0.0.1:8765），
   除 `GET /health`、`/webhooks/*`、`/oauth/*` 外都要 `X-API-Key` 请求头。
+  任务实时链路：`GET /tasks/{id}/trace`（已结束任务的快照）与
+  `GET /tasks/{id}/trace/stream`（运行中任务的 SSE：先 snapshot 再逐条
+  entry，终态发 done 关流；命令台详情的链路区用 fetch 流式读它，
+  EventSource 不能带 X-API-Key 头故不用）。
 
 ## GitHub PR/CI 的外部条件（真实 PR 仍缺这两样）
 
diff --git a/cloud/console.html b/cloud/console.html
index c863f22..b6cbbf9 100644
--- a/cloud/console.html
+++ b/cloud/console.html
@@ -137,6 +137,8 @@
   <span id="actionMsg" class="muted"></span>
 </div>
 <pre id="detailBody" style="display:none"></pre>
+<h3>实时链路</h3>
+<div id="traceBox" class="muted">点任务行查看详情后，这里显示过程链路</div>
 <div id="costBox"></div>
 
 <script>
@@ -144,6 +146,10 @@
 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_LIMIT = 1000;
+var TRACE_TERMINAL = ["completed", "failed", "interrupted", "merged", "merge-failed"];
 
 function esc(s) {
   return String(s == null ? "" : s).replace(/&/g, "&amp;").replace(/</g, "&lt;")
@@ -256,7 +262,106 @@ async function loadTasks() {
   } catch (e) { $("#taskRows").innerHTML = "<tr><td colspan='7' class='err'>加载失败：" + esc(e.message) + "</td></tr>"; }
 }
 
+function stopTrace() {
+  if (traceAbort) { try { traceAbort.abort(); } catch (e) {} traceAbort = null; }
+}
+function traceLine(e) {
+  var k = TRACE_KIND[e.kind] || e.kind || "?";
+  var body = "";
+  if (e.kind === "exec") body = esc(e.command || "") + "（exit=" + esc(e.exitCode) + "，" + esc(e.durationMs) + "ms，cwd=" + esc(e.cwd || "") + "）";
+  else if (e.kind === "file") body = esc(e.path || "") + " " + esc(e.change || "") + " " + esc(e.status || "");
+  else if (e.kind === "message") body = (e.phase ? "[" + esc(e.phase) + "] " : "") + esc(e.text || "");
+  else if (e.kind === "reasoning") body = esc(e.summary || "");
+  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 body = esc(JSON.stringify(e).slice(0, 300));
+  return "[" + k + "] " + body;
+}
+function renderTrace(entries, note, doneStatus) {
+  var box = $("#traceBox");
+  var html = "";
+  if (note) html += "<span class='err'>" + esc(note) + "</span><br>";
+  if (!entries.length) html += "（暂无链路条目）";
+  else {
+    var shown = entries.slice(-TRACE_LIMIT);
+    if (entries.length > TRACE_LIMIT) html += "（仅显示最近 " + TRACE_LIMIT + " 条，共 " + entries.length + " 条）<br>";
+    html += shown.map(traceLine).join("<br>");
+  }
+  if (doneStatus) html += "<br><span class='muted'>链路已结束（" + esc(doneStatus) + "）</span>";
+  box.innerHTML = html;
+}
+async function startTrace(taskId, status) {
+  stopTrace();
+  var box = $("#traceBox");
+  if (TRACE_TERMINAL.indexOf(status) >= 0) {
+    try {
+      var d = await api("/tasks/" + encodeURIComponent(taskId) + "/trace", "GET");
+      renderTrace(d.entries || [], d.note || "", d.status);
+    } catch (e) { box.innerHTML = "<span class='err'>链路加载失败：" + esc(e.message) + "</span>"; }
+    return;
+  }
+  // 运行中：fetch 流式读 SSE（EventSource 不能带 X-API-Key 头，故不用它）。
+  var entries = [];
+  var index = {};
+  var note = "";
+  var done = false;
+  var doneStatus = "";
+  box.innerHTML = "链路连接中…";
+  var ctrl = new AbortController();
+  traceAbort = ctrl;
+  try {
+    var headers = {};
+    var key = getKey();
+    if (key) headers["X-API-Key"] = key;
+    var r = await fetch("/tasks/" + encodeURIComponent(taskId) + "/trace/stream", { headers: headers, signal: ctrl.signal });
+    if (!r.ok) {
+      var msg = "HTTP " + r.status;
+      try { var dj = await r.json(); if (dj && dj.error) msg = dj.error; } catch (e) {}
+      if (r.status === 401) msg += "（key 未填或无效）";
+      throw new Error(msg);
+    }
+    var reader = r.body.getReader();
+    var decoder = new TextDecoder();
+    var buf = "";
+    while (true) {
+      var chunk = await reader.read();
+      if (chunk.done) break;
+      buf += decoder.decode(chunk.value, { stream: true });
+      var parts = buf.split("\n\n");
+      buf = parts.pop();
+      parts.forEach(function (part) {
+        var line = part.split("\n").filter(function (l) { return l.indexOf("data: ") === 0; }).map(function (l) { return l.slice(6); }).join("\n");
+        if (!line) return;
+        try {
+          var m = JSON.parse(line);
+          if (m.type === "snapshot") {
+            entries = m.entries || []; index = {};
+            entries.forEach(function (e, i) { index[e.id] = i; });
+            note = m.note || "";
+            renderTrace(entries, note, "");
+          } 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, "");
+          } else if (m.type === "done") {
+            done = true; doneStatus = m.status;
+            renderTrace(entries, note, doneStatus);
+          }
+        } catch (e) {}
+      });
+      if (done) break;
+    }
+  } catch (e) {
+    if (e && e.name === "AbortError") return;
+    box.innerHTML = "<span class='err'>链路流中断：" + esc(e.message) + "</span>";
+    return;
+  }
+  if (done) { loadTasks(); showDetail(taskId); }
+}
+
 async function showDetail(taskId) {
+  stopTrace();
   try {
     var t = await api("/tasks/" + encodeURIComponent(taskId), "GET");
     currentTask = t;
@@ -280,6 +385,7 @@ async function showDetail(taskId) {
     $("#detailActions").style.display = "flex";
     $("#actionMsg").textContent = "";
     $("#costBox").innerHTML = "";
+    startTrace(taskId, t.status);
   } catch (e) { $("#detail").innerHTML = "<span class='err'>加载失败：" + esc(e.message) + "</span>"; }
 }
 
diff --git a/cloud/server.py b/cloud/server.py
index 1b52077..2ccd3f0 100644
--- a/cloud/server.py
+++ b/cloud/server.py
@@ -42,6 +42,7 @@ import html
 import json
 import logging
 import os
+import re
 import socket
 import sqlite3
 import threading
@@ -126,6 +127,53 @@ _server_log = logging.getLogger("cloud.server")
 TERMINAL_STATUSES = frozenset(
     {"completed", "failed", "interrupted", "merged", "merge-failed"}
 )
+# 实时链路（任务过程事件流）：host 在回合进行中把 app-server
+# 通知摘要成条目追加写 jsonl，服务端 tail 文件经 SSE 推送。
+TRACE_STREAM_MAX_SECONDS = 3600
+TRACE_STREAM_POLL_SECONDS = 1.0
+TRACE_DEFAULT_DIR = "/opt/codex-app-server/work/host-traces"
+
+
+def _trace_file_name(task_id):
+    """云任务 id -> 链路文件名：与 cloud.sandbox 的 _host_name
+    同款清洗（只保留 [A-Za-z0-9_.-]，其余替 -），两边算出同一名字。"""
+    return re.sub(r"[^A-Za-z0-9_.-]", "-", str(task_id))
+
+
+def _trace_dir_for(server):
+    """链路目录解析：server 属性（单测注入）> 环境变量 > 默认路径。"""
+    d = getattr(server, "trace_dir", None)
+    if d:
+        return str(d)
+    return os.environ.get("CODEX_TRACE_DIR") or TRACE_DEFAULT_DIR
+
+
+def _read_trace_entries(path):
+    """读 host 写的 jsonl 链路文件：跳过坏行；按 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
+                eid = entry["id"]
+                if eid in index:
+                    entries[index[eid]] = entry
+                else:
+                    index[eid] = len(entries)
+                    entries.append(entry)
+    except OSError:
+        return []
+    return entries
 DEFAULT_SWEEP_THRESHOLD_SECONDS = 300
 # 限流默认值：每 IP / 每 API key 每分钟 120 次；环境变量
 # CLOUD_RATE_LIMIT_PER_MINUTE / CLOUD_RATE_LIMIT_WINDOW_SECONDS 可调。
@@ -848,6 +896,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._get_task(user_id, rest)
             elif rest.endswith("/cost") and method == "GET":
                 self._task_cost(user_id, rest[: -len("/cost")])
+            elif rest.endswith("/trace/stream") and method == "GET":
+                self._task_trace_stream(
+                    user_id, rest[: -len("/trace/stream")])
+            elif rest.endswith("/trace") and method == "GET":
+                self._task_trace(user_id, rest[: -len("/trace")])
             elif rest.endswith("/cancel") and method == "POST":
                 self._cancel_task(user_id, rest[: -len("/cancel")])
             elif rest.endswith("/merge") and method == "POST":
@@ -1553,6 +1606,105 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 return
         self._send_json(200, task)
 
+    # -- 任务实时链路 --------------------------------------------------
+    def _trace_snapshot(self, task):
+        """当前链路快照 (entries, note)：worktree 在别机时读不到
+        文件，返回空链路 + 说明（多机诚实降级，不谎报无过程）。"""
+        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), ""
+
+    def _task_trace(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
+        entries, note = self._trace_snapshot(task)
+        self._send_json(200, {"task_id": task_id, "status": task["status"],
+                              "entries": entries, "note": note})
+
+    def _task_trace_stream(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
+        entries, note = self._trace_snapshot(task)
+        try:
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.send_header("Cache-Control", "no-cache")
+            self.send_header("Connection", "keep-alive")
+            self.end_headers()
+        except (BrokenPipeError, ConnectionResetError):
+            return
+        # SSE 无 Content-Length，done 后靠关连接界定流结束；注意
+        # send_header("Connection", "keep-alive") 会把
+        # close_connection 改回 False，所以必须在 end_headers 之后
+        # 再置 True（CPython send_header 的 Connection 分支行为）。
+        self.close_connection = True
+        sent = {}
+
+        def _emit(obj):
+            try:
+                self.wfile.write(
+                    ("data: " + json.dumps(obj, ensure_ascii=False,
+                                           default=_json_default) +
+                     "\n\n").encode("utf-8"))
+                self.wfile.flush()
+            except (BrokenPipeError, ConnectionResetError):
+                return False
+            return True
+
+        if not _emit({"type": "snapshot", "task_id": task_id,
+                      "status": task["status"], "entries": entries,
+                      "note": note}):
+            return
+        for entry in entries:
+            sent[entry["id"]] = json.dumps(entry, sort_keys=True,
+                                           ensure_ascii=False)
+        if task["status"] in TERMINAL_STATUSES:
+            _emit({"type": "done", "status": task["status"]})
+            return
+        deadline = time.time() + TRACE_STREAM_MAX_SECONDS
+        try:
+            while time.time() < deadline:
+                time.sleep(TRACE_STREAM_POLL_SECONDS)
+                row = self._fetch_task(task_id)
+                if row is None:
+                    _emit({"type": "done", "status": "deleted"})
+                    return
+                task = _task_to_dict(row)
+                entries, _note = self._trace_snapshot(task)
+                for entry in entries:
+                    blob = json.dumps(entry, sort_keys=True,
+                                      ensure_ascii=False)
+                    if sent.get(entry["id"]) != blob:
+                        sent[entry["id"]] = blob
+                        if not _emit({"type": "entry", "entry": entry}):
+                            return
+                if task["status"] in TERMINAL_STATUSES:
+                    _emit({"type": "done", "status": task["status"]})
+                    return
+        except (BrokenPipeError, ConnectionResetError):
+            return
+        _emit({"type": "done", "status": task["status"],
+               "reason": "timeout"})
+
     def _cancel_task(self, user_id, task_id):
         row = self._fetch_task(task_id)
         if row is None:
diff --git a/cloud/tests/test_task_trace.py b/cloud/tests/test_task_trace.py
new file mode 100644
index 0000000..adaf6e1
--- /dev/null
+++ b/cloud/tests/test_task_trace.py
@@ -0,0 +1,178 @@
+"""任务实时链路（服务端）单测：GET /tasks/{id}/trace 与 /trace/stream。
+
+覆盖：鉴权（401/403/404）、快照去重与坏行跳过、worktree 在别机
+时的空链路 + 说明、已结束任务的流（snapshot 后立刻 done 关流）。
+不跑真实模型，不启动 worker；链路目录经 server.trace_dir 注入。
+"""
+
+import http.client
+import json
+import os
+import socket
+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
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.auth import AuthStore
+    from cloud.server import create_server
+
+
+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.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.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 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, [])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/host.py b/host.py
index 6774c5b..6e3115a 100644
--- a/host.py
+++ b/host.py
@@ -118,6 +118,72 @@ def default_codex_bin():
     return "codex"
 
 
+# 实时链路：单条通知里 item 的上限行数（防失控回合把文件写爆）。
+TRACE_MAX_LINES = 5000
+
+
+def _summarize_trace_item(method, params):
+    """app-server 通知 → 链路条目列表（实时链路用）。
+
+    只取过程事实，不取大文本：命令截 500 字、消息截 300 字、
+    思考摘要截 200 字；fileChange 一项一变化拆多条。
+    返回 [] 表示该通知不进链路（userMessage 等）。
+    ts 取通知的 completedAtMs/startedAtMs（毫秒 epoch），
+    没有时为 None，服务端展示时按无时间处理。
+    """
+    item = (params or {}).get("item")
+    if not isinstance(item, dict):
+        return []
+    item_type = item.get("type")
+    item_id = item.get("id") or ""
+    ts = params.get("completedAtMs") or params.get("startedAtMs")
+    if item_type == "commandExecution":
+        return [{"kind": "exec", "id": item_id, "ts": ts,
+                 "command": (item.get("command") or "")[:500],
+                 "cwd": item.get("cwd") or "",
+                 "exitCode": item.get("exitCode"),
+                 "durationMs": item.get("durationMs")}]
+    if item_type == "fileChange":
+        entries = []
+        for ch in item.get("changes") or []:
+            if not isinstance(ch, dict):
+                continue
+            path = ch.get("path") or ""
+            entries.append({"kind": "file",
+                            "id": item_id + ":" + path, "ts": ts,
+                            "path": path, "change": ch.get("kind") or "",
+                            "status": item.get("status") or ""})
+        return entries
+    if item_type == "agentMessage":
+        text = item.get("text") or ""
+        if not text:
+            return []
+        return [{"kind": "message", "id": item_id, "ts": ts,
+                 "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,
+                 "summary": text[:200]}]
+    if item_type == "mcpToolCall":
+        return [{"kind": "tool", "id": item_id, "ts": ts,
+                 "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,
+                 "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,
+                 "text": text[:300]}]
+    return []
+
+
 class Server:
     """单个 app-server 进程，多 thread 并发。"""
 
@@ -133,8 +199,48 @@ class Server:
         self.events = []
         self.events_lock = threading.Lock()
         self.restart_count = 0
+        # 实时链路文件：set_trace_file 未调时为 None，回合本身
+        # 不受影响（写链路失败也绝不打断 _read_loop）。
+        self.trace_path = None
+        self._trace_lines = 0
         self._start()
 
+    def set_trace_file(self, state_path, task_name):
+        """链路文件定到 <state_dir>/host-traces/<task_name>.jsonl。
+
+        state_dir 与 state 文件同目录：本地 worker 是
+        /opt/codex-app-server/work，容器内是 /app/work（经挂载
+        落到宿主同一目录），服务端按同一规则读。目录建不出
+        时 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_lines = 0
+        except Exception:
+            self.trace_path = None
+
+    def _append_trace(self, msg):
+        """通知 → 摘要条目 → 追加写链路文件（每事件开合一次文件，
+        进程被杀也不丢已写行；行数到 TRACE_MAX_LINES 即停）。"""
+        if self.trace_path is None or \
+                self._trace_lines >= TRACE_MAX_LINES:
+            return
+        entries = _summarize_trace_item(msg.get("method"),
+                                        msg.get("params") or {})
+        if not entries:
+            return
+        with open(self.trace_path, "a", encoding="utf-8") as f:
+            for entry in entries:
+                if self._trace_lines >= TRACE_MAX_LINES:
+                    break
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+                self._trace_lines += 1
+
     def _start(self):
         env = dict(os.environ)
         if self.codex_home:
@@ -206,6 +312,11 @@ class Server:
                     thread_id = params["thread"].get("id")
                 with self.events_lock:
                     self.events.append(msg)
+                if self.trace_path is not None:
+                    try:
+                        self._append_trace(msg)
+                    except Exception:
+                        pass
                 if msg.get("method") == "turn/completed" and thread_id:
                     q = self.turn_queues.get(thread_id)
                     if q is not None:
@@ -1708,6 +1819,16 @@ def main():
         return
     srv = Server(codex_bin=args.codex_bin, codex_home=args.codex_home,
                  approval=args.approval)
+    # 实时链路：只给跑 turn 的命令设链路文件（dispatch/run/goal/
+    # delegate）；只读命令不设。名字映射与 worker 派发一致：
+    # dispatch/goal/delegate 用 --name，run 用 --task。
+    _trace_name = None
+    if args.cmd in ("dispatch", "goal", "delegate"):
+        _trace_name = getattr(args, "name", None)
+    elif args.cmd == "run":
+        _trace_name = getattr(args, "task", None)
+    if _trace_name:
+        srv.set_trace_file(args.state, _trace_name)
     try:
         srv.initialize()
         if args.cmd == "dispatch":
diff --git a/tests/test_host_trace.py b/tests/test_host_trace.py
new file mode 100644
index 0000000..79f7474
--- /dev/null
+++ b/tests/test_host_trace.py
@@ -0,0 +1,135 @@
+"""实时链路（host 侧）单测：通知摘要器 + 链路文件追加。
+
+不启动 app-server：Server 实例用 object.__new__ 绕开 __init__
+（__init__ 会拉起 codex 子进程），只测与进程无关的方法。
+"""
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+try:
+    import host
+    from host import Server, TRACE_MAX_LINES, _summarize_trace_item
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    import host
+    from host import Server, TRACE_MAX_LINES, _summarize_trace_item
+
+
+def _msg(item, ts=1700000000000):
+    return {"method": "item/completed",
+            "params": {"item": item, "completedAtMs": ts}}
+
+
+class SummarizeTraceItemTestCase(unittest.TestCase):
+    def test_exec_truncates_command(self):
+        entries = _summarize_trace_item(
+            "item/completed",
+            {"item": {"type": "commandExecution", "id": "c1",
+                      "command": "x" * 600, "cwd": "/w",
+                      "exitCode": 0, "durationMs": 12},
+             "completedAtMs": 1700000000000})
+        self.assertEqual(len(entries), 1)
+        self.assertEqual(entries[0]["kind"], "exec")
+        self.assertEqual(len(entries[0]["command"]), 500)
+        self.assertEqual(entries[0]["exitCode"], 0)
+        self.assertEqual(entries[0]["ts"], 1700000000000)
+
+    def test_file_change_split_per_change(self):
+        entries = _summarize_trace_item(
+            "item/completed",
+            {"item": {"type": "fileChange", "id": "f1", "status": "completed",
+                      "changes": [{"path": "a.py", "kind": "update"},
+                                  {"path": "b.py", "kind": "add"}]},
+             "startedAtMs": 1700000000001})
+        self.assertEqual([e["kind"] for e in entries], ["file", "file"])
+        self.assertEqual(entries[0]["id"], "f1:a.py")
+        self.assertEqual(entries[1]["path"], "b.py")
+
+    def test_message_reasoning_tool_call_plan(self):
+        cases = [
+            ({"type": "agentMessage", "id": "m1", "phase": "final",
+              "text": "done"}, "message"),
+            ({"type": "reasoning", "id": "r1",
+              "summary": ["a", "b"]}, "reasoning"),
+            ({"type": "mcpToolCall", "id": "t1", "server": "s",
+              "tool": "read", "status": "completed"}, "tool"),
+            ({"type": "functionCallOutput", "id": "o1",
+              "name": "read_file"}, "call"),
+            ({"type": "plan", "id": "p1", "text": "step"}, "plan"),
+        ]
+        for item, kind in cases:
+            entries = _summarize_trace_item("item/completed",
+                                            {"item": item})
+            self.assertEqual(len(entries), 1)
+            self.assertEqual(entries[0]["kind"], kind)
+
+    def test_skipped_notifications(self):
+        self.assertEqual(_summarize_trace_item(
+            "item/completed", {"item": {"type": "userMessage", "id": "u1",
+                                        "text": "hi"}}), [])
+        self.assertEqual(_summarize_trace_item(
+            "item/completed", {"item": {"type": "agentMessage", "id": "m2",
+                                        "text": ""}}), [])
+        self.assertEqual(_summarize_trace_item("item/completed", {}), [])
+        self.assertEqual(_summarize_trace_item(
+            "item/completed", {"item": {"type": "unknownThing",
+                                        "id": "z"}}), [])
+
+
+class TraceFileTestCase(unittest.TestCase):
+    def _bare_server(self):
+        srv = object.__new__(Server)
+        srv.trace_path = None
+        srv._trace_lines = 0
+        return srv
+
+    def test_set_trace_file_and_append(self):
+        with tempfile.TemporaryDirectory() as d:
+            state_path = os.path.join(d, "host-state.json")
+            srv = self._bare_server()
+            srv.set_trace_file(state_path, "abc/def?id=1")
+            self.assertEqual(
+                srv.trace_path,
+                os.path.join(d, "host-traces", "abc-def-id-1.jsonl"))
+            srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
+                                    "command": "ls", "cwd": "/w",
+                                    "exitCode": 0, "durationMs": 3}))
+            srv._append_trace(_msg({"type": "userMessage", "id": "u1",
+                                    "text": "hi"}))
+            with open(srv.trace_path, encoding="utf-8") as f:
+                lines = f.read().strip().split("\n")
+            self.assertEqual(len(lines), 1)
+            self.assertEqual(json.loads(lines[0])["kind"], "exec")
+
+    def test_set_trace_file_failure_keeps_none(self):
+        with tempfile.TemporaryDirectory() as d:
+            # host-traces 已被普通文件占住，建目录失败时不抛、不影响回合。
+            blocker = os.path.join(d, "host-traces")
+            with open(blocker, "w", encoding="utf-8") as f:
+                f.write("x")
+            srv = self._bare_server()
+            srv.set_trace_file(os.path.join(d, "host-state.json"), "t1")
+            self.assertIsNone(srv.trace_path)
+
+    def test_append_stops_at_max_lines(self):
+        with tempfile.TemporaryDirectory() as d:
+            srv = self._bare_server()
+            srv.trace_path = os.path.join(d, "t.jsonl")
+            srv._trace_lines = TRACE_MAX_LINES - 1
+            srv._append_trace(_msg({"type": "commandExecution", "id": "c1",
+                                    "command": "a"}))
+            srv._append_trace(_msg({"type": "commandExecution", "id": "c2",
+                                    "command": "b"}))
+            with open(srv.trace_path, encoding="utf-8") as f:
+                lines = f.read().strip().split("\n")
+            self.assertEqual(len(lines), 1)
+            self.assertEqual(json.loads(lines[0])["id"], "c1")
+
+
+if __name__ == "__main__":
+    unittest.main()