org_loop_route.py
7.24 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
"""c_org_loop_route (03-C §3.11) + the validating-state exit that isn't
covered by any named controller (04-P: validating->done on validator_pass,
action "关闭任务" -- no c_* name given). Built together because they're the
same decision point (did the Validator's verdict PASS or FAIL) that U-015
already produces one half of (FAIL -> root-cause judgment); this module is
where task.status actually reaches 'done' or gets routed back into the loop.
Judgment (which route, or PASS/FAIL) already happened via the Validator --
this module is purely the deterministic code side ("判断用 Agent，路由用代码"
per 03-C §3.11): given a decided_route, apply the matching p_org_task_lifecycle
(04-P) transition.
**Simplification, noted rather than silently assumed**: 04-P's table models
validating->routed_fast_lane->coding as two hops (an observable
'routed_fast_lane'/'routed_full_replan' status, then a second 'route_applied'
transition to the final state). c_org_loop_route's own 03-C output_contract
returns {new_status, new_assigned_org_node} -- the *final* state directly, not
the intermediate one -- and no controller is named for the second hop. This
implementation collapses both into one atomic call (validating straight to
coding/planning_brd_prd_sdd), recording the route decision via an explicit
`execution_event(event_type=loop_routed)` instead of ever persisting
'routed_fast_lane'/'routed_full_replan' as a literal task.status row. Neither
REQ-05 nor REQ-09's acceptance wording checks for that literal intermediate
status, so this doesn't affect Done -- but a future PH3 dashboard wanting to
*see* that transient state would need this revisited.
core_engine routing (03-C: routes to p_deploy_supervisor_blue_green instead
of fast_lane/full_replan) is PH5 scope (U-034+), not attempted here --
core_engine isn't a valid bug_record.root_cause_category yet either (U-001's
schema deliberately excludes it, see src/bug_record.py).
"""
import json
import uuid
_NEW_STATUS = {"fast_lane": "coding", "full_replan": "planning_brd_prd_sdd"}
_NEW_ORG_NODE = {"fast_lane": "coder", "full_replan": "planner"}
_LOOP_LIMIT = 3
class LoopLimitReachedError(Exception):
"""LOOP_LIMIT_REACHED -- 03-C §3.11."""
class TaskNotBlockedError(Exception):
"""manual_resume called on a task that isn't blocked_pending_human (REQ-09
failure case: calling it on an already-'done' task must error, not
silently no-op or corrupt the finished task)."""
class TaskNotValidatingError(Exception):
"""close_task_done / route_org_loop(fast_lane|full_replan) called on a
task that isn't currently 'validating'."""
def _write_status_event(conn, task_id, project_id, from_status, to_status):
conn.execute(
"INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
"VALUES (?, ?, ?, 'task_status_changed', ?)",
(str(uuid.uuid4()), project_id, task_id, json.dumps({"from_status": from_status, "to_status": to_status})),
)
def _get_task(conn, task_id):
row = conn.execute("SELECT project_id, status, loop_round FROM task WHERE task_id = ?", (task_id,)).fetchone()
if row is None:
raise ValueError("no such task: " + task_id)
return {"project_id": row[0], "status": row[1], "loop_round": row[2]}
def close_task_done(conn, task_id):
"""04-P: validating -> done (validator_pass, "全部 Benchmark/测试用例通过")."""
task = _get_task(conn, task_id)
if task["status"] != "validating":
raise TaskNotValidatingError(
"task {} is status={}, not validating -- can't close to done".format(task_id, task["status"])
)
conn.execute("UPDATE task SET status = 'done', updated_at = CURRENT_TIMESTAMP WHERE task_id = ?", (task_id,))
_write_status_event(conn, task_id, task["project_id"], "validating", "done")
conn.commit()
return {"task_id": task_id, "new_status": "done"}
def route_org_loop(conn, task_id, decided_route):
"""decided_route: 'fast_lane' | 'full_replan' | 'manual_resume'.
Returns {"task_id", "new_status", "new_assigned_org_node"}.
"""
if decided_route not in ("fast_lane", "full_replan", "manual_resume"):
raise ValueError(decided_route)
task = _get_task(conn, task_id)
if decided_route == "manual_resume":
return _manual_resume(conn, task_id, task)
if task["status"] != "validating":
raise TaskNotValidatingError(
"task {} is status={}, not validating -- fast_lane/full_replan routing only applies "
"right after a Validator fail".format(task_id, task["status"])
)
if task["loop_round"] >= _LOOP_LIMIT:
conn.execute(
"UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'loop_limit_reached', "
"updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
(task_id,),
)
_write_status_event(conn, task_id, task["project_id"], task["status"], "blocked_pending_human")
conn.commit()
raise LoopLimitReachedError(
"task {} has loop_round={} >= {}, forced to blocked_pending_human".format(
task_id, task["loop_round"], _LOOP_LIMIT
)
)
new_status = _NEW_STATUS[decided_route]
new_org_node = _NEW_ORG_NODE[decided_route]
conn.execute(
"UPDATE task SET status = ?, assigned_org_node = ?, loop_round = loop_round + 1, "
"updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
(new_status, new_org_node, task_id),
)
conn.execute(
"INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
"VALUES (?, ?, ?, 'loop_routed', ?)",
(str(uuid.uuid4()), task["project_id"], task_id, json.dumps({"decided_route": decided_route})),
)
_write_status_event(conn, task_id, task["project_id"], task["status"], new_status)
conn.commit()
return {"task_id": task_id, "new_status": new_status, "new_assigned_org_node": new_org_node}
def _manual_resume(conn, task_id, task):
if task["status"] != "blocked_pending_human":
raise TaskNotBlockedError(
"task {} is status={}, not blocked_pending_human -- manual_resume only applies to "
"blocked tasks".format(task_id, task["status"])
)
rows = conn.execute(
"SELECT payload FROM execution_event WHERE task_id = ? AND event_type = 'task_status_changed' "
"ORDER BY rowid DESC",
(task_id,),
).fetchall()
restore_status = None
for (payload_json,) in rows:
payload = json.loads(payload_json)
if payload.get("to_status") == "blocked_pending_human":
restore_status = payload.get("from_status")
break
if restore_status is None:
raise ValueError(
"could not find the pre-blocked status for task {} from execution_event history".format(task_id)
)
conn.execute(
"UPDATE task SET status = ?, blocked_reason = NULL, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
(restore_status, task_id),
)
_write_status_event(conn, task_id, task["project_id"], "blocked_pending_human", restore_status)
conn.commit()
new_org_node = conn.execute("SELECT assigned_org_node FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
return {"task_id": task_id, "new_status": restore_status, "new_assigned_org_node": new_org_node}