"""U-025 tests: c_task_list / c_task_get (03-C §3.2), real sqlite (no mocking needed). Run: python3 -m unittest src.test_task_query -v """ import json import unittest from src import schema, task_query class TaskQueryTestBase(unittest.TestCase): def setUp(self): self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) self.conn.commit() def tearDown(self): self.conn.close() def _insert_task(self, task_id, status="queued", assigned_org_node="planner", goal_text="goal"): self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES (?, 'proj-1', ?, 'goal', ?, ?)", (task_id, goal_text, status, assigned_org_node), ) self.conn.commit() class ListTasksTest(TaskQueryTestBase): def test_lists_tasks_for_project(self): self._insert_task("task-1") self._insert_task("task-2") result = task_query.list_tasks(self.conn, "proj-1") self.assertEqual({t["task_id"] for t in result}, {"task-1", "task-2"}) def test_other_projects_excluded(self): self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-2', 'P2', '/tmp/proj-2', 1, 'active')" ) self.conn.commit() self._insert_task("task-1") self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-2', 'proj-2', 'g', 'goal', 'queued', 'planner')" ) self.conn.commit() result = task_query.list_tasks(self.conn, "proj-1") self.assertEqual([t["task_id"] for t in result], ["task-1"]) def test_status_filter(self): self._insert_task("task-1", status="queued") self._insert_task("task-2", status="done") result = task_query.list_tasks(self.conn, "proj-1", status="done") self.assertEqual([t["task_id"] for t in result], ["task-2"]) def test_org_node_filter(self): self._insert_task("task-1", assigned_org_node="coder") self._insert_task("task-2", assigned_org_node="validator") result = task_query.list_tasks(self.conn, "proj-1", org_node="coder") self.assertEqual([t["task_id"] for t in result], ["task-1"]) def test_keyword_filter_matches_goal_text_substring(self): self._insert_task("task-1", goal_text="implement snake game") self._insert_task("task-2", goal_text="fix login bug") result = task_query.list_tasks(self.conn, "proj-1", keyword="snake") self.assertEqual([t["task_id"] for t in result], ["task-1"]) def test_empty_project_returns_empty_list(self): self.assertEqual(task_query.list_tasks(self.conn, "proj-1"), []) class GetTaskTest(TaskQueryTestBase): def test_nonexistent_task_raises(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.get_task(self.conn, "no-such-task") def test_returns_task_fields(self): self._insert_task("task-1", status="coding", assigned_org_node="coder") result = task_query.get_task(self.conn, "task-1") self.assertEqual(result["task_id"], "task-1") self.assertEqual(result["status"], "coding") self.assertEqual(result["assigned_org_node"], "coder") self.assertEqual(result["artifacts"], []) self.assertEqual(result["bug_records"], []) self.assertEqual(result["work_graph_runs"], []) self.assertEqual(result["planner_review_events"], []) def test_blocked_detail_absent_when_not_blocked(self): self._insert_task("task-1", status="coding") result = task_query.get_task(self.conn, "task-1") self.assertNotIn("blocked_detail", result) def test_blocked_detail_none_when_no_detail_recorded(self): self._insert_task("task-1", status="blocked_pending_human") result = task_query.get_task(self.conn, "task-1") self.assertIsNone(result["blocked_detail"]) def test_blocked_detail_from_most_recent_blocking_event(self): # 2026-08-13 (real incident): task.blocked_reason is only ever a # coarse category ('execution_error') -- the actual cause lives in # the matching task_status_changed event's own detail text. self._insert_task("task-1", status="blocked_pending_human") self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', " "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\", " "\"detail\": \"first real reason\"}')" ) self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES ('evt-2', 'proj-1', 'task-1', 'task_status_changed', " "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\", " "\"detail\": \"second, more recent reason\"}')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(result["blocked_detail"], "second, more recent reason") def test_includes_artifacts(self): self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES ('run-1', 'task-1', '{}', 'done')" ) self.conn.execute( "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, " "git_commit_sha) VALUES ('art-1', 'brd', 'task-1', 'run-1', 'docs/brd.md', 'sha1')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(len(result["artifacts"]), 1) self.assertEqual(result["artifacts"][0]["artifact_id"], "art-1") def test_includes_bug_records(self): self._insert_task("task-1") self.conn.execute( "INSERT INTO bug_record (bug_id, task_id, source, description, status) " "VALUES ('bug-1', 'task-1', 'human', 'x', 'open')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(len(result["bug_records"]), 1) self.assertEqual(result["bug_records"][0]["bug_id"], "bug-1") def test_includes_work_graph_runs_with_parsed_graph_manifest_and_nodes(self): self._insert_task("task-1") manifest = {"nodes": [{"node_id": "n1", "artifact_type": "brd", "depends_on": []}], "edges": []} self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', ?, 'running')", (json.dumps(manifest),), ) self.conn.execute( "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " "VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', 'new', 'active')" ) self.conn.execute( "INSERT 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, retry_count) " "VALUES ('node-1', 'run-1', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', " "'{\"x\": 1}', 'done', 0)" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(len(result["work_graph_runs"]), 1) run = result["work_graph_runs"][0] self.assertEqual(run["graph_manifest"], manifest) self.assertEqual(len(run["nodes"]), 1) self.assertEqual(run["nodes"][0]["node_id"], "n1") self.assertIsNone(run["parent_run_id"]) self.assertIsNone(run["fork_of_node_exec_id"]) def test_includes_loop_back_events_for_the_run(self): # 2026-08-08 (Work Graph self-planning + loop-back): v_graph_visualizer # needs this to draw an on_failure edge/retry badge -- not derivable # from graph_manifest or work_node_execution alone. self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', '{}', 'running')" ) self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) " "VALUES ('evt-1', 'proj-1', 'task-1', 'run-1', 'loop_back_triggered', ?)", (json.dumps({ "source_node_id": "benchmark", "target_node_id": "benchmark", "attempt_number": 1, "feedback": ["not enough detail"], }),), ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") run = result["work_graph_runs"][0] self.assertEqual(len(run["loop_back_events"]), 1) event = run["loop_back_events"][0] self.assertEqual(event["source_node_id"], "benchmark") self.assertEqual(event["target_node_id"], "benchmark") self.assertEqual(event["attempt_number"], 1) self.assertEqual(event["feedback"], ["not enough detail"]) def test_no_loop_back_events_is_an_empty_list_not_missing_key(self): self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES ('run-1', 'task-1', '{}', 'done')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(result["work_graph_runs"][0]["loop_back_events"], []) def test_forked_run_surfaces_parent_and_source_node(self): # D (2026-08-06): v_task_detail must be able to tell a fork branch # apart from the original run it descends from. self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-orig', 'task-1', '{}', 'done')" ) self.conn.execute( "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " "VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', 'new', 'active')" ) self.conn.execute( "INSERT 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 ('node-1', 'run-orig', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', '{}', 'done')" ) self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, " "parent_run_id, fork_of_node_exec_id) " "VALUES ('run-fork', 'task-1', '{}', 'done', 'run-orig', 'node-1')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") by_run_id = {r["run_id"]: r for r in result["work_graph_runs"]} self.assertIsNone(by_run_id["run-orig"]["parent_run_id"]) self.assertEqual(by_run_id["run-fork"]["parent_run_id"], "run-orig") self.assertEqual(by_run_id["run-fork"]["fork_of_node_exec_id"], "node-1") def test_work_node_includes_quota_wait_fields_for_v_graph_visualizer(self): # U-026: v_graph_visualizer's waiting_for_quota state needs # quota_limit_type + how long it's been waiting (02-V §2.7) self._insert_task("task-1") self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', '{}', 'running')" ) self.conn.execute( "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')" ) self.conn.execute( "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " "input_schema_ref, output_schema_ref, input_payload, status, quota_limit_type, " "quota_wait_started_at) " "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', " "'waiting_for_quota', 'session', '2026-07-31 09:00:00')" ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") node = result["work_graph_runs"][0]["nodes"][0] self.assertEqual(node["quota_limit_type"], "session") self.assertEqual(node["quota_wait_started_at"], "2026-07-31 09:00:00") def test_includes_planner_review_events_for_the_task(self): # 2026-08-13 (Org Graph unification Phase 3, FR-5a): planner_reviewer # has no work_graph_run of its own -- its verdicts live in # execution_event (event_type='planner_review_verdict'), see # e2e_pipeline.py::_record_planner_review_event. self._insert_task("task-1") self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES ('evt-1', 'proj-1', 'task-1', 'planner_review_verdict', ?)", (json.dumps({"attempt": 1, "verdict": "REJECTED", "used_existing_artifacts": False}),), ) self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES ('evt-2', 'proj-1', 'task-1', 'planner_review_verdict', ?)", (json.dumps({"attempt": 2, "verdict": "PASS", "used_existing_artifacts": True}),), ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") events = result["planner_review_events"] self.assertEqual(len(events), 2) self.assertEqual(events[0]["attempt"], 1) self.assertEqual(events[0]["verdict"], "REJECTED") self.assertEqual(events[0]["used_existing_artifacts"], False) self.assertEqual(events[1]["attempt"], 2) self.assertEqual(events[1]["verdict"], "PASS") self.assertEqual(events[1]["used_existing_artifacts"], True) def test_no_planner_review_events_is_an_empty_list_not_missing_key(self): self._insert_task("task-1") result = task_query.get_task(self.conn, "task-1") self.assertEqual(result["planner_review_events"], []) def test_other_task_planner_review_events_excluded(self): self._insert_task("task-1") self._insert_task("task-2") self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES ('evt-1', 'proj-1', 'task-2', 'planner_review_verdict', ?)", (json.dumps({"attempt": 1, "verdict": "PASS", "used_existing_artifacts": False}),), ) self.conn.commit() result = task_query.get_task(self.conn, "task-1") self.assertEqual(result["planner_review_events"], []) class ListTasksSortTest(TaskQueryTestBase): def setUp(self): super().setUp() self._insert_task("task-a", goal_text="banana") self.conn.execute( "UPDATE task SET created_at = '2026-01-01 00:00:00', updated_at = '2026-01-01 00:00:00' " "WHERE task_id = 'task-a'" ) self._insert_task("task-b", goal_text="apple") self.conn.execute( "UPDATE task SET created_at = '2026-01-02 00:00:00', updated_at = '2026-01-02 00:00:00' " "WHERE task_id = 'task-b'" ) self._insert_task("task-c", goal_text="cherry") self.conn.execute( "UPDATE task SET created_at = '2026-01-03 00:00:00', updated_at = '2026-01-03 00:00:00' " "WHERE task_id = 'task-c'" ) self.conn.commit() def test_default_sort_is_updated_at_desc_unchanged(self): # existing v_task_board behavior must not shift under it result = task_query.list_tasks(self.conn, "proj-1") self.assertEqual([t["task_id"] for t in result], ["task-c", "task-b", "task-a"]) def test_sort_by_created_at_asc(self): result = task_query.list_tasks(self.conn, "proj-1", sort="created_at", order="asc") self.assertEqual([t["task_id"] for t in result], ["task-a", "task-b", "task-c"]) def test_sort_by_goal_text_asc(self): result = task_query.list_tasks(self.conn, "proj-1", sort="goal_text", order="asc") self.assertEqual([t["task_id"] for t in result], ["task-b", "task-a", "task-c"]) # apple, banana, cherry def test_invalid_sort_column_rejected(self): with self.assertRaises(task_query.InvalidParamError): task_query.list_tasks(self.conn, "proj-1", sort="task_id; DROP TABLE task") class UpdateTaskGoalTextTest(TaskQueryTestBase): def test_queued_task_can_be_edited(self): self._insert_task("task-1", status="queued", goal_text="old") result = task_query.update_task_goal_text(self.conn, "task-1", "new goal text") self.assertEqual(result["goal_text"], "new goal text") def test_nonexistent_task_raises_not_found(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.update_task_goal_text(self.conn, "no-such-task", "x") def test_in_progress_task_cannot_be_edited(self): self._insert_task("task-1", status="coding") with self.assertRaises(task_query.TaskNotEditableError): task_query.update_task_goal_text(self.conn, "task-1", "new text") class UpdateTaskExecutionPriorityTest(TaskQueryTestBase): """2026-08-13 (real incident retrospective): unlike goal_text, this is NOT gated to status='queued' -- it's a scheduling/display hint, not a planning input.""" def test_in_progress_task_can_still_be_edited(self): self._insert_task("task-1", status="coding") result = task_query.update_task_execution_priority(self.conn, "task-1", 2.1) self.assertEqual(result["execution_priority"], 2.1) def test_serial_integer_value(self): self._insert_task("task-1", status="queued") result = task_query.update_task_execution_priority(self.conn, "task-1", 1) self.assertEqual(result["execution_priority"], 1) def test_none_clears_it(self): self._insert_task("task-1", status="queued") task_query.update_task_execution_priority(self.conn, "task-1", 3) result = task_query.update_task_execution_priority(self.conn, "task-1", None) self.assertIsNone(result["execution_priority"]) def test_nonexistent_task_raises_not_found(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.update_task_execution_priority(self.conn, "no-such-task", 1) def test_non_numeric_value_raises_invalid_param(self): self._insert_task("task-1", status="queued") with self.assertRaises(task_query.InvalidParamError): task_query.update_task_execution_priority(self.conn, "task-1", "not a number") class DeleteTaskTest(TaskQueryTestBase): def test_queued_task_can_be_deleted(self): self._insert_task("task-1", status="queued") task_query.delete_task(self.conn, "task-1") self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = 'task-1'").fetchone()[0], 0) def test_nonexistent_task_raises_not_found(self): with self.assertRaises(task_query.TaskNotFoundError): task_query.delete_task(self.conn, "no-such-task") def test_done_task_cannot_be_deleted(self): self._insert_task("task-1", status="done") with self.assertRaises(task_query.TaskNotEditableError): task_query.delete_task(self.conn, "task-1") self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = 'task-1'").fetchone()[0], 1) def test_real_task_create_task_can_still_be_deleted(self): # 2026-08-03 regression: _insert_task (this file's own raw-INSERT # helper) never writes an execution_event row, so it never caught # the real bug -- task_create.create_task() always writes one # execution_event in the same transaction as the task itself (01-M's # own FK, no ON DELETE CASCADE), so even a never-touched queued task # already has one referencing it, and a real DELETE FROM task raised # sqlite3.IntegrityError until delete_task() also cleared it. from src import task_create result = task_create.create_task(self.conn, "proj-1", "a real goal") self.assertEqual( self.conn.execute( "SELECT COUNT(*) FROM execution_event WHERE task_id = ?", (result["task_id"],) ).fetchone()[0], 1, ) task_query.delete_task(self.conn, result["task_id"]) self.assertEqual( self.conn.execute("SELECT COUNT(*) FROM task WHERE task_id = ?", (result["task_id"],)).fetchone()[0], 0 ) self.assertEqual( self.conn.execute( "SELECT COUNT(*) FROM execution_event WHERE task_id = ?", (result["task_id"],) ).fetchone()[0], 0, ) if __name__ == "__main__": unittest.main()