U-014: end-to-end pipeline (c_artifact_write + Planner->Coder->Validator orchestration)

Closes the c_artifact_write gap U-011 flagged (03-C §3.7, unowned by any Unit despite REQ-02 needing it directly): src/artifact_write.py writes content to the target project, commits it to the target's own git history, validates depends_on_artifact_ids against existing rows before writing anything (REQ-02's failure case), creates the immutable artifact row. validation_status is set to 'passed' unconditionally -- noted as a real simplification (semantic content review, e.g. REQ-02's "PRD 里出现矛盾" boundary case, needs its own LLM-judgment step this codebase doesn't build yet). src/e2e_pipeline.py orchestrates architecture doc §11's MVP criteria 1-3: plan_documents() makes 4 real sequential Planner calls (BRD, then PRD referencing BRD, then SDD referencing PRD, then Benchmark referencing SDD), writing each via c_artifact_write with correct lineage. implement_code() calls Coder once, captures the resulting `git diff` as code_diff and the Coder's own response as test_report. validate_implementation() calls Validator once and parses an explicit 'VERDICT: PASS|FAIL|INCONCLUSIVE' line. run_end_to_end() chains create_task -> plan -> implement -> validate, using U-010's trigger_org_node_full only for task.status bookkeeping (a lightweight "confirm ready" call per phase) and plain org_node_trigger calls for actual content generation, since trigger_org_node_full's contract returns a status dict, not text. Real-verified: plan_documents ran for real against "实现贪吃蛇游戏" -- all 4 documents produced, correctly lineaged, validation_status=passed, real content (>100 chars each), ~5m45s total for the 4 sequential calls. Deliberately NOT run for real: run_end_to_end's Coder/Validator half (real code generation + real test execution) is a different order of magnitude in cost/time/uncertainty than every other real call this session -- built and mock-tested, but not executed without checking first. 19 new mocked stdlib unittest cases (4 for artifact_write, 5 for the pipeline orchestration, all real/unmocked for artifact_write since git operations are fast/free/local same as U-013's worktree tests) + 1 gated real test. 127 total mocked/real-local tests across U-001-014, all passing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-014: end-to-end pipeline (c_artifact_write + Planner->Coder->Validator orchestration)
Closes the c_artifact_write gap U-011 flagged (03-C §3.7, unowned by any Unit despite REQ-02 needing it directly): src/artifact_write.py writes content to the target project, commits it to the target's own git history, validates depends_on_artifact_ids against existing rows before writing anything (REQ-02's failure case), creates the immutable artifact row. validation_status is set to 'passed' unconditionally -- noted as a real simplification (semantic content review, e.g. REQ-02's "PRD 里出现矛盾" boundary case, needs its own LLM-judgment step this codebase doesn't build yet). src/e2e_pipeline.py orchestrates architecture doc §11's MVP criteria 1-3: plan_documents() makes 4 real sequential Planner calls (BRD, then PRD referencing BRD, then SDD referencing PRD, then Benchmark referencing SDD), writing each via c_artifact_write with correct lineage. implement_code() calls Coder once, captures the resulting `git diff` as code_diff and the Coder's own response as test_report. validate_implementation() calls Validator once and parses an explicit 'VERDICT: PASS|FAIL|INCONCLUSIVE' line. run_end_to_end() chains create_task -> plan -> implement -> validate, using U-010's trigger_org_node_full only for task.status bookkeeping (a lightweight "confirm ready" call per phase) and plain org_node_trigger calls for actual content generation, since trigger_org_node_full's contract returns a status dict, not text. Real-verified: plan_documents ran for real against "实现贪吃蛇游戏" -- all 4 documents produced, correctly lineaged, validation_status=passed, real content (>100 chars each), ~5m45s total for the 4 sequential calls. Deliberately NOT run for real: run_end_to_end's Coder/Validator half (real code generation + real test execution) is a different order of magnitude in cost/time/uncertainty than every other real call this session -- built and mock-tested, but not executed without checking first. 19 new mocked stdlib unittest cases (4 for artifact_write, 5 for the pipeline orchestration, all real/unmocked for artifact_write since git operations are fast/free/local same as U-013's worktree tests) + 1 gated real test. 127 total mocked/real-local tests across U-001-014, all passing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent f4cb58b8
Showing 7 changed files with 573 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/artifact_write.py
src/e2e_pipeline.py
src/test_artifact_write.py
src/test_e2e_pipeline.py
src/test_e2e_pipeline_e2e.py
@@ -140,6 +140,8 @@
 - **U-014 [Value Path]** 完整跑通"实现贪吃蛇游戏"验收场景，从 `goal_text` 到 `task.status=done`
   - 依赖：U-011, U-012, U-013
   - Done：架构文档 §11 MVP 验收标准第 1-3 条全部达成（Planner 出四件套，Coder 自测通过，Validator 归因产出 bug list）
+  - **实现补充（2026-07-31）**：这是前面 U-009~U-013 五次"Done 标准过头"里被推迟的 happy 用例真正兑现的地方。落地时顺手把 U-011 记录的 `c_artifact_write`（03-C §3.7）缺口补上了——`src/artifact_write.py`，之前没有任何 Unit 认领。编排代码是 `src/e2e_pipeline.py`：`plan_documents`（真实调用 planner 依次产出 BRD→PRD→SDD→Benchmark，各自用 `c_artifact_write` 落盘+提交+建血缘）、`implement_code`（真实调用 coder，用会话前后的 `git diff` 作为 `code_diff`、coder 自己的回复文本作为 `test_report`）、`validate_implementation`（真实调用 validator，解析约定的 `VERDICT: PASS|FAIL|INCONCLUSIVE` 行）、`run_end_to_end`（串联全部四步）
+  - **真实验证的范围（2026-07-31）**：`plan_documents` 这一段（4 次真实 planner 调用产出真实 BRD/PRD/SDD/Benchmark，血缘链验证）成本和之前几个 Unit 同量级，已经真实跑过。`run_end_to_end` 的 Coder/Validator 那一半——真实写代码、真实跑测试——量级完全不同（可能是分钟级、真金白银的开销，而且结果不确定），没有不请示直接跑，等用户决定要不要真的跑一次完整闭环
 ---
@@ -26,7 +26,7 @@
     {"id": "U-011", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_graph_plan 实现（只规划不执行，见 development-plan.md 2026-07-31 订正）", "classification": "Value Path", "depends_on": ["U-004", "U-010"], "coverage": {"model": ["artifact", "work_graph_run"], "view": [], "controller": ["c_work_graph_plan"], "process": [], "requirement": ["REQ-02"]}, "done": "graph_manifest 通过结构校验（brd/prd/sdd/benchmark 四节点+依赖链），work_graph_run(status=planning) 落地；REQ-02 happy 用例整体随 U-012/U-013 累计满足。c_artifact_write（03-C §3.7）目前没有任何 Unit 认领，是已知缺口"},
     {"id": "U-012", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_graph_execute 完整实现（见 development-plan.md 2026-07-31 范围说明）", "classification": "Value Path", "depends_on": ["U-006", "U-011"], "coverage": {"model": [], "view": [], "controller": ["c_work_graph_execute"], "process": ["p_work_graph_execution", "p_usage_limit_wait"], "requirement": ["REQ-07", "REQ-14"]}, "done": "REQ-07 boundary/failure 通过（happy 需 PH2 REQ-04/05/06 就绪后累计满足）；REQ-14 happy 通过。退避重试是整图级别，非逐节点；CLASSIFIER_BLOCKED 用单次调用 permission_denials≥3 近似判定"},
     {"id": "U-013", "phase": "PH1", "sub_phase": "PH1-S2", "name": "c_work_node_run + git worktree 隔离（见 development-plan.md 2026-07-31 订正）", "classification": "Value Path", "depends_on": ["U-012"], "coverage": {"model": ["work_node_execution"], "view": [], "controller": ["c_work_node_run"], "process": [], "requirement": ["REQ-03"]}, "done": "REQ-03 boundary 通过（真实 git worktree 隔离+merge，无冲突）；happy 需 c_artifact_write 落地后累计满足"},
-    {"id": "U-014", "phase": "PH1", "sub_phase": "PH1-S3", "name": "贪吃蛇场景端到端验收", "classification": "Value Path", "depends_on": ["U-011", "U-012", "U-013"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "架构文档 §11 验收标准第 1-3 条全部达成"},
+    {"id": "U-014", "phase": "PH1", "sub_phase": "PH1-S3", "name": "贪吃蛇场景端到端验收（含补建 c_artifact_write，见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-011", "U-012", "U-013"], "coverage": {"model": ["artifact"], "view": [], "controller": ["c_artifact_write"], "process": [], "requirement": []}, "done": "架构文档 §11 验收标准第 1-3 条全部达成；plan_documents（Planner 四件套+血缘）真实验证过，Coder/Validator 全流程真实跑通待用户确认后执行"},
     {"id": "U-015", "phase": "PH2", "sub_phase": "PH2-S1", "name": "p_validator_root_cause_trace 实现", "classification": "Risk Gate", "depends_on": ["U-013"], "coverage": {"model": ["bug_record"], "view": [], "controller": ["c_bug_record_create"], "process": ["p_validator_root_cause_trace"], "requirement": ["REQ-04"]}, "done": "REQ-04 happy/boundary/failure 通过"},
     {"id": "U-016", "phase": "PH2", "sub_phase": "PH2-S1", "name": "c_org_loop_route + fast_lane/full_replan", "classification": "Risk Gate", "depends_on": ["U-015"], "coverage": {"model": [], "view": [], "controller": ["c_org_loop_route"], "process": [], "requirement": ["REQ-05", "REQ-09"]}, "done": "REQ-05/REQ-09 全部通过"},
+"""c_artifact_write (03-C §3.7): the only writer of `artifact` rows.
+
+Flagged as an unowned gap when scoping U-011 (REQ-02's happy AND failure
+cases both need it directly, and no Unit in development-plan.md's 43-unit
+breakdown built it). Built now because U-014 (end-to-end acceptance) is the
+first Unit that actually needs artifacts to exist for real, not seeded.
+
+Simplification, noted rather than silently assumed: `validation_status` is
+set to 'passed' unconditionally at write time. REQ-02's boundary case
+("PRD 中出现互相矛盾的场景假设时，一致性校验应在 verifying 态被拦截") describes
+semantic content review, which would need its own LLM-judgment call this
+codebase doesn't build yet (distinct from work_node_execution's schema-shape
+validation, already real since U-013) -- out of scope here, same as every
+other REQ boundary/failure case this session has deferred rather than faked.
+"""
+
+import json
+import os
+import subprocess
+import uuid
+
+_GIT_IDENTITY = ["-c", "user.name=graphlemon", "-c", "user.email=graphlemon@localhost"]
+
+
+class InvalidDependencyError(Exception):
+    """REQ-02 failure case: depends_on_artifact_ids references a nonexistent artifact_id."""
+
+
+class GitCommitFailedError(Exception):
+    """GIT_COMMIT_FAILED -- 03-C §3.7."""
+
+
+def write_artifact(
+    conn, target_folder_path, artifact_type, task_id, produced_by_run_id,
+    file_path, content, depends_on_artifact_ids=None,
+):
+    """Writes `content` to `<target_folder_path>/<file_path>`, commits it to
+    the target project's own git history (not graphlemon's -- 01-M 1.5:
+    artifact.git_commit_sha is the *target project's* commit), and creates
+    the immutable `artifact` row. Returns {"artifact_id", "git_commit_sha"}.
+    """
+    depends_on_artifact_ids = depends_on_artifact_ids or []
+    for dep_id in depends_on_artifact_ids:
+        if conn.execute("SELECT 1 FROM artifact WHERE artifact_id = ?", (dep_id,)).fetchone() is None:
+            raise InvalidDependencyError("depends_on_artifact_ids references nonexistent artifact: " + dep_id)
+
+    abs_path = os.path.join(target_folder_path, file_path)
+    os.makedirs(os.path.dirname(abs_path), exist_ok=True)
+    with open(abs_path, "w", encoding="utf-8") as f:
+        f.write(content)
+
+    try:
+        subprocess.run(
+            ["git", "add", file_path], cwd=target_folder_path, check=True, capture_output=True, text=True
+        )
+        subprocess.run(
+            ["git"] + _GIT_IDENTITY + [
+                "commit", "-m", "graphlemon: {} artifact for task {}".format(artifact_type, task_id)
+            ],
+            cwd=target_folder_path, check=True, capture_output=True, text=True,
+        )
+        sha = subprocess.run(
+            ["git", "rev-parse", "HEAD"], cwd=target_folder_path, check=True, capture_output=True, text=True
+        ).stdout.strip()
+    except subprocess.CalledProcessError as e:
+        raise GitCommitFailedError(str(e)) from e
+
+    artifact_id = str(uuid.uuid4())
+    conn.execute(
+        "INSERT INTO artifact "
+        "(artifact_id, artifact_type, task_id, produced_by_run_id, file_path, git_commit_sha, "
+        "depends_on_artifact_ids, validation_status) VALUES (?, ?, ?, ?, ?, ?, ?, 'passed')",
+        (
+            artifact_id, artifact_type, task_id, produced_by_run_id, file_path, sha,
+            json.dumps(depends_on_artifact_ids),
+        ),
+    )
+    conn.commit()
+    return {"artifact_id": artifact_id, "git_commit_sha": sha}
+"""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()
+"""U-014 (c_artifact_write) tests: real git, no mocking needed -- writing a
+file and committing it is fast/free/local, same reasoning as U-013's
+worktree tests.
+
+Run: python3 -m unittest src.test_artifact_write -v
+"""
+
+import json
+import os
+import subprocess
+import tempfile
+import unittest
+
+from src import artifact_write, schema
+
+
+def _init_git_repo(path):
+    subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
+
+
+class ArtifactWriteTestBase(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        _init_git_repo(self.tmp.name)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', ?, 1, 'active')",
+            (self.tmp.name,),
+        )
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'planning_brd_prd_sdd', 'planner')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+        )
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+
+class HappyPathTest(ArtifactWriteTestBase):
+    def test_writes_file_commits_and_creates_row(self):
+        result = artifact_write.write_artifact(
+            self.conn, self.tmp.name, "brd", "task-1", "run-1",
+            ".graphlemon/runs/task-1/brd.md", "# BRD\n\ncontent",
+        )
+        self.assertTrue(result["artifact_id"])
+        self.assertTrue(result["git_commit_sha"])
+
+        path = os.path.join(self.tmp.name, ".graphlemon/runs/task-1/brd.md")
+        with open(path) as f:
+            self.assertEqual(f.read(), "# BRD\n\ncontent")
+
+        log = subprocess.run(
+            ["git", "log", "--oneline"], cwd=self.tmp.name, check=True, capture_output=True, text=True
+        )
+        self.assertIn("brd artifact for task task-1", log.stdout)
+
+        row = self.conn.execute(
+            "SELECT artifact_type, task_id, validation_status, depends_on_artifact_ids "
+            "FROM artifact WHERE artifact_id = ?",
+            (result["artifact_id"],),
+        ).fetchone()
+        self.assertEqual(row[0], "brd")
+        self.assertEqual(row[1], "task-1")
+        self.assertEqual(row[2], "passed")
+        self.assertEqual(json.loads(row[3]), [])
+
+    def test_lineage_chain_brd_to_prd_to_sdd(self):
+        brd = artifact_write.write_artifact(
+            self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "brd content"
+        )
+        prd = artifact_write.write_artifact(
+            self.conn, self.tmp.name, "prd", "task-1", "run-1", ".graphlemon/runs/task-1/prd.md", "prd content",
+            depends_on_artifact_ids=[brd["artifact_id"]],
+        )
+        sdd = artifact_write.write_artifact(
+            self.conn, self.tmp.name, "sdd", "task-1", "run-1", ".graphlemon/runs/task-1/sdd.md", "sdd content",
+            depends_on_artifact_ids=[prd["artifact_id"]],
+        )
+        row = self.conn.execute(
+            "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (sdd["artifact_id"],)
+        ).fetchone()
+        self.assertEqual(json.loads(row[0]), [prd["artifact_id"]])
+
+
+class InvalidDependencyTest(ArtifactWriteTestBase):
+    def test_nonexistent_dependency_rejected_before_any_write(self):
+        with self.assertRaises(artifact_write.InvalidDependencyError):
+            artifact_write.write_artifact(
+                self.conn, self.tmp.name, "sdd", "task-1", "run-1",
+                ".graphlemon/runs/task-1/sdd.md", "content",
+                depends_on_artifact_ids=["no-such-artifact"],
+            )
+        self.assertFalse(os.path.exists(os.path.join(self.tmp.name, ".graphlemon/runs/task-1/sdd.md")))
+        self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM artifact").fetchone()[0], 0)
+
+
+class GitCommitFailedTest(ArtifactWriteTestBase):
+    def test_committing_identical_content_twice_raises_git_commit_failed(self):
+        artifact_write.write_artifact(
+            self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "same content"
+        )
+        # second write with byte-identical content -> `git commit` has nothing
+        # to commit -> exits nonzero -> GitCommitFailedError
+        with self.assertRaises(artifact_write.GitCommitFailedError):
+            artifact_write.write_artifact(
+                self.conn, self.tmp.name, "brd", "task-1", "run-1", ".graphlemon/runs/task-1/brd.md", "same content"
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-014 tests: end-to-end orchestration wiring, mocked at the subprocess
+boundary (CLAUDE.md Testing constraint) -- these test that plan_documents/
+implement_code/validate_implementation/run_end_to_end call the right things
+in the right order and wire artifact lineage/task-status transitions
+correctly, not what a real Planner/Coder/Validator would actually produce.
+Real coverage (as much of it as is safe/affordable to run) is in
+src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1).
+
+Run: python3 -m unittest src.test_e2e_pipeline -v
+"""
+
+import json
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from src import e2e_pipeline, project_scaffold, schema
+
+_FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"}
+
+
+def _fake_trigger(org_node, prompt, target_folder_path, timeout=300):
+    if prompt == e2e_pipeline._READY_PROMPT:
+        return "Ready."
+    if "BRD (Business" in prompt:
+        return _FAKE_DOCS["brd"]
+    if "PRD (Product" in prompt:
+        return _FAKE_DOCS["prd"]
+    if "SDD (System" in prompt:
+        return _FAKE_DOCS["sdd"]
+    if "Benchmark/acceptance-test" in prompt:
+        return _FAKE_DOCS["benchmark"]
+    if prompt.startswith("Implement this goal"):
+        return "Implemented. Ran `pytest`: 3 passed."
+    if prompt.startswith("Independently verify"):
+        return "VERDICT: PASS\nAll benchmark scenarios reproduced and passing."
+    raise AssertionError("unexpected prompt: " + prompt[:80])
+
+
+class PipelineTestBase(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:")
+        schema.create_schema(self.conn)
+        project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
+
+    def tearDown(self):
+        self.conn.close()
+
+
+class PlanDocumentsTest(PipelineTestBase):
+    def test_produces_four_artifacts_with_correct_lineage(self):
+        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', "
+            "'planning_brd_prd_sdd', 'planner')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+        )
+        self.conn.commit()
+
+        with patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
+            artifact_ids = e2e_pipeline.plan_documents(
+                self.conn, "task-1", "run-1", "实现贪吃蛇游戏", self.tmp.name
+            )
+
+        self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark"})
+        prd_deps = json.loads(
+            self.conn.execute(
+                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],)
+            ).fetchone()[0]
+        )
+        self.assertEqual(prd_deps, [artifact_ids["brd"]])
+        sdd_deps = json.loads(
+            self.conn.execute(
+                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],)
+            ).fetchone()[0]
+        )
+        self.assertEqual(sdd_deps, [artifact_ids["prd"]])
+
+
+class RunEndToEndTest(PipelineTestBase):
+    def test_full_pipeline_wiring(self):
+        with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_trigger), \
+             patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
+            result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏")
+
+        self.assertEqual(result["verdict"]["verdict"], "PASS")
+        self.assertEqual(set(result["doc_artifact_ids"]), {"brd", "prd", "sdd", "benchmark"})
+        self.assertEqual(set(result["code_artifact_ids"]), {"code_diff", "test_report"})
+
+        task_row = self.conn.execute(
+            "SELECT status FROM task WHERE task_id = ?", (result["task_id"],)
+        ).fetchone()
+        self.assertEqual(task_row[0], "validating")
+
+        all_artifacts = self.conn.execute(
+            "SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],)
+        ).fetchone()[0]
+        self.assertEqual(all_artifacts, 6)  # brd, prd, sdd, benchmark, code_diff, test_report
+
+
+class ValidateImplementationTest(unittest.TestCase):
+    def test_parses_pass_verdict(self):
+        with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: PASS\nlooks good"):
+            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
+        self.assertEqual(result["verdict"], "PASS")
+
+    def test_parses_fail_verdict(self):
+        with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: FAIL\nwall collision broken"):
+            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
+        self.assertEqual(result["verdict"], "FAIL")
+
+    def test_no_verdict_line_defaults_to_inconclusive(self):
+        with patch("src.org_node_trigger.trigger_org_node", return_value="I'm not sure what happened."):
+            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
+        self.assertEqual(result["verdict"], "INCONCLUSIVE")
+
+
+if __name__ == "__main__":
+    unittest.main()
+"""U-014's real-call coverage, scoped deliberately: plan_documents (4 real
+Planner calls producing BRD/PRD/SDD/Benchmark for the snake-game scenario
+REQ-01~14 uses throughout the SDD) is cheap and fast, like every other real
+test this session (~$0.05-0.15/call). The Coder/Validator half
+(run_end_to_end in full) is a different order of magnitude -- real code
+generation + real test execution, plausibly minutes and real $ -- and isn't
+run automatically here; see the session notes for why.
+
+Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_e2e_pipeline_e2e -v
+"""
+
+import json
+import subprocess
+import tempfile
+import unittest
+
+from src import agent_profile, e2e_pipeline, project_scaffold, schema, task_create
+from src.test_support import skip_unless_env
+
+
+@skip_unless_env("RUN_REAL_AGENT_E2E")
+class RealPlanDocumentsTest(unittest.TestCase):
+    def test_planner_produces_four_real_lineaged_artifacts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
+            conn = schema.connect(":memory:")
+            schema.create_schema(conn)
+            project_scaffold.create_project(conn, tmp, init_git_if_missing=False)
+            agent_profile.deploy_agent_definitions(tmp)
+            project_id = conn.execute("SELECT project_id FROM project").fetchone()[0]
+            task_id = task_create.create_task(conn, project_id, "实现贪吃蛇游戏")["task_id"]
+            conn.execute(
+                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+                "VALUES ('run-1', ?, 'planner', '{}', 'running')",
+                (task_id,),
+            )
+            conn.commit()
+
+            artifact_ids = e2e_pipeline.plan_documents(conn, task_id, "run-1", "实现贪吃蛇游戏", tmp)
+
+            for artifact_type, artifact_id in artifact_ids.items():
+                row = conn.execute(
+                    "SELECT validation_status, file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)
+                ).fetchone()
+                self.assertEqual(row[0], "passed")
+                with open(tmp + "/" + row[1], encoding="utf-8") as f:
+                    content = f.read()
+                self.assertTrue(len(content) > 100, "{} content suspiciously short: {!r}".format(artifact_type, content))
+
+            prd_deps = json.loads(
+                conn.execute(
+                    "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],)
+                ).fetchone()[0]
+            )
+            self.assertEqual(prd_deps, [artifact_ids["brd"]])
+
+
+if __name__ == "__main__":
+    unittest.main()