From c18e4d4112f911ffadaab68826de2f872ed46ef0 Mon Sep 17 00:00:00 2001 From: graphlemon Date: Tue, 4 Aug 2026 16:44:10 +0800 Subject: [PATCH] graphlemon: auto-commit uncommitted changes for 08b89f13-2c70-4f72-b743-6c6e81702b85 --- src/backend_server.py | 15 +++++++++++++-- src/test_backend_server.py | 13 +++++++++++++ src/test_work_graph_orchestrate.py | 100 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/work_graph_orchestrate.py | 56 ++++++++++++++++++++++++++++++++++++++++++++++++-------- 4 files changed, 173 insertions(+), 11 deletions(-) diff --git a/src/backend_server.py b/src/backend_server.py index 8716596..2b442d4 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -821,8 +821,19 @@ class _Handler(BaseHTTPRequestHandler): 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() + # execution_log_file rows are created up front (work_node_run.py) but + # the backing file itself is only ever written by execution_log. + # append_log() -- a node that errored before its first log write (or + # a file removed out-of-band) leaves a row with no file on disk yet; + # that's a clean 404, not an uncaught FileNotFoundError bubbling up + # as a bare 500 (bug A2, same root cause as A1: see execution_log.py + # and work_graph_orchestrate.py's DU-01 notes). + try: + with open(file_path, "rb") as f: + body = f.read() + except FileNotFoundError: + self._send_json(404, {"error": "LOG_FILE_MISSING"}) + return self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Content-Disposition", 'attachment; filename="' + node_exec_id + '.log"') diff --git a/src/test_backend_server.py b/src/test_backend_server.py index daad261..c156be1 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -1145,6 +1145,19 @@ class ExecutionLogRouteTest(unittest.TestCase): self.assertEqual(status, 404) self.assertEqual(json.loads(body)["error"], "LOG_NOT_FOUND") + def test_download_when_backing_file_missing_returns_clean_404_not_uncaught_exception(self): + # Bug A2: execution_log_file row exists (inserted in setUp) but its + # file_path was never actually written to disk -- append_log() is + # what creates the file, and this test deliberately never calls it, + # matching the exact real-world state before DU-01's fix (a node + # that errored before its first log write, or any other row whose + # backing file went missing). _send_file_download() must catch this + # itself, not let a bare FileNotFoundError escape as an uncaught 500. + self.assertFalse(os.path.exists(os.path.join(self.tmp.name, "node-1.jsonl"))) + status, body, _ = self._get("/api/work-nodes/node-1/log?download=true") + self.assertEqual(status, 404) + self.assertEqual(json.loads(body)["error"], "LOG_FILE_MISSING") + 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'") diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py index 31c724a..a9c4ceb 100644 --- a/src/test_work_graph_orchestrate.py +++ b/src/test_work_graph_orchestrate.py @@ -7,10 +7,13 @@ spike -- this module's own job is orchestration, not re-proving that). Run: python3 -m unittest src.test_work_graph_orchestrate -v """ +import json +import os +import tempfile import unittest from unittest.mock import patch -from src import paths, schema, work_graph_execute, work_graph_orchestrate as wgo, work_node_run +from src import execution_log, paths, schema, work_graph_execute, work_graph_orchestrate as wgo, work_node_run class TopologicalOrderTest(unittest.TestCase): @@ -64,6 +67,17 @@ class TopologicalOrderTest(unittest.TestCase): class RunGraphTestBase(unittest.TestCase): def setUp(self): + # DU-01: run_fn now really calls execution_log.append_log(), which + # writes to paths.log_file_path() (default ~/.graphlemon/logs/...) -- + # isolate that under a throwaway tempdir so real (non-mocked) + # run_work_node/run_fn calls in this test module never touch the + # developer's actual home directory, same pattern as test_paths.py. + self.home_tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.home_tmp.cleanup) + self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") + os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name + self.addCleanup(self._restore_graphlemon_home) + self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( @@ -80,6 +94,12 @@ class RunGraphTestBase(unittest.TestCase): ) self.conn.commit() + def _restore_graphlemon_home(self): + if self.old_graphlemon_home is None: + os.environ.pop("GRAPHLEMON_HOME", None) + else: + os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home + def tearDown(self): self.conn.close() @@ -306,5 +326,83 @@ class SelfTargetForcesWorktreeTest(RunGraphTestBase): self.assertTrue(mock_run.call_args.kwargs["needs_worktree"]) +class RunFnLogsExecutionEventsTest(RunGraphTestBase): + """DU-01 (bug A1/A2 root cause): `_make_run_fn()`'s `run_fn` closure must + actually call `execution_log.append_log()` around the real + `execute_work_graph()` call, so `execution_log_file`'s backing file on + disk genuinely has content -- not just a DB row tracking its lifecycle.""" + + def _node_exec_id(self, node_id="brd"): + return self.conn.execute( + "SELECT node_exec_id FROM work_node_execution WHERE run_id = 'run-1' AND node_id = ?", (node_id,) + ).fetchone()[0] + + def _log_lines(self, node_exec_id): + content = execution_log.get_log(self.conn, node_exec_id)["content"] + return [json.loads(line) for line in content.splitlines() if line] + + def test_success_writes_node_start_then_node_completed(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "brd text"}): + wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, + ) + + events = self._log_lines(self._node_exec_id()) + self.assertEqual([e["event"] for e in events], ["node_start", "node_completed"]) + self.assertEqual(events[1]["output"], {"content": "brd text"}) + + # also exercise the real HTTP-facing read path end to end (not just + # the module function) -- an empty file was exactly bug A1. + row = self.conn.execute( + "SELECT size_bytes FROM execution_log_file WHERE node_exec_id = ?", (self._node_exec_id(),) + ).fetchone() + self.assertGreater(row[0], 0) + + def test_classifier_blocked_writes_node_failed_and_still_propagates(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object( + work_graph_execute, "execute_work_graph", + side_effect=work_graph_execute.ClassifierBlockedError("blocked by classifier"), + ): + with self.assertRaises(work_graph_execute.ClassifierBlockedError): + wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, + ) + + events = self._log_lines(self._node_exec_id()) + self.assertEqual([e["event"] for e in events], ["node_start", "node_failed"]) + self.assertIn("blocked by classifier", events[1]["error"]) + + def test_usage_limit_writes_quota_limit_then_retries(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + attempts = {"n": 0} + + def fake_execute(cwd, name, args, output_schema, timeout=600): + attempts["n"] += 1 + if attempts["n"] == 1: + raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session") + return {"content": "recovered"} + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + outputs = wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, + sleep=lambda s: None, now=lambda: 0.0, + ) + self.assertEqual(outputs["brd"], {"content": "recovered"}) + + events = self._log_lines(self._node_exec_id()) + self.assertEqual( + [e["event"] for e in events], ["node_start", "quota_limit", "node_start", "node_completed"] + ) + self.assertEqual(events[1]["quota_limit_type"], "session") + + if __name__ == "__main__": unittest.main() diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py index fb0b95b..d2562f5 100644 --- a/src/work_graph_orchestrate.py +++ b/src/work_graph_orchestrate.py @@ -39,13 +39,42 @@ subgraph IS just itself), escalate to `work_graph_run.status='blocked'` + raise if the regenerated content still fails. Not a general multi-node re-planning mechanism -- that would be new, unscoped complexity 04-P itself doesn't describe either. + +**DU-01 (2026-08-04, bug A1/A2 root-cause fix)**: `_make_run_fn()`'s `run_fn` +closure is the natural place to call `execution_log.append_log()` -- it's +the one spot that both has `node_exec_id` and wraps the actual blocking +`execute_work_graph()` call, and `work_node_run.run_work_node()` already +guarantees an `execution_log_file` row (status='active') exists before +`run_fn` is ever invoked. Before this, `append_log()` was fully implemented +(U-013/U-027) but never called by any production code path, so +`execution_log_file` rows existed with zero bytes of real content forever +-- the console's "暂无日志内容" was a correct rendering of genuinely empty +data, and downloading it crashed `backend_server._send_file_download()` +with an uncaught `FileNotFoundError` (also fixed here, see that function). +This is one blocking `agent()` call per node (03-C's own dynamic-workflow +design, see this module's own docstring above) with no mid-flight +progress to report, so logging is 4 discrete events around that one call, +not a real character-by-character stream: `node_start` (before the call), +`node_completed` (after a successful return), `node_failed` (any other +exception, re-raised unchanged after logging), `quota_limit` (the +`UsageLimitReachedError` case, logged before translating to +`work_node_run`'s own `_UsageLimitSignal`). A node retried by +`run_work_node`'s own retry loop calls `run_fn` again, appending another +`node_start`/... pair per attempt -- that's an accurate log of what +actually happened (N attempts), not a bug. """ import json import time import uuid -from src import paths, work_graph_execute, work_node_run +from src import execution_log, paths, work_graph_execute, work_node_run + + +def _log_event(conn, node_exec_id, event_type, **payload): + execution_log.append_log( + conn, node_exec_id, json.dumps({"event": event_type, **payload}) + "\n" + ) class ConsistencyCheckFailedError(Exception): @@ -138,19 +167,28 @@ def _render_node_script(script_name, org_node, prompt, output_schema, model): ) -def _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout): +def _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout): script_name = "graphlemon-node-" + node_exec_id.replace("-", "") def run_fn(cwd, model=None): + _log_event(conn, node_exec_id, "node_start", org_node=org_node, model=model) script_content = _render_node_script(script_name, org_node, prompt, output_schema, model) work_graph_execute.write_workflow_script(cwd, script_name, script_content) try: - return work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout) + output = work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout) except work_graph_execute.UsageLimitReachedError as e: + _log_event(conn, node_exec_id, "quota_limit", quota_limit_type=e.quota_limit_type) work_node_run.raise_usage_limit_signal(e.quota_limit_type) - # ClassifierBlockedError / generic WorkGraphExecuteError propagate - # as-is -- run_work_node's own `except Exception` clause marks the - # node 'blocked' and re-raises, no special handling needed here. + except Exception as e: + # ClassifierBlockedError / generic WorkGraphExecuteError -- logged + # then propagated as-is; run_work_node's own `except Exception` + # clause marks the node 'blocked' and re-raises, no special + # handling needed here beyond recording what actually happened. + _log_event(conn, node_exec_id, "node_failed", error=str(e)) + raise + else: + _log_event(conn, node_exec_id, "node_completed", output=output) + return output return run_fn @@ -215,7 +253,7 @@ def run_graph( {"node_id": node_id}, ) prompt = build_prompt(node, outputs) - run_fn = _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout) + run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout) output = work_node_run.run_work_node( conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn, needs_worktree=effective_needs_worktree, repo_path=target_folder_path, @@ -253,7 +291,9 @@ def _verify_and_maybe_regenerate( prompt + "\n\nA consistency reviewer flagged issues with your previous draft: " + "; ".join(feedback or []) + "\nProduce a corrected version resolving these issues." ) - reflect_run_fn = _make_run_fn(target_folder_path, reflect_node_exec_id, org_node, reflect_prompt, output_schema, timeout) + reflect_run_fn = _make_run_fn( + conn, target_folder_path, reflect_node_exec_id, org_node, reflect_prompt, output_schema, timeout + ) regenerated = work_node_run.run_work_node( conn, reflect_node_exec_id, {"node_id": node_id}, input_schema, output_schema, reflect_run_fn, needs_worktree=needs_worktree, repo_path=target_folder_path, -- libgit2 0.21.4