"""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_orchestrate, 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": [], } # Phase 2c (2026-08-12): Validator now self-plans too (validate_ # implementation_with_own_graph() calls plan_work_graph(conn, "validator", # ..., append_to_run_id=...), through the SAME mocked subprocess boundary # as Planner's/Coder's own planning calls) -- a single node_type= # 'final_verdict' node is the minimal real shape this mechanism accepts # (see work_graph_plan._VALIDATOR_NODE_CATALOG's own catalog entry). # depends_on ["sdd"] for the same "avoid landing in the same in-degree-zero # topological layer as an already-'done' node" reason _FAKE_CODER_MANIFEST # above documents. _FAKE_VALIDATOR_MANIFEST = { "nodes": [{ "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict", "node_type": "final_verdict", "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, }], "edges": [], } _real_subprocess_run = subprocess.run # captured before any test patches subprocess.run globally def _fake_plan_subprocess_run_dispatch(*args, **kwargs): """Real plan_work_graph() dispatches by the --agent argv value -- this mock does the same, so Planner's/Coder's/Validator's planning calls (which all now share the same mocked subprocess.run patch target) each get their own realistic fake manifest instead of one hardcoded response serving all three callers. Same dispatch-by-argv + real-passthrough-for- non-'claude'-calls pattern as test_e2e_pipeline.py's helper of the same name -- see that module's docstring for the full rationale (this file's own git add/commit/diff calls inside the `with patch(...)` blocks below would otherwise also get intercepted, since `subprocess` is one shared module object).""" argv = args[0] if not argv or argv[0] != "claude": return _real_subprocess_run(*args, **kwargs) org_node = argv[argv.index("--agent") + 1] manifest = _FAKE_VALIDATOR_MANIFEST if org_node == "validator" else _FAKE_CODER_MANIFEST payload = {"is_error": False, "result": "```json\n" + json.dumps(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, graph_manifest, status) " "VALUES (?, ?, ?, 'done')", (run_id, task_id, manifest), ) # Phase 2c (2026-08-12, found while adding Validator self-planning # tests): a real Planner run_graph() call ALSO leaves a real # work_node_execution row behind for this node -- without one here # too, a later self-planning append onto this SAME run_id (Coder's/ # Validator's own plan_work_graph(..., append_to_run_id=run_id)) # finds no existing (run_id, node_id) row for `artifact_type` and # spuriously RE-EXECUTES it for real via the mocked execute_work_ # graph, burning an extra mock_execute.side_effect slot the calling # test never accounted for (masked before this fix only because # those tests happened to provide enough spare list items and never # asserted on the exact content that ended up threaded through). work_graph_orchestrate.ensure_placeholder_skill(self.conn, "planner-doc-writing", "planner") self.conn.execute( "INSERT OR IGNORE INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, " "skill_ref, input_schema_ref, output_schema_ref, input_payload, output_payload, status) " "VALUES (?, ?, ?, 'planner', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')", ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})), ) 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_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: # validator's self-planned final_verdict node (Phase 2c) mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] 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 self-planned node ran -- no coder call mock_execute.assert_called_once() 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_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") 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": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] 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() # one real org_node_trigger.trigger_org_node call: planner_reviewer's # judgment -- Validator's own verdict is now a self-planned node (Phase # 2c), read via mock_execute above, not a flat org_node_trigger call. mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer") 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_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") 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": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] 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). Validator's # own verdict is now a self-planned node (Phase 2c), read via # mock_execute above, not a second org_node_trigger call. mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer") 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_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") 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": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] # 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_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node 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 validator's self-planned node ran mock_execute.assert_called_once() # Phase 2c: Validator appends its final_verdict node into the SAME # shared run the sdd artifact points at (via _shared_run_id_for_task), # not a brand-new work_graph_run row of its own -- confirm the merge # landed on "run-sdd" and no extra row was created for this task. row = self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-sdd'").fetchone() node_ids = {n["node_id"] for n in json.loads(row[0])["nodes"]} self.assertIn("verdict", node_ids) total_rows = self.conn.execute( "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'" ).fetchone()[0] self.assertEqual(total_rows, 5) 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 # Real "nodes"/"edges" shape + a matching 'done' work_node_execution # row -- see TaskExecutionTestBase._write_artifact's own comment # (same file) for why both are needed once Validator self-plans and # appends onto this run via plan_work_graph(..., append_to_run_id=...). manifest = json.dumps({ "nodes": [{"node_id": artifact_type, "artifact_type": artifact_type, "depends_on": []}], "edges": [], }) conn.execute( "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES (?, ?, ?, 'done')", (run_id, task_id, manifest), ) work_graph_orchestrate.ensure_placeholder_skill(conn, "planner-doc-writing", "planner") conn.execute( "INSERT OR IGNORE INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, " "skill_ref, input_schema_ref, output_schema_ref, input_payload, output_payload, status) " "VALUES (?, ?, ?, 'planner', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')", ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})), ) 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_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node 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 (org_node_trigger.trigger_org_node_ # structured/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_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node 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_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=RuntimeError("boom")): # Phase 2c: Validator's own verdict is now a self-planned node, # not a flat org_node_trigger.trigger_org_node call -- the # failure this test cares about (a real exception surfacing as # blocked_pending_human/execution_error) is injected at its own # node's real execution instead. 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()