task_query.py
9.88 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
"""c_task_list / c_task_get (03-C §3.2): read-only task queries for the
console (v_task_board, v_task_detail). Unowned by any Unit until now --
built here because U-025 is the first Unit that actually needs them.
c_task_get's output goes beyond 03-C's literal "task + artifact 列表 + bug_record
列表" wording: v_task_detail (02-V §2.6) also needs the task's work_graph_run(s)
(for the to-do tree built from graph_manifest, 01-M 1.3's "补充说明") and each
run's work_node_execution rows (status per to-do node, expandable
input/output/reflection_notes) -- 03-C's own wording undersells what the view
it feeds actually requires, so this returns the fuller shape rather than
forcing the frontend to make N more round trips.
"""
import json
_SORT_COLUMNS = ("created_at", "updated_at", "goal_text", "status")
class InvalidParamError(Exception):
"""INVALID_PARAM -- e.g. an unrecognized `sort` column."""
def list_tasks(conn, project_id, status=None, org_node=None, keyword=None, sort=None, order=None):
"""`sort`/`order` (2026-08-03, v_goal_list): defaults preserve the
original `updated_at DESC` behavior every existing caller (v_task_board)
already depends on. `sort` is validated against a fixed column
whitelist and never interpolated as anything other than one of those
literal strings -- this is the one query in this module built as a
dynamic ORDER BY, so it's worth being explicit that user input never
reaches the SQL text itself, only a pre-approved column name."""
query = (
"SELECT task_id, goal_text, task_type, status, assigned_org_node, loop_round, "
"created_at, updated_at FROM task WHERE project_id = ?"
)
params = [project_id]
if status:
query += " AND status = ?"
params.append(status)
if org_node:
query += " AND assigned_org_node = ?"
params.append(org_node)
if keyword:
query += " AND goal_text LIKE ?"
params.append("%" + keyword + "%")
sort_column = sort or "updated_at"
if sort_column not in _SORT_COLUMNS:
raise InvalidParamError("sort must be one of {}, got {!r}".format(_SORT_COLUMNS, sort))
sort_direction = "ASC" if (order or "desc").lower() == "asc" else "DESC"
query += " ORDER BY {} {}".format(sort_column, sort_direction)
rows = conn.execute(query, params).fetchall()
return [
{
"task_id": r[0], "goal_text": r[1], "task_type": r[2], "status": r[3],
"assigned_org_node": r[4], "loop_round": r[5], "created_at": r[6], "updated_at": r[7],
}
for r in rows
]
class TaskNotFoundError(Exception):
"""NOT_FOUND -- 03-C §3.2: c_task_get on a nonexistent task_id."""
class TaskNotEditableError(Exception):
"""A goal's `goal_text` can only be edited or deleted while it's still
`queued` -- once Planner/Coder/Validator has started real work from it
(real artifacts, work_graph_runs, bug_records all reference task_id),
changing or removing the goal out from under that history would leave
those records pointing at a goal that no longer matches what was
actually built, which is exactly the kind of untraceable drift this
codebase otherwise goes out of its way to avoid."""
def update_task_goal_text(conn, task_id, goal_text):
row = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()
if row is None:
raise TaskNotFoundError(task_id)
if row[0] != "queued":
raise TaskNotEditableError(
"task {} is status={}, goal_text can only be edited while queued".format(task_id, row[0])
)
conn.execute(
"UPDATE task SET goal_text = ?, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?", (goal_text, task_id)
)
conn.commit()
return get_task(conn, task_id)
def delete_task(conn, task_id):
row = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()
if row is None:
raise TaskNotFoundError(task_id)
if row[0] != "queued":
raise TaskNotEditableError(
"task {} is status={}, can only be deleted while queued".format(task_id, row[0])
)
# a real DELETE against a real queued task raised sqlite3.IntegrityError
# (2026-08-03, found on first real verification pass) -- task_create.
# create_task() always writes an execution_event row in the same
# transaction as the task itself (01-M's own FK, no ON DELETE CASCADE),
# so even a task that has never been touched by Planner/Coder/Validator
# already has at least one execution_event referencing it. Those rows
# are this task's own creation-time bookkeeping, not independent audit
# content worth preserving once the task itself is gone -- a queued
# task can't yet have any artifact/bug_record/work_graph_run rows
# (those only start existing once real work begins), so this is the
# only FK that can actually be in the way here.
conn.execute("DELETE FROM execution_event WHERE task_id = ?", (task_id,))
conn.execute("DELETE FROM task WHERE task_id = ?", (task_id,))
conn.commit()
def get_task(conn, task_id):
row = conn.execute(
"SELECT task_id, project_id, goal_text, task_type, origin_ref, created_via, status, route, "
"loop_round, blocked_reason, assigned_org_node, created_at, updated_at FROM task WHERE task_id = ?",
(task_id,),
).fetchone()
if row is None:
raise TaskNotFoundError(task_id)
task = {
"task_id": row[0], "project_id": row[1], "goal_text": row[2], "task_type": row[3],
"origin_ref": row[4], "created_via": row[5], "status": row[6], "route": row[7],
"loop_round": row[8], "blocked_reason": row[9], "assigned_org_node": row[10],
"created_at": row[11], "updated_at": row[12],
}
artifacts = conn.execute(
"SELECT artifact_id, artifact_type, produced_by_run_id, file_path, git_commit_sha, "
"validation_status, created_at FROM artifact WHERE task_id = ? ORDER BY created_at",
(task_id,),
).fetchall()
task["artifacts"] = [
{
"artifact_id": a[0], "artifact_type": a[1], "produced_by_run_id": a[2], "file_path": a[3],
"git_commit_sha": a[4], "validation_status": a[5], "created_at": a[6],
}
for a in artifacts
]
bugs = conn.execute(
"SELECT bug_id, source, description, root_cause_category, suggested_route, status, repro_steps "
"FROM bug_record WHERE task_id = ?",
(task_id,),
).fetchall()
task["bug_records"] = [
{
"bug_id": b[0], "source": b[1], "description": b[2], "root_cause_category": b[3],
"suggested_route": b[4], "status": b[5], "repro_steps": b[6],
}
for b in bugs
]
runs = conn.execute(
"SELECT run_id, org_node, graph_manifest, status, started_at, completed_at, "
"parent_run_id, fork_of_node_exec_id "
"FROM work_graph_run WHERE task_id = ? ORDER BY started_at",
(task_id,),
).fetchall()
task["work_graph_runs"] = [
{
"run_id": w[0], "org_node": w[1],
"graph_manifest": json.loads(w[2]) if w[2] else None,
"status": w[3], "started_at": w[4], "completed_at": w[5],
# D (2026-08-06): both None for a normal run; both set when this
# run was forked from a historical node (src/work_graph_fork.py).
"parent_run_id": w[6], "fork_of_node_exec_id": w[7],
"nodes": _work_nodes_for_run(conn, w[0]),
# 2026-08-08 (Work Graph self-planning + loop-back): the console's
# v_graph_visualizer needs this to actually draw a loop_back's
# on_failure edge / a retry_self node's "sent back Nx" badge --
# neither is derivable from graph_manifest/work_node_execution
# alone (loop_back_triggered is execution_event-only by design,
# see docs/todo/work-graph-self-planning-and-loop-back-design.md §7).
"loop_back_events": _loop_back_events_for_run(conn, w[0]),
}
for w in runs
]
return task
def _loop_back_events_for_run(conn, run_id):
rows = conn.execute(
"SELECT payload, created_at FROM execution_event "
"WHERE run_id = ? AND event_type = 'loop_back_triggered' ORDER BY created_at",
(run_id,),
).fetchall()
events = []
for payload, created_at in rows:
data = json.loads(payload)
events.append({
"source_node_id": data["source_node_id"], "target_node_id": data["target_node_id"],
"attempt_number": data["attempt_number"], "feedback": data.get("feedback") or [],
"created_at": created_at,
})
return events
def _work_nodes_for_run(conn, run_id):
# input_schema_ref/output_schema_ref/quota_limit_type/quota_wait_started_at
# added for U-026's hover tooltip + waiting_for_quota display (02-V §2.7:
# "tooltip 展示 input_schema_ref/output_schema_ref/retry_count 摘要...
# waiting_for_quota 态额外展示 quota_limit_type 和已等待时长") -- not
# needed by U-025's to-do tree, which is why they weren't here yet.
rows = conn.execute(
"SELECT node_exec_id, node_id, subagent_type, status, retry_count, input_payload, "
"output_payload, reflection_notes, git_worktree_ref, input_schema_ref, output_schema_ref, "
"quota_limit_type, quota_wait_started_at "
"FROM work_node_execution WHERE run_id = ? ORDER BY node_exec_id",
(run_id,),
).fetchall()
return [
{
"node_exec_id": n[0], "node_id": n[1], "subagent_type": n[2], "status": n[3],
"retry_count": n[4],
"input_payload": json.loads(n[5]) if n[5] else None,
"output_payload": json.loads(n[6]) if n[6] else None,
"reflection_notes": n[7], "git_worktree_ref": n[8],
"input_schema_ref": n[9], "output_schema_ref": n[10],
"quota_limit_type": n[11], "quota_wait_started_at": n[12],
}
for n in rows
]