"""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}