test_task_query.py
19.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
"""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_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()