org_node_lifecycle.py
6.23 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
"""c_org_node_trigger (03-C §3.3): full implementation.
U-010 scope: per 05 §5.1 4.4's staged rollout (05-实施约束与依赖.md §5.5:
"手工调用验证 Planner/Coder/Validator 三个 Org Agent 分别可被触发" -- NOT
architecture doc §11, which doesn't contain that phrase; development-plan.md's
citation was wrong and is corrected in this session's notes), this drives the
three DIRECT forward transitions in `p_org_task_lifecycle` (04-P) that a single
`c_org_node_trigger` call causes:
queued -> planning_brd_prd_sdd (trigger planner)
planning_brd_prd_sdd -> coding (trigger coder)
coding -> validating (trigger validator)
with TASK_NOT_READY / CONCURRENCY_LIMIT_REACHED / USAGE_LIMIT_REACHED guards.
Everything past that -- planning_failed/coding_failed retries, the
validator_pass/validator_fail decision and bug_record-driven routing,
blocked_pending_human -- needs real Work Graph artifact production (U-011+)
or Org-level Loop routing (U-015/U-016, PH2) that doesn't exist yet, and is
not attempted here. The artifact-validation guards below are real, testable
logic even though nothing produces passing artifacts for real yet -- tests
seed the precondition rows directly, same as any consumer of an
upstream contract that hasn't been built.
"""
import json
import uuid
from src.org_node_trigger import OrgNodeTriggerError, trigger_org_node
_PRE_STATE = {
"planner": "queued",
"planner_reviewer": "planning_brd_prd_sdd",
"coder": "reviewing_planner_output",
"validator": "coding",
}
_POST_STATE = {
"planner": "planning_brd_prd_sdd",
"planner_reviewer": "reviewing_planner_output",
"coder": "coding",
"validator": "validating",
}
# Org Graph unification Phase 1 (2026-08-08): _REQUIRED_ARTIFACTS["coder"] used to be
# ("brd", "prd", "sdd", "benchmark"), but artifact_write.py sets validation_status='passed'
# unconditionally on write (no real semantic check exists yet -- see that module's own
# docstring), so this gate never actually blocked anything. planner_reviewer is now the
# first real semantic gate before Coder starts (see e2e_pipeline.review_planner_output).
# Only `sdd` is kept here, as pure existence-defense: implement_code() requires a real
# sdd_content positional argument, and a missing sdd artifact should fail clearly here
# rather than crash deeper inside implement_code() with a harder-to-diagnose error.
_REQUIRED_ARTIFACTS = {
"coder": ("sdd",),
"validator": ("code_diff", "test_report"),
}
# REQ-14 happy case cites "You've hit your session limit" as the real signal
# text; weekly/generic phrasing added defensively. Best-effort until we've
# observed the real error text from an actual exhausted-quota call (U-020).
_USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit")
class TaskNotReadyError(Exception):
"""TASK_NOT_READY -- 03-C §3.3."""
class ConcurrencyLimitReachedError(Exception):
"""CONCURRENCY_LIMIT_REACHED -- 03-C §3.3."""
class UsageLimitReachedError(Exception):
"""USAGE_LIMIT_REACHED -- 03-C §3.3 / 04-P p_usage_limit_wait. Detection
only here; the wait-and-retry loop itself is U-012/U-020's job."""
def _get_task(conn, task_id):
row = conn.execute("SELECT status, project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()
if row is None:
raise TaskNotReadyError("no such task: " + task_id)
return {"status": row[0], "project_id": row[1]}
def _artifacts_ready(conn, task_id, artifact_types):
for artifact_type in artifact_types:
row = conn.execute(
"SELECT 1 FROM artifact WHERE task_id = ? AND artifact_type = ? AND validation_status = 'passed'",
(task_id, artifact_type),
).fetchone()
if row is None:
return False
return True
def _check_concurrency(conn):
row = conn.execute(
"SELECT max_concurrent_goals FROM org_concurrency_config WHERE config_id = 'default'"
).fetchone()
limit = row[0] if row else 1 # 01-M 1.11 default
active = conn.execute(
"SELECT COUNT(*) FROM task WHERE status NOT IN ('queued', 'done', 'blocked_pending_human')"
).fetchone()[0]
if active >= limit:
raise ConcurrencyLimitReachedError(
"max_concurrent_goals={} reached ({} active)".format(limit, active)
)
def _looks_like_usage_limit(message):
lowered = message.lower()
return any(phrase in lowered for phrase in _USAGE_LIMIT_PHRASES)
def trigger_org_node_full(conn, org_node, task_id, target_folder_path, prompt, timeout=300):
"""Full c_org_node_trigger: guards + real subprocess call (src.org_node_trigger)
+ task.status transition, all in one call. Returns the updated
{"task_id", "status", "assigned_org_node"}.
"""
if org_node not in _PRE_STATE:
raise ValueError(org_node)
task = _get_task(conn, task_id)
if task["status"] != _PRE_STATE[org_node]:
raise TaskNotReadyError(
"task {} is in status={}, not ready for {} (needs {})".format(
task_id, task["status"], org_node, _PRE_STATE[org_node]
)
)
if org_node in _REQUIRED_ARTIFACTS and not _artifacts_ready(conn, task_id, _REQUIRED_ARTIFACTS[org_node]):
raise TaskNotReadyError("required artifacts not validation_status=passed yet for task " + task_id)
if org_node == "planner":
_check_concurrency(conn)
try:
trigger_org_node(org_node, prompt, target_folder_path, timeout=timeout)
except OrgNodeTriggerError as e:
if _looks_like_usage_limit(str(e)):
raise UsageLimitReachedError(str(e)) from e
raise
old_status = task["status"]
new_status = _POST_STATE[org_node]
conn.execute(
"UPDATE task SET status = ?, assigned_org_node = ?, updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
(new_status, org_node, task_id),
)
conn.execute(
"INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
"VALUES (?, ?, ?, 'task_status_changed', ?)",
(
str(uuid.uuid4()),
task["project_id"],
task_id,
json.dumps({"from_status": old_status, "to_status": new_status}),
),
)
conn.commit()
return {"task_id": task_id, "status": new_status, "assigned_org_node": org_node}