'
+ '
'
+ "retry_count: " + wn.retry_count
+ (wn.reflection_notes ? "
reflection_notes: " + escapeHtml(wn.reflection_notes) : "")
- + '
原始日志查看器（v_log_viewer）随 U-027 补上
'
+ + '
'
+ "
"
) : "";
return (
@@ -769,6 +870,7 @@ function mountTaskDetail(taskId) {
}
function renderDetail(task) {
+ closeAllLogViewers();
const activeSet = new Set([task.assigned_org_node]);
for (const run of task.work_graph_runs) {
if (run.status !== "done" && run.status !== "failed") activeSet.add(run.org_node);
@@ -815,7 +917,6 @@ function mountTaskDetail(taskId) {
+ '
'
+ '
关联 Artifact
' + artifactsHtml + ""
+ '
关联 Bug 记录
' + bugsHtml + ""
- + '
原始执行日志
v_log_viewer 随 U-027 补上
'
);
for (const row of content.querySelectorAll(".todo-row[data-node-exec-id]")) {
@@ -823,7 +924,13 @@ function mountTaskDetail(taskId) {
const nodeExecId = row.dataset.nodeExecId;
if (!nodeExecId) return;
const detail = document.getElementById("todo-detail-" + nodeExecId);
- if (detail) detail.classList.toggle("open");
+ if (!detail) return;
+ detail.classList.toggle("open");
+ if (detail.classList.contains("open") && !activeLogViewers.has(nodeExecId)) {
+ const slot = document.getElementById("log-slot-" + nodeExecId);
+ slot.innerHTML = renderLogViewerHtml(nodeExecId);
+ activeLogViewers.set(nodeExecId, mountLogViewer(nodeExecId, detail.dataset.status));
+ }
});
}
mountGraphVisualizerTooltips(content);
@@ -856,5 +963,8 @@ function mountTaskDetail(taskId) {
onEvent: () => refresh(),
onStatusChange: () => {},
});
- registerRouteCleanup(() => socket.close());
+ registerRouteCleanup(() => {
+ socket.close();
+ closeAllLogViewers();
+ });
}
diff --git a/frontend/style.css b/frontend/style.css
index af870cc..b057afb 100644
--- a/frontend/style.css
+++ b/frontend/style.css
@@ -89,6 +89,13 @@ table.data-table tr[data-task-id]:hover td { background: #f9fafb; }
.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; }
.todo-detail.open { display: block; }
+/* U-027: v_log_viewer -- class names match docs/wireframes/v_task_detail.html */
+.log-toolbar { display: flex; justify-content: space-between; align-items: center; margin: 8px 0; font-size: 12px; }
+.log-toolbar label { display: flex; align-items: center; gap: 6px; color: #6b7280; }
+.log-viewer { background: #0b1120; color: #d1d5db; font-family: ui-monospace, monospace; font-size: 12px; border-radius: 8px; padding: 14px; max-height: 220px; overflow: auto; }
+.log-viewer .line { white-space: pre-wrap; line-height: 1.6; }
+.log-viewer .hint { color: #9ca3af; }
+
/* U-026: v_graph_visualizer -- class/data-* names match
docs/wireframes/v_graph_visualizer.html */
.work-graph { position: relative; border: 1px dashed #d1d5db; border-radius: 8px; padding: 32px; background: #f9fafb; }
diff --git a/src/backend_server.py b/src/backend_server.py
index 4abc07d..e8560dc 100644
--- a/src/backend_server.py
+++ b/src/backend_server.py
@@ -12,10 +12,12 @@ route until now) for `v_project_setup`'s submit flow, `c_task_create`/
`c_bug_report_create` (03-C §3.1/§3.1.1, U-023) for `v_goal_composer`/
`v_bug_feedback`'s submit flows, `c_ws_subscribe` (03-C §3.14, U-024,
hand-rolled RFC 6455 WebSocket handshake + framing in src/ws_subscribe.py --
-no `websockets` dependency) for real-time `execution_event` push, and
-serving the U-008 frontend shell as static files (no build step, 05 §5.7:
-whitelisted by exact path, not a generic directory walker, so there's no
-path-traversal surface to reason about).
+no `websockets` dependency) for real-time `execution_event` push,
+`c_execution_log_get`/`c_ws_log_tail` (03-C §3.6.1/§3.6.2, U-027, the
+latter reusing the same hand-rolled WS machinery) for raw execution log
+viewing, and serving the U-008 frontend shell as static files (no build
+step, 05 §5.7: whitelisted by exact path, not a generic directory walker,
+so there's no path-traversal surface to reason about).
"""
import json
@@ -29,8 +31,8 @@ import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from src import (
- budget_guardrail, bug_record, org_loop_route, paths, project_scaffold,
- schema, task_create, task_query, ws_subscribe,
+ budget_guardrail, bug_record, execution_log, org_loop_route, paths,
+ project_scaffold, schema, task_create, task_query, ws_subscribe,
)
_FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
@@ -112,14 +114,19 @@ def _pick_folder_linux():
_WS_PROJECTS_PREFIX = "/api/ws/projects/"
-
-
_TASKS_PREFIX = "/api/tasks/"
+_WORK_NODES_PREFIX = "/api/work-nodes/"
+_WS_WORK_NODES_PREFIX = "/api/ws/work-nodes/"
+_LOG_SUFFIX = "/log"
class _Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urllib.parse.urlsplit(self.path)
+ if parsed.path.startswith(_WS_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
+ node_exec_id = parsed.path[len(_WS_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
+ self._handle_ws_log_tail(node_exec_id)
+ return
if parsed.path.startswith(_WS_PROJECTS_PREFIX):
self._handle_ws_subscribe(parsed.path, parsed.query)
return
@@ -134,6 +141,10 @@ class _Handler(BaseHTTPRequestHandler):
if parsed.path == "/api/settings/budget":
self._handle_budget_list()
return
+ if parsed.path.startswith(_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
+ node_exec_id = parsed.path[len(_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
+ self._handle_execution_log_get(node_exec_id, parsed.query)
+ return
entry = _STATIC_FILES.get(self.path)
if entry is None:
@@ -314,6 +325,34 @@ class _Handler(BaseHTTPRequestHandler):
self._send_json(404, {"error": "PROJECT_NOT_FOUND"})
return
+ self._send_ws_handshake_response(ws_key)
+
+ with self.server.conn_lock:
+ after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
+ self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
+ try:
+ while True:
+ try:
+ opcode, _payload = ws_subscribe.decode_client_frame(self.connection)
+ if opcode == ws_subscribe.OPCODE_CLOSE:
+ break
+ if opcode == ws_subscribe.OPCODE_PING:
+ self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_PONG))
+ except socket.timeout:
+ pass # normal: nothing from the client this poll tick
+
+ with self.server.conn_lock:
+ events, after_rowid = ws_subscribe.poll_new_events(
+ self.server.conn, project_id, after_rowid, task_id=task_id
+ )
+ for event in events:
+ self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
+ except (ConnectionError, OSError):
+ pass # client disconnected -- nothing more to send
+ except sqlite3.ProgrammingError:
+ pass # server shutting down closed the shared conn out from under this poll loop
+
+ def _send_ws_handshake_response(self, ws_key):
accept_key = ws_subscribe.compute_accept_key(ws_key)
self.connection.sendall(
(
@@ -325,8 +364,60 @@ class _Handler(BaseHTTPRequestHandler):
)
self.close_connection = True
+ def _handle_execution_log_get(self, node_exec_id, query_string):
+ params = urllib.parse.parse_qs(query_string)
+ download = params.get("download", ["false"])[0].lower() in ("true", "1")
+ try:
+ if download:
+ with self.server.conn_lock:
+ file_path = execution_log.get_log_file_path(self.server.conn, node_exec_id)
+ self._send_file_download(file_path, node_exec_id)
+ return
+ offset = int(params.get("offset", ["0"])[0])
+ limit_param = params.get("limit", [None])[0]
+ limit = int(limit_param) if limit_param is not None else None
+ with self.server.conn_lock:
+ result = execution_log.get_log(self.server.conn, node_exec_id, offset=offset, limit=limit)
+ except execution_log.LogNotFoundError:
+ self._send_json(404, {"error": "LOG_NOT_FOUND"})
+ except execution_log.LogPurgedError:
+ self._send_json(410, {"error": "LOG_PURGED"})
+ else:
+ self._send_json(200, result)
+
+ def _send_file_download(self, file_path, node_exec_id):
+ with open(file_path, "rb") as f:
+ body = f.read()
+ self.send_response(200)
+ self.send_header("Content-Type", "application/octet-stream")
+ self.send_header("Content-Disposition", 'attachment; filename="' + node_exec_id + '.log"')
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def _handle_ws_log_tail(self, node_exec_id):
+ """c_ws_log_tail (03-C §3.6.2): WS /api/ws/work-nodes/:node_exec_id/log.
+ The server actively closes the connection once the node reaches a
+ terminal status (done/blocked/failed) -- 03-C's own wording, not
+ left for the client to keep polling forever after there's nothing
+ more to send; the frontend switches to c_execution_log_get's static
+ mode on close."""
+ ws_key = self.headers.get("Sec-WebSocket-Key")
+ if self.headers.get("Upgrade", "").lower() != "websocket" or not ws_key:
+ self._send_json(400, {"error": "INVALID_INPUT", "detail": "expected a WebSocket upgrade request"})
+ return
+
+ try:
+ with self.server.conn_lock:
+ execution_log.node_exec_status(self.server.conn, node_exec_id)
+ except execution_log.NodeExecNotFoundError:
+ self._send_json(404, {"error": "NODE_EXEC_NOT_FOUND"})
+ return
+
+ self._send_ws_handshake_response(ws_key)
+
with self.server.conn_lock:
- after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
+ after_rowid = execution_log.latest_log_event_rowid(self.server.conn, node_exec_id)
self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
try:
while True:
@@ -340,11 +431,13 @@ class _Handler(BaseHTTPRequestHandler):
pass # normal: nothing from the client this poll tick
with self.server.conn_lock:
- events, after_rowid = ws_subscribe.poll_new_events(
- self.server.conn, project_id, after_rowid, task_id=task_id
- )
+ events, after_rowid = execution_log.poll_log_events(self.server.conn, node_exec_id, after_rowid)
+ status = execution_log.node_exec_status(self.server.conn, node_exec_id)
for event in events:
self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
+ if execution_log.is_terminal_status(status):
+ self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_CLOSE))
+ break
except (ConnectionError, OSError):
pass # client disconnected -- nothing more to send
except sqlite3.ProgrammingError:
@@ -377,7 +470,13 @@ class _Server(ThreadingHTTPServer):
def server_close(self):
super().server_close()
- self.conn.close()
+ # a WS handler thread can still be mid-poll on another thread at
+ # shutdown (its next tick would just get a clean ProgrammingError,
+ # already caught) -- closing conn without this lock races the C
+ # sqlite3 connection object across threads with no synchronization
+ # at all, which segfaults rather than raising cleanly.
+ with self.conn_lock:
+ self.conn.close()
def create_server(host="127.0.0.1", port=0, db_path=None):
diff --git a/src/execution_log.py b/src/execution_log.py
new file mode 100644
index 0000000..f68cba3
--- /dev/null
+++ b/src/execution_log.py
@@ -0,0 +1,213 @@
+"""c_execution_log_get + c_ws_log_tail (03-C §3.6.1/§3.6.2) + p_log_retention_cleanup
+(04-P §4.5): raw execution log storage, retrieval, real-time streaming, and
+scheduled cleanup -- REQ-13.
+
+U-013 (work_node_run.py) already manages execution_log_file's row lifecycle
+(insert status='active' -> update status='closed') but nothing ever wrote
+actual log content -- append_log() is that missing write path, meant to be
+called from inside a run_fn as it works (this module doesn't know or care
+what a node's actual work is, same boundary work_node_run.py itself draws
+around run_fn).
+
+p_log_retention_cleanup is a scheduled job (04-P: "定时触发...具体周期由
+平台调度配置，非用户可配") -- no scheduler daemon exists in this codebase
+yet and building one is out of this Unit's scope (coverage only lists
+c_execution_log_get/c_ws_log_tail); run_log_retention_cleanup() is the pure,
+directly-testable function a future scheduler calls periodically.
+"""
+
+import json
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+
+class LogNotFoundError(Exception):
+ """LOG_NOT_FOUND -- 03-C §3.6.1: node_exec_id has no execution_log_file row."""
+
+
+class LogPurgedError(Exception):
+ """LOG_PURGED -- 03-C §3.6.1: the log was already cleaned up by retention."""
+
+
+class NodeExecNotFoundError(Exception):
+ """NODE_EXEC_NOT_FOUND -- 03-C §3.6.2: c_ws_log_tail subscribe target doesn't exist."""
+
+
+_TERMINAL_NODE_STATUSES = {"done", "blocked", "failed"}
+
+
+def is_terminal_status(status):
+ """02-V §2.7.1 / 03-C §3.6.2: once a node reaches one of these, no more
+ log content will ever be appended -- c_ws_log_tail actively closes."""
+ return status in _TERMINAL_NODE_STATUSES
+
+
+def append_log(conn, node_exec_id, content):
+ """Appends `content` to the node's log file on disk, bumps
+ execution_log_file.size_bytes, and writes
+ execution_event(event_type='log_appended') so c_ws_log_tail subscribers
+ see it in realtime mode (02-V §2.7.1)."""
+ row = conn.execute(
+ "SELECT log_file_id, project_id, file_path FROM execution_log_file WHERE node_exec_id = ?",
+ (node_exec_id,),
+ ).fetchone()
+ if row is None:
+ raise LogNotFoundError(node_exec_id)
+ log_file_id, project_id, file_path = row
+
+ os.makedirs(os.path.dirname(file_path), exist_ok=True)
+ with open(file_path, "a", encoding="utf-8") as f:
+ f.write(content)
+ new_size = os.path.getsize(file_path)
+ conn.execute("UPDATE execution_log_file SET size_bytes = ? WHERE log_file_id = ?", (new_size, log_file_id))
+
+ task_id = conn.execute(
+ "SELECT wgr.task_id FROM work_node_execution wne "
+ "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id WHERE wne.node_exec_id = ?",
+ (node_exec_id,),
+ ).fetchone()[0]
+ conn.execute(
+ "INSERT INTO execution_event (event_id, project_id, task_id, node_exec_id, event_type, payload) "
+ "VALUES (?, ?, ?, ?, 'log_appended', ?)",
+ (str(uuid.uuid4()), project_id, task_id, node_exec_id, json.dumps({"appended_content": content})),
+ )
+ conn.commit()
+
+
+def get_log(conn, node_exec_id, offset=0, limit=None):
+ """c_execution_log_get (03-C §3.6.1) static-mode read, paginated by
+ byte offset/limit so a large file isn't loaded all at once."""
+ row = conn.execute(
+ "SELECT file_path, format, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
+ ).fetchone()
+ if row is None:
+ raise LogNotFoundError(node_exec_id)
+ file_path, fmt, status = row
+ if status == "purged":
+ raise LogPurgedError(node_exec_id)
+
+ if not os.path.isfile(file_path):
+ return {"content": "", "format": fmt, "total_size_bytes": 0, "status": status}
+
+ total_size = os.path.getsize(file_path)
+ with open(file_path, "rb") as f:
+ f.seek(offset)
+ chunk = f.read(limit) if limit is not None else f.read()
+ return {"content": chunk.decode("utf-8", errors="replace"), "format": fmt, "total_size_bytes": total_size, "status": status}
+
+
+def get_log_file_path(conn, node_exec_id):
+ """For the download=true response path (send the raw file, not JSON)."""
+ row = conn.execute(
+ "SELECT file_path, status FROM execution_log_file WHERE node_exec_id = ?", (node_exec_id,)
+ ).fetchone()
+ if row is None:
+ raise LogNotFoundError(node_exec_id)
+ file_path, status = row
+ if status == "purged":
+ raise LogPurgedError(node_exec_id)
+ return file_path
+
+
+def node_exec_status(conn, node_exec_id):
+ row = conn.execute("SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,)).fetchone()
+ if row is None:
+ raise NodeExecNotFoundError(node_exec_id)
+ return row[0]
+
+
+def latest_log_event_rowid(conn, node_exec_id):
+ row = conn.execute(
+ "SELECT MAX(rowid) FROM execution_event WHERE node_exec_id = ? AND event_type = 'log_appended'",
+ (node_exec_id,),
+ ).fetchone()
+ return row[0] or 0
+
+
+def poll_log_events(conn, node_exec_id, after_rowid):
+ """c_ws_log_tail's push source -- same rowid-increment polling pattern
+ as ws_subscribe.poll_new_events, scoped to one node_exec_id's
+ log_appended events."""
+ rows = conn.execute(
+ "SELECT rowid, payload, created_at FROM execution_event "
+ "WHERE node_exec_id = ? AND event_type = 'log_appended' AND rowid > ? ORDER BY rowid",
+ (node_exec_id, after_rowid),
+ ).fetchall()
+ events = [
+ {
+ "node_exec_id": node_exec_id,
+ "appended_content": json.loads(r[1])["appended_content"],
+ "created_at": r[2],
+ }
+ for r in rows
+ ]
+ new_after_rowid = rows[-1][0] if rows else after_rowid
+ return events, new_after_rowid
+
+
+def run_log_retention_cleanup(conn, now=None, config_id="default"):
+ """p_log_retention_cleanup (04-P §4.5): idle -> scanning -> purging -> idle.
+
+ Age-expired 'closed' logs are always candidates. If max_total_size_mb is
+ set and the remaining (non-expired) total still exceeds it, the
+ oldest-first remaining logs are added as candidates too, until the kept
+ total drops back under the cap. A file that fails to delete (e.g. still
+ open elsewhere) is recorded as a failure and left status='closed' for
+ the next cycle to retry -- it must not block purging the rest of the
+ batch (REQ-13 failure case).
+
+ Returns {"purged": [log_file_id, ...], "failed": [log_file_id, ...]}.
+ """
+ config_row = conn.execute(
+ "SELECT retention_days, max_total_size_mb FROM log_retention_config WHERE config_id = ?", (config_id,)
+ ).fetchone()
+ if config_row is None:
+ raise ValueError("no such log_retention_config: " + config_id)
+ retention_days, max_total_size_mb = config_row
+
+ now = now if now is not None else datetime.now(timezone.utc)
+ cutoff_str = (now - timedelta(days=retention_days)).strftime("%Y-%m-%d %H:%M:%S")
+
+ expired = conn.execute(
+ "SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
+ "WHERE status = 'closed' AND created_at < ? ORDER BY created_at",
+ (cutoff_str,),
+ ).fetchall()
+ kept = conn.execute(
+ "SELECT log_file_id, file_path, size_bytes FROM execution_log_file "
+ "WHERE status = 'closed' AND created_at >= ? ORDER BY created_at",
+ (cutoff_str,),
+ ).fetchall()
+
+ candidates = list(expired)
+ if max_total_size_mb is not None:
+ limit_bytes = max_total_size_mb * 1024 * 1024
+ kept_total = sum(row[2] for row in kept)
+ for row in kept:
+ if kept_total <= limit_bytes:
+ break
+ candidates.append(row)
+ kept_total -= row[2]
+
+ purged, failed = [], []
+ for log_file_id, file_path, _size in candidates:
+ try:
+ if os.path.isfile(file_path):
+ os.remove(file_path)
+ elif os.path.exists(file_path):
+ # something's occupying the path but isn't a plain file
+ # (e.g. a directory) -- can't silently treat that as "already
+ # gone", it's a real delete failure
+ raise OSError("not a regular file: " + file_path)
+ except OSError:
+ failed.append(log_file_id)
+ continue
+ conn.execute(
+ "UPDATE execution_log_file SET status = 'purged', purged_at = CURRENT_TIMESTAMP WHERE log_file_id = ?",
+ (log_file_id,),
+ )
+ purged.append(log_file_id)
+ conn.commit()
+
+ return {"purged": purged, "failed": failed}
diff --git a/src/test_backend_server.py b/src/test_backend_server.py
index 1442057..22e4184 100644
--- a/src/test_backend_server.py
+++ b/src/test_backend_server.py
@@ -20,7 +20,7 @@ import time
import unittest
from unittest.mock import patch
-from src import backend_server, ws_subscribe
+from src import backend_server, execution_log, ws_subscribe
class ServerStartStopTest(unittest.TestCase):
@@ -536,16 +536,23 @@ def _read_http_response_head(sock):
return buf.decode("ascii", errors="replace")
-def _read_server_frame(sock):
+def _read_server_frame_raw(sock):
"""Server->client frames are never masked (RFC 6455 §5.1) -- simpler
- decode than ws_subscribe.decode_client_frame, which expects masking."""
+ decode than ws_subscribe.decode_client_frame, which expects masking.
+ Returns (opcode, payload_bytes)."""
first_two = _recv_exact(sock, 2)
+ opcode = first_two[0] & 0x0F
length = first_two[1] & 0x7F
if length == 126:
length = int.from_bytes(_recv_exact(sock, 2), "big")
elif length == 127:
length = int.from_bytes(_recv_exact(sock, 8), "big")
payload = _recv_exact(sock, length)
+ return opcode, payload
+
+
+def _read_server_frame(sock):
+ _opcode, payload = _read_server_frame_raw(sock)
return json.loads(payload.decode("utf-8"))
@@ -665,6 +672,147 @@ class WsSubscribeRouteTest(unittest.TestCase):
self.assertEqual(event["event_id"], "evt-related")
+class ExecutionLogRouteTest(unittest.TestCase):
+ """U-027: c_execution_log_get (GET /api/work-nodes/:node_exec_id/log)
+ and c_ws_log_tail (WS /api/ws/work-nodes/:node_exec_id/log)."""
+
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.server = backend_server.create_server(db_path=":memory:")
+ self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+ self.thread.start()
+ conn = self.server.conn
+ conn.execute(
+ "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+ "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+ )
+ conn.execute(
+ "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+ "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
+ )
+ conn.execute(
+ "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+ "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+ )
+ conn.execute(
+ "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+ "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
+ )
+ conn.execute(
+ "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+ "input_schema_ref, output_schema_ref, input_payload, status) "
+ "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
+ )
+ conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
+ "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
+ (os.path.join(self.tmp.name, "node-1.jsonl"),),
+ )
+ conn.commit()
+
+ def tearDown(self):
+ self.server.shutdown()
+ self.server.server_close()
+ self.thread.join(timeout=5)
+
+ def _get(self, path):
+ host, port = self.server.server_address
+ conn = http.client.HTTPConnection(host, port, timeout=5)
+ conn.request("GET", path)
+ resp = conn.getresponse()
+ body = resp.read()
+ headers = dict(resp.getheaders())
+ conn.close()
+ return resp.status, body, headers
+
+ def test_returns_static_content(self):
+ with self.server.conn_lock:
+ execution_log.append_log(self.server.conn, "node-1", "hello world")
+ status, body, _ = self._get("/api/work-nodes/node-1/log")
+ self.assertEqual(status, 200)
+ parsed = json.loads(body)
+ self.assertEqual(parsed["content"], "hello world")
+ self.assertEqual(parsed["total_size_bytes"], len("hello world"))
+
+ def test_pagination(self):
+ with self.server.conn_lock:
+ execution_log.append_log(self.server.conn, "node-1", "0123456789")
+ status, body, _ = self._get("/api/work-nodes/node-1/log?offset=2&limit=3")
+ self.assertEqual(status, 200)
+ self.assertEqual(json.loads(body)["content"], "234")
+
+ def test_download_returns_raw_file_with_attachment_headers(self):
+ with self.server.conn_lock:
+ execution_log.append_log(self.server.conn, "node-1", "raw log bytes")
+ status, body, headers = self._get("/api/work-nodes/node-1/log?download=true")
+ self.assertEqual(status, 200)
+ self.assertEqual(body, b"raw log bytes")
+ self.assertIn("attachment", headers.get("Content-Disposition", ""))
+
+ def test_nonexistent_node_returns_404_log_not_found(self):
+ status, body, _ = self._get("/api/work-nodes/no-such-node/log")
+ self.assertEqual(status, 404)
+ self.assertEqual(json.loads(body)["error"], "LOG_NOT_FOUND")
+
+ def test_purged_log_returns_410(self):
+ with self.server.conn_lock:
+ self.server.conn.execute("UPDATE execution_log_file SET status = 'purged' WHERE log_file_id = 'log-1'")
+ self.server.conn.commit()
+ status, body, _ = self._get("/api/work-nodes/node-1/log")
+ self.assertEqual(status, 410)
+ self.assertEqual(json.loads(body)["error"], "LOG_PURGED")
+
+ def _connect(self):
+ host, port = self.server.server_address
+ return socket.create_connection((host, port), timeout=5)
+
+ def test_ws_handshake_succeeds_for_existing_node(self):
+ sock = self._connect()
+ self.addCleanup(sock.close)
+ key = base64.b64encode(b"0123456789012345").decode("ascii")
+ sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
+ head = _read_http_response_head(sock)
+ self.assertIn("101 Switching Protocols", head.splitlines()[0])
+
+ def test_ws_nonexistent_node_rejected_before_upgrade(self):
+ sock = self._connect()
+ self.addCleanup(sock.close)
+ key = base64.b64encode(b"0123456789012345").decode("ascii")
+ sock.sendall(_ws_handshake_request("/api/ws/work-nodes/no-such-node/log", key))
+ head = _read_http_response_head(sock)
+ self.assertIn("404", head.splitlines()[0])
+
+ def test_ws_pushes_appended_log_content(self):
+ sock = self._connect()
+ self.addCleanup(sock.close)
+ key = base64.b64encode(b"0123456789012345").decode("ascii")
+ sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
+ _read_http_response_head(sock)
+
+ with self.server.conn_lock:
+ execution_log.append_log(self.server.conn, "node-1", "streamed chunk")
+
+ sock.settimeout(5)
+ event = _read_server_frame(sock)
+ self.assertEqual(event["appended_content"], "streamed chunk")
+
+ def test_ws_closes_when_node_reaches_terminal_status(self):
+ sock = self._connect()
+ self.addCleanup(sock.close)
+ key = base64.b64encode(b"0123456789012345").decode("ascii")
+ sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
+ _read_http_response_head(sock)
+
+ with self.server.conn_lock:
+ self.server.conn.execute("UPDATE work_node_execution SET status = 'done' WHERE node_exec_id = 'node-1'")
+ self.server.conn.commit()
+
+ sock.settimeout(5)
+ opcode, _payload = _read_server_frame_raw(sock)
+ self.assertEqual(opcode, ws_subscribe.OPCODE_CLOSE)
+
+
class ZeroThirdPartyDependencyTest(unittest.TestCase):
"""05 §5.7: no Flask/FastAPI/Express/Node.js -- checked statically so this
doesn't silently rot if someone adds an import later."""
diff --git a/src/test_execution_log.py b/src/test_execution_log.py
new file mode 100644
index 0000000..8e0c31f
--- /dev/null
+++ b/src/test_execution_log.py
@@ -0,0 +1,287 @@
+"""U-027 tests: c_execution_log_get + c_ws_log_tail's data layer +
+p_log_retention_cleanup (REQ-13 happy/boundary/failure). Real sqlite + real
+tempfile-backed log files, no mocking needed.
+
+Run: python3 -m unittest src.test_execution_log -v
+"""
+
+import json
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from src import execution_log, schema
+
+
+class ExecutionLogTestBase(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ self.conn = schema.connect(":memory:")
+ schema.create_schema(self.conn)
+ self.conn.execute(
+ "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+ "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+ )
+ self.conn.execute(
+ "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+ "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
+ )
+ self.conn.execute(
+ "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+ "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+ )
+ self.conn.execute(
+ "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+ "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
+ )
+ self.conn.execute(
+ "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+ "input_schema_ref, output_schema_ref, input_payload, status) "
+ "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
+ )
+ self.conn.commit()
+
+ def tearDown(self):
+ self.conn.close()
+
+ def _log_path(self):
+ return os.path.join(self.tmp.name, "node-1.jsonl")
+
+ def _insert_log_file(self, status="active"):
+ self.conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
+ "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', ?)",
+ (self._log_path(), status),
+ )
+ self.conn.commit()
+
+
+class AppendLogTest(ExecutionLogTestBase):
+ def test_appends_content_to_file_and_bumps_size(self):
+ self._insert_log_file()
+ execution_log.append_log(self.conn, "node-1", "line one\n")
+ execution_log.append_log(self.conn, "node-1", "line two\n")
+ with open(self._log_path()) as f:
+ self.assertEqual(f.read(), "line one\nline two\n")
+ size = self.conn.execute("SELECT size_bytes FROM execution_log_file WHERE log_file_id = 'log-1'").fetchone()[0]
+ self.assertEqual(size, len("line one\nline two\n"))
+
+ def test_writes_log_appended_execution_event(self):
+ self._insert_log_file()
+ execution_log.append_log(self.conn, "node-1", "hello\n")
+ row = self.conn.execute(
+ "SELECT event_type, task_id, payload FROM execution_event WHERE node_exec_id = 'node-1'"
+ ).fetchone()
+ self.assertEqual(row[0], "log_appended")
+ self.assertEqual(row[1], "task-1")
+ self.assertEqual(json.loads(row[2]), {"appended_content": "hello\n"})
+
+ def test_creates_parent_directory_if_missing(self):
+ nested_path = os.path.join(self.tmp.name, "a", "b", "node-1.jsonl")
+ self.conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
+ "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
+ (nested_path,),
+ )
+ self.conn.commit()
+ execution_log.append_log(self.conn, "node-1", "x")
+ self.assertTrue(os.path.isfile(nested_path))
+
+ def test_nonexistent_node_exec_raises(self):
+ with self.assertRaises(execution_log.LogNotFoundError):
+ execution_log.append_log(self.conn, "no-such-node", "x")
+
+
+class GetLogTest(ExecutionLogTestBase):
+ def test_returns_full_content_and_size(self):
+ self._insert_log_file(status="closed")
+ with open(self._log_path(), "w") as f:
+ f.write("all the log content")
+ result = execution_log.get_log(self.conn, "node-1")
+ self.assertEqual(result["content"], "all the log content")
+ self.assertEqual(result["total_size_bytes"], len("all the log content"))
+ self.assertEqual(result["status"], "closed")
+
+ def test_pagination_offset_and_limit(self):
+ self._insert_log_file(status="closed")
+ with open(self._log_path(), "w") as f:
+ f.write("0123456789")
+ result = execution_log.get_log(self.conn, "node-1", offset=3, limit=4)
+ self.assertEqual(result["content"], "3456")
+ self.assertEqual(result["total_size_bytes"], 10)
+
+ def test_missing_physical_file_returns_empty_not_error(self):
+ self._insert_log_file(status="closed")
+ result = execution_log.get_log(self.conn, "node-1")
+ self.assertEqual(result["content"], "")
+ self.assertEqual(result["total_size_bytes"], 0)
+
+ def test_nonexistent_node_exec_raises_log_not_found(self):
+ with self.assertRaises(execution_log.LogNotFoundError):
+ execution_log.get_log(self.conn, "no-such-node")
+
+ def test_purged_log_raises(self):
+ self._insert_log_file(status="purged")
+ with self.assertRaises(execution_log.LogPurgedError):
+ execution_log.get_log(self.conn, "node-1")
+
+
+class NodeExecStatusTest(ExecutionLogTestBase):
+ def test_returns_status(self):
+ self.assertEqual(execution_log.node_exec_status(self.conn, "node-1"), "running")
+
+ def test_nonexistent_raises(self):
+ with self.assertRaises(execution_log.NodeExecNotFoundError):
+ execution_log.node_exec_status(self.conn, "no-such-node")
+
+ def test_is_terminal_status(self):
+ for status in ("done", "blocked", "failed"):
+ self.assertTrue(execution_log.is_terminal_status(status))
+ for status in ("pending", "running", "reflecting", "retrying", "waiting_for_quota"):
+ self.assertFalse(execution_log.is_terminal_status(status))
+
+
+class PollLogEventsTest(ExecutionLogTestBase):
+ def test_polls_only_new_log_appended_events_for_this_node(self):
+ self._insert_log_file()
+ after = execution_log.latest_log_event_rowid(self.conn, "node-1")
+ execution_log.append_log(self.conn, "node-1", "chunk-1")
+ events, new_after = execution_log.poll_log_events(self.conn, "node-1", after)
+ self.assertEqual(len(events), 1)
+ self.assertEqual(events[0]["appended_content"], "chunk-1")
+ self.assertGreater(new_after, after)
+
+ events_again, after2 = execution_log.poll_log_events(self.conn, "node-1", new_after)
+ self.assertEqual(events_again, [])
+ self.assertEqual(after2, new_after)
+
+ def test_other_node_events_excluded(self):
+ self.conn.execute(
+ "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+ "input_schema_ref, output_schema_ref, input_payload, status) "
+ "VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
+ )
+ self.conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
+ "VALUES ('log-2', 'node-2', 'proj-1', ?, 'jsonl', 'active')",
+ (os.path.join(self.tmp.name, "node-2.jsonl"),),
+ )
+ self._insert_log_file()
+ self.conn.commit()
+ execution_log.append_log(self.conn, "node-2", "not mine")
+ events, _ = execution_log.poll_log_events(self.conn, "node-1", 0)
+ self.assertEqual(events, [])
+
+
+class LogRetentionCleanupTest(ExecutionLogTestBase):
+ def setUp(self):
+ super().setUp()
+ self.conn.execute(
+ "INSERT INTO log_retention_config (config_id, retention_days, updated_by) "
+ "VALUES ('default', 1, 'test')"
+ )
+ self.conn.commit()
+
+ def _insert_closed_log(self, log_file_id, created_at, size_bytes=100, file_content="x"):
+ node_exec_id = log_file_id + "-node"
+ self.conn.execute(
+ "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+ "input_schema_ref, output_schema_ref, input_payload, status) "
+ "VALUES (?, 'run-1', ?, 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')",
+ (node_exec_id, log_file_id + "-n"),
+ )
+ path = os.path.join(self.tmp.name, log_file_id + ".jsonl")
+ with open(path, "w") as f:
+ f.write(file_content)
+ self.conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
+ "status, size_bytes, created_at) VALUES (?, ?, 'proj-1', ?, 'jsonl', 'closed', ?, ?)",
+ (log_file_id, node_exec_id, path, size_bytes, created_at),
+ )
+ self.conn.commit()
+ return path
+
+ def test_expired_closed_log_purged_and_file_deleted(self):
+ old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
+ path = self._insert_closed_log("log-old", old_ts)
+ result = execution_log.run_log_retention_cleanup(self.conn)
+ self.assertEqual(result["purged"], ["log-old"])
+ status, purged_at = self.conn.execute(
+ "SELECT status, purged_at FROM execution_log_file WHERE log_file_id = 'log-old'"
+ ).fetchone()
+ self.assertEqual(status, "purged")
+ self.assertIsNotNone(purged_at)
+ self.assertFalse(os.path.isfile(path))
+
+ def test_not_yet_expired_log_untouched(self):
+ recent_ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+ path = self._insert_closed_log("log-fresh", recent_ts)
+ result = execution_log.run_log_retention_cleanup(self.conn)
+ self.assertEqual(result["purged"], [])
+ status = self.conn.execute(
+ "SELECT status FROM execution_log_file WHERE log_file_id = 'log-fresh'"
+ ).fetchone()[0]
+ self.assertEqual(status, "closed")
+ self.assertTrue(os.path.isfile(path))
+
+ def test_viewing_purged_log_raises_purged_not_error(self):
+ old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
+ self.conn.execute(
+ "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+ "input_schema_ref, output_schema_ref, input_payload, status) "
+ "VALUES ('node-old', 'run-1', 'n-old', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')"
+ )
+ path = os.path.join(self.tmp.name, "log-old2.jsonl")
+ with open(path, "w") as f:
+ f.write("x")
+ self.conn.execute(
+ "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
+ "status, created_at) VALUES ('log-old2', 'node-old', 'proj-1', ?, 'jsonl', 'closed', ?)",
+ (path, old_ts),
+ )
+ self.conn.commit()
+ execution_log.run_log_retention_cleanup(self.conn)
+ with self.assertRaises(execution_log.LogPurgedError):
+ execution_log.get_log(self.conn, "node-old")
+
+ def test_max_total_size_pulls_in_oldest_non_expired_candidates(self):
+ self.conn.execute(
+ "UPDATE log_retention_config SET retention_days = 30, max_total_size_mb = 1 WHERE config_id = 'default'"
+ )
+ self.conn.commit()
+ now = datetime.now(timezone.utc)
+ one_mb = 1024 * 1024
+ # both within retention_days=30, but combined over the 1MB cap --
+ # the older of the two should get pulled in as a candidate too
+ self._insert_closed_log("log-a", (now - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
+ self._insert_closed_log("log-b", (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
+ result = execution_log.run_log_retention_cleanup(self.conn, now=now)
+ self.assertEqual(result["purged"], ["log-a"])
+ status_b = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-b'").fetchone()[0]
+ self.assertEqual(status_b, "closed")
+
+ def test_delete_failure_recorded_and_others_still_purged(self):
+ old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
+ self._insert_closed_log("log-ok", old_ts)
+ bad_path = self._insert_closed_log("log-bad", old_ts)
+ os.remove(bad_path) # simulate "file missing/inaccessible" -- os.remove will raise
+ os.mkdir(bad_path) # replace it with a directory so os.remove() fails with OSError
+ self.addCleanup(lambda: os.path.isdir(bad_path) and os.rmdir(bad_path))
+
+ result = execution_log.run_log_retention_cleanup(self.conn)
+ self.assertEqual(result["purged"], ["log-ok"])
+ self.assertEqual(result["failed"], ["log-bad"])
+ status = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-bad'").fetchone()[0]
+ self.assertEqual(status, "closed")
+
+ def test_missing_config_raises(self):
+ self.conn.execute("DELETE FROM log_retention_config")
+ self.conn.commit()
+ with self.assertRaises(ValueError):
+ execution_log.run_log_retention_cleanup(self.conn)
+
+
+if __name__ == "__main__":
+ unittest.main()