code_diff.patch
20.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
diff --git a/docs/reference/graph-engineering-1000-agents-5-steps.diff-with-graphlemon.md b/docs/reference/graph-engineering-1000-agents-5-steps.diff-with-graphlemon.md
new file mode 100644
index 0000000..dfe4954
--- /dev/null
+++ b/docs/reference/graph-engineering-1000-agents-5-steps.diff-with-graphlemon.md
@@ -0,0 +1,50 @@
+# 对照笔记:《Graph Engineering:一个提示词,跑起 1000+ 的智能体循环》vs graphlemon 架构
+
+- **对照对象**:[graph-engineering-1000-agents-5-steps.md](./graph-engineering-1000-agents-5-steps.md)(本文简称"5 步版")
+- **背景**:仓库里已有同主题的 [14 步版](../Graph%20Engineering%20with%20Claude_%2014-Step%20roadmap%20from%200%20to%20graph%20architect%20(Full%20Course).md),`docs/graphlemon_architecture.md` 第 10 节已对 14 步版做过逐条映射,并在 10.1 节据此决策"Work Graph 执行层直接复用 Claude Code 原生动态工作流"。5 步版是同一方法论的精简版,本笔记只记录 5 步版带来的**新增信息**和**尚待跟进的开放项**,不重复 14 步版已覆盖的内容。
+- **状态**:讨论草稿,待后续会话继续展开,非最终结论。
+- **整理日期**:2026-08-04
+
+---
+
+## 1. 已经吻合、且已在代码里落地的部分
+
+| 5 步版观点 | graphlemon 现状 | 依据 |
+|---|---|---|
+| 验证器(Verifier)必须用**干净的独立上下文**,不能复用执行器的对话,否则"验证"只是自己给自己点头 | Validator 是独立的 Org Agent,独立 `claude -p` 调用,不共享 Coder 的会话上下文 | `docs/graphlemon_architecture.md` §6.3 根因溯源节点设计;`.claude/agents/validator.md` |
+| 并行智能体必须用 **git worktree** 隔离工作区,否则会互相踩(文中举 Bun 团队真实翻车案例) | 并行 Coder 节点强制走独立 worktree,self-target(自举)场景无条件强制 | `src/work_node_run.py::create_worktree` / `ensure_worktree_committed`;CLAUDE.md「Isolation」小节 |
+| 图需要**锚点(Anchors)**:真跑过的测试、基于证据的判定、不可被智能体自行调整的冻结规则 | Validator 只出 PASS / FAIL / INCONCLUSIVE 三态,且要求独立复核证据,不采信 Coder 自称"测试通过" | `docs/graphlemon_architecture.md` §9.1(qa profile 映射) |
+| Work Graph 层直接用 Claude Code 原生 `agent()`/`parallel()`/`pipeline()`,编排本身零 token | 已决策并做过实测验证 | `docs/graphlemon_architecture.md` §10.1;`spikes/a1-dynamic-workflows/` |
+
+## 2. 5 步版提到、但 graphlemon 仍是"已知简化项"的部分
+
+这几条是本轮对照读出来的**待办候选**,还没有决定要不要做、什么时候做:
+
+1. **loop-until-dry(连续 K 轮无新发现才停止)**
+ 现状:Org-level Loop 固定 3 轮上限,超过未收敛直接转 `blocked_pending_human`。
+ 出处:`docs/sdd/04-P-流程与状态流转.md`(明确标注为"未来优化候选");`docs/sdd/06-验收与测试锚点.md`。
+ 5 步版的场景(第 4 步"未知规模的发现")建议连续两轮无新发现才停,比固定轮次更适合"扫描规模未知"的任务——这跟 Validator 未来如果要承担"未知规模缺陷扫描"类任务是同一个开放问题。
+
+2. **按节点分层选模型**(便宜模型跑重复/有界节点,贵模型做综合裁决)
+ 现状:`src/agent_profile.py` 里 Planner / Coder / Validator 三个 Org Agent 目前**全部固定** `claude-sonnet-5`,Work Graph 内部节点也没有差异化选型。
+ 出处:`docs/graphlemon_architecture.md` §10 表格第 12 行早就标注"应补进第 7.5 节预算护栏",至今未落地。
+
+3. **1000 智能体 / 16 并发的量级验证**
+ graphlemon 目前的 Work Graph 场景(自举开发自己)是个位数到十几个节点的规模,从未真正测过大规模 fan-out。5 步版给的"先 20 个探路,看单次运行花多少钱再放开"的做法,如果以后要做"审计整个代码库"这类任务,可以直接借用作为验证节奏。
+
+## 3. 一处架构定位上的差异(不是缺口,是设计选择)
+
+5 步版第 5 步专门讲"什么时候不该用 Graph"(任务太小/太孤立、需要人紧盯每一步、还在探索阶段、步骤间真正强依赖 → 别上图,用一个 Loop 就够)。
+
+这个判断题在 graphlemon 里**不是留给每次任务临场判断的**,而是直接被架构的两层设计固化成了硬边界:
+
+- **Org Graph**(Planner → Coder → Validator)人工固定设计,永远不是"动态生成的图"——恰好对应 5 步版说的"步骤间真的互相依赖 + 需要锚点 + 不该让智能体自己改架构"这一类场景。
+- **Work Graph** 才是 5 步版描述的那种"动态生成、可扇形展开"的图,且作用域被限定在单次 Org Agent 调用内部(Planner 一次规划、Coder 一次实现、Validator 一次验收)。
+
+也就是说,graphlemon 把"要不要上图"做成了架构里的一条硬边界,而不是运行时的判断逻辑——这一点比 5 步版单纯给"判断标准"更进一步,值得在后续讨论里明确写进架构文档,作为对 5 步版方法论的一个延伸。
+
+## 4. 待讨论的开放问题(供下一轮会话展开)
+
+- [ ] loop-until-dry 是否要在 Validator 的"未知规模缺陷扫描"场景里落地?优先级几何?
+- [ ] 按节点分层选模型:Org Graph 三节点要不要差异化?Work Graph 内部节点(Planning Skill 生成的编排脚本)要不要在 Prompt 里显式要求按任务复杂度选模型?
+- [ ] 是否需要在 `docs/graphlemon_architecture.md` 里补一节,显式写明"Org Graph 固定 / Work Graph 动态"这条边界与 5 步版"何时不该用 Graph"判断标准的对应关系?
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,