From 7c46a2f281a9ad93c5abfd4fc56e7473efec00eb Mon Sep 17 00:00:00 2001 From: bboonnee Date: Wed, 12 Aug 2026 21:28:13 +0800 Subject: [PATCH] Phase 2b: Coder self-plans its own implementation graph via append_to_run_id --- src/e2e_pipeline.py | 218 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------ src/task_execution.py | 2 +- src/test_e2e_pipeline.py | 182 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------- src/test_task_execution.py | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- src/test_work_graph_fork.py | 44 ++++++++++++++++++++++++++++++++++++++++++++ src/test_work_graph_plan.py | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/work_graph_fork.py | 53 ++++++++++++++++++++++++++++++++++++++--------------- src/work_graph_plan.py | 280 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------------------- 8 files changed, 857 insertions(+), 113 deletions(-) diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 9a22ac3..778e226 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -140,6 +140,12 @@ _DOC_PROMPTS = { _CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} _CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]} +# 2026-08-12 (Phase 2 §6): retired as the per-node prompt now that Coder +# self-plans (implement_code() no longer runs one synthetic "implement" +# node) -- its guidance is folded into _CODER_PLANNING_CONTEXT below, +# which feeds plan_work_graph()'s planning call instead of a single +# execution call. Kept here, unused, only because deleting it isn't this +# change's job and a future revert would want it back verbatim. _CODER_PROMPT = ( "Implement this goal per the approved SDD below. Write real code and real tests, run the " "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts " @@ -160,6 +166,32 @@ _CODER_PROMPT = ( "Test script:\n\n{test_script_content}" ) +# 2026-08-12 (Phase 2 §6): fed as `goal_text` into plan_work_graph(conn, +# "coder", ..., append_to_run_id=run_id) -- reuses the existing +# {goal_text} substitution point in _generic_prompt_template() rather than +# adding a new parameter to plan_work_graph() just for Coder's extra +# context (SDD/Benchmark/test dataset/script), which Planner's own +# _PROMPT_TEMPLATE call never needed. Same .replace("{goal_text}", ...) +# convention _PROMPT_TEMPLATE/_generic_prompt_template() already use (not +# .format() -- see work_graph_plan.py's own comment on why), so this +# string's remaining braces (none currently, but keep this in mind if +# edited) must stay literal, not format fields. +_CODER_PLANNING_CONTEXT = ( + "{goal_text}\n\n" + "Approved SDD:\n\n{sdd_content}\n\n" + "Benchmark (test dataset/script may be 'N/A' if this goal has none -- that's expected, not " + "an error):\n\n{benchmark_content}\n\n" + "Test dataset:\n\n{test_dataset_content}\n\n" + "Test script:\n\n{test_script_content}\n\n" + "Guidance: write real code and real tests, run the tests yourself as part of your own " + "nodes' work, and don't consider the goal done until they pass for real (or you've " + "exhausted reasonable attempts and a node explains why not). If the Planner's test script " + "above is real (not N/A), materialize it as a real file under this project's own test " + "conventions and make it pass for real -- it is not optional, it's part of this goal's " + "acceptance bar; the Planner's script is a floor, write additional tests of your own if it " + "doesn't cover everything the SDD requires." +) + _VALIDATOR_PROMPT = ( "Independently verify the implementation for this goal against the Benchmark below: run " "the tests yourself, don't just trust the Coder's claims. Reply with a verdict line " @@ -486,8 +518,30 @@ def _failed_node_ids(exc): return set() +def _resolve_node_artifact_type(node, node_artifact_types): + """A node promotes to an `artifact` row if EITHER: it's in the caller's + precomputed `node_artifact_types` dict (Planner's/Validator's fixed + catalogs, where every node is a promotable artifact by construction -- + unchanged behavior), OR it declares `artifact_type` directly on + itself, OR (2026-08-12, Phase 2 §2.3) it declares `promotes_to_ + artifact: {"artifact_type": ...}` (Coder's/Validator's generic + self-planned nodes -- most of which should declare NEITHER, meaning + "this is an internal graph step, not a final deliverable"). Returns + None for a node that isn't promoting anything -- not an error, the + expected case for most self-planned nodes.""" + if node["node_id"] in node_artifact_types: + return node_artifact_types[node["node_id"]] + if "artifact_type" in node: + return node["artifact_type"] + promotes = node.get("promotes_to_artifact") + if promotes: + return promotes.get("artifact_type") + return None + + def _write_completed_node_artifacts( - conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, run_graph_error=None, + conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, + run_graph_error=None, only_node_ids=None, ): """Writes an artifact for every node that's genuinely done AND verify_fn-accepted, sourced from `work_node_execution` in the DB -- @@ -511,6 +565,15 @@ def _write_completed_node_artifacts( nodes are excluded outright: their row is schema-valid ('done') but verify_fn never accepted either attempt, so neither is trustworthy content to persist. + + 2026-08-12 (Phase 2 §2.3/§6): a node with no resolvable artifact_type + (see `_resolve_node_artifact_type`) is silently skipped, not an + error -- most Coder/Validator self-planned nodes are internal graph + steps, not final deliverables. `only_node_ids` (default None = every + node in `graph_manifest`, unchanged behavior) restricts processing to + a subset -- needed once a caller works over a MERGED graph (Coder's + `append_to_run_id`) and must avoid re-processing/re-promoting nodes an + EARLIER stage already wrote artifacts for. """ done_rows = conn.execute( "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'", @@ -529,7 +592,11 @@ def _write_completed_node_artifacts( node_id = node["node_id"] if node_id not in done_outputs: continue - artifact_type = node_artifact_types[node_id] + if only_node_ids is not None and node_id not in only_node_ids: + continue + artifact_type = _resolve_node_artifact_type(node, node_artifact_types) + if artifact_type is None: + continue deps = [ artifact_ids[node_artifact_types[dep_id]] for dep_id in (node.get("depends_on") or []) @@ -743,7 +810,19 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ artifact_ids = task_execution._latest_artifact_ids( conn, task_id, task_execution._DOC_ARTIFACT_TYPES ) - plan_result = {"run_id": None, "artifact_ids": artifact_ids} + # Phase 2 §6.2 (2026-08-12, found while wiring implement_code()'s + # new append_to_run_id requirement): this used to hardcode + # run_id=None -- harmless for Phase 1 (nothing downstream read + # it), but Phase 2's Coder self-planning needs a REAL run_id to + # append into. Every real artifact row already has one + # (produced_by_run_id is NOT NULL) -- read it off the sdd + # artifact (required a few lines below anyway, guaranteed + # present whenever `existing` is truthy) rather than inventing + # a new lookup path. + sdd_run_id = conn.execute( + "SELECT produced_by_run_id FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],) + ).fetchone()[0] + plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids} else: plan_result = plan_documents(conn, task_id, goal_text + feedback_suffix, target_folder_path) artifact_ids = plan_result["artifact_ids"] @@ -792,57 +871,124 @@ def plan_and_review_documents(conn, task_id, goal_text, target_folder_path, time def implement_code( - conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200, + conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200, benchmark_content="N/A: not provided", test_dataset_content="N/A: not provided", test_script_content="N/A: not provided", ): - """Architecture §11 criterion 2: a real Coder call via c_work_node_run - (work_graph_orchestrate.run_graph with a synthetic single-node - manifest -- see module docstring for why there's no real Coder graph - planner), then registers whatever the Coder actually committed as - code_diff (the diff itself) and test_report (the Coder's own response - text, which per its migrated SOUL.md is required to include real test - commands + real output) artifacts. + """Phase 2 §6 (2026-08-12): Coder now self-plans its own implementation + graph via `plan_work_graph(..., append_to_run_id=run_id)` -- `run_id` + is Planner's own work_graph_run (created by `plan_and_review_ + documents()`), no longer a fresh row this function creates itself. + Coder's self-planned nodes get appended into that SAME run, not a + separate one -- see work_graph_plan.py's own append_to_run_id + docstring for the merge mechanics. Replaces the old synthetic + single-node "implement" manifest (architecture §11 criterion 2's + original, pre-self-planning shape). + + `code_diff` stays a Python-computed artifact (a real `git diff` across + whatever Coder's whole self-planned graph actually changed on disk, + end to end) -- not something any self-planned node's own `content` + field could meaningfully hold, since no node knows the diff ahead of + running it. `test_report` is promoted from whichever self-planned + node(s) declared `promotes_to_artifact: {"artifact_type": + "test_report"}` (the catalog's own "integration_check" entry suggests + this); if the LLM's plan didn't include one (the catalog is + suggestions, not requirements), falls back to concatenating every + self-planned node's own content in topological order -- same spirit + as the old single-node manifest's `response` field always being + Coder's own narrative of what it did, just synthesized instead of + directly given by one node. + + `needs_worktree=True` passed to run_graph() (2026-08-12, found while + doing this rework -- NOT previously passed here at all, even though + Coder is the one org_node that actually writes real code): with a + single synthetic node this was harmless (nothing to collide with + inside one run), but a genuinely self-planned graph can have 2+ + code-writing nodes in the SAME topological layer running concurrently + -- 03-C §3.6's "并行写代码的节点必须先申请 git_worktree_ref" applies for + real now, not just in theory. 2026-08-06: benchmark_content/test_dataset_content/test_script_content (default 'N/A: not provided' for callers that predate this, e.g. a - resumed task from before this change) are threaded into _CODER_PROMPT - so the Coder must make the Planner's real test script pass for real, - not just write whatever tests it invents on its own. + resumed task from before this change) are folded into + _CODER_PLANNING_CONTEXT so Coder's planning call sees the Planner's + real test script and must make it pass for real, not just write + whatever tests it invents on its own. """ before_sha = _current_commit(target_folder_path) - run_id = str(uuid.uuid4()) - manifest = {"nodes": [{"node_id": "implement", "depends_on": []}], "edges": []} - conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, 'coder', ?, 'running')", - (run_id, task_id, json.dumps(manifest)), - ) - conn.commit() - outputs = work_graph_orchestrate.run_graph( - conn, run_id, "coder", "coder-implementation", target_folder_path, manifest, - lambda node, outputs: _CODER_PROMPT.format( - goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content, - test_dataset_content=test_dataset_content, test_script_content=test_script_content, - ), - _CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout, + planning_context = _CODER_PLANNING_CONTEXT.format( + goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content, + test_dataset_content=test_dataset_content, test_script_content=test_script_content, + ) + plan_result = work_graph_plan.plan_work_graph( + conn, "coder", task_id, planning_context, target_folder_path, + timeout=300, append_to_run_id=run_id, + ) + graph_manifest = plan_result["graph_manifest"] + new_node_ids = set(plan_result["new_node_ids"]) + + # Covers BOTH node shapes that can now coexist in this merged graph: + # Planner's own nodes (artifact_type) and Coder's just-appended ones + # (promotes_to_artifact) -- used for both _planner_prompt_builder's + # prompt construction (only its own artifact_type entries are ever + # queried) and _write_completed_node_artifacts' artifact-type/deps + # resolution below. + node_artifact_types = {} + for n in graph_manifest["nodes"]: + if "artifact_type" in n: + node_artifact_types[n["node_id"]] = n["artifact_type"] + elif n.get("promotes_to_artifact"): + node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"] + + build_prompt = work_graph_plan.dispatch_build_prompt( + _planner_prompt_builder(goal_text, node_artifact_types), work_graph_plan.generic_node_prompt_builder, ) - response = outputs["implement"]["response"] + run_graph_error = None + outputs = None + try: + outputs = work_graph_orchestrate.run_graph( + conn, run_id, "coder", "coder-implementation", target_folder_path, graph_manifest, + build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout, + needs_worktree=True, mechanical_check_registry={"check_script_compiles": check_script_compiles}, + ) + except Exception as e: + run_graph_error = e diff = subprocess.run( ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True ).stdout - diff_result = artifact_write.write_artifact( conn, target_folder_path, "code_diff", task_id, run_id, ".graphlemon/runs/{}/code_diff.patch".format(task_id), diff or "(no diff -- see test_report)", ) - report_result = artifact_write.write_artifact( - conn, target_folder_path, "test_report", task_id, run_id, - ".graphlemon/runs/{}/test_report.md".format(task_id), response, + + artifact_ids = _write_completed_node_artifacts( + conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, + _planner_ext_for, run_graph_error, only_node_ids=new_node_ids, ) - return {"code_diff": diff_result["artifact_id"], "test_report": report_result["artifact_id"]} + + if run_graph_error is not None: + raise run_graph_error + + if "test_report" not in artifact_ids: + # The self-planned graph didn't include a node declaring + # promotes_to_artifact: {"artifact_type": "test_report"} -- + # synthesize one so callers can still rely on this key always + # being present (same contract the old single-node manifest gave + # unconditionally via its one node's `response` field). + report_content = "\n\n---\n\n".join( + "### {}\n\n{}".format(node_id, (outputs or {}).get(node_id, {}).get("content", "")) + for node_id in new_node_ids + if node_id in (outputs or {}) + ) or "(Coder's self-planned graph produced no synthesizable report content)" + report_result = artifact_write.write_artifact( + conn, target_folder_path, "test_report", task_id, run_id, + ".graphlemon/runs/{}/test_report.md".format(task_id), report_content, + ) + artifact_ids["test_report"] = report_result["artifact_id"] + + return {"code_diff": diff_result["artifact_id"], "test_report": artifact_ids["test_report"]} def _parse_verdict_line(response): @@ -1093,7 +1239,7 @@ def run_task_lifecycle(conn, task_id, target_folder_path, goal_text): org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, _READY_PROMPT) code_artifact_ids = implement_code( - conn, task_id, sdd_content, goal_text, target_folder_path, + conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path, benchmark_content=benchmark_content, test_dataset_content=test_dataset_content, test_script_content=test_script_content, ) diff --git a/src/task_execution.py b/src/task_execution.py index f29f561..b213327 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -165,7 +165,7 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): _check_not_paused(conn, task_id) org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT) code_artifact_ids = e2e_pipeline.implement_code( - conn, task_id, sdd_content, goal_text, target_folder_path, + conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path, benchmark_content=benchmark_content, test_dataset_content=test_dataset_content, test_script_content=test_script_content, ) diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 6ae54db..cf7f9d8 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -99,6 +99,56 @@ def _fake_plan_subprocess_run(*args, **kwargs): return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") +# Phase 2 §6 (2026-08-12): Coder now self-plans too (implement_code() calls +# plan_work_graph(conn, "coder", ..., append_to_run_id=...), going through +# the SAME mocked subprocess boundary as Planner's own planning call) -- +# depends_on Planner's sdd node so it lands in a LATER topological layer +# than Planner's already-'done' nodes, not the same one (both being +# in-degree-zero would trigger run_graph()'s concurrent ThreadPoolExecutor +# layer path purely as a mocking artifact, unrelated to what this test is +# actually checking). +_FAKE_CODER_GRAPH_MANIFEST = { + "nodes": [{ + "node_id": "impl", "depends_on": ["sdd"], "task_prompt": "implement it", + "promotes_to_artifact": {"artifact_type": "test_report"}, + }], + "edges": [], +} + + +_real_subprocess_run = subprocess.run # captured before any test patches subprocess.run globally + + +def _fake_plan_subprocess_run_dispatch(*args, **kwargs): + """Real plan_work_graph() dispatches by the --agent argv value -- this + mock does the same, so both Planner's and Coder's planning calls (which + now share the same mocked subprocess.run patch target) each get their + own realistic fake manifest instead of one hardcoded response serving + both callers (which would hand Coder Planner's own brd/prd/... nodes, + colliding with the ones already in the run it's appending onto). + + patch("src.work_graph_plan.subprocess.run", ...) patches subprocess.run + GLOBALLY, not just within work_graph_plan.py -- `subprocess` is the same + singleton module object everywhere it's imported, so this side_effect + also intercepts every OTHER real subprocess.run call this test's own + call chain makes (artifact_write.write_artifact's `git add`/`git + commit`, e2e_pipeline's `git diff`/`git rev-parse`, etc). Those aren't + `claude --agent ... -p ...` calls -- pass them through to the REAL + subprocess.run instead of returning a canned fake "claude planning + response" CompletedProcess for them too (which happened to not visibly + break anything before this dispatch function existed, only because + nothing downstream read those specific calls' .stdout in a way that + exposed it -- still wrong to keep doing on purpose now that this + function actually inspects argv).""" + argv = args[0] + if not argv or argv[0] != "claude": + return _real_subprocess_run(*args, **kwargs) + org_node = argv[argv.index("--agent") + 1] + manifest = _FAKE_CODER_GRAPH_MANIFEST if org_node == "coder" else _FAKE_GRAPH_MANIFEST + payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"} + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") + + def _fake_plan_subprocess_run_with_test_script(*args, **kwargs): payload = { "is_error": False, @@ -117,7 +167,11 @@ class PipelineTestBase(unittest.TestCase): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True) - self.conn = schema.connect(":memory:") + # check_same_thread=False (2026-08-12, Phase 2 §6) -- see + # test_task_execution.py's TaskExecutionTestBase for why: a merged + # graph's concurrent-layer path can now be reached even in tests + # that don't otherwise care about G's parallel execution. + self.conn = schema.connect(":memory:", check_same_thread=False) schema.create_schema(self.conn) project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False) @@ -125,6 +179,94 @@ class PipelineTestBase(unittest.TestCase): self.conn.close() +class WriteCompletedNodeArtifactsTest(PipelineTestBase): + """Phase 2 §2.3/§6 (2026-08-12) direct coverage: promotes_to_artifact + resolution and only_node_ids scoping, tested directly rather than only + incidentally through plan_documents()/implement_code() integration + tests.""" + + def setUp(self): + super().setUp() + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', (SELECT project_id FROM project), 'goal', '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 ('coder-implementation', 'coder', 'skills/coder-implementation/', 'unknown', 'new', 'active')" + ) + self.conn.commit() + + def _mark_done(self, node_id, content): + 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, output_payload, status) " + "VALUES (?, 'run-1', ?, 'coder', 'coder-implementation', 'in.json', 'out.json', '{}', ?, 'done')", + ("node-exec-" + node_id, node_id, json.dumps({"content": content})), + ) + self.conn.commit() + + def test_node_with_no_artifact_type_or_promotes_to_artifact_is_skipped(self): + # most self-planned Coder nodes are internal graph steps, not final + # deliverables -- must be silently skipped, not an error. + manifest = {"nodes": [{"node_id": "explore", "depends_on": []}], "edges": []} + self._mark_done("explore", "some intermediate reasoning") + artifact_ids = e2e_pipeline._write_completed_node_artifacts( + self.conn, self.tmp.name, "task-1", "run-1", manifest, {}, lambda t: "md", + ) + self.assertEqual(artifact_ids, {}) + count = self.conn.execute("SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1'").fetchone()[0] + self.assertEqual(count, 0) + + def test_promotes_to_artifact_node_is_persisted(self): + manifest = { + "nodes": [{ + "node_id": "final_report", "depends_on": [], + "promotes_to_artifact": {"artifact_type": "test_report"}, + }], + "edges": [], + } + self._mark_done("final_report", "# Real report content") + node_artifact_types = {"final_report": "test_report"} + artifact_ids = e2e_pipeline._write_completed_node_artifacts( + self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md", + ) + self.assertEqual(set(artifact_ids), {"test_report"}) + content = self.conn.execute( + "SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_ids["test_report"],) + ).fetchone()[0] + with open(os.path.join(self.tmp.name, content)) as f: + self.assertEqual(f.read(), "# Real report content") + + def test_only_node_ids_scopes_to_new_nodes_skipping_earlier_promotable_ones(self): + # a merged graph (Planner's already-artifact'd nodes + Coder's just- + # appended ones) must not re-promote/re-process nodes an EARLIER + # stage already wrote artifacts for, even though they'd otherwise + # be individually promotable. + manifest = { + "nodes": [ + {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}, + { + "node_id": "impl", "depends_on": ["sdd"], + "promotes_to_artifact": {"artifact_type": "test_report"}, + }, + ], + "edges": [], + } + self._mark_done("sdd", "# SDD content") + self._mark_done("impl", "# Report content") + node_artifact_types = {"sdd": "sdd", "impl": "test_report"} + artifact_ids = e2e_pipeline._write_completed_node_artifacts( + self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md", + only_node_ids={"impl"}, + ) + self.assertEqual(set(artifact_ids), {"test_report"}) # NOT "sdd" -- outside only_node_ids + + class PlanDocumentsTest(PipelineTestBase): def test_produces_five_artifacts_with_correct_lineage(self): self.conn.execute( @@ -228,7 +370,7 @@ class PlanDocumentsTest(PipelineTestBase): class RunEndToEndTest(PipelineTestBase): def test_full_pipeline_wiring(self): with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \ - patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \ + patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(work_graph_execute, "write_workflow_script"), \ patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."): @@ -237,7 +379,11 @@ class RunEndToEndTest(PipelineTestBase): {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, {"content": _FAKE_DOCS["test_dataset"]}, - {"response": "Implemented. Ran `pytest`: 3 passed."}, + # Phase 2 §6 (2026-08-12): Coder self-plans now -- one + # self-planned "impl" node (_FAKE_CODER_GRAPH_MANIFEST), + # content-keyed like every other self-planned node, not the + # old single synthetic node's response-keyed output. + {"content": "Implemented. Ran `pytest`: 3 passed."}, # Validator's OWN independent benchmark/dataset (2026-08-06) -- # deliberately different content from the Planner's, to also prove # nothing here accidentally reuses the Planner's copies. @@ -265,15 +411,31 @@ class RunEndToEndTest(PipelineTestBase): # (test_script/validator_test_script disabled, see _VALIDATOR_MANIFEST) self.assertEqual(all_artifacts, 9) - # planner, coder, AND validator (2026-08-06: validator now has its own - # real work_graph_run too, not a flat direct call) each got their own - # work_graph_run (not sharing one) - run_org_nodes = { + # Phase 2 §6 (2026-08-12): Coder no longer gets its own work_graph_run + # -- implement_code() appends its self-planned nodes onto Planner's + # SAME run (plan_work_graph(..., append_to_run_id=...)), so only TWO + # rows exist now, not three: the shared Planner+Coder run (still + # labeled org_node='planner' -- that column reflects who ORIGINATED + # the run, not every org_node that has since contributed nodes to it; + # Phase 2c is expected to fold Validator into the same run too, at + # which point this column stops being a reliable "who's involved" + # signal at all, see design doc §2.1) and Validator's own separate run. + run_rows = self.conn.execute( + "SELECT run_id, org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],) + ).fetchall() + self.assertEqual(len(run_rows), 2) + self.assertEqual({row[1] for row in run_rows}, {"planner", "validator"}) + + # Coder's self-planned "impl" node genuinely landed in the SAME run + # as Planner's nodes, not a separate one. + planner_run_id = next(row[0] for row in run_rows if row[1] == "planner") + node_ids = { row[0] for row in self.conn.execute( - "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],) + "SELECT node_id FROM work_node_execution WHERE run_id = ?", (planner_run_id,) ).fetchall() } - self.assertEqual(run_org_nodes, {"planner", "coder", "validator"}) + self.assertIn("impl", node_ids) + self.assertIn("sdd", node_ids) class LooksLikeNarrationTest(unittest.TestCase): @@ -437,7 +599,7 @@ class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase): def test_prd_and_benchmark_missing_does_not_raise_key_error(self): captured = {} - def fake_implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, **kwargs): + def fake_implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, **kwargs): captured["benchmark_content"] = kwargs.get("benchmark_content") return {"code_diff": "diff-1", "test_report": "report-1"} diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 2a79007..12c1b3a 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -14,7 +14,10 @@ import threading import unittest from unittest.mock import patch -from src import e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, task_execution, work_graph_execute +from src import ( + e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, + task_execution, work_graph_execute, work_graph_plan, +) def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300): @@ -22,12 +25,46 @@ def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300): return "Ready." +# Phase 2 §6 (2026-08-12): implement_code() now self-plans via +# plan_work_graph(conn, "coder", ..., append_to_run_id=...) instead of +# running one hardcoded synthetic node -- these tests aren't about self- +# planning mechanics (that's test_e2e_pipeline.py's job), so this fakes +# the planning subprocess with the smallest possible one-node manifest, +# keeping the rest of each test's mock_execute.side_effect shape close to +# what it was before this change (one node execution slot for Coder, +# {"content": ...} instead of the old single-node manifest's {"response": ...}). +_FAKE_CODER_MANIFEST = { + "nodes": [{ + "node_id": "impl", "depends_on": [], "task_prompt": "implement it", + "promotes_to_artifact": {"artifact_type": "test_report"}, + }], + "edges": [], +} + + +def _fake_coder_plan_subprocess_run(*args, **kwargs): + payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_CODER_MANIFEST) + "\n```"} + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") + + class TaskExecutionTestBase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True) - self.conn = schema.connect(":memory:") + # check_same_thread=False (2026-08-12, Phase 2 §6): a merged graph + # can legitimately put an already-'done' node (Planner's, short- + # circuiting instantly) in the SAME topological layer as a newly- + # appended node with no depends_on between them (e.g. Coder's own + # self-planned node happens not to depend on Planner's) -- + # work_graph_orchestrate.run_graph()'s multi-node-layer path always + # uses a ThreadPoolExecutor once len(pending) > 1, regardless of how + # fast one of them finishes. Real production connections already do + # this (scheduler.py/task_execution.py's own start_task_execution_ + # in_background, same "G" reasoning) -- this test fixture just + # hadn't needed it before self-planning could produce concurrent + # layers spanning old + new nodes. + self.conn = schema.connect(":memory:", check_same_thread=False) schema.create_schema(self.conn) project = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False) self.project_id = project["project_id"] @@ -45,9 +82,18 @@ class TaskExecutionTestBase(unittest.TestCase): def _write_artifact(self, task_id, artifact_type, content): run_id = "run-" + artifact_type + # Phase 2 §6 (2026-08-12): graph_manifest is no longer a throwaway + # '{}' -- implement_code() now reads it back (plan_work_graph's + # append_to_run_id mode) whenever it appends onto the run the sdd + # artifact points at, so it needs a real "nodes"/"edges" shape, not + # an empty object real production data would never actually have. + manifest = json.dumps({ + "nodes": [{"node_id": artifact_type, "artifact_type": artifact_type, "depends_on": []}], + "edges": [], + }) self.conn.execute( "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id), + "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest), ) rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type) abs_path = os.path.join(self.tmp.name, rel_path) @@ -122,12 +168,13 @@ class ResumeFromPlanningTest(TaskExecutionTestBase): self._write_artifact("task-1", "benchmark", "# Benchmark content") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ patch.object(work_graph_execute, "write_workflow_script"), \ patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ - {"response": "done, tests pass"}, # coder's implement node + {"content": "done, tests pass"}, # coder's self-planned "impl" node # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06) {"content": "# Validator's own Benchmark"}, {"content": "N/A: no dataset needed"}, @@ -169,12 +216,13 @@ class ResumeFromReviewingTest(TaskExecutionTestBase): return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout) with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=tracking_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ patch.object(work_graph_execute, "write_workflow_script"), \ patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ - {"response": "done, tests pass"}, # coder's implement node + {"content": "done, tests pass"}, # coder's self-planned "impl" node {"content": "# Validator's own Benchmark"}, {"content": "N/A: no dataset needed"}, {"content": "import sys\nsys.exit(0)\n"}, @@ -209,16 +257,22 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): # writes a real, validation_status='passed' sdd artifact, since # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] checks the real # `artifact` table, not just this function's Python return value. + # run_id matches what _write_artifact's own test fixture convention + # actually creates ("run-" + artifact_type) -- a real plan_documents() + # call's return value would likewise point at the real work_graph_run + # row it just inserted, which Phase 2's implement_code() now needs to + # append into. self._write_artifact(task_id, "sdd", "# SDD\nfreshly regenerated") - return {"run_id": "run-x", "artifact_ids": {"sdd": "art-sdd"}} + return {"run_id": "run-sdd", "artifact_ids": {"sdd": "art-sdd"}} with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ patch.object(work_graph_execute, "write_workflow_script"), \ patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"), \ patch.object(e2e_pipeline, "plan_documents", side_effect=fake_plan_documents) as mock_plan: mock_execute.side_effect = [ - {"response": "done, tests pass"}, # coder's implement node + {"content": "done, tests pass"}, # coder's self-planned "impl" node {"content": "# Validator's own Benchmark"}, {"content": "N/A: no dataset needed"}, {"content": "import sys\nsys.exit(0)\n"}, diff --git a/src/test_work_graph_fork.py b/src/test_work_graph_fork.py index f5f4613..25dea98 100644 --- a/src/test_work_graph_fork.py +++ b/src/test_work_graph_fork.py @@ -139,6 +139,50 @@ class ForkFromPrdAcceptanceCriterion1Test(ForkTestBase): self.assertEqual(run_row, ("done", "run-orig", node_exec_ids["prd"])) +class ForkFromMixedShapeGraphTest(ForkTestBase): + """Phase 2 §2.1.2 (2026-08-12) regression: a graph_manifest can now + genuinely mix Planner-shaped nodes ('artifact_type') and Coder-shaped + nodes ('task_prompt') side by side, once plan_work_graph(..., + append_to_run_id=...) appends Coder's self-planned nodes onto + Planner's original run. Forking from the artifact_type node must + still correctly reconstruct a build_prompt for the DOWNSTREAM + task_prompt node -- before this fix, _reconstruct_build_prompt + required the WHOLE graph to be one uniform shape and would have + raised ForkGraphShapeUnsupportedError here.""" + + def test_downstream_generic_node_reexecutes_after_forking_upstream_planner_node(self): + manifest = { + "nodes": [ + {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}, + { + "node_id": "impl", "depends_on": ["sdd"], "task_prompt": "implement per the sdd", + "promotes_to_artifact": {"artifact_type": "test_report"}, + }, + ], + "edges": [], + } + contents = {"sdd": {"content": "original sdd"}, "impl": {"content": "original impl"}} + node_exec_ids = self._seed_completed_run("run-orig", "planner", manifest, contents) + + execute_calls = [] + + def fake_execute(cwd, name, args, output_schema, timeout=600): + execute_calls.append(name) + return {"content": "regenerated for " + name} + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + result = work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["sdd"], {"content": "a completely different sdd"}, "/tmp/proj-1", + ) + + # only "impl" (downstream of the fork point) genuinely re-ran. + self.assertEqual(len(execute_calls), 1) + outputs = result["outputs"] + self.assertEqual(outputs["sdd"], {"content": "a completely different sdd"}) + self.assertNotEqual(outputs["impl"], {"content": "original impl"}) + + class OriginalRunUnaffectedTest(ForkTestBase): """Acceptance standard 2: the original run's records are completely untouched by a fork.""" diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index 1ddd862..f82210e 100644 --- a/src/test_work_graph_plan.py +++ b/src/test_work_graph_plan.py @@ -399,6 +399,129 @@ class GenericPlanningPromptTest(PlanWorkGraphTestBase): self.assertNotIn("BRD, PRD, SDD, Benchmark", prompt) self.assertIn("task_prompt", prompt) + def test_coder_prompt_includes_its_own_suggested_catalog_not_validators(self): + # Phase 2 §3 (2026-08-12): each org_node gets its OWN suggested + # node-type catalog, not a shared one -- "starting points, not + # requirements", same contract _PLANNER_NODE_CATALOG already uses. + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)} + ) + work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1") + prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1] + self.assertIn("Suggested node types for this goal", prompt) + self.assertIn('"node_type": "unit_test"', prompt) + self.assertIn('"node_type": "integration_check"', prompt) + self.assertNotIn('"node_type": "independent_benchmark"', prompt) # validator-only + self.assertIn("promotes_to_artifact", prompt) + self.assertIn("dynamic_loop_back", prompt) # explicitly told NOT to use it + + def test_validator_prompt_includes_its_own_catalog_not_coders(self): + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)} + ) + work_graph_plan.plan_work_graph(self.conn, "validator", "task-1", "goal", "/tmp/proj-1") + prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1] + self.assertIn('"node_type": "independent_benchmark"', prompt) + self.assertIn('"node_type": "test_script"', prompt) + self.assertNotIn('"node_type": "unit_test"', prompt) # coder-only + + def test_unknown_org_node_gets_no_catalog_section_at_all(self): + # An org_node this hasn't been extended for yet (Phase 1's + # planner_reviewer, for instance) still gets the generic path -- + # just with no suggested catalog, same as before catalogs existed. + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)} + ) + work_graph_plan.plan_work_graph(self.conn, "planner_reviewer", "task-1", "goal", "/tmp/proj-1") + prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1] + self.assertNotIn("Suggested node types for this goal", prompt) + self.assertIn("task_prompt", prompt) # still the generic template itself + + +class AppendToRunIdTest(PlanWorkGraphTestBase): + """Phase 2 §6.1 (2026-08-12): plan_work_graph(..., append_to_run_id=...) + merges new nodes into an EXISTING work_graph_run instead of INSERTing a + fresh row.""" + + def _seed_run(self, run_id, manifest): + self.conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES (?, 'task-1', 'planner', ?, 'running')", + (run_id, json.dumps(manifest)), + ) + self.conn.commit() + + def test_merges_new_nodes_into_existing_run_not_a_new_row(self): + self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []}) + new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []} + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)} + ) + result = work_graph_plan.plan_work_graph( + self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1", + ) + + self.assertEqual(result["run_id"], "run-1") + self.assertEqual(result["new_node_ids"], ["impl"]) + merged_node_ids = {n["node_id"] for n in result["graph_manifest"]["nodes"]} + self.assertEqual(merged_node_ids, {"sdd", "impl"}) + + # exactly one work_graph_run row for this task -- no second row created. + row_count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'").fetchone()[0] + self.assertEqual(row_count, 1) + persisted = json.loads( + self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] + ) + self.assertEqual({n["node_id"] for n in persisted["nodes"]}, {"sdd", "impl"}) + + def test_existing_node_ids_are_injected_into_the_prompt(self): + self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []}) + new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []} + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)} + ) + work_graph_plan.plan_work_graph( + self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1", + ) + prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1] + self.assertIn("already has these nodes", prompt) + self.assertIn("sdd", prompt) + self.assertIn("ADDITIONAL nodes only", prompt) + + def test_node_id_collision_with_existing_node_raises(self): + self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []}) + # the "new" plan reuses "sdd" as its own node_id -- a real collision. + colliding_manifest = {"nodes": [{"node_id": "sdd", "depends_on": [], "task_prompt": "oops"}], "edges": []} + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(colliding_manifest)} + ) + with self.assertRaises(work_graph_plan.PlanningFailedError): + work_graph_plan.plan_work_graph( + self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1", + ) + # the original run's manifest must be untouched by the failed append. + persisted = json.loads( + self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] + ) + self.assertEqual([n["node_id"] for n in persisted["nodes"]], ["sdd"]) + + def test_append_to_nonexistent_run_id_raises(self): + new_manifest = {"nodes": [{"node_id": "impl", "depends_on": [], "task_prompt": "do it"}], "edges": []} + with patch("src.work_graph_plan.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)} + ) + with self.assertRaises(ValueError): + work_graph_plan.plan_work_graph( + self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="no-such-run", + ) + class CoderThreeIndependentModulesScenarioTest(PlanWorkGraphTestBase): """E acceptance criterion 1: a Coder scenario that naturally splits into diff --git a/src/work_graph_fork.py b/src/work_graph_fork.py index 14a1f1f..edfbedf 100644 --- a/src/work_graph_fork.py +++ b/src/work_graph_fork.py @@ -94,23 +94,46 @@ def _downstream_node_ids(graph_manifest, start_node_id): def _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest): """Rebuilds a build_prompt(node, outputs) usable for the re-executed downstream subgraph, purely from persisted state -- no access to the - original call site's Python closures. Two supported shapes (design doc - §2): Planner's canonical doc chain (every node has 'artifact_type', + original call site's Python closures. Two supported node shapes + (design doc §2): Planner's canonical doc chain ('artifact_type', prompt reconstructed via e2e_pipeline's own builder + task.goal_text) - and E's generic dynamic path (every node carries its own - self-contained 'task_prompt'). Anything else raises rather than - silently building broken prompts.""" + and E's generic dynamic path ('task_prompt'). A node with neither + raises rather than silently building a broken prompt. + + 2026-08-12 (Phase 2 §2.1.2, found while designing append_to_run_id): + this used to require the WHOLE graph to be one uniform shape (all + nodes 'artifact_type', or all 'task_prompt') -- correct when every + graph_manifest genuinely was one shape throughout, which stopped being + true the moment Coder's self-planned nodes (task_prompt) can get + appended onto Planner's original run (artifact_type) via + plan_work_graph(..., append_to_run_id=...). Dispatches per-NODE now + (work_graph_plan.dispatch_build_prompt) instead of once for the whole + graph -- every pre-Phase-2 graph_manifest still has just one shape + throughout, so this is a strict generalization, not a behavior change + for anything that isn't a merged graph. `org_node` is unused by this + per-node dispatch itself (kept as a parameter -- callers still pass it + for other purposes, e.g. run_graph()'s own org_node argument; + removing it from work_graph_run/this signature is Phase 2c's job + alongside the actual org_node column removal, not this fix's).""" nodes = graph_manifest["nodes"] - if org_node == "planner" and nodes and all("artifact_type" in n for n in nodes): - goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0] - node_artifact_types = {n["node_id"]: n["artifact_type"] for n in nodes} - return e2e_pipeline._planner_prompt_builder(goal_text, node_artifact_types) - if nodes and all("task_prompt" in n for n in nodes): - return work_graph_plan.generic_node_prompt_builder - raise ForkGraphShapeUnsupportedError( - "graph_manifest nodes must either all have 'artifact_type' (Planner's canonical chain) " - "or all have 'task_prompt' (the generic dynamic-planning path) to reconstruct prompts for a fork" - ) + if not nodes: + raise ForkGraphShapeUnsupportedError("graph_manifest has no nodes to reconstruct a build_prompt for") + goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0] + node_artifact_types = {n["node_id"]: n["artifact_type"] for n in nodes if "artifact_type" in n} + planner_builder = e2e_pipeline._planner_prompt_builder(goal_text, node_artifact_types) + dispatch = work_graph_plan.dispatch_build_prompt(planner_builder, work_graph_plan.generic_node_prompt_builder) + + def build_prompt(node, outputs): + try: + return dispatch(node, outputs) + except ValueError: + raise ForkGraphShapeUnsupportedError( + "node {} has neither 'artifact_type' (Planner's canonical chain) nor 'task_prompt' " + "(the generic dynamic-planning path) -- can't reconstruct a prompt for a fork".format( + node.get("node_id") + ) + ) + return build_prompt def create_fork(conn, source_node_exec_id, new_output_payload): diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index 3f26150..c4f78bb 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -41,10 +41,17 @@ GRAPH_MANIFEST_SCHEMA = { "required": ["node_id", "artifact_type", "depends_on"], "properties": { "node_id": {"type": "string"}, + # "artifact_type" (Planner's fixed-catalog nodes) and "task_prompt" + # (Coder/Validator's generic self-planned nodes, _generic_prompt_ + # template()) are mutually exclusive node shapes in practice -- + # "required" above only reflects Planner's shape (this schema + # predates the generic path); see work_graph_plan.py's own dispatch + # helpers for how callers tell the two apart per-node. "artifact_type": { "type": "string", "enum": ["brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"], }, + "task_prompt": {"type": "string"}, "depends_on": {"type": "array", "items": {"type": "string"}}, # 2026-08-08 (Work Graph self-planning + loop-back) -- both optional, # documentation only (this schema isn't jsonschema-enforced, see @@ -60,10 +67,20 @@ GRAPH_MANIFEST_SCHEMA = { "on_fail": { "type": "object", "properties": { - "action": {"type": "string", "enum": ["retry_self", "loop_back", "hard_fail"]}, + "action": { + "type": "string", + "enum": ["retry_self", "loop_back", "hard_fail", "dynamic_loop_back"], + }, "max_attempts": {"type": "integer"}, }, }, + # 2026-08-12 (Phase 2 §2.3/§4): also optional/documentation-only. + "promotes_to_artifact": { + "type": "object", + "properties": {"artifact_type": {"type": "string"}}, + }, + "subagent_type": {"type": "string"}, + "skill_ref": {"type": "string"}, }, }, }, @@ -126,13 +143,19 @@ _PLANNER_NODE_CATALOG = [ }, ] -# NOTE: no `str.format()` involved in building or filling this template -# (unlike _GENERIC_PROMPT_TEMPLATE) -- the embedded catalog JSON and the -# instructional JSON examples below (e.g. {"kind": "none"}) are full of -# their own literal braces, which .format() would try to parse as format -# fields and crash on. plan_work_graph() fills in goal_text with a plain +# NOTE: no `str.format()` involved in building or filling this template -- +# the embedded catalog JSON and the instructional JSON examples below +# (e.g. {"kind": "none"}) are full of their own literal braces, which +# .format() would try to parse as format fields and crash on. +# plan_work_graph() fills in goal_text with a plain # `.replace("{goal_text}", goal_text)` instead, which treats every other # brace in this string as inert literal text -- no escaping needed at all. +# 2026-08-12: _generic_prompt_template() (below) now follows this SAME +# convention for the same reason -- its own catalog JSON has the same +# literal-brace problem .format() would choke on, which the original +# _GENERIC_PROMPT_TEMPLATE constant used to dodge by having no embedded +# JSON at all; now that it does (Phase 2 §3's node catalogs), it needed +# the same fix this template already had. _PROMPT_TEMPLATE = ( "Plan (do not execute) the Work Graph for this goal: {goal_text}\n" "Available node types (use any subset, in any order that makes sense -- this goal may not " @@ -163,29 +186,107 @@ _PROMPT_TEMPLATE = ( # _PROMPT_TEMPLATE above unchanged (zero regression risk, see # docs/done/dynamic-work-graph-design-skill-design.md §3): that chain is a # fixed architectural constant (brd->prd->sdd->benchmark->...), not -# something that should vary call to call, whereas Coder/Validator's -# sub-tasks have no such fixed small vocabulary of node types. Each node -# here carries its own self-contained `task_prompt` (see the -# design-dynamic-work-graph Skill this prompt points at) instead of an -# `artifact_type` that a fixed lookup table like _DOC_PROMPTS could resolve. -_GENERIC_PROMPT_TEMPLATE = ( - "Plan (do not execute) a Work Graph for this goal: {goal_text}\n" - "Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge " - "when a node's task_prompt genuinely needs another node's output content -- nodes with " - "no real data dependency between them will be scheduled to run CONCURRENTLY, so adding " - "an edge just to make the graph look ordered actively slows execution down. If this goal " - "is small enough to finish in one call, output a single-node graph_manifest rather than " - "forcing a split.\n" - "Output a graph_manifest whose nodes each have: node_id (a short readable identifier), " - "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt " - "for that node's own execution -- assume the agent running it sees nothing except this " - "task_prompt and its dependencies' outputs, not this planning conversation). Every edge " - "needs edge_type: 'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for " - "anything requiring the agent to actually reason about content.\n" - "End your reply with the graph_manifest as a single fenced ```json code block " - "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key." +# something that should vary call to call. Each node here carries its own +# self-contained `task_prompt` (see the design-dynamic-work-graph Skill +# this prompt points at) instead of an `artifact_type` that a fixed lookup +# table like _DOC_PROMPTS could resolve. +# +# 2026-08-12 (Phase 2 §3, user-confirmed 2026-08-08 -- reverses this +# comment's own prior "no fixed small vocabulary of node types" framing): +# Coder/Validator's sub-tasks DO get a suggested (not required) catalog +# now, same "starting points, not requirements" contract _PLANNER_NODE_ +# CATALOG already uses -- the LLM can ignore every entry and invent its +# own node_ids/task_prompts freely, same as before this change. +_NODE_REVIEW_JUDGE_PROMPT = ( + "Judge whether this node's output is genuinely what it claims to be, not a narrative " + "description of an action taken (e.g. 'I ran/produced/completed X') standing in for the real " + "content itself -- this is a known recurring failure mode in nodes that explore with tools " + "before producing a final answer. ok=false if the content is narration/progress-reporting " + "prose rather than the real deliverable (code, test output, structured analysis, etc.), or if " + "it's otherwise clearly incomplete/wrong for what was asked. List each concrete problem in " + "issues.\n\nContent:\n\n{content}" ) +_CODER_NODE_CATALOG = [ + {"node_type": "impl", "description": "Actual code changes implementing the SDD."}, + { + "node_type": "unit_test", + "description": "New/updated unit tests covering the change -- same repo convention as existing tests.", + "suggested_judgment": {"kind": "llm", "judge_prompt": _NODE_REVIEW_JUDGE_PROMPT}, + }, + { + "node_type": "integration_check", + "description": "Actually runs the real test suite/benchmark script against the change and reports " + "pass/fail -- not a description of doing so.", + "suggested_judgment": {"kind": "llm", "judge_prompt": _NODE_REVIEW_JUDGE_PROMPT}, + "suggested_promotes_to_artifact": {"artifact_type": "test_report"}, + }, +] + +_VALIDATOR_NODE_CATALOG = [ + { + "node_type": "independent_benchmark", + "description": "Validator's own held-out acceptance scenarios -- NOT reused from Planner's/Coder's " + "benchmark, a genuine second opinion.", + }, + {"node_type": "test_dataset", "description": "Concrete test data, if this goal has a natural one."}, + { + "node_type": "test_script", + "description": "A real, runnable script mechanically checking the independent benchmark's scenarios.", + "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"}, + }, +] + +_GENERIC_NODE_CATALOGS = {"coder": _CODER_NODE_CATALOG, "validator": _VALIDATOR_NODE_CATALOG} + + +def _generic_prompt_template(org_node): + """Builds _GENERIC_PROMPT_TEMPLATE's replacement, parameterized by + org_node so each gets its own suggested catalog (or none, for any + future org_node this hasn't been extended for yet -- catalog_section + is simply empty then, same as before this change existed). + node_type is a label in the SUGGESTION data only, not a required + graph_manifest node field -- node_id/task_prompt stay exactly as + freeform as before.""" + catalog = _GENERIC_NODE_CATALOGS.get(org_node, []) + catalog_section = ( + "Suggested node types for this goal (use any subset, in any order, invent your own if " + "none fit -- these are starting points, not requirements):\n" + json.dumps(catalog) + "\n" + if catalog else "" + ) + return ( + "Plan (do not execute) a Work Graph for this goal: {goal_text}\n" + + catalog_section + + "Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge " + "when a node's task_prompt genuinely needs another node's output content -- nodes with " + "no real data dependency between them will be scheduled to run CONCURRENTLY, so adding " + "an edge just to make the graph look ordered actively slows execution down. If this goal " + "is small enough to finish in one call, output a single-node graph_manifest rather than " + "forcing a split.\n" + "Output a graph_manifest whose nodes each have: node_id (a short readable identifier), " + "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt " + "for that node's own execution -- assume the agent running it sees nothing except this " + "task_prompt and its dependencies' outputs, not this planning conversation), plus:\n" + " - judgment (optional): how to tell if this node's output is acceptable -- " + "{\"kind\": \"none\"} if nothing needs checking (the default if omitted), {\"kind\": " + "\"llm\", \"judge_prompt\": \"...{content}...\"} for your own judgment prompt with a " + "{content} placeholder.\n" + " - on_fail (only meaningful if judgment.kind != \"none\"): {\"action\": \"retry_self\"}, " + "{\"action\": \"hard_fail\"}, or {\"action\": \"loop_back\"} with a matching on_failure " + "edge to a genuine upstream dependency (same rule _PLANNER_NODE_CATALOG's own nodes " + "follow) -- do NOT use \"dynamic_loop_back\", that's reserved for the terminal review " + "stage outside this graph, not for nodes you plan yourself.\n" + " - promotes_to_artifact (optional): {\"artifact_type\": \"test_report\"} ONLY for a " + "node producing this task's actual final deliverable text -- most nodes should omit " + "this; it marks a node's content for persistence beyond this graph's own execution. " + "(code_diff, if relevant, is derived separately from real file changes, not from any " + "node's own content -- never declare artifact_type: \"code_diff\" here.)\n" + "Every edge needs edge_type: 'zero_token' for pure aggregation/reduce steps, " + "'needs_judgment' for anything requiring the agent to actually reason about content.\n" + "End your reply with the graph_manifest as a single fenced ```json code block " + "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key." + ) + _JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL) @@ -333,22 +434,61 @@ def _normalize_graph_manifest(manifest): return manifest -def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300): - """Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row - (status='planning') -- moving it to 'running'/executing it is out of scope. +def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None): + """Returns {"run_id", "graph_manifest", "new_node_ids"}. Creates one + `work_graph_run` row (status='planning') -- moving it to 'running'/ + executing it is out of scope. 2026-08-08: `org_node == 'planner'` gets `_PROMPT_TEMPLATE` (now a genuine self-planning prompt over `_PLANNER_NODE_CATALOG`, see that template's own comment for why goal_text is filled in via `.replace()` rather than `.format()`). Every other org_node gets - `_GENERIC_PROMPT_TEMPLATE` -- the cross-Org-Agent dynamic planning path - described in the design-dynamic-work-graph Skill, whose nodes carry - their own task_prompt instead of a fixed artifact_type. Function - signature/return shape are unchanged either way.""" + `_generic_prompt_template(org_node)` -- the cross-Org-Agent dynamic + planning path described in the design-dynamic-work-graph Skill, whose + nodes carry their own task_prompt instead of a fixed artifact_type. + + 2026-08-12 (Phase 2 §6.1, user-confirmed 2026-08-08: a parameter on + this function, not a separate plan_work_graph_append()): with + `append_to_run_id=None` (default), behavior is unchanged from before + this parameter existed -- a fresh INSERT. With `append_to_run_id= + `, this call plans ADDITIONAL nodes for an + already-existing work_graph_run (e.g. Coder appending its own + implementation nodes onto Planner's run) instead of creating a new + row: the existing manifest's node_ids are injected into the prompt so + new nodes can `depends_on` them, a `node_id` collision between the + newly-planned nodes and the existing ones raises PlanningFailedError + (never silently renamed or overwritten), and the row is UPDATEd with + the merged manifest instead of a new row being INSERTed. + `new_node_ids` (only meaningfully non-empty in append mode, though + always present for a uniform return shape) is the set of node_ids this + specific call added -- callers that promote node output to `artifact` + rows (e2e_pipeline.py's `_write_completed_node_artifacts`) need this + to avoid re-processing nodes an EARLIER stage already promoted; the + merged manifest itself doesn't distinguish "planned this call" from + "already existed" once merged.""" + existing_manifest = None + if append_to_run_id: + row = conn.execute( + "SELECT graph_manifest FROM work_graph_run WHERE run_id = ?", (append_to_run_id,) + ).fetchone() + if row is None: + raise ValueError("no such work_graph_run to append to: " + append_to_run_id) + existing_manifest = json.loads(row[0]) + if org_node == "planner": prompt = _PROMPT_TEMPLATE.replace("{goal_text}", goal_text) else: - prompt = _GENERIC_PROMPT_TEMPLATE.format(goal_text=goal_text) + prompt = _generic_prompt_template(org_node).replace("{goal_text}", goal_text) + + if existing_manifest is not None: + existing_ids = [n["node_id"] for n in existing_manifest["nodes"]] + prompt = prompt.replace( + "Plan (do not execute)", + "This graph already has these nodes from earlier stages (you may depends_on them, but do " + "NOT redeclare or duplicate them -- pick different node_ids for your own new nodes): " + "{}. Plan (do not execute) ADDITIONAL nodes only".format(existing_ids), + 1, + ) result = subprocess.run( [ @@ -374,24 +514,48 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time payload = json.loads(result.stdout) if payload.get("is_error"): raise PlanningFailedError("plan failed for task {}: {}".format(task_id, payload)) - graph_manifest = _extract_graph_manifest(payload.get("result", "")) - if not graph_manifest: + new_manifest = _extract_graph_manifest(payload.get("result", "")) + if not new_manifest: raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload)) - graph_manifest = _normalize_graph_manifest(graph_manifest) + new_manifest = _normalize_graph_manifest(new_manifest) + + if existing_manifest is None: + run_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES (?, ?, ?, ?, 'planning')", + (run_id, task_id, org_node, json.dumps(new_manifest)), + ) + conn.commit() + return {"run_id": run_id, "graph_manifest": new_manifest, "new_node_ids": []} - run_id = str(uuid.uuid4()) + new_node_ids = [n["node_id"] for n in new_manifest["nodes"]] + existing_ids_set = {n["node_id"] for n in existing_manifest["nodes"]} + collision = existing_ids_set & set(new_node_ids) + if collision: + raise PlanningFailedError( + "append_to_run_id={} node_id collision with existing nodes: {}".format(append_to_run_id, collision) + ) + merged = { + "nodes": existing_manifest["nodes"] + new_manifest["nodes"], + "edges": existing_manifest.get("edges", []) + new_manifest.get("edges", []), + } + # _validate_loop_back_targets over the MERGED graph, not just the new + # nodes -- a new node's on_fail.loop_back could (in principle) target + # an existing node, and _normalize_graph_manifest already validated + # new_manifest in isolation above, not against the merged shape. + _validate_loop_back_targets(merged) conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, ?, ?, 'planning')", - (run_id, task_id, org_node, json.dumps(graph_manifest)), + "UPDATE work_graph_run SET graph_manifest = ? WHERE run_id = ?", + (json.dumps(merged), append_to_run_id), ) conn.commit() - return {"run_id": run_id, "graph_manifest": graph_manifest} + return {"run_id": append_to_run_id, "graph_manifest": merged, "new_node_ids": new_node_ids} def generic_node_prompt_builder(node, outputs): """E (2026-08-06): the `build_prompt(node, outputs)` run_graph() needs - for a graph_manifest produced via _GENERIC_PROMPT_TEMPLATE -- lives + for a graph_manifest produced via `_generic_prompt_template()` -- lives here, not in e2e_pipeline.py, because it's tightly coupled to the `task_prompt` field convention that prompt asks for (keeping both in the same module means a field rename can't silently desync producer @@ -399,3 +563,31 @@ def generic_node_prompt_builder(node, outputs): per the design-dynamic-work-graph Skill's contract, so `outputs` (already-completed dependency content) isn't needed here.""" return node["task_prompt"] + + +def dispatch_build_prompt(planner_builder, generic_builder): + """2026-08-12 (Phase 2 §6.1): a build_prompt(node, outputs) that picks + per-NODE between planner_builder (Planner's artifact_type-keyed nodes, + e.g. e2e_pipeline._planner_prompt_builder's return value) and + generic_builder (task_prompt-keyed nodes, e.g. generic_node_prompt_ + builder above) -- needed once a single graph_manifest can genuinely + contain BOTH shapes side by side, which append_to_run_id makes real + for the first time (Planner's original nodes + Coder's/Validator's + later self-planned ones, merged into one run). Every pre-Phase-2 + caller only ever had ONE shape for a whole graph and dispatched once + up front; this replaces that with a per-node check so a caller doesn't + need to know in advance whether ANY given node in a merged graph is + Planner-shaped or generic-shaped. Shared here (not duplicated in + e2e_pipeline.py/work_graph_fork.py, its two real callers) so the two + don't drift on what "neither shape" should do.""" + def build_prompt(node, outputs): + if "artifact_type" in node: + return planner_builder(node, outputs) + if "task_prompt" in node: + return generic_builder(node, outputs) + raise ValueError( + "node {} has neither 'artifact_type' nor 'task_prompt' -- can't dispatch a build_prompt".format( + node.get("node_id") + ) + ) + return build_prompt -- libgit2 0.21.4