graphlemon: merge 08b89f13-2c70-4f72-b743-6c6e81702b85

graphlemon
2 parents 2cf260f8 c18e4d41
Showing 4 changed files with 173 additions and 11 deletions
src/backend_server.py
src/test_backend_server.py
src/test_work_graph_orchestrate.py
src/work_graph_orchestrate.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"')
@@ -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'")
@@ -7,10 +7,13 @@ spike -- this module&#39;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()
@@ -39,13 +39,42 @@ subgraph IS just itself), escalate to `work_graph_run.status=&#39;blocked&#39;` +
 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,