e2e_pipeline.py
12.2 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
"""U-014: end-to-end orchestration for architecture doc §11 MVP acceptance
criteria 1-3 (Planner produces validated BRD/PRD/SDD/Benchmark; Coder
implements against the SDD and self-tests pass; Validator runs
benchmark/tests and produces a root-cause-attributed verdict). Criteria 4-7
(Org-level Loop, breakpoint recovery, ...) are explicitly out of this Unit's
scope -- PH2 (U-017/U-021).
**Rewired 2026-08-01 (U-031 regression audit)**: `plan_documents` used to
call `org_node_trigger.trigger_org_node()` directly, bypassing
`work_graph_plan.plan_work_graph()`/`work_graph_execute.execute_work_graph()`
/`work_node_run.run_work_node()` entirely -- those three modules were each
fully unit-tested (and even real-API-validated) in isolation since PH1/PH2,
but nothing in production ever called any of them; this was the one real,
validated closed loop, and it used a completely separate, simpler mechanism.
REQ-02's own coverage line names `c_work_graph_plan`/`c_work_graph_execute`
explicitly, so that gap is now closed: `plan_documents` calls
`plan_work_graph()` for a real graph_manifest, then
`work_graph_orchestrate.run_graph()` executes each node for real (see that
module's docstring for why each node gets its own one-node dynamic-workflow
script rather than one big multi-node script -- REQ-06 breakpoint recovery
needs it). `implement_code` (Coder) goes through the same per-node
mechanism with a synthetic single-node manifest, since REQ-03's own
coverage only names `c_work_node_run`, not graph planning -- there is no
real "plan a Coder module graph" LLM capability built (a much bigger,
separate undertaking; planning docs and planning code modules are
different problems), so a one-node graph is the honest minimum that makes
`c_work_node_run` genuinely load-bearing for Coder too.
`validate_implementation` (Validator) is unchanged -- REQ-04's own
coverage never named the Work Graph controllers at all.
Status-transition bookkeeping (task.status, precondition guards) stays
U-010's job (org_node_lifecycle.trigger_org_node_full) -- called once per
phase with a lightweight "confirm ready" prompt.
"""
import json
import os
import subprocess
import uuid
from src import (
artifact_write, org_node_lifecycle, org_node_trigger, task_create,
work_graph_orchestrate, work_graph_plan,
)
_READY_PROMPT = "Confirm you're ready to begin. Take no other action yet."
_DOC_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
_DOC_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]}
_DOC_PROMPTS = {
"brd": (
"Write a BRD (Business Requirement Document) for this goal, following your own "
"SOUL.md methodology: {goal_text}\n"
"Reply with the document content (Markdown) via the required schema -- no preamble, "
"no meta-commentary about what you're about to do."
),
"prd": (
"Here is the approved BRD for goal '{goal_text}':\n\n{brd_content}\n\n"
"Write the PRD (Product Requirement Document) based on this BRD. "
"Reply with the document content (Markdown) via the required schema."
),
"sdd": (
"Here is the approved PRD for goal '{goal_text}':\n\n{prd_content}\n\n"
"Write the SDD (System Design Document) based on this PRD, including a Development "
"Unit breakdown. Reply with the document content (Markdown) via the required schema."
),
"benchmark": (
"Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n"
"Write the Benchmark/acceptance-test document based on this SDD: concrete test "
"scenarios (happy/boundary/failure) with observable pass criteria. "
"Reply with the document content (Markdown) via the required schema."
),
}
_CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
_CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]}
_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 "
"and can explain why not). Reply via the required schema with a `response` field describing "
"what you did and the real test commands/output. Goal: {goal_text}\n\nSDD:\n\n{sdd_content}"
)
_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 "
"'VERDICT: PASS' or 'VERDICT: FAIL' or 'VERDICT: INCONCLUSIVE' followed by your reasoning "
"and, if FAIL, a root_cause_category (planning|coder|skill_defect) with detail. "
"Goal: {goal_text}\n\nBenchmark:\n\n{benchmark_content}"
)
def _planner_prompt_builder(goal_text, node_artifact_types):
"""node_artifact_types: dict[node_id] -> artifact_type, read off the
real graph_manifest before running it (build_prompt only ever sees one
node/outputs-so-far pair at a time, not the whole manifest)."""
def build_prompt(node, outputs):
node_id = node["node_id"]
artifact_type = node_artifact_types.get(node_id, node_id)
content_kwargs = {}
for dep_id in (node.get("depends_on") or []):
dep_output = outputs.get(dep_id)
if dep_output is not None:
dep_type = node_artifact_types.get(dep_id, dep_id)
content_kwargs[dep_type + "_content"] = dep_output["content"]
return _DOC_PROMPTS[artifact_type].format(goal_text=goal_text, **content_kwargs)
return build_prompt
def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
"""Architecture §11 criterion 1: a real Planner Work Graph
(c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark,
each written via c_artifact_write with correct lineage.
Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}.
"""
plan_result = work_graph_plan.plan_work_graph(
conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout
)
run_id = plan_result["run_id"]
graph_manifest = plan_result["graph_manifest"]
node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}
outputs = work_graph_orchestrate.run_graph(
conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
_planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
)
artifact_ids = {}
for node in work_graph_orchestrate.topological_order(graph_manifest):
node_id = node["node_id"]
artifact_type = node_artifact_types[node_id]
deps = [
artifact_ids[node_artifact_types[dep_id]]
for dep_id in (node.get("depends_on") or [])
if node_artifact_types.get(dep_id) in artifact_ids
]
result = artifact_write.write_artifact(
conn, target_folder_path, artifact_type, task_id, run_id,
".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), outputs[node_id]["content"],
depends_on_artifact_ids=deps,
)
artifact_ids[artifact_type] = result["artifact_id"]
return {"run_id": run_id, "artifact_ids": artifact_ids}
def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200):
"""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.
"""
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),
_CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout,
)
response = outputs["implement"]["response"]
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,
)
return {"code_diff": diff_result["artifact_id"], "test_report": report_result["artifact_id"]}
def validate_implementation(task_id, benchmark_content, goal_text, target_folder_path, timeout=1200):
"""Architecture §11 criterion 3: real Validator call, parses the
'VERDICT: PASS|FAIL|INCONCLUSIVE' line this Unit's prompt asks for.
Returns {"verdict", "raw_response"}. Unchanged by the 2026-08-01
rewiring -- REQ-04's own coverage never named the Work Graph
controllers (c_bug_record_create/p_validator_root_cause_trace only),
so the simpler flat trigger_org_node call is already faithful to what
this REQ actually requires.
"""
response = org_node_trigger.trigger_org_node(
"validator",
_VALIDATOR_PROMPT.format(goal_text=goal_text, benchmark_content=benchmark_content),
target_folder_path, timeout=timeout,
)
verdict = "INCONCLUSIVE"
for line in response.splitlines():
stripped = line.strip().upper()
if stripped.startswith("VERDICT:"):
for candidate in ("PASS", "FAIL", "INCONCLUSIVE"):
if candidate in stripped:
verdict = candidate
break
break
return {"verdict": verdict, "raw_response": response}
def _current_commit(repo_path):
return subprocess.run(
["git", "rev-parse", "HEAD"], cwd=repo_path, capture_output=True, text=True
).stdout.strip()
def run_end_to_end(conn, target_folder_path, goal_text):
"""Full orchestration: create_task -> plan_documents -> implement_code ->
validate_implementation. Returns a dict with every intermediate result,
for inspection/debugging rather than hiding it behind a single verdict.
"""
project_id = conn.execute(
"SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
).fetchone()[0]
task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT)
plan_result = plan_documents(conn, task_id, goal_text, target_folder_path)
artifact_ids = plan_result["artifact_ids"]
sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
benchmark_content = _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"])
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)
org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT)
verdict = validate_implementation(task_id, benchmark_content, goal_text, target_folder_path)
return {
"task_id": task_id,
"doc_artifact_ids": artifact_ids,
"code_artifact_ids": code_artifact_ids,
"verdict": verdict,
}
def _read_artifact_content(conn, target_folder_path, artifact_id):
file_path = conn.execute("SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)).fetchone()[0]
with open(os.path.join(target_folder_path, file_path), encoding="utf-8") as f:
return f.read()