宜博 / lemongraph

Commit b1c19805eb36e8a7da8ef1db825acd18117e4e8e

Authored by bboonnee
1 parent b72b0384

U-027: execution_log_file + c_execution_log_get + c_ws_log_tail + v_log_viewer 

execution_log_file 的行生命周期 U-013 早就管好了，但从来没有代码真的往
日志文件里写过内容——src/execution_log.py 新增 append_log() 补上这条
缺失的写路径，设计成 run_fn 执行过程中随时可调用，不关心节点在做什么。
get_log() 按 offset/limit 分页读，download=true 时走原始文件流。

c_ws_log_tail（WS /api/ws/work-nodes/:node_exec_id/log）复用 U-024 手写
的 RFC 6455 机制，轮询 execution_event(log_appended)；节点一旦转终态
（done/blocked/failed）服务端主动发关闭帧断开，不让客户端一直轮询。

p_log_retention_cleanup 实现成纯函数 run_log_retention_cleanup()：过期
扫描 + max_total_size_mb 超限按最旧优先追加候选 + 单个文件删除失败只
记录不阻塞其余项，没有配调度器（不在本 Unit 覆盖范围），留给未来的
调度进程周期调用。

过程中意外抓到一个真的并发 bug：_Server.server_close() 关闭共享
sqlite 连接时原来没拿 conn_lock，如果这时候正好有个 WS 轮询线程在
另一边持锁执行 SQL，两个线程无同步地并发操作同一个 C 级 sqlite3 连接
对象，实测直接把解释器崩了（SIGSEGV）——U-024 就已经有这个隐患，这次
高频 WS 测试把时序窗口撞开了。修好后连续跑几十次全部稳定通过，记录
在 05 §5.4.1.3。

frontend/app.js 的 v_log_viewer 挂在 v_task_detail 每个 to-do 节点的
展开详情里，首次展开才建连接（懒加载），运行中态走 c_ws_log_tail 实时
滚动、终态走 c_execution_log_get 静态拉取，活跃连接在任务级刷新/离开
页面时统一关闭。真实服务器 + Node 原生 WebSocket/fetch 端到端验证过
实时追加与终态自动断开的完整链路。

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Inline Side-by-side
Showing 9 changed files with 895 additions and 22 deletions
docs/development-plan.md
  View file @b1c1980
... ... @@ -256,6 +256,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
256 256 - **U-027 [Value Path]** `execution_log_file` + `c_execution_log_get` + `c_ws_log_tail` + `v_log_viewer`
257 257 - 依赖：U-024
258 258 - Done：REQ-13 happy/boundary/failure 通过
  259 + - **实现补充（2026-07-31）**：`execution_log_file` 的行生命周期（`active`→`closed`）U-013 早就管好了，但从来没有代码真的往日志文件里写过内容——`src/execution_log.py` 新增 `append_log()` 补上这条缺失的写路径，设计成 `run_fn`（真正干活的那个函数）在执行过程中随时可调用，不关心节点在做什么，和 `work_node_run.py` 自己对 `run_fn` 划的边界一致。`get_log()` 按 offset/limit 分页读，`download=true` 时走原始文件流。`c_ws_log_tail`（`WS /api/ws/work-nodes/:node_exec_id/log`）复用 U-024 手写的 RFC 6455 机制，同样是轮询 `execution_event(event_type=log_appended)`；按 03-C §3.6.2 原文要求，节点一旦转终态（`done`/`blocked`/`failed`）服务端主动发关闭帧断开连接，不是让客户端一直轮询下去。`p_log_retention_cleanup`（04-P §4.5）实现成一个纯函数 `run_log_retention_cleanup()`——过期扫描 + `max_total_size_mb` 超限时按最旧优先追加候选 + 单个文件删除失败只记录不阻塞其余项，全部按 04-P 原文逻辑写；没有配调度器/cron（那不在本 Unit 覆盖范围内，`c_execution_log_get`/`c_ws_log_tail` 才是），留给未来的调度进程周期调用。**过程中意外抓到一个真的并发 bug**：`backend_server.py` 的 `_Server.server_close()` 关闭共享 sqlite 连接时原来没拿 `conn_lock`，如果这时候正好有个 WS 轮询线程在另一边持锁执行 SQL，两个线程无同步地并发操作同一个 C 级 sqlite3 连接对象，实测直接把解释器崩了（`exit code 139`，SIGSEGV）——不是新代码引入的，U-024 就已经有这个隐患，只是这次高频率的 WS 测试把它的时序窗口撞开了；修好后连续跑几十次全部稳定通过。前端：`frontend/app.js` 的 `v_log_viewer` 挂在 `v_task_detail` 每个 to-do 节点的展开详情里（不是页面底部一个孤立面板，`node_exec_id` 对应哪个节点这份日志就该长在哪个节点旁边），首次展开才建连接（懒加载），运行中态走 `c_ws_log_tail` 实时滚动、终态走 `c_execution_log_get` 一次性静态拉取；活跃的 log-tail 连接会在任务级刷新/离开页面时统一关闭，避免连接泄漏。真实服务器 + Node 原生 `WebSocket`/`fetch` 端到端验证过实时追加与终态自动断开的完整链路（静态回填先拿到已有内容，WS 只推增量，二者拼起来等于最终静态内容）。
259 260 - 覆盖：01-M `execution_log_file`/`log_retention_config`；02-V `v_log_viewer`；03-C `c_execution_log_get`/`c_ws_log_tail`；REQ-13
260 261  
261 262 - **U-028 [Value Path]** `v_artifact_browser`
... ...
docs/sdd/05-实施约束与依赖.md
  View file @b1c1980
... ... @@ -46,6 +46,14 @@
46 46 - **实际后果**：U-004 迁移 Validator 人设时，原本想用 `disallowedTools: [Edit, Write, Bash(git commit:*), Bash(git push:*), Bash(git reset:*), Bash(git clean:*)]` 把"不准改代码"和"不准动 git 历史"都在工具层锁死，结果是 Validator 真实调用时完全没有 Bash 权限，无法运行任何测试——这个问题一直存在到 U-019 之后才通过跑 U-014 完整真实闭环时暴露出来（Validator 给出 INCONCLUSIVE 判定时明确说"没有 Bash 工具"）。已订正为 `disallowedTools: [Edit, Write]`（不再限定 Bash 子命令），"不准动 git 历史"这条约束现在只靠 AGENTS.md 里的文字指令，不是工具层强制——一个跑不了任何测试的 Validator 比一个理论上可能跑 `git commit` 的 Validator 风险大得多，两权相害取其轻
47 47 - 同 §5.4.1.1：这条限制没在官方文档里找到说明，是实测发现的，不排除后续版本行为会变——升级 Claude Code 版本后值得重新验证
48 48  
  49 +## 5.4.1.3 `ThreadingHTTPServer` + 共享 sqlite 连接：关闭连接必须持有同一把锁（发现，2026-07-31，U-027）
  50 +
  51 +- 这不是 Claude Code 的行为，是 `backend_server.py` 自己的并发 bug，记录在这里是因为它和上面两条一样"没写测试根本发现不了、实测才炸出来"，属于同一类"实现约束"。
  52 +- 背景：U-022 起 `backend_server.py` 给 `ThreadingHTTPServer`（每个连接一个线程）配了一个跨线程共享的 sqlite 连接（`schema.connect(..., check_same_thread=False)`），配了 `server.conn_lock` 要求所有读写都在锁内进行。U-024/U-027 又加了长连接 WebSocket 处理器，在后台线程里用同一把锁循环轮询。
  53 +- **实测确认**：`_Server.server_close()`（测试 `tearDown`/进程正常关闭时都会走到）原来是 `self.conn.close()` 直接关，没拿 `conn_lock`。如果这一刻恰好有个 WS 轮询线程正在锁内执行 SQL，两个线程在没有任何同步的情况下并发操作同一个 C 级 sqlite3 连接对象——不是干净地抛 `ProgrammingError`，是直接把 Python 解释器崩了（`exit code 139` / SIGSEGV）。命中概率取决于时序窗口，单独跑某个测试文件时几乎必现，跑整个套件时因为时序被摊薄，時有时无，不修的话会是一个典型的"本地测试大多数时候是绿的，CI 偶尔莫名其妙崩溃"式 flaky 源。
  54 +- **应对方式**：`self.conn.close()` 也必须包在 `with self.conn_lock:` 里——"关闭"和"读写"对同一个连接对象而言都是需要互斥的操作，不能默认关闭是安全的。已在 U-027 修复，修复后连续跑几十次全部稳定通过。
  55 +- 这条不需要等 Claude Code 升级重新验证（跟它无关），但如果以后这里改用连接池/每线程一个连接，这条约束本身会自然消失，到时候可以直接删掉这一节。
  56 +
49 57 ## 5.4.2 Claude Code plan 额度等待重试参数（架构文档 7.5 节决策，2026-07-29）
50 58  
51 59 - 退避轮询间隔（5→10→20→30 分钟，上限 30 分钟）、超时兜底阈值（session 8 小时、weekly 9 天）是 MVP 阶段的硬编码默认值，不在管理后台开放用户配置，和 `log_retention_config` 的"保留天数可配、清理频率不可配"是同一个 MVP 简化原则
... ...
docs/sdd/unit_matrix.json
  View file @b1c1980
... ... @@ -42,7 +42,7 @@
42 42 {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（2026-07-30 新增）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准"},
43 43 {"id": "U-025", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_dashboard + v_task_board + v_task_detail（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_dashboard", "v_task_board", "v_task_detail"], "controller": ["c_task_list", "c_task_get", "c_org_loop_route"], "process": [], "requirement": []}, "done": "三个页面可正常展示实时数据；补上此前无 Unit 认领的 c_task_list/c_task_get HTTP 路由，c_org_loop_route 首次接上 HTTP；v_graph_visualizer/v_log_viewer 子模块占位等 U-026/U-027"},
44 44 {"id": "U-026", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_graph_visualizer（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_graph_visualizer"], "controller": [], "process": [], "requirement": []}, "done": "Work Graph DAG 分层视图含 waiting_for_quota 信息态；Org Graph 层复用 v_task_detail 已有的 org-strip 面板，不重复渲染"},
45   - {"id": "U-027", "phase": "PH3", "sub_phase": "PH3-S3", "name": "execution_log_file + 日志查看器", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": ["execution_log_file", "log_retention_config"], "view": ["v_log_viewer"], "controller": ["c_execution_log_get", "c_ws_log_tail"], "process": [], "requirement": ["REQ-13"]}, "done": "REQ-13 全部通过"},
  45 + {"id": "U-027", "phase": "PH3", "sub_phase": "PH3-S3", "name": "execution_log_file + 日志查看器（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": ["execution_log_file", "log_retention_config"], "view": ["v_log_viewer"], "controller": ["c_execution_log_get", "c_ws_log_tail"], "process": ["p_log_retention_cleanup"], "requirement": ["REQ-13"]}, "done": "REQ-13 happy/boundary/failure 通过；补上此前从未存在的 append_log 写路径；顺带修了 U-024 就存在的 server_close() 并发 segfault 隐患（conn_lock 未覆盖）"},
46 46 {"id": "U-028", "phase": "PH3", "sub_phase": "PH3-S3", "name": "v_artifact_browser", "classification": "Value Path", "depends_on": ["U-011"], "coverage": {"model": [], "view": ["v_artifact_browser"], "controller": [], "process": [], "requirement": []}, "done": "Artifact 列表+预览+血缘链可用"},
47 47 {"id": "U-029", "phase": "PH3", "sub_phase": "PH3-S3", "name": "v_skill_library + v_skill_audit_log", "classification": "Value Path", "depends_on": ["U-018"], "coverage": {"model": [], "view": ["v_skill_library", "v_skill_audit_log"], "controller": [], "process": [], "requirement": []}, "done": "Skill 库与变更审计日志可查看"},
48 48 {"id": "U-030", "phase": "PH3", "sub_phase": "PH3-S4", "name": "并发/预算/日志保留配置页", "classification": "Value Path", "depends_on": ["U-019", "U-021"], "coverage": {"model": ["org_concurrency_config"], "view": ["v_concurrency_settings", "v_budget_settings", "v_log_retention_settings"], "controller": ["c_concurrency_config_update", "c_budget_config_update", "c_log_retention_config_update"], "process": [], "requirement": []}, "done": "三个配置页可用且改动生效"},
... ...
frontend/app.js
  View file @b1c1980
... ... @@ -58,6 +58,22 @@ function connectEventSocket(projectId, taskId, { onEvent, onStatusChange }) {
58 58 };
59 59 }
60 60  
  61 +// c_ws_log_tail (03-C §3.6.2, U-027): live tail for one Work Node's raw log.
  62 +// No auto-reconnect here, unlike connectEventSocket -- 03-C is explicit that
  63 +// the server closes this connection on purpose once the node reaches a
  64 +// terminal status, and the caller (mountLogViewer) is expected to switch to
  65 +// c_execution_log_get's static mode on close, not treat it as a dropped
  66 +// connection to retry.
  67 +function connectLogTailSocket(nodeExecId, { onChunk, onClose }) {
  68 + const wsProtocol = window.location.protocol === "https:" ? "wss:" : "ws:";
  69 + const url = wsProtocol + "//" + window.location.host + "/api/ws/work-nodes/" + encodeURIComponent(nodeExecId) + "/log";
  70 + const socket = new WebSocket(url);
  71 + socket.onmessage = (event) => onChunk(JSON.parse(event.data).appended_content);
  72 + socket.onclose = () => onClose();
  73 + socket.onerror = () => socket.close();
  74 + return { close: () => socket.close() };
  75 +}
  76 +
61 77 const routes = {
62 78 "#/": { render: renderDashboard, mount: mountDashboard },
63 79 "#/tasks": { render: renderTaskBoard, mount: mountTaskBoard },
... ... @@ -716,6 +732,82 @@ function formatWaitDuration(sqliteTimestamp) {
716 732 return minutes < 60 ? minutes + " 分钟" : Math.round(minutes / 60) + " 小时";
717 733 }
718 734  
  735 +// ---- v_log_viewer (02-V §2.7.1): embedded raw execution log ----
  736 +// Non-terminal node status -> realtime scroll mode via c_ws_log_tail;
  737 +// terminal status -> one-shot static fetch via c_execution_log_get.
  738 +// Embedded per node_exec_id inside v_task_detail's to-do tree, mounted
  739 +// lazily the first time a to-do row is expanded (mountTaskDetail wires the
  740 +// click handler and tracks cleanup for whichever ones got opened).
  741 +
  742 +function renderLogViewerHtml(nodeExecId) {
  743 + return (
  744 + '<div class="log-toolbar">'
  745 + + '<label><input type="checkbox" class="log-autoscroll" checked> 自动滚动到最新</label>'
  746 + + '<a class="btn" href="/api/work-nodes/' + encodeURIComponent(nodeExecId) + '/log?download=true">下载完整日志</a>'
  747 + + "</div>"
  748 + + '<div class="log-viewer" id="log-viewer-' + nodeExecId + '"><div class="hint">加载中…</div></div>'
  749 + );
  750 +}
  751 +
  752 +function mountLogViewer(nodeExecId, initialStatus) {
  753 + const viewerEl = document.getElementById("log-viewer-" + nodeExecId);
  754 + const autoScrollEl = document.querySelector('#todo-detail-' + nodeExecId + " .log-autoscroll");
  755 + if (!viewerEl) return () => {};
  756 +
  757 + function appendLine(text) {
  758 + const line = document.createElement("div");
  759 + line.className = "line";
  760 + line.textContent = text;
  761 + viewerEl.appendChild(line);
  762 + if (autoScrollEl && autoScrollEl.checked) viewerEl.scrollTop = viewerEl.scrollHeight;
  763 + }
  764 +
  765 + async function loadStatic() {
  766 + try {
  767 + const resp = await fetch("/api/work-nodes/" + encodeURIComponent(nodeExecId) + "/log");
  768 + if (resp.status === 404) {
  769 + viewerEl.innerHTML = '<div class="hint">日志暂不可用</div>';
  770 + return;
  771 + }
  772 + if (resp.status === 410) {
  773 + viewerEl.innerHTML = '<div class="hint">该日志已超过保留期限被清理，保留天数可在设置中调整</div>';
  774 + return;
  775 + }
  776 + const body = await resp.json();
  777 + viewerEl.innerHTML = "";
  778 + if (!body.content) {
  779 + viewerEl.innerHTML = '<div class="hint">暂无日志内容</div>';
  780 + return;
  781 + }
  782 + for (const line of body.content.split("\n")) appendLine(line);
  783 + } catch (e) {
  784 + viewerEl.innerHTML = '<div class="hint">日志暂不可用</div>';
  785 + }
  786 + }
  787 +
  788 + if (isTerminalNodeStatus(initialStatus)) {
  789 + loadStatic();
  790 + return () => {};
  791 + }
  792 +
  793 + // live mode still needs to show everything written so far, not just
  794 + // chunks appended after this connection opens -- backfill via the same
  795 + // static read before tailing new content (mounting is re-triggered on
  796 + // every task-level refresh, e.g. from v_task_detail's own WS feed, so
  797 + // this also re-populates content a reconnect would otherwise have wiped)
  798 + loadStatic();
  799 + const socket = connectLogTailSocket(nodeExecId, {
  800 + onChunk: (chunk) => { for (const line of chunk.split("\n")) if (line) appendLine(line); },
  801 + onClose: () => loadStatic(), // server closes on terminal status (03-C §3.6.2) -- switch to static mode
  802 + });
  803 + return () => socket.close();
  804 +}
  805 +
  806 +const TERMINAL_NODE_STATUSES = new Set(["done", "blocked", "failed"]);
  807 +function isTerminalNodeStatus(status) {
  808 + return TERMINAL_NODE_STATUSES.has(status);
  809 +}
  810 +
719 811 // ---- v_task_detail (02-V §2.6): task detail / process inspection ----
720 812 // Embeds a to-do tree built from graph_manifest (01-M 1.3's "补充说明")
721 813 // and the v_graph_visualizer DAG view above. The raw-log v_log_viewer is a
... ... @@ -737,6 +829,15 @@ function mountTaskDetail(taskId) {
737 829 if (!projectId) return;
738 830 const content = document.getElementById("td_content");
739 831  
  832 + // v_log_viewer instances (02-V §2.7.1) mounted lazily on first to-do
  833 + // expand -- tracked here so a re-render (WS-triggered refresh) or leaving
  834 + // the page closes any live c_ws_log_tail connections instead of leaking them.
  835 + const activeLogViewers = new Map();
  836 + function closeAllLogViewers() {
  837 + for (const close of activeLogViewers.values()) close();
  838 + activeLogViewers.clear();
  839 + }
  840 +
740 841 function renderTodoTree(run) {
741 842 if (!run.graph_manifest) return "";
742 843 const nodesById = {};
... ... @@ -745,10 +846,10 @@ function mountTaskDetail(taskId) {
745 846 const wn = nodesById[gn.node_id];
746 847 const status = wn ? wn.status : "pending";
747 848 const detail = wn ? (
748   - '<div class="todo-detail" id="todo-detail-' + wn.node_exec_id + '">'
  849 + '<div class="todo-detail" id="todo-detail-' + wn.node_exec_id + '" data-status="' + status + '">'
749 850 + "retry_count: " + wn.retry_count
750 851 + (wn.reflection_notes ? "<br>reflection_notes: " + escapeHtml(wn.reflection_notes) : "")
751   - + '<div class="hint">原始日志查看器（v_log_viewer）随 U-027 补上</div>'
  852 + + '<div class="log-viewer-slot" id="log-slot-' + wn.node_exec_id + '"></div>'
752 853 + "</div>"
753 854 ) : "";
754 855 return (
... ... @@ -769,6 +870,7 @@ function mountTaskDetail(taskId) {
769 870 }
770 871  
771 872 function renderDetail(task) {
  873 + closeAllLogViewers();
772 874 const activeSet = new Set([task.assigned_org_node]);
773 875 for (const run of task.work_graph_runs) {
774 876 if (run.status !== "done" && run.status !== "failed") activeSet.add(run.org_node);
... ... @@ -815,7 +917,6 @@ function mountTaskDetail(taskId) {
815 917 + '<div class="tooltip" id="graph-tooltip"></div>'
816 918 + '<div class="panel"><h2>关联 Artifact</h2>' + artifactsHtml + "</div>"
817 919 + '<div class="panel"><h2>关联 Bug 记录</h2>' + bugsHtml + "</div>"
818   - + '<div class="panel"><h2>原始执行日志</h2><div class="hint">v_log_viewer 随 U-027 补上</div></div>'
819 920 );
820 921  
821 922 for (const row of content.querySelectorAll(".todo-row[data-node-exec-id]")) {
... ... @@ -823,7 +924,13 @@ function mountTaskDetail(taskId) {
823 924 const nodeExecId = row.dataset.nodeExecId;
824 925 if (!nodeExecId) return;
825 926 const detail = document.getElementById("todo-detail-" + nodeExecId);
826   - if (detail) detail.classList.toggle("open");
  927 + if (!detail) return;
  928 + detail.classList.toggle("open");
  929 + if (detail.classList.contains("open") && !activeLogViewers.has(nodeExecId)) {
  930 + const slot = document.getElementById("log-slot-" + nodeExecId);
  931 + slot.innerHTML = renderLogViewerHtml(nodeExecId);
  932 + activeLogViewers.set(nodeExecId, mountLogViewer(nodeExecId, detail.dataset.status));
  933 + }
827 934 });
828 935 }
829 936 mountGraphVisualizerTooltips(content);
... ... @@ -856,5 +963,8 @@ function mountTaskDetail(taskId) {
856 963 onEvent: () => refresh(),
857 964 onStatusChange: () => {},
858 965 });
859   - registerRouteCleanup(() => socket.close());
  966 + registerRouteCleanup(() => {
  967 + socket.close();
  968 + closeAllLogViewers();
  969 + });
860 970 }
... ...
frontend/style.css
  View file @b1c1980
... ... @@ -89,6 +89,13 @@ table.data-table tr[data-task-id]:hover td { background: #f9fafb; }
89 89 .todo-detail { display: none; margin: 6px 0 0 17px; padding: 8px 10px; background: #fff; border: 1px solid #e5e7eb; border-radius: 6px; font-size: 12px; color: #6b7280; }
90 90 .todo-detail.open { display: block; }
91 91  
  92 +/* U-027: v_log_viewer -- class names match docs/wireframes/v_task_detail.html */
  93 +.log-toolbar { display: flex; justify-content: space-between; align-items: center; margin: 8px 0; font-size: 12px; }
  94 +.log-toolbar label { display: flex; align-items: center; gap: 6px; color: #6b7280; }
  95 +.log-viewer { background: #0b1120; color: #d1d5db; font-family: ui-monospace, monospace; font-size: 12px; border-radius: 8px; padding: 14px; max-height: 220px; overflow: auto; }
  96 +.log-viewer .line { white-space: pre-wrap; line-height: 1.6; }
  97 +.log-viewer .hint { color: #9ca3af; }
  98 +
92 99 /* U-026: v_graph_visualizer -- class/data-* names match
93 100 docs/wireframes/v_graph_visualizer.html */
94 101 .work-graph { position: relative; border: 1px dashed #d1d5db; border-radius: 8px; padding: 32px; background: #f9fafb; }
... ...
src/backend_server.py
  View file @b1c1980
... ... @@ -12,10 +12,12 @@ route until now) for `v_project_setup`&#39;s submit flow, `c_task_create`/
12 12 `c_bug_report_create` (03-C §3.1/§3.1.1, U-023) for `v_goal_composer`/
13 13 `v_bug_feedback`'s submit flows, `c_ws_subscribe` (03-C §3.14, U-024,
14 14 hand-rolled RFC 6455 WebSocket handshake + framing in src/ws_subscribe.py --
15   -no `websockets` dependency) for real-time `execution_event` push, and
16   -serving the U-008 frontend shell as static files (no build step, 05 §5.7:
17   -whitelisted by exact path, not a generic directory walker, so there's no
18   -path-traversal surface to reason about).
  15 +no `websockets` dependency) for real-time `execution_event` push,
  16 +`c_execution_log_get`/`c_ws_log_tail` (03-C §3.6.1/§3.6.2, U-027, the
  17 +latter reusing the same hand-rolled WS machinery) for raw execution log
  18 +viewing, and serving the U-008 frontend shell as static files (no build
  19 +step, 05 §5.7: whitelisted by exact path, not a generic directory walker,
  20 +so there's no path-traversal surface to reason about).
19 21 """
20 22  
21 23 import json
... ... @@ -29,8 +31,8 @@ import urllib.parse
29 31 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
30 32  
31 33 from src import (
32   - budget_guardrail, bug_record, org_loop_route, paths, project_scaffold,
33   - schema, task_create, task_query, ws_subscribe,
  34 + budget_guardrail, bug_record, execution_log, org_loop_route, paths,
  35 + project_scaffold, schema, task_create, task_query, ws_subscribe,
34 36 )
35 37  
36 38 _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
... ... @@ -112,14 +114,19 @@ def _pick_folder_linux():
112 114  
113 115  
114 116 _WS_PROJECTS_PREFIX = "/api/ws/projects/"
115   -
116   -
117 117 _TASKS_PREFIX = "/api/tasks/"
  118 +_WORK_NODES_PREFIX = "/api/work-nodes/"
  119 +_WS_WORK_NODES_PREFIX = "/api/ws/work-nodes/"
  120 +_LOG_SUFFIX = "/log"
118 121  
119 122  
120 123 class _Handler(BaseHTTPRequestHandler):
121 124 def do_GET(self):
122 125 parsed = urllib.parse.urlsplit(self.path)
  126 + if parsed.path.startswith(_WS_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
  127 + node_exec_id = parsed.path[len(_WS_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
  128 + self._handle_ws_log_tail(node_exec_id)
  129 + return
123 130 if parsed.path.startswith(_WS_PROJECTS_PREFIX):
124 131 self._handle_ws_subscribe(parsed.path, parsed.query)
125 132 return
... ... @@ -134,6 +141,10 @@ class _Handler(BaseHTTPRequestHandler):
134 141 if parsed.path == "/api/settings/budget":
135 142 self._handle_budget_list()
136 143 return
  144 + if parsed.path.startswith(_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
  145 + node_exec_id = parsed.path[len(_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
  146 + self._handle_execution_log_get(node_exec_id, parsed.query)
  147 + return
137 148  
138 149 entry = _STATIC_FILES.get(self.path)
139 150 if entry is None:
... ... @@ -314,6 +325,34 @@ class _Handler(BaseHTTPRequestHandler):
314 325 self._send_json(404, {"error": "PROJECT_NOT_FOUND"})
315 326 return
316 327  
  328 + self._send_ws_handshake_response(ws_key)
  329 +
  330 + with self.server.conn_lock:
  331 + after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
  332 + self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
  333 + try:
  334 + while True:
  335 + try:
  336 + opcode, _payload = ws_subscribe.decode_client_frame(self.connection)
  337 + if opcode == ws_subscribe.OPCODE_CLOSE:
  338 + break
  339 + if opcode == ws_subscribe.OPCODE_PING:
  340 + self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_PONG))
  341 + except socket.timeout:
  342 + pass # normal: nothing from the client this poll tick
  343 +
  344 + with self.server.conn_lock:
  345 + events, after_rowid = ws_subscribe.poll_new_events(
  346 + self.server.conn, project_id, after_rowid, task_id=task_id
  347 + )
  348 + for event in events:
  349 + self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
  350 + except (ConnectionError, OSError):
  351 + pass # client disconnected -- nothing more to send
  352 + except sqlite3.ProgrammingError:
  353 + pass # server shutting down closed the shared conn out from under this poll loop
  354 +
  355 + def _send_ws_handshake_response(self, ws_key):
317 356 accept_key = ws_subscribe.compute_accept_key(ws_key)
318 357 self.connection.sendall(
319 358 (
... ... @@ -325,8 +364,60 @@ class _Handler(BaseHTTPRequestHandler):
325 364 )
326 365 self.close_connection = True
327 366  
  367 + def _handle_execution_log_get(self, node_exec_id, query_string):
  368 + params = urllib.parse.parse_qs(query_string)
  369 + download = params.get("download", ["false"])[0].lower() in ("true", "1")
  370 + try:
  371 + if download:
  372 + with self.server.conn_lock:
  373 + file_path = execution_log.get_log_file_path(self.server.conn, node_exec_id)
  374 + self._send_file_download(file_path, node_exec_id)
  375 + return
  376 + offset = int(params.get("offset", ["0"])[0])
  377 + limit_param = params.get("limit", [None])[0]
  378 + limit = int(limit_param) if limit_param is not None else None
  379 + with self.server.conn_lock:
  380 + result = execution_log.get_log(self.server.conn, node_exec_id, offset=offset, limit=limit)
  381 + except execution_log.LogNotFoundError:
  382 + self._send_json(404, {"error": "LOG_NOT_FOUND"})
  383 + except execution_log.LogPurgedError:
  384 + self._send_json(410, {"error": "LOG_PURGED"})
  385 + else:
  386 + self._send_json(200, result)
  387 +
  388 + def _send_file_download(self, file_path, node_exec_id):
  389 + with open(file_path, "rb") as f:
  390 + body = f.read()
  391 + self.send_response(200)
  392 + self.send_header("Content-Type", "application/octet-stream")
  393 + self.send_header("Content-Disposition", 'attachment; filename="' + node_exec_id + '.log"')
  394 + self.send_header("Content-Length", str(len(body)))
  395 + self.end_headers()
  396 + self.wfile.write(body)
  397 +
  398 + def _handle_ws_log_tail(self, node_exec_id):
  399 + """c_ws_log_tail (03-C §3.6.2): WS /api/ws/work-nodes/:node_exec_id/log.
  400 + The server actively closes the connection once the node reaches a
  401 + terminal status (done/blocked/failed) -- 03-C's own wording, not
  402 + left for the client to keep polling forever after there's nothing
  403 + more to send; the frontend switches to c_execution_log_get's static
  404 + mode on close."""
  405 + ws_key = self.headers.get("Sec-WebSocket-Key")
  406 + if self.headers.get("Upgrade", "").lower() != "websocket" or not ws_key:
  407 + self._send_json(400, {"error": "INVALID_INPUT", "detail": "expected a WebSocket upgrade request"})
  408 + return
  409 +
  410 + try:
  411 + with self.server.conn_lock:
  412 + execution_log.node_exec_status(self.server.conn, node_exec_id)
  413 + except execution_log.NodeExecNotFoundError:
  414 + self._send_json(404, {"error": "NODE_EXEC_NOT_FOUND"})
  415 + return
  416 +
  417 + self._send_ws_handshake_response(ws_key)
  418 +
328 419 with self.server.conn_lock:
329   - after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
  420 + after_rowid = execution_log.latest_log_event_rowid(self.server.conn, node_exec_id)
330 421 self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
331 422 try:
332 423 while True:
... ... @@ -340,11 +431,13 @@ class _Handler(BaseHTTPRequestHandler):
340 431 pass # normal: nothing from the client this poll tick
341 432  
342 433 with self.server.conn_lock:
343   - events, after_rowid = ws_subscribe.poll_new_events(
344   - self.server.conn, project_id, after_rowid, task_id=task_id
345   - )
  434 + events, after_rowid = execution_log.poll_log_events(self.server.conn, node_exec_id, after_rowid)
  435 + status = execution_log.node_exec_status(self.server.conn, node_exec_id)
346 436 for event in events:
347 437 self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
  438 + if execution_log.is_terminal_status(status):
  439 + self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_CLOSE))
  440 + break
348 441 except (ConnectionError, OSError):
349 442 pass # client disconnected -- nothing more to send
350 443 except sqlite3.ProgrammingError:
... ... @@ -377,7 +470,13 @@ class _Server(ThreadingHTTPServer):
377 470  
378 471 def server_close(self):
379 472 super().server_close()
380   - self.conn.close()
  473 + # a WS handler thread can still be mid-poll on another thread at
  474 + # shutdown (its next tick would just get a clean ProgrammingError,
  475 + # already caught) -- closing conn without this lock races the C
  476 + # sqlite3 connection object across threads with no synchronization
  477 + # at all, which segfaults rather than raising cleanly.
  478 + with self.conn_lock:
  479 + self.conn.close()
381 480  
382 481  
383 482 def create_server(host="127.0.0.1", port=0, db_path=None):
... ...
src/execution_log.py 0 → 100644
  View file @b1c1980
  1 +"""c_execution_log_get + c_ws_log_tail (03-C §3.6.1/§3.6.2) + p_log_retention_cleanup
  2 +(04-P §4.5): raw execution log storage, retrieval, real-time streaming, and
  3 +scheduled cleanup -- REQ-13.
  4 +
  5 +U-013 (work_node_run.py) already manages execution_log_file's row lifecycle
  6 +(insert status='active' -> update status='closed') but nothing ever wrote
  7 +actual log content -- append_log() is that missing write path, meant to be
  8 +called from inside a run_fn as it works (this module doesn't know or care
  9 +what a node's actual work is, same boundary work_node_run.py itself draws
  10 +around run_fn).
  11 +
  12 +p_log_retention_cleanup is a scheduled job (04-P: "定时触发...具体周期由
  13 +平台调度配置，非用户可配") -- no scheduler daemon exists in this codebase
  14 +yet and building one is out of this Unit's scope (coverage only lists
  15 +c_execution_log_get/c_ws_log_tail); run_log_retention_cleanup() is the pure,
  16 +directly-testable function a future scheduler calls periodically.
  17 +"""
  18 +
  19 +import json
  20 +import os
  21 +import uuid
  22 +from datetime import datetime, timedelta, timezone
  23 +
  24 +
  25 +class LogNotFoundError(Exception):
  26 + """LOG_NOT_FOUND -- 03-C §3.6.1: node_exec_id has no execution_log_file row."""
  27 +
  28 +
  29 +class LogPurgedError(Exception):
  30 + """LOG_PURGED -- 03-C §3.6.1: the log was already cleaned up by retention."""
  31 +
  32 +
  33 +class NodeExecNotFoundError(Exception):
  34 + """NODE_EXEC_NOT_FOUND -- 03-C §3.6.2: c_ws_log_tail subscribe target doesn't exist."""
  35 +
  36 +
  37 +_TERMINAL_NODE_STATUSES = {"done", "blocked", "failed"}
  38 +
  39 +
  40 +def is_terminal_status(status):
  41 + """02-V §2.7.1 / 03-C §3.6.2: once a node reaches one of these, no more
  42 + log content will ever be appended -- c_ws_log_tail actively closes."""
  43 + return status in _TERMINAL_NODE_STATUSES
  44 +
  45 +
  46 +def append_log(conn, node_exec_id, content):
  47 + """Appends `content` to the node's log file on disk, bumps
  48 + execution_log_file.size_bytes, and writes
  49 + execution_event(event_type='log_appended') so c_ws_log_tail subscribers
  50 + see it in realtime mode (02-V §2.7.1)."""
  51 + row = conn.execute(
  52 + "SELECT log_file_id, project_id, file_path FROM execution_log_file WHERE node_exec_id = ?",
  53 + (node_exec_id,),
  54 + ).fetchone()
  55 + if row is None:
  56 + raise LogNotFoundError(node_exec_id)
  57 + log_file_id, project_id, file_path = row
  58 +
  59 + os.makedirs(os.path.dirname(file_path), exist_ok=True)
  60 + with open(file_path, "a", encoding="utf-8") as f:
  61 + f.write(content)
  62 + new_size = os.path.getsize(file_path)
  63 + conn.execute("UPDATE execution_log_file SET size_bytes = ? WHERE log_file_id = ?", (new_size, log_file_id))
  64 +
  65 + task_id = conn.execute(
  66 + "SELECT wgr.task_id FROM work_node_execution wne "
  67 + "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id WHERE wne.node_exec_id = ?",
  68 + (node_exec_id,),
  69 + ).fetchone()[0]
  70 + conn.execute(
  71 + "INSERT INTO execution_event (event_id, project_id, task_id, node_exec_id, event_type, payload) "
  72 + "VALUES (?, ?, ?, ?, 'log_appended', ?)",
  73 + (str(uuid.uuid4()), project_id, task_id, node_exec_id, json.dumps({"appended_content": content})),
  74 + )
  75 + conn.commit()
  76 +
  77 +
  78 +def get_log(conn, node_exec_id, offset=0, limit=None):
  79 + """c_execution_log_get (03-C §3.6.1) static-mode read, paginated by
  80 + byte offset/limit so a large file isn't loaded all at once."""
  81 + row = conn.execute(
  82 + "SELECT file_path, format, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
  83 + ).fetchone()
  84 + if row is None:
  85 + raise LogNotFoundError(node_exec_id)
  86 + file_path, fmt, status = row
  87 + if status == "purged":
  88 + raise LogPurgedError(node_exec_id)
  89 +
  90 + if not os.path.isfile(file_path):
  91 + return {"content": "", "format": fmt, "total_size_bytes": 0, "status": status}
  92 +
  93 + total_size = os.path.getsize(file_path)
  94 + with open(file_path, "rb") as f:
  95 + f.seek(offset)
  96 + chunk = f.read(limit) if limit is not None else f.read()
  97 + return {"content": chunk.decode("utf-8", errors="replace"), "format": fmt, "total_size_bytes": total_size, "status": status}
  98 +
  99 +
  100 +def get_log_file_path(conn, node_exec_id):
  101 + """For the download=true response path (send the raw file, not JSON)."""
  102 + row = conn.execute(
  103 + "SELECT file_path, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
  104 + ).fetchone()
  105 + if row is None:
  106 + raise LogNotFoundError(node_exec_id)
  107 + file_path, status = row
  108 + if status == "purged":
  109 + raise LogPurgedError(node_exec_id)
  110 + return file_path
  111 +
  112 +
  113 +def node_exec_status(conn, node_exec_id):
  114 + row = conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,)).fetchone()
  115 + if row is None:
  116 + raise NodeExecNotFoundError(node_exec_id)
  117 + return row[0]
  118 +
  119 +
  120 +def latest_log_event_rowid(conn, node_exec_id):
  121 + row = conn.execute(
  122 + "SELECT MAX(rowid) FROM execution_event WHERE node_exec_id = ? AND event_type = 'log_appended'",
  123 + (node_exec_id,),
  124 + ).fetchone()
  125 + return row[0] or 0
  126 +
  127 +
  128 +def poll_log_events(conn, node_exec_id, after_rowid):
  129 + """c_ws_log_tail's push source -- same rowid-increment polling pattern
  130 + as ws_subscribe.poll_new_events, scoped to one node_exec_id's
  131 + log_appended events."""
  132 + rows = conn.execute(
  133 + "SELECT rowid, payload, created_at FROM execution_event "
  134 + "WHERE node_exec_id = ? AND event_type = 'log_appended' AND rowid > ? ORDER BY rowid",
  135 + (node_exec_id, after_rowid),
  136 + ).fetchall()
  137 + events = [
  138 + {
  139 + "node_exec_id": node_exec_id,
  140 + "appended_content": json.loads(r[1])["appended_content"],
  141 + "created_at": r[2],
  142 + }
  143 + for r in rows
  144 + ]
  145 + new_after_rowid = rows[-1][0] if rows else after_rowid
  146 + return events, new_after_rowid
  147 +
  148 +
  149 +def run_log_retention_cleanup(conn, now=None, config_id="default"):
  150 + """p_log_retention_cleanup (04-P §4.5): idle -> scanning -> purging -> idle.
  151 +
  152 + Age-expired 'closed' logs are always candidates. If max_total_size_mb is
  153 + set and the remaining (non-expired) total still exceeds it, the
  154 + oldest-first remaining logs are added as candidates too, until the kept
  155 + total drops back under the cap. A file that fails to delete (e.g. still
  156 + open elsewhere) is recorded as a failure and left status='closed' for
  157 + the next cycle to retry -- it must not block purging the rest of the
  158 + batch (REQ-13 failure case).
  159 +
  160 + Returns {"purged": [log_file_id, ...], "failed": [log_file_id, ...]}.
  161 + """
  162 + config_row = conn.execute(
  163 + "SELECT retention_days, max_total_size_mb FROM log_retention_config WHERE config_id = ?", (config_id,)
  164 + ).fetchone()
  165 + if config_row is None:
  166 + raise ValueError("no such log_retention_config: " + config_id)
  167 + retention_days, max_total_size_mb = config_row
  168 +
  169 + now = now if now is not None else datetime.now(timezone.utc)
  170 + cutoff_str = (now - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S")
  171 +
  172 + expired = conn.execute(
  173 + "SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
  174 + "WHERE status = 'closed' AND created_at < ? ORDER BY created_at",
  175 + (cutoff_str,),
  176 + ).fetchall()
  177 + kept = conn.execute(
  178 + "SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
  179 + "WHERE status = 'closed' AND created_at >= ? ORDER BY created_at",
  180 + (cutoff_str,),
  181 + ).fetchall()
  182 +
  183 + candidates = list(expired)
  184 + if max_total_size_mb is not None:
  185 + limit_bytes = max_total_size_mb * 1024 * 1024
  186 + kept_total = sum(row[2] for row in kept)
  187 + for row in kept:
  188 + if kept_total <= limit_bytes:
  189 + break
  190 + candidates.append(row)
  191 + kept_total -= row[2]
  192 +
  193 + purged, failed = [], []
  194 + for log_file_id, file_path, _size in candidates:
  195 + try:
  196 + if os.path.isfile(file_path):
  197 + os.remove(file_path)
  198 + elif os.path.exists(file_path):
  199 + # something's occupying the path but isn't a plain file
  200 + # (e.g. a directory) -- can't silently treat that as "already
  201 + # gone", it's a real delete failure
  202 + raise OSError("not a regular file: " + file_path)
  203 + except OSError:
  204 + failed.append(log_file_id)
  205 + continue
  206 + conn.execute(
  207 + "UPDATE execution_log_file SET status = 'purged', purged_at = CURRENT_TIMESTAMP WHERE log_file_id = ?",
  208 + (log_file_id,),
  209 + )
  210 + purged.append(log_file_id)
  211 + conn.commit()
  212 +
  213 + return {"purged": purged, "failed": failed}
... ...
src/test_backend_server.py
  View file @b1c1980
... ... @@ -20,7 +20,7 @@ import time
20 20 import unittest
21 21 from unittest.mock import patch
22 22  
23   -from src import backend_server, ws_subscribe
  23 +from src import backend_server, execution_log, ws_subscribe
24 24  
25 25  
26 26 class ServerStartStopTest(unittest.TestCase):
... ... @@ -536,16 +536,23 @@ def _read_http_response_head(sock):
536 536 return buf.decode("ascii", errors="replace")
537 537  
538 538  
539   -def _read_server_frame(sock):
  539 +def _read_server_frame_raw(sock):
540 540 """Server->client frames are never masked (RFC 6455 §5.1) -- simpler
541   - decode than ws_subscribe.decode_client_frame, which expects masking."""
  541 + decode than ws_subscribe.decode_client_frame, which expects masking.
  542 + Returns (opcode, payload_bytes)."""
542 543 first_two = _recv_exact(sock, 2)
  544 + opcode = first_two[0] & 0x0F
543 545 length = first_two[1] & 0x7F
544 546 if length == 126:
545 547 length = int.from_bytes(_recv_exact(sock, 2), "big")
546 548 elif length == 127:
547 549 length = int.from_bytes(_recv_exact(sock, 8), "big")
548 550 payload = _recv_exact(sock, length)
  551 + return opcode, payload
  552 +
  553 +
  554 +def _read_server_frame(sock):
  555 + _opcode, payload = _read_server_frame_raw(sock)
549 556 return json.loads(payload.decode("utf-8"))
550 557  
551 558  
... ... @@ -665,6 +672,147 @@ class WsSubscribeRouteTest(unittest.TestCase):
665 672 self.assertEqual(event["event_id"], "evt-related")
666 673  
667 674  
  675 +class ExecutionLogRouteTest(unittest.TestCase):
  676 + """U-027: c_execution_log_get (GET /api/work-nodes/:node_exec_id/log)
  677 + and c_ws_log_tail (WS /api/ws/work-nodes/:node_exec_id/log)."""
  678 +
  679 + def setUp(self):
  680 + self.tmp = tempfile.TemporaryDirectory()
  681 + self.addCleanup(self.tmp.cleanup)
  682 + self.server = backend_server.create_server(db_path=":memory:")
  683 + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
  684 + self.thread.start()
  685 + conn = self.server.conn
  686 + conn.execute(
  687 + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
  688 + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
  689 + )
  690 + conn.execute(
  691 + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
  692 + "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
  693 + )
  694 + conn.execute(
  695 + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
  696 + "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
  697 + )
  698 + conn.execute(
  699 + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
  700 + "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
  701 + )
  702 + conn.execute(
  703 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  704 + "input_schema_ref, output_schema_ref, input_payload, status) "
  705 + "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
  706 + )
  707 + conn.execute(
  708 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
  709 + "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
  710 + (os.path.join(self.tmp.name, "node-1.jsonl"),),
  711 + )
  712 + conn.commit()
  713 +
  714 + def tearDown(self):
  715 + self.server.shutdown()
  716 + self.server.server_close()
  717 + self.thread.join(timeout=5)
  718 +
  719 + def _get(self, path):
  720 + host, port = self.server.server_address
  721 + conn = http.client.HTTPConnection(host, port, timeout=5)
  722 + conn.request("GET", path)
  723 + resp = conn.getresponse()
  724 + body = resp.read()
  725 + headers = dict(resp.getheaders())
  726 + conn.close()
  727 + return resp.status, body, headers
  728 +
  729 + def test_returns_static_content(self):
  730 + with self.server.conn_lock:
  731 + execution_log.append_log(self.server.conn, "node-1", "hello world")
  732 + status, body, _ = self._get("/api/work-nodes/node-1/log")
  733 + self.assertEqual(status, 200)
  734 + parsed = json.loads(body)
  735 + self.assertEqual(parsed["content"], "hello world")
  736 + self.assertEqual(parsed["total_size_bytes"], len("hello world"))
  737 +
  738 + def test_pagination(self):
  739 + with self.server.conn_lock:
  740 + execution_log.append_log(self.server.conn, "node-1", "0123456789")
  741 + status, body, _ = self._get("/api/work-nodes/node-1/log?offset=2&limit=3")
  742 + self.assertEqual(status, 200)
  743 + self.assertEqual(json.loads(body)["content"], "234")
  744 +
  745 + def test_download_returns_raw_file_with_attachment_headers(self):
  746 + with self.server.conn_lock:
  747 + execution_log.append_log(self.server.conn, "node-1", "raw log bytes")
  748 + status, body, headers = self._get("/api/work-nodes/node-1/log?download=true")
  749 + self.assertEqual(status, 200)
  750 + self.assertEqual(body, b"raw log bytes")
  751 + self.assertIn("attachment", headers.get("Content-Disposition", ""))
  752 +
  753 + def test_nonexistent_node_returns_404_log_not_found(self):
  754 + status, body, _ = self._get("/api/work-nodes/no-such-node/log")
  755 + self.assertEqual(status, 404)
  756 + self.assertEqual(json.loads(body)["error"], "LOG_NOT_FOUND")
  757 +
  758 + def test_purged_log_returns_410(self):
  759 + with self.server.conn_lock:
  760 + self.server.conn.execute("UPDATE execution_log_file SET status = 'purged' WHERE log_file_id = 'log-1'")
  761 + self.server.conn.commit()
  762 + status, body, _ = self._get("/api/work-nodes/node-1/log")
  763 + self.assertEqual(status, 410)
  764 + self.assertEqual(json.loads(body)["error"], "LOG_PURGED")
  765 +
  766 + def _connect(self):
  767 + host, port = self.server.server_address
  768 + return socket.create_connection((host, port), timeout=5)
  769 +
  770 + def test_ws_handshake_succeeds_for_existing_node(self):
  771 + sock = self._connect()
  772 + self.addCleanup(sock.close)
  773 + key = base64.b64encode(b"0123456789012345").decode("ascii")
  774 + sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
  775 + head = _read_http_response_head(sock)
  776 + self.assertIn("101 Switching Protocols", head.splitlines()[0])
  777 +
  778 + def test_ws_nonexistent_node_rejected_before_upgrade(self):
  779 + sock = self._connect()
  780 + self.addCleanup(sock.close)
  781 + key = base64.b64encode(b"0123456789012345").decode("ascii")
  782 + sock.sendall(_ws_handshake_request("/api/ws/work-nodes/no-such-node/log", key))
  783 + head = _read_http_response_head(sock)
  784 + self.assertIn("404", head.splitlines()[0])
  785 +
  786 + def test_ws_pushes_appended_log_content(self):
  787 + sock = self._connect()
  788 + self.addCleanup(sock.close)
  789 + key = base64.b64encode(b"0123456789012345").decode("ascii")
  790 + sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
  791 + _read_http_response_head(sock)
  792 +
  793 + with self.server.conn_lock:
  794 + execution_log.append_log(self.server.conn, "node-1", "streamed chunk")
  795 +
  796 + sock.settimeout(5)
  797 + event = _read_server_frame(sock)
  798 + self.assertEqual(event["appended_content"], "streamed chunk")
  799 +
  800 + def test_ws_closes_when_node_reaches_terminal_status(self):
  801 + sock = self._connect()
  802 + self.addCleanup(sock.close)
  803 + key = base64.b64encode(b"0123456789012345").decode("ascii")
  804 + sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
  805 + _read_http_response_head(sock)
  806 +
  807 + with self.server.conn_lock:
  808 + self.server.conn.execute("UPDATE work_node_execution SET status = 'done' WHERE node_exec_id = 'node-1'")
  809 + self.server.conn.commit()
  810 +
  811 + sock.settimeout(5)
  812 + opcode, _payload = _read_server_frame_raw(sock)
  813 + self.assertEqual(opcode, ws_subscribe.OPCODE_CLOSE)
  814 +
  815 +
668 816 class ZeroThirdPartyDependencyTest(unittest.TestCase):
669 817 """05 §5.7: no Flask/FastAPI/Express/Node.js -- checked statically so this
670 818 doesn't silently rot if someone adds an import later."""
... ...
src/test_execution_log.py 0 → 100644
  View file @b1c1980
  1 +"""U-027 tests: c_execution_log_get + c_ws_log_tail's data layer +
  2 +p_log_retention_cleanup (REQ-13 happy/boundary/failure). Real sqlite + real
  3 +tempfile-backed log files, no mocking needed.
  4 +
  5 +Run: python3 -m unittest src.test_execution_log -v
  6 +"""
  7 +
  8 +import json
  9 +import os
  10 +import tempfile
  11 +import unittest
  12 +from datetime import datetime, timedelta, timezone
  13 +
  14 +from src import execution_log, schema
  15 +
  16 +
  17 +class ExecutionLogTestBase(unittest.TestCase):
  18 + def setUp(self):
  19 + self.tmp = tempfile.TemporaryDirectory()
  20 + self.addCleanup(self.tmp.cleanup)
  21 + self.conn = schema.connect(":memory:")
  22 + schema.create_schema(self.conn)
  23 + self.conn.execute(
  24 + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
  25 + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
  26 + )
  27 + self.conn.execute(
  28 + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
  29 + "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
  30 + )
  31 + self.conn.execute(
  32 + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
  33 + "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
  34 + )
  35 + self.conn.execute(
  36 + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
  37 + "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
  38 + )
  39 + self.conn.execute(
  40 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  41 + "input_schema_ref, output_schema_ref, input_payload, status) "
  42 + "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
  43 + )
  44 + self.conn.commit()
  45 +
  46 + def tearDown(self):
  47 + self.conn.close()
  48 +
  49 + def _log_path(self):
  50 + return os.path.join(self.tmp.name, "node-1.jsonl")
  51 +
  52 + def _insert_log_file(self, status="active"):
  53 + self.conn.execute(
  54 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
  55 + "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', ?)",
  56 + (self._log_path(), status),
  57 + )
  58 + self.conn.commit()
  59 +
  60 +
  61 +class AppendLogTest(ExecutionLogTestBase):
  62 + def test_appends_content_to_file_and_bumps_size(self):
  63 + self._insert_log_file()
  64 + execution_log.append_log(self.conn, "node-1", "line one\n")
  65 + execution_log.append_log(self.conn, "node-1", "line two\n")
  66 + with open(self._log_path()) as f:
  67 + self.assertEqual(f.read(), "line one\nline two\n")
  68 + size = self.conn.execute("SELECT size_bytes FROM execution_log_file WHERE log_file_id = 'log-1'").fetchone()[0]
  69 + self.assertEqual(size, len("line one\nline two\n"))
  70 +
  71 + def test_writes_log_appended_execution_event(self):
  72 + self._insert_log_file()
  73 + execution_log.append_log(self.conn, "node-1", "hello\n")
  74 + row = self.conn.execute(
  75 + "SELECT event_type, task_id, payload FROM execution_event WHERE node_exec_id = 'node-1'"
  76 + ).fetchone()
  77 + self.assertEqual(row[0], "log_appended")
  78 + self.assertEqual(row[1], "task-1")
  79 + self.assertEqual(json.loads(row[2]), {"appended_content": "hello\n"})
  80 +
  81 + def test_creates_parent_directory_if_missing(self):
  82 + nested_path = os.path.join(self.tmp.name, "a", "b", "node-1.jsonl")
  83 + self.conn.execute(
  84 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
  85 + "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
  86 + (nested_path,),
  87 + )
  88 + self.conn.commit()
  89 + execution_log.append_log(self.conn, "node-1", "x")
  90 + self.assertTrue(os.path.isfile(nested_path))
  91 +
  92 + def test_nonexistent_node_exec_raises(self):
  93 + with self.assertRaises(execution_log.LogNotFoundError):
  94 + execution_log.append_log(self.conn, "no-such-node", "x")
  95 +
  96 +
  97 +class GetLogTest(ExecutionLogTestBase):
  98 + def test_returns_full_content_and_size(self):
  99 + self._insert_log_file(status="closed")
  100 + with open(self._log_path(), "w") as f:
  101 + f.write("all the log content")
  102 + result = execution_log.get_log(self.conn, "node-1")
  103 + self.assertEqual(result["content"], "all the log content")
  104 + self.assertEqual(result["total_size_bytes"], len("all the log content"))
  105 + self.assertEqual(result["status"], "closed")
  106 +
  107 + def test_pagination_offset_and_limit(self):
  108 + self._insert_log_file(status="closed")
  109 + with open(self._log_path(), "w") as f:
  110 + f.write("0123456789")
  111 + result = execution_log.get_log(self.conn, "node-1", offset=3, limit=4)
  112 + self.assertEqual(result["content"], "3456")
  113 + self.assertEqual(result["total_size_bytes"], 10)
  114 +
  115 + def test_missing_physical_file_returns_empty_not_error(self):
  116 + self._insert_log_file(status="closed")
  117 + result = execution_log.get_log(self.conn, "node-1")
  118 + self.assertEqual(result["content"], "")
  119 + self.assertEqual(result["total_size_bytes"], 0)
  120 +
  121 + def test_nonexistent_node_exec_raises_log_not_found(self):
  122 + with self.assertRaises(execution_log.LogNotFoundError):
  123 + execution_log.get_log(self.conn, "no-such-node")
  124 +
  125 + def test_purged_log_raises(self):
  126 + self._insert_log_file(status="purged")
  127 + with self.assertRaises(execution_log.LogPurgedError):
  128 + execution_log.get_log(self.conn, "node-1")
  129 +
  130 +
  131 +class NodeExecStatusTest(ExecutionLogTestBase):
  132 + def test_returns_status(self):
  133 + self.assertEqual(execution_log.node_exec_status(self.conn, "node-1"), "running")
  134 +
  135 + def test_nonexistent_raises(self):
  136 + with self.assertRaises(execution_log.NodeExecNotFoundError):
  137 + execution_log.node_exec_status(self.conn, "no-such-node")
  138 +
  139 + def test_is_terminal_status(self):
  140 + for status in ("done", "blocked", "failed"):
  141 + self.assertTrue(execution_log.is_terminal_status(status))
  142 + for status in ("pending", "running", "reflecting", "retrying", "waiting_for_quota"):
  143 + self.assertFalse(execution_log.is_terminal_status(status))
  144 +
  145 +
  146 +class PollLogEventsTest(ExecutionLogTestBase):
  147 + def test_polls_only_new_log_appended_events_for_this_node(self):
  148 + self._insert_log_file()
  149 + after = execution_log.latest_log_event_rowid(self.conn, "node-1")
  150 + execution_log.append_log(self.conn, "node-1", "chunk-1")
  151 + events, new_after = execution_log.poll_log_events(self.conn, "node-1", after)
  152 + self.assertEqual(len(events), 1)
  153 + self.assertEqual(events[0]["appended_content"], "chunk-1")
  154 + self.assertGreater(new_after, after)
  155 +
  156 + events_again, after2 = execution_log.poll_log_events(self.conn, "node-1", new_after)
  157 + self.assertEqual(events_again, [])
  158 + self.assertEqual(after2, new_after)
  159 +
  160 + def test_other_node_events_excluded(self):
  161 + self.conn.execute(
  162 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  163 + "input_schema_ref, output_schema_ref, input_payload, status) "
  164 + "VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
  165 + )
  166 + self.conn.execute(
  167 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
  168 + "VALUES ('log-2', 'node-2', 'proj-1', ?, 'jsonl', 'active')",
  169 + (os.path.join(self.tmp.name, "node-2.jsonl"),),
  170 + )
  171 + self._insert_log_file()
  172 + self.conn.commit()
  173 + execution_log.append_log(self.conn, "node-2", "not mine")
  174 + events, _ = execution_log.poll_log_events(self.conn, "node-1", 0)
  175 + self.assertEqual(events, [])
  176 +
  177 +
  178 +class LogRetentionCleanupTest(ExecutionLogTestBase):
  179 + def setUp(self):
  180 + super().setUp()
  181 + self.conn.execute(
  182 + "INSERT INTO log_retention_config (config_id, retention_days, updated_by) "
  183 + "VALUES ('default', 1, 'test')"
  184 + )
  185 + self.conn.commit()
  186 +
  187 + def _insert_closed_log(self, log_file_id, created_at, size_bytes=100, file_content="x"):
  188 + node_exec_id = log_file_id + "-node"
  189 + self.conn.execute(
  190 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  191 + "input_schema_ref, output_schema_ref, input_payload, status) "
  192 + "VALUES (?, 'run-1', ?, 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')",
  193 + (node_exec_id, log_file_id + "-n"),
  194 + )
  195 + path = os.path.join(self.tmp.name, log_file_id + ".jsonl")
  196 + with open(path, "w") as f:
  197 + f.write(file_content)
  198 + self.conn.execute(
  199 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
  200 + "status, size_bytes, created_at) VALUES (?, ?, 'proj-1', ?, 'jsonl', 'closed', ?, ?)",
  201 + (log_file_id, node_exec_id, path, size_bytes, created_at),
  202 + )
  203 + self.conn.commit()
  204 + return path
  205 +
  206 + def test_expired_closed_log_purged_and_file_deleted(self):
  207 + old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
  208 + path = self._insert_closed_log("log-old", old_ts)
  209 + result = execution_log.run_log_retention_cleanup(self.conn)
  210 + self.assertEqual(result["purged"], ["log-old"])
  211 + status, purged_at = self.conn.execute(
  212 + "SELECT status, purged_at FROM execution_log_file WHERE log_file_id = 'log-old'"
  213 + ).fetchone()
  214 + self.assertEqual(status, "purged")
  215 + self.assertIsNotNone(purged_at)
  216 + self.assertFalse(os.path.isfile(path))
  217 +
  218 + def test_not_yet_expired_log_untouched(self):
  219 + recent_ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
  220 + path = self._insert_closed_log("log-fresh", recent_ts)
  221 + result = execution_log.run_log_retention_cleanup(self.conn)
  222 + self.assertEqual(result["purged"], [])
  223 + status = self.conn.execute(
  224 + "SELECT status FROM execution_log_file WHERE log_file_id = 'log-fresh'"
  225 + ).fetchone()[0]
  226 + self.assertEqual(status, "closed")
  227 + self.assertTrue(os.path.isfile(path))
  228 +
  229 + def test_viewing_purged_log_raises_purged_not_error(self):
  230 + old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
  231 + self.conn.execute(
  232 + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
  233 + "input_schema_ref, output_schema_ref, input_payload, status) "
  234 + "VALUES ('node-old', 'run-1', 'n-old', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')"
  235 + )
  236 + path = os.path.join(self.tmp.name, "log-old2.jsonl")
  237 + with open(path, "w") as f:
  238 + f.write("x")
  239 + self.conn.execute(
  240 + "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
  241 + "status, created_at) VALUES ('log-old2', 'node-old', 'proj-1', ?, 'jsonl', 'closed', ?)",
  242 + (path, old_ts),
  243 + )
  244 + self.conn.commit()
  245 + execution_log.run_log_retention_cleanup(self.conn)
  246 + with self.assertRaises(execution_log.LogPurgedError):
  247 + execution_log.get_log(self.conn, "node-old")
  248 +
  249 + def test_max_total_size_pulls_in_oldest_non_expired_candidates(self):
  250 + self.conn.execute(
  251 + "UPDATE log_retention_config SET retention_days = 30, max_total_size_mb = 1 WHERE config_id = 'default'"
  252 + )
  253 + self.conn.commit()
  254 + now = datetime.now(timezone.utc)
  255 + one_mb = 1024 * 1024
  256 + # both within retention_days=30, but combined over the 1MB cap --
  257 + # the older of the two should get pulled in as a candidate too
  258 + self._insert_closed_log("log-a", (now - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
  259 + self._insert_closed_log("log-b", (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
  260 + result = execution_log.run_log_retention_cleanup(self.conn, now=now)
  261 + self.assertEqual(result["purged"], ["log-a"])
  262 + status_b = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-b'").fetchone()[0]
  263 + self.assertEqual(status_b, "closed")
  264 +
  265 + def test_delete_failure_recorded_and_others_still_purged(self):
  266 + old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
  267 + self._insert_closed_log("log-ok", old_ts)
  268 + bad_path = self._insert_closed_log("log-bad", old_ts)
  269 + os.remove(bad_path) # simulate "file missing/inaccessible" -- os.remove will raise
  270 + os.mkdir(bad_path) # replace it with a directory so os.remove() fails with OSError
  271 + self.addCleanup(lambda: os.path.isdir(bad_path) and os.rmdir(bad_path))
  272 +
  273 + result = execution_log.run_log_retention_cleanup(self.conn)
  274 + self.assertEqual(result["purged"], ["log-ok"])
  275 + self.assertEqual(result["failed"], ["log-bad"])
  276 + status = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-bad'").fetchone()[0]
  277 + self.assertEqual(status, "closed")
  278 +
  279 + def test_missing_config_raises(self):
  280 + self.conn.execute("DELETE FROM log_retention_config")
  281 + self.conn.commit()
  282 + with self.assertRaises(ValueError):
  283 + execution_log.run_log_retention_cleanup(self.conn)
  284 +
  285 +
  286 +if __name__ == "__main__":
  287 + unittest.main()
... ...