work_graph_run_query.py
4.63 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
"""c_work_graph_run_list (`GET /api/work-graph-runs`), the one net-new
backend capability from the `v_graph_console` SDD (2026-08-03): a
cross-task "graph run history" list for the FR-5 Planner-output-overview
panel. Every other data need on that page is served by existing
`task_query`/`artifact_query` functions -- this is the only query in this
codebase that lists `work_graph_run` rows across all tasks in a project
rather than for one `task_id` at a time, which is why it lives in its own
module instead of `task_query.py` (that module's own docstring frames it
as "built around a single task_id").
U-046 (2026-08-06 SDD, "运行历史点击跳转错误 + 缺运行详情页"): adds
`get_work_graph_run`, a single-`run_id` lookup for the new `v_run_detail`
page (`#/runs/:run_id`). This deliberately mirrors the shape of
`task_query.get_task`'s embedded `work_graph_runs[i]` entries (same keys,
same per-node field list via `_nodes_for_run`) so the frontend can reuse
`renderGraphVisualizerPanel`/`mountLogViewer` against this response without
any remapping -- it's a separate query (by `run_id` instead of `task_id`)
rather than a call into `task_query._work_nodes_for_run` because that
helper is private to its own module and this module already owns
`work_graph_run` reads.
"""
import json
_VALID_ORG_NODES = ("planner", "coder", "validator")
class InvalidParamError(Exception):
"""INVALID_PARAM -- missing project_id, org_node outside the enum, or
limit outside [1, 100]."""
def list_work_graph_runs(conn, project_id, org_node=None, limit=20):
if not project_id:
raise InvalidParamError("project_id is required")
if org_node is not None and org_node not in _VALID_ORG_NODES:
raise InvalidParamError("org_node must be one of {}, got {!r}".format(_VALID_ORG_NODES, org_node))
if limit is None:
limit = 20
try:
limit = int(limit)
except (TypeError, ValueError):
raise InvalidParamError("limit must be an integer, got {!r}".format(limit))
if limit < 1 or limit > 100:
raise InvalidParamError("limit must be in [1, 100], got {!r}".format(limit))
query = (
"SELECT wgr.run_id, wgr.task_id, wgr.org_node, wgr.status, wgr.started_at, wgr.completed_at, "
"(SELECT COUNT(*) FROM work_node_execution wne WHERE wne.run_id = wgr.run_id) AS node_count "
"FROM work_graph_run wgr JOIN task t ON t.task_id = wgr.task_id WHERE t.project_id = ?"
)
params = [project_id]
if org_node:
query += " AND wgr.org_node = ?"
params.append(org_node)
query += " ORDER BY wgr.started_at DESC LIMIT ?"
params.append(limit)
rows = conn.execute(query, params).fetchall()
return [
{
"run_id": r[0], "task_id": r[1], "org_node": r[2], "status": r[3],
"started_at": r[4], "completed_at": r[5], "node_count": r[6],
}
for r in rows
]
class WorkGraphRunNotFoundError(Exception):
"""NOT_FOUND -- c_work_graph_run_get on a nonexistent run_id."""
def get_work_graph_run(conn, run_id):
row = conn.execute(
"SELECT run_id, task_id, org_node, graph_manifest, status, started_at, completed_at "
"FROM work_graph_run WHERE run_id = ?",
(run_id,),
).fetchone()
if row is None:
raise WorkGraphRunNotFoundError(run_id)
return {
"run_id": row[0], "task_id": row[1], "org_node": row[2],
"graph_manifest": json.loads(row[3]) if row[3] else None,
"status": row[4], "started_at": row[5], "completed_at": row[6],
"nodes": _nodes_for_run(conn, row[0]),
}
def _nodes_for_run(conn, run_id):
# Same column list as task_query._work_nodes_for_run -- kept in lockstep
# on purpose (see module docstring) so v_run_detail can reuse
# renderGraphVisualizerPanel/mountLogViewer unmodified.
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
]