e2e_pipeline.py
8.71 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
"""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).
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. Content generation is
separate, plain org_node_trigger.trigger_org_node calls (no status
side-effects) so each document/step's real text is available to write as an
artifact -- trigger_org_node_full's own contract (a status dict, no text)
isn't changed to accommodate this.
"""
import os
import subprocess
from src import artifact_write, org_node_lifecycle, org_node_trigger, task_create
_READY_PROMPT = "Confirm you're ready to begin. Take no other action yet."
_DOC_PROMPTS = {
"brd": (
"Write a BRD (Business Requirement Document) for this goal, following your own "
"SOUL.md methodology: {goal_text}\n"
"Output ONLY the BRD content in Markdown -- 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. "
"Output ONLY the PRD content in Markdown."
),
"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. Output ONLY the SDD content in Markdown."
),
"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. "
"Output ONLY the Benchmark content in Markdown."
),
}
_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). 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 plan_documents(conn, task_id, run_id, goal_text, target_folder_path, timeout=300):
"""Architecture §11 criterion 1: real Planner calls producing BRD->PRD->SDD
->Benchmark, each written via c_artifact_write with correct lineage.
Returns {"brd": artifact_id, "prd": ..., "sdd": ..., "benchmark": ...}.
"""
artifact_ids = {}
contents = {}
prev_type = None
for artifact_type in ("brd", "prd", "sdd", "benchmark"):
prompt = _DOC_PROMPTS[artifact_type].format(
goal_text=goal_text, **{k + "_content": v for k, v in contents.items()}
)
content = org_node_trigger.trigger_org_node("planner", prompt, target_folder_path, timeout=timeout)
deps = [artifact_ids[prev_type]] if prev_type else []
result = artifact_write.write_artifact(
conn, target_folder_path, artifact_type, task_id, run_id,
".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), content,
depends_on_artifact_ids=deps,
)
artifact_ids[artifact_type] = result["artifact_id"]
contents[artifact_type] = content
prev_type = artifact_type
return artifact_ids
def implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200):
"""Architecture §11 criterion 2: real Coder call, 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)
response = org_node_trigger.trigger_org_node(
"coder", _CODER_PROMPT.format(goal_text=goal_text, sdd_content=sdd_content),
target_folder_path, timeout=timeout,
)
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"}.
"""
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)
# trigger_org_node_full itself doesn't create a work_graph_run row (that's
# U-011's plan_work_graph, a separate call this pipeline doesn't invoke,
# since REQ-02's graph_manifest and this pipeline's actual document
# content are orthogonal concerns) -- artifact.produced_by_run_id is a
# real FK, so a row has to exist; this one just records that the e2e
# pipeline (not a real planned Work Graph) produced these artifacts
run_id = "e2e-" + task_id
conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES (?, ?, 'planner', '{}', 'running')",
(run_id, task_id),
)
conn.commit()
artifact_ids = plan_documents(conn, task_id, run_id, goal_text, target_folder_path)
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, run_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()