"""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 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 def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300): assert prompt == e2e_pipeline._READY_PROMPT return "Ready." 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) self.conn = schema.connect(":memory:") 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 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), ) 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 run Coder then Validator, not redo Planner.""" def test_runs_coder_then_validator(self): self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner") # trigger_org_node_full("coder", ...) requires ALL FOUR doc artifacts # validation_status='passed' (org_node_lifecycle._REQUIRED_ARTIFACTS), # not just the two this module's own resume logic reads content from. 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_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 = [ {"response": "done, tests pass"}, # coder's implement 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"]) mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args[0][0], "validator") 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_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): 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") 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()