work_graph_run_query.py
6.7 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
"""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
from src import agent_profile
# H (2026-08-06): was its own hardcoded ('planner', 'coder', 'validator')
# tuple -- a real dependency the requirements doc's own risk analysis
# missed (see docs/todo/graph-engineering-open-items.md's H review note).
# Now reuses agent_profile.ORG_NODES so this validation stays data-driven
# too, not a second place someone has to remember to update for a new role.
_VALID_ORG_NODES = agent_profile.ORG_NODES
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):
"""`org_node` (2026-08-12, Org Graph unification Phase 2c §2.1.1):
work_graph_run.org_node no longer exists (a merged run genuinely spans
several org_nodes) -- this filter now means "runs with at least one
node whose subagent_type matches", via an EXISTS subquery against
work_node_execution instead of a plain column match. Each returned row
carries "org_nodes_involved" (a list, via GROUP_CONCAT DISTINCT over
the same table) instead of the old single-value "org_node" -- Phase 3's
Org-level meta-graph (FR-5a) needs exactly this "which stages does this
run span" signal anyway."""
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.status, wgr.started_at, wgr.completed_at, "
"(SELECT COUNT(*) FROM work_node_execution wne WHERE wne.run_id = wgr.run_id) AS node_count, "
"(SELECT GROUP_CONCAT(DISTINCT wne.subagent_type) FROM work_node_execution wne "
" WHERE wne.run_id = wgr.run_id) AS org_nodes_involved, "
"wgr.parent_run_id, wgr.fork_of_node_exec_id "
"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 EXISTS (SELECT 1 FROM work_node_execution wne "
"WHERE wne.run_id = wgr.run_id AND wne.subagent_type = ?)"
)
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], "status": r[2],
"started_at": r[3], "completed_at": r[4], "node_count": r[5],
"org_nodes_involved": r[6].split(",") if r[6] else [],
# 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": r[7], "fork_of_node_exec_id": r[8],
}
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):
"""`org_nodes_involved` (2026-08-12, Phase 2c §2.1.1): same replacement
for the removed single-value `org_node` as list_work_graph_runs above
-- computed from this one run's own nodes (already fetched via
_nodes_for_run below) rather than a second query."""
row = conn.execute(
"SELECT run_id, task_id, graph_manifest, status, started_at, completed_at, "
"parent_run_id, fork_of_node_exec_id "
"FROM work_graph_run WHERE run_id = ?",
(run_id,),
).fetchone()
if row is None:
raise WorkGraphRunNotFoundError(run_id)
nodes = _nodes_for_run(conn, row[0])
org_nodes_involved = sorted({n["subagent_type"] for n in nodes if n["subagent_type"]})
return {
"run_id": row[0], "task_id": row[1],
"graph_manifest": json.loads(row[2]) if row[2] else None,
"status": row[3], "started_at": row[4], "completed_at": row[5],
"parent_run_id": row[6], "fork_of_node_exec_id": row[7],
"org_nodes_involved": org_nodes_involved,
"nodes": nodes,
}
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
]