宜博 / lemongraph

"""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).

**Rewired 2026-08-01 (U-031 regression audit)**: plan_documents/
implement_code now go through work_graph_plan.plan_work_graph +
work_graph_orchestrate.run_graph (which calls work_graph_execute.
execute_work_graph per node) instead of a flat org_node_trigger.
trigger_org_node call -- see e2e_pipeline.py's module docstring for why.
plan_work_graph is mocked at its own subprocess boundary (same pattern as
test_work_graph_plan.py) so its real graph_manifest-parsing logic and its
real work_graph_run INSERT both still run for real; execute_work_graph is
mocked directly (one level higher than its own subprocess boundary) since
this file's job is testing e2e_pipeline's wiring, not re-proving
execute_work_graph's own mechanics (covered in test_work_graph_execute.py)
or work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py).

Run: python3 -m unittest src.test_e2e_pipeline -v
"""

import json
import os
import subprocess
import tempfile
import unittest
from unittest.mock import patch

from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute

_FAKE_DOCS = {
    "brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake",
    "test_dataset": "N/A: no natural dataset for this fake goal",
    "test_script": "assert True  # fake but real, compilable script",
}

_FAKE_GRAPH_MANIFEST = {
    "nodes": [
        {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
        {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
        {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]},
        {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]},
        {"node_id": "test_dataset", "artifact_type": "test_dataset", "depends_on": ["benchmark"]},
        {"node_id": "test_script", "artifact_type": "test_script", "depends_on": ["test_dataset"]},
    ],
    "edges": [
        {"from": "brd", "to": "prd", "edge_type": "needs_judgment"},
        {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"},
        {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"},
        {"from": "benchmark", "to": "test_dataset", "edge_type": "needs_judgment"},
        {"from": "test_dataset", "to": "test_script", "edge_type": "needs_judgment"},
    ],
}


def _fake_plan_subprocess_run(*args, **kwargs):
    payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_GRAPH_MANIFEST) + "\n```"}
    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")


def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
    assert prompt == e2e_pipeline._READY_PROMPT
    return "Ready."


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_six_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.commit()

        with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
             patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
            mock_execute.side_effect = [
                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                {"content": _FAKE_DOCS["test_dataset"]}, {"content": _FAKE_DOCS["test_script"]},
            ]
            plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name)

        artifact_ids = plan_result["artifact_ids"]
        self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"})
        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"]])

        # plan_work_graph's own real work_graph_run INSERT actually ran (not bypassed)
        run_row = self.conn.execute(
            "SELECT org_node, status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
        ).fetchone()
        self.assertEqual(run_row, ("planner", "planning"))


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.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."):
            mock_execute.side_effect = [
                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                {"content": _FAKE_DOCS["test_dataset"]}, {"content": _FAKE_DOCS["test_script"]},
                {"response": "Implemented. Ran `pytest`: 3 passed."},
                # Validator's OWN independent benchmark/dataset/script (2026-08-06) --
                # deliberately different content from the Planner's, to also prove
                # nothing here accidentally reuses the Planner's copies.
                {"content": "# Validator's own Benchmark\nfake"},
                {"content": "N/A: validator judged no dataset needed either"},
                {"content": "assert True  # validator's own real, compilable script"},
            ]
            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", "test_dataset", "test_script"}
        )
        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]
        # brd, prd, sdd, benchmark, test_dataset, test_script, code_diff, test_report,
        # validator_benchmark, validator_test_dataset, validator_test_script
        self.assertEqual(all_artifacts, 11)

        # 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 = {
            row[0] for row in self.conn.execute(
                "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
            ).fetchall()
        }
        self.assertEqual(run_org_nodes, {"planner", "coder", "validator"})


class PrdConsistencyVerifyFnTest(unittest.TestCase):
    """REQ-02 boundary: only the PRD node gets the consistency check."""

    def test_non_prd_node_passes_through_without_a_call(self):
        node_artifact_types = {"brd": "brd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
            ok, feedback = verify_fn({"node_id": "brd"}, {"content": "brd text"})
        self.assertTrue(ok)
        self.assertIsNone(feedback)
        mock_execute.assert_not_called()

    def test_string_false_from_a_real_call_is_not_treated_as_truthy(self):
        # regression: a real call (2026-08-02) returned "consistent": "false"
        # as a JSON string, not a boolean -- `if "false":` is truthy in
        # Python, silently inverting the check until a real test caught it.
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": "false", "issues": ["contradiction found"]},
             ):
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
        self.assertFalse(ok)

    def test_string_true_from_a_real_call_is_treated_as_consistent(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": "true", "issues": []},
             ):
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "consistent prd"})
        self.assertTrue(ok)

    def test_prd_node_triggers_a_real_check_call(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": False, "issues": ["contradiction found"]},
             ) as mock_execute:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
        self.assertFalse(ok)
        self.assertEqual(feedback, ["contradiction found"])
        mock_execute.assert_called_once()
        args = mock_execute.call_args[0]
        self.assertEqual(args[0], "/tmp/proj-1")


class CheckScriptCompilesTest(unittest.TestCase):
    """2026-08-06: the one mechanical (not LLM-judged) verify_fn -- real,
    unmocked subprocess calls to py_compile/node --check, no fixtures to
    fake since this is testing an actual compiler's real opinion."""

    def test_na_content_passes_without_compiling(self):
        ok, detail = e2e_pipeline.check_script_compiles("N/A: this goal has no automatable check")
        self.assertTrue(ok)
        self.assertIsNone(detail)

    def test_valid_python_passes(self):
        ok, detail = e2e_pipeline.check_script_compiles("def test_thing():\n    assert 1 + 1 == 2\n")
        self.assertTrue(ok)

    def test_valid_javascript_passes(self):
        ok, detail = e2e_pipeline.check_script_compiles("function testThing() { return 1 + 1; }\n")
        self.assertTrue(ok)

    def test_narrative_prose_fails_both(self):
        # this is the exact failure mode that motivated this check -- a
        # narrative sentence about a script, not a real script.
        ok, detail = e2e_pipeline.check_script_compiles(
            "The graphlemon planner Work Node workflow completed successfully and produced "
            "a test script covering 18 scenarios across four areas."
        )
        self.assertFalse(ok)
        self.assertIn("py_compile", detail)
        self.assertIn("node --check", detail)

    def test_syntactically_broken_python_and_js_fails(self):
        ok, detail = e2e_pipeline.check_script_compiles("def broken(:\n    this is not valid anything\n")
        self.assertFalse(ok)


class TestScriptVerifyFnTest(unittest.TestCase):
    def test_non_test_script_node_passes_through_without_compiling(self):
        node_artifact_types = {"brd": "brd"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "brd"}, {"content": "not even close to code {{{"})
        self.assertTrue(ok)
        self.assertIsNone(detail)

    def test_test_script_node_with_broken_content_fails(self):
        node_artifact_types = {"script": "test_script"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "script"}, {"content": "this is narrative, not code, no way"})
        self.assertFalse(ok)

    def test_test_script_node_with_real_code_passes(self):
        node_artifact_types = {"script": "test_script"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "script"}, {"content": "assert True\n"})
        self.assertTrue(ok)


class CombinedPlannerVerifyFnTest(unittest.TestCase):
    """Dispatches to the right per-node-type check -- both real checks
    (PRD consistency and script-compiles) are exercised elsewhere in
    isolation; this only tests the dispatch itself."""

    def test_prd_node_dispatches_to_prd_check(self):
        node_artifact_types = {"n1": "prd"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            ok, detail = verify_fn({"node_id": "n1"}, {"content": "a prd"})
        self.assertTrue(ok)
        mock_execute.assert_called_once()

    def test_test_script_node_dispatches_to_compile_check(self):
        node_artifact_types = {"n1": "test_script"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        ok, detail = verify_fn({"node_id": "n1"}, {"content": "not code at all, just words"})
        self.assertFalse(ok)

    def test_other_node_types_pass_through(self):
        node_artifact_types = {"n1": "test_dataset"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        ok, detail = verify_fn({"node_id": "n1"}, {"content": "anything at all"})
        self.assertTrue(ok)
        self.assertIsNone(detail)


class PrdConsistencySelfTargetWorktreeTest(unittest.TestCase):
    """2026-08-03 regression: a real self-hosting run showed _check_prd_
    consistency() calling execute_work_graph() directly against
    target_folder_path, bypassing worktree isolation entirely -- when
    self-targeting, its own agentType='planner'+memory:project call writes
    a real, uncommitted change straight to the main checkout's
    .claude/agent-memory/planner/MEMORY.md, which then made the NEXT
    node's own (correctly worktree-isolated) merge fail with "local
    changes would be overwritten by merge". This call must isolate the
    same way work_node_run.run_work_node() already does for every other
    self-target call."""

    def test_self_target_isolates_the_check_in_its_own_worktree(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/repo")
        with patch.object(e2e_pipeline.paths, "is_self_target", return_value=True), \
             patch.object(e2e_pipeline.work_node_run, "create_worktree", return_value="/repo/.graphlemon-worktrees/x") as mock_create, \
             patch.object(e2e_pipeline.work_node_run, "ensure_worktree_committed") as mock_ensure, \
             patch.object(e2e_pipeline.work_node_run, "merge_worktree") as mock_merge, \
             patch.object(e2e_pipeline.work_node_run, "remove_worktree") as mock_remove, \
             patch.object(work_graph_execute, "write_workflow_script") as mock_write, \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "some prd"})

        self.assertTrue(ok)
        mock_create.assert_called_once_with("/repo", unittest.mock.ANY)
        # both the script write and the real execute call must use the
        # WORKTREE path as cwd, never the main checkout, when self-targeting
        self.assertEqual(mock_write.call_args[0][0], "/repo/.graphlemon-worktrees/x")
        self.assertEqual(mock_execute.call_args[0][0], "/repo/.graphlemon-worktrees/x")
        # called twice: once for the worktree itself, once for the main
        # checkout (worktrees share project identity, so a real memory
        # write can land on the main checkout regardless of cwd -- see
        # work_node_run.ensure_worktree_committed's own docstring)
        self.assertEqual(mock_ensure.call_count, 2)
        ensure_paths = [call.args[0] for call in mock_ensure.call_args_list]
        self.assertIn("/repo/.graphlemon-worktrees/x", ensure_paths)
        self.assertIn("/repo", ensure_paths)
        mock_merge.assert_called_once()
        mock_remove.assert_called_once()

    def test_non_self_target_does_not_use_a_worktree_at_all(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(e2e_pipeline.paths, "is_self_target", return_value=False), \
             patch.object(e2e_pipeline.work_node_run, "create_worktree") as mock_create, \
             patch.object(work_graph_execute, "write_workflow_script") as mock_write, \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            verify_fn({"node_id": "prd"}, {"content": "some prd"})

        mock_create.assert_not_called()
        self.assertEqual(mock_write.call_args[0][0], "/tmp/proj-1")
        self.assertEqual(mock_execute.call_args[0][0], "/tmp/proj-1")


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")


class RunTestScriptForRealTest(unittest.TestCase):
    """2026-08-06: real, unmocked execution -- these are testing what an
    actual interpreter actually does, same reasoning as CheckScriptCompilesTest."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)

    def test_na_content_is_not_executed(self):
        result = e2e_pipeline.run_test_script_for_real(
            "N/A: no automatable check for this goal", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertFalse(result["ran"])
        self.assertIsNone(result["exit_code"])

    def test_real_passing_python_script(self):
        result = e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(0)\n", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertTrue(result["ran"])
        self.assertEqual(result["exit_code"], 0)

    def test_real_failing_python_script(self):
        result = e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(1)\n", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertTrue(result["ran"])
        self.assertEqual(result["exit_code"], 1)

    def test_script_runs_with_real_cwd_so_it_sees_the_real_repo(self):
        with open(os.path.join(self.tmp.name, "marker.txt"), "w") as f:
            f.write("real file")
        script = (
            "import os, sys\n"
            "sys.exit(0 if os.path.exists('marker.txt') else 1)\n"
        )
        result = e2e_pipeline.run_test_script_for_real(script, self.tmp.name, "task-1", "validator_test_script")
        self.assertEqual(result["exit_code"], 0)

    def test_materializes_a_real_file_under_graphlemon_runs(self):
        e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(0)\n", self.tmp.name, "task-42", "validator_test_script"
        )
        expected = os.path.join(self.tmp.name, ".graphlemon", "runs", "task-42", "validator_test_script.py")
        self.assertTrue(os.path.exists(expected))


class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
    """2026-08-06: the safety-critical piece -- a real non-zero exit from
    Validator's own script must force FAIL no matter what the LLM's verdict
    line says (user-confirmed hard gate)."""

    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)
        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', 'validating', 'validator')"
        )
        self.conn.commit()

    def tearDown(self):
        self.conn.close()

    def _run(self, script_content, verdict_line):
        with 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_line):
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": script_content},
            ]
            return e2e_pipeline.validate_implementation_with_own_graph(
                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
            )

    def test_llm_says_pass_and_script_really_passes_stays_pass(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
        self.assertEqual(result["verdict"], "PASS")
        self.assertTrue(result["script_execution"]["ran"])
        self.assertEqual(result["script_execution"]["exit_code"], 0)

    def test_llm_says_pass_but_script_really_fails_is_force_overridden_to_fail(self):
        # the exact case the user's hard-gate rule exists for: a script
        # that objectively fails must not be argued away by the LLM's own
        # verdict line.
        result = self._run("import sys\nsys.exit(1)\n", "VERDICT: PASS\nlooks good to me")
        self.assertEqual(result["verdict"], "FAIL")

    def test_llm_says_fail_and_script_fails_stays_fail(self):
        result = self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")
        self.assertEqual(result["verdict"], "FAIL")

    def test_na_script_leaves_llm_judgment_untouched(self):
        result = self._run("N/A: no mechanical check possible", "VERDICT: INCONCLUSIVE\nneeds human review")
        self.assertEqual(result["verdict"], "INCONCLUSIVE")
        self.assertFalse(result["script_execution"]["ran"])

    def test_writes_three_validator_artifacts_with_lineage(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
        self.assertEqual(
            set(result["artifact_ids"]),
            {"validator_benchmark", "validator_test_dataset", "validator_test_script"},
        )
        script_deps = json.loads(
            self.conn.execute(
                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
                (result["artifact_ids"]["validator_test_script"],),
            ).fetchone()[0]
        )
        self.assertIn(result["artifact_ids"]["validator_benchmark"], script_deps)
        self.assertIn(result["artifact_ids"]["validator_test_dataset"], script_deps)

    def test_creates_its_own_work_graph_run_row(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
        run_row = self.conn.execute(
            "SELECT org_node FROM work_graph_run WHERE task_id = 'task-1'"
        ).fetchone()
        self.assertEqual(run_row[0], "validator")


if __name__ == "__main__":
    unittest.main()