宜博 / lemongraph

"""c_task_execute / c_task_pause tests (2026-08-03): mocked at the
subprocess boundary (CLAUDE.md Testing constraint), same pattern as
test_e2e_pipeline.py -- this file tests task_execution.py's own resume/
pause wiring, not e2e_pipeline's phase mechanics again.

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

import json
import os
import subprocess
import tempfile
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, work_graph_plan,
)


def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
    assert prompt == e2e_pipeline._READY_PROMPT
    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)
        # 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"]

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

    def _insert_task(self, task_id, status, assigned_org_node="planner", loop_round=0):
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, loop_round) "
            "VALUES (?, ?, 'a goal', 'goal', ?, ?, ?)",
            (task_id, self.project_id, status, assigned_org_node, loop_round),
        )
        self.conn.commit()

    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, manifest),
        )
        rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
        abs_path = os.path.join(self.tmp.name, rel_path)
        os.makedirs(os.path.dirname(abs_path), exist_ok=True)
        with open(abs_path, "w") as f:
            f.write(content)
        subprocess.run(["git", "add", "-A"], cwd=self.tmp.name, check=True, capture_output=True)
        subprocess.run(
            ["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-m", "x"],
            cwd=self.tmp.name, check=True, capture_output=True,
        )
        sha = subprocess.run(
            ["git", "rev-parse", "HEAD"], cwd=self.tmp.name, capture_output=True, text=True
        ).stdout.strip()
        self.conn.execute(
            "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
            "git_commit_sha, validation_status) VALUES (?, ?, ?, ?, ?, ?, 'passed')",
            ("art-" + artifact_type, artifact_type, task_id, run_id, rel_path, sha),
        )
        self.conn.commit()


class ResumeFromCodingTest(TaskExecutionTestBase):
    """Coder already ran (real code_diff/test_report artifacts exist) --
    resuming from 'coding' must skip straight to Validator, not redo
    Planner or Coder."""

    def test_skips_to_validator_only(self):
        self._insert_task("task-1", status="coding", assigned_org_node="coder")
        self._write_artifact("task-1", "prd", "# PRD content")
        self._write_artifact("task-1", "sdd", "# SDD content")
        self._write_artifact("task-1", "benchmark", "# Benchmark content")
        self._write_artifact("task-1", "code_diff", "diff content")
        self._write_artifact("task-1", "test_report", "report content")

        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
             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:
            # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06)
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff")
        # only the validator's final-verdict real call happened -- no coder call
        mock_trigger.assert_called_once()
        self.assertEqual(mock_trigger.call_args[0][0], "validator")
        row = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()
        self.assertEqual(row[0], "validating")


class ResumeFromPlanningTest(TaskExecutionTestBase):
    """Planner already ran (real doc artifacts exist) -- resuming from
    'planning_brd_prd_sdd' must trigger planner_reviewer's ready handshake,
    review (reusing the existing artifacts, not re-planning), then run
    Coder then Validator, not redo Planner."""

    def test_runs_reviewer_then_coder_then_validator(self):
        self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner")
        # sdd alone is what org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] now
        # requires (2026-08-08, Org Graph unification Phase 1 -- narrowed from
        # all four); brd/prd/benchmark are written too so the review step has
        # real content to reuse instead of falling through to plan_documents().
        self._write_artifact("task-1", "brd", "# BRD content")
        self._write_artifact("task-1", "prd", "# PRD content")
        self._write_artifact("task-1", "sdd", "# SDD content")
        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 = [
                {"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"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        self.assertIn("code_diff", result["code_artifact_ids"])
        # existing artifacts were reused -- plan_documents() must NOT have been
        # called again (that's the whole point of the existence-probing retry loop).
        mock_plan.assert_not_called()
        # two real org_node_trigger.trigger_org_node calls: planner_reviewer's
        # judgment, then validator's final verdict -- not just validator anymore.
        self.assertEqual(mock_trigger.call_count, 2)
        self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer")
        self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator")


class ResumeFromReviewingTest(TaskExecutionTestBase):
    """2026-08-08 (Org Graph unification Phase 1): resuming from
    'reviewing_planner_output' means planner_reviewer's own ready handshake
    already succeeded (that's how the task reached this status) -- must
    NOT re-trigger planner or planner_reviewer's ready handshake a second
    time, but MUST still run the real review judgment call, then Coder,
    then Validator."""

    def test_skips_reviewer_ready_handshake_runs_review_then_coder_then_validator(self):
        self._insert_task("task-1", status="reviewing_planner_output", assigned_org_node="planner_reviewer")
        self._write_artifact("task-1", "brd", "# BRD content")
        self._write_artifact("task-1", "prd", "# PRD content")
        self._write_artifact("task-1", "sdd", "# SDD content")
        self._write_artifact("task-1", "benchmark", "# Benchmark content")

        ready_calls = []

        def tracking_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
            ready_calls.append(org_node)
            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 = [
                {"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"},
            ]
            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        mock_plan.assert_not_called()
        # planner_reviewer's own ready handshake must NOT fire again -- only
        # coder's and validator's (its judgment already happened once to get
        # this task into 'reviewing_planner_output' in the first place).
        self.assertEqual(ready_calls, ["coder", "validator"])
        # but the real review judgment call DOES still run (that's the actual
        # gap this status resumes into -- see the module docstring).
        self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer")
        self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator")


class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase):
    """P0 bug regression (task-execution-resume-assumes-planner-artifacts-
    exist.md): a task can reach 'planning_brd_prd_sdd' with ZERO doc
    artifacts ever persisted (the Planner's Work Graph failed on its very
    first node). Resuming from here must NOT raise -- it must fall through
    to a real plan_documents() call, not assume something already exists."""

    def test_resume_with_no_artifacts_falls_through_to_replanning(self):
        self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner")
        # deliberately: no _write_artifact() calls at all.

        def fake_plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
            # simulates what a real plan_documents() call would actually do --
            # 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-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 = [
                {"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"},
            ]
            # must not raise (this is the actual regression: it used to KeyError/
            # TaskExecutionError before ever reaching here)
            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

        mock_plan.assert_called_once()
        self.assertEqual(result["verdict"]["verdict"], "PASS")


class ResumeFromValidatingTest(TaskExecutionTestBase):
    """2026-08-06 real incident: a task stuck in 'validating' for 15+ hours
    with no validator work_graph_run ever created, because nothing in
    production polls scheduler.run_scheduler_once() and whatever one-off
    call drove it that far had already exited. Resuming from 'validating'
    must skip BOTH Planner and Coder (already done) AND the
    trigger_org_node_full("validator", ...) readiness call (already ran
    once -- that's literally how status became 'validating'), going
    straight to validate_implementation_with_own_graph()."""

    def test_skips_readiness_trigger_runs_validation_only(self):
        self._insert_task("task-1", status="validating", assigned_org_node="validator")
        self._write_artifact("task-1", "prd", "# PRD content")
        self._write_artifact("task-1", "sdd", "# SDD content")
        self._write_artifact("task-1", "benchmark", "# Benchmark content")
        self._write_artifact("task-1", "code_diff", "diff content")
        self._write_artifact("task-1", "test_report", "report content")

        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger) as mock_ready, \
             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:
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff")
        # the readiness trigger (trigger_org_node_full's own PRE_STATE-guarded
        # call) must NOT fire a second time -- it already succeeded once to
        # get the task into 'validating' in the first place, and calling it
        # again would hit org_node_lifecycle's own "needs status=coding" guard.
        mock_ready.assert_not_called()
        # only the real work_graph_run's own final call happened
        mock_trigger.assert_called_once()
        self.assertEqual(mock_trigger.call_args[0][0], "validator")
        # a brand-new validator work_graph_run was created (none existed before)
        rows = self.conn.execute(
            "SELECT run_id FROM work_graph_run WHERE task_id = 'task-1' AND org_node = 'validator'"
        ).fetchall()
        self.assertEqual(len(rows), 1)


class ResumeUnresumableStatusTest(TaskExecutionTestBase):
    def test_done_task_raises(self):
        self._insert_task("task-1", status="done")
        with self.assertRaises(task_execution.TaskExecutionError):
            task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")

    def test_blocked_task_raises(self):
        self._insert_task("task-1", status="blocked_pending_human")
        with self.assertRaises(task_execution.TaskExecutionError):
            task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")


class PauseMidResumeTest(TaskExecutionTestBase):
    """If task.status flips to blocked_pending_human WHILE this thread is
    between phases (checked right before each org_node trigger),
    ExecutionPausedError must stop cleanly without triggering the next
    phase."""

    def test_pause_before_validator_stops_cleanly(self):
        self._insert_task("task-1", status="coding", assigned_org_node="coder")
        self._write_artifact("task-1", "prd", "# PRD content")
        self._write_artifact("task-1", "sdd", "# SDD content")
        self._write_artifact("task-1", "benchmark", "# Benchmark content")
        self._write_artifact("task-1", "code_diff", "diff content")
        self._write_artifact("task-1", "test_report", "report content")

        real_lookup = task_execution._latest_artifact_ids

        def lookup_then_pause(conn, task_id, artifact_types):
            # simulates a concurrent request_pause() landing right after
            # this (real, unmocked) lookup runs -- the last step before
            # _check_not_paused() is reached in the real code path.
            result = real_lookup(conn, task_id, artifact_types)
            conn.execute(
                "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'user_paused' "
                "WHERE task_id = 'task-1'"
            )
            conn.commit()
            return result

        with patch.object(task_execution, "_latest_artifact_ids", side_effect=lookup_then_pause), \
             patch.object(org_node_trigger, "trigger_org_node") as mock_trigger:
            with self.assertRaises(task_execution.ExecutionPausedError):
                task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
        mock_trigger.assert_not_called()


class CheckNotPausedTest(TaskExecutionTestBase):
    def test_raises_when_blocked(self):
        self._insert_task("task-1", status="blocked_pending_human")
        with self.assertRaises(task_execution.ExecutionPausedError):
            task_execution._check_not_paused(self.conn, "task-1")

    def test_does_not_raise_when_not_blocked(self):
        self._insert_task("task-1", status="coding")
        task_execution._check_not_paused(self.conn, "task-1")  # no raise


class RequestPauseTest(TaskExecutionTestBase):
    def test_pauses_a_coding_task(self):
        self._insert_task("task-1", status="coding")
        result = task_execution.request_pause(self.conn, "task-1", "user_paused")
        self.assertEqual(result["status"], "blocked_pending_human")
        row = self.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
        self.assertEqual(row, ("blocked_pending_human", "user_paused"))

    def test_stops_a_planning_task(self):
        self._insert_task("task-1", status="planning_brd_prd_sdd")
        result = task_execution.request_pause(self.conn, "task-1", "user_stopped")
        self.assertEqual(result["blocked_reason"], "user_stopped")

    def test_pauses_a_reviewing_task(self):
        # 2026-08-08 (Org Graph unification Phase 1): 'reviewing_planner_output'
        # stays task.status for the whole duration of the (up to 3-attempt)
        # review retry loop, same "in-flight under one status name" shape as
        # 'coding'/'validating' -- must be pausable too.
        self._insert_task("task-1", status="reviewing_planner_output")
        result = task_execution.request_pause(self.conn, "task-1", "user_paused")
        self.assertEqual(result["status"], "blocked_pending_human")

    def test_pauses_a_validating_task(self):
        # 2026-08-06: 'validating' stays task.status for the whole duration
        # of a resumed validation run (validate_implementation_with_own_graph
        # never updates task.status itself), so it must be pausable too.
        self._insert_task("task-1", status="validating")
        result = task_execution.request_pause(self.conn, "task-1", "user_paused")
        self.assertEqual(result["status"], "blocked_pending_human")

    def test_cannot_pause_a_queued_task(self):
        self._insert_task("task-1", status="queued")
        with self.assertRaises(task_execution.TaskExecutionError):
            task_execution.request_pause(self.conn, "task-1", "user_paused")

    def test_cannot_pause_a_done_task(self):
        self._insert_task("task-1", status="done")
        with self.assertRaises(task_execution.TaskExecutionError):
            task_execution.request_pause(self.conn, "task-1", "user_paused")

    def test_nonexistent_task_raises(self):
        with self.assertRaises(task_execution.TaskExecutionError):
            task_execution.request_pause(self.conn, "no-such-task", "user_paused")


class StartTaskExecutionInBackgroundTest(unittest.TestCase):
    """Real threading (own connection per thread, same as scheduler.py) --
    needs a real file db_path, not ':memory:' (a fresh ':memory:' connection
    per thread would be a completely separate, empty database)."""

    def setUp(self):
        # Defense-in-depth against the 2026-08-06 pollution incident (see
        # test_second_call_while_in_flight_raises_no_second_thread's own
        # comment): `_in_flight_task_ids` is a process-wide global, and
        # "task-1" is reused across many unrelated tests/files -- if any
        # test's mocking is incomplete enough that a spawned thread outlives
        # its own join(timeout=...), the claim leaks and silently fails
        # OTHER tests depending on run order. Unconditionally clearing here
        # makes that failure mode impossible regardless of what any
        # individual test does right or wrong.
        task_execution._release("task-1")
        self.addCleanup(lambda: task_execution._release("task-1"))
        self.repo = tempfile.TemporaryDirectory()
        self.addCleanup(self.repo.cleanup)
        subprocess.run(["git", "init"], cwd=self.repo.name, check=True, capture_output=True)
        self.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
        self.tmp_db.close()
        self.db_path = self.tmp_db.name
        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))

        conn = schema.connect(self.db_path)
        schema.create_schema(conn)
        project = project_scaffold.create_project(conn, self.repo.name, init_git_if_missing=False)
        self.project_id = project["project_id"]
        conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', ?, 'a goal', 'goal', 'coding', 'coder')", (self.project_id,)
        )
        conn.commit()
        conn.close()

    def _write_artifact(self, task_id, artifact_type, content):
        conn = schema.connect(self.db_path)
        run_id = "run-" + artifact_type
        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),
        )
        rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
        abs_path = os.path.join(self.repo.name, rel_path)
        os.makedirs(os.path.dirname(abs_path), exist_ok=True)
        with open(abs_path, "w") as f:
            f.write(content)
        subprocess.run(["git", "add", "-A"], cwd=self.repo.name, check=True, capture_output=True)
        subprocess.run(
            ["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-m", "x"],
            cwd=self.repo.name, check=True, capture_output=True,
        )
        sha = subprocess.run(
            ["git", "rev-parse", "HEAD"], cwd=self.repo.name, capture_output=True, text=True
        ).stdout.strip()
        conn.execute(
            "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
            "git_commit_sha, validation_status) VALUES (?, ?, ?, ?, ?, ?, 'passed')",
            ("art-" + artifact_type, artifact_type, task_id, run_id, rel_path, sha),
        )
        conn.commit()
        conn.close()

    def test_real_thread_completes_and_closes_task(self):
        self._write_artifact("task-1", "prd", "# PRD")
        self._write_artifact("task-1", "sdd", "# SDD")
        self._write_artifact("task-1", "benchmark", "# Benchmark")
        self._write_artifact("task-1", "code_diff", "diff")
        self._write_artifact("task-1", "test_report", "report")

        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
             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"):
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            thread = task_execution.start_task_execution_in_background(
                self.db_path, "task-1", self.repo.name, "a goal",
            )
            thread.join(timeout=10)

        conn = schema.connect(self.db_path)
        status = conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
        conn.close()
        self.assertEqual(status, "done")
        # 2026-08-06 (task_reconciliation): the in-flight guard must release
        # once the thread is done, so a LATER dispatch of the same task_id
        # (e.g. a future reconciliation pass) isn't permanently blocked by a
        # completed run.
        self.assertFalse(task_execution.is_in_flight("task-1"))

    def test_second_call_while_in_flight_raises_no_second_thread(self):
        # 2026-08-06 (task_reconciliation): guards against the console's
        # Start button and a reconciliation pass racing to dispatch the SAME
        # task_id concurrently -- the second call must fail fast, not spawn
        # a second thread that would run the same task twice.
        #
        # Fully mocks the same real work the OTHER StartTaskExecutionIn
        # BackgroundTest tests mock (write_workflow_script/execute_work_graph/
        # org_node_trigger.trigger_org_node), not just the readiness trigger
        # this test blocks on -- an earlier version of this test only
        # patched the readiness trigger, so once unblocked the thread went on
        # to attempt REAL subprocess/file work, outlived this test's own
        # join(timeout=10), and left "task-1" permanently claimed in the
        # process-wide `_in_flight_task_ids` set -- silently failing every
        # OTHER test in the suite that reuses the literal task_id "task-1"
        # (nearly all of them), depending on run order. Caught by running
        # the full discover -s src suite, not by running this file alone.
        self._write_artifact("task-1", "prd", "# PRD")
        self._write_artifact("task-1", "sdd", "# SDD")
        self._write_artifact("task-1", "benchmark", "# Benchmark")
        self._write_artifact("task-1", "code_diff", "diff")
        self._write_artifact("task-1", "test_report", "report")

        release_first_call = threading.Event()
        entered_first_call = threading.Event()

        def blocking_trigger(org_node, prompt, target_folder_path, timeout=300):
            entered_first_call.set()
            release_first_call.wait(timeout=10)
            return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout)

        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=blocking_trigger), \
             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"):
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            thread = task_execution.start_task_execution_in_background(
                self.db_path, "task-1", self.repo.name, "a goal",
            )
            entered_first_call.wait(timeout=10)
            self.assertTrue(task_execution.is_in_flight("task-1"))
            with self.assertRaises(task_execution.TaskExecutionError):
                task_execution.start_task_execution_in_background(
                    self.db_path, "task-1", self.repo.name, "a goal",
                )
            release_first_call.set()
            thread.join(timeout=10)

        self.assertFalse(task_execution.is_in_flight("task-1"))

    def test_generic_failure_marks_blocked(self):
        self._write_artifact("task-1", "prd", "# PRD")
        self._write_artifact("task-1", "sdd", "# SDD")
        self._write_artifact("task-1", "benchmark", "# Benchmark")
        self._write_artifact("task-1", "code_diff", "diff")
        self._write_artifact("task-1", "test_report", "report")

        with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
             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", side_effect=RuntimeError("boom")):
            # validator's own benchmark/dataset/script graph completes fine --
            # the failure this test cares about is the FINAL verdict call
            # (org_node_trigger.trigger_org_node), which happens after.
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            thread = task_execution.start_task_execution_in_background(
                self.db_path, "task-1", self.repo.name, "a goal",
            )
            thread.join(timeout=10)

        conn = schema.connect(self.db_path)
        row = conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
        payload = conn.execute(
            "SELECT payload FROM execution_event WHERE task_id = 'task-1' ORDER BY rowid DESC LIMIT 1"
        ).fetchone()[0]
        conn.close()
        self.assertEqual(row, ("blocked_pending_human", "execution_error"))
        # 2026-08-03: 'execution_error' alone gives zero diagnostic signal --
        # the real exception message must still be recoverable somewhere.
        self.assertEqual(json.loads(payload)["detail"], "boom")


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