scheduler.py
9.33 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
"""U-032: the scheduler 04-P/01-M/06-验收与测试锚点.md repeatedly presume but
no Unit before this one ever built. REQ-09's boundary case ("多个任务同时有
一个 blocked_pending_human、其余 queued 时，调度器应继续调度 queued 任务")
and 01-M's own indexing rationale ("索引...用于调度器扫描") both assume an
autonomous dispatcher exists; before this it didn't.
`org_node_lifecycle._check_concurrency()` (U-010) already gates a *single*
trigger call against `max_concurrent_goals`, but nothing ever called it
across a real queue of multiple Goals, and `e2e_pipeline.run_end_to_end()`
(U-014) drives exactly one task, synchronously, and was only ever called
from tests.
Found and closed in the same pass (2026-08-02): `run_end_to_end()` itself
deliberately stops at 'validating' with a computed-but-unapplied verdict
("Criteria 4-7 ... explicitly out of this Unit's scope" per its own
docstring) -- a scheduler whose job is "drive Goals to completion" can't
leave every task sitting at 'validating' forever, so `finish_task_lifecycle()`
below applies the PASS -> `close_task_done` / FAIL -> `org_level_loop.
run_fix_loop` step `run_end_to_end` always left undone, without changing
`run_end_to_end`'s own tested behavior or scope.
Threading model: each worker thread opens its OWN sqlite connection
(`schema.connect(db_path)`) rather than sharing one `conn` object across
threads -- U-027 already found the hard way (SIGSEGV) that a shared sqlite3
connection needs a hand-held lock around every operation including close();
replicating that discipline here would just serialize all DB work anyway and
defeat the point of concurrency. Each connection sets a `busy_timeout` so
concurrent writers on the shared file retry instead of raising "database is
locked".
**Second gap found and closed in the same pass (2026-08-02, PH4-closure
audit before starting PH5)**: `work_node_run.CorruptPersistedStateError`
(REQ-06 failure), `work_graph_execute.ClassifierBlockedError` (REQ-07
failure), and `work_node_run.UsageLimitWaitTimeoutError` (REQ-14 failure)
were each correctly raised and each correctly marked their own
`work_node_execution.status='blocked'` (the last one via U-031's catch-all
in `work_node_run.run_work_node`) -- but nothing ever caught them to apply
the *task*-level consequence 06-验收与测试锚点.md's failure text for all
three literally names: `task.status='blocked_pending_human'` with a specific
`blocked_reason`. `work_node_run.py`'s own docstring already flagged this as
a deliberate layering choice ("callers should route it to
task.status='blocked_pending_human'") -- but no caller ever did. This is
exactly the layer `budget_guardrail.check_budget()` already does the
equivalent translation at for its own case (REQ-10), so `_block_task()`
below follows that same established pattern (UPDATE + `task_status_changed`
execution_event) rather than inventing a new one.
"""
import json
import uuid
import concurrent.futures
from src import (
e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config,
task_create, work_graph_execute, work_graph_orchestrate, work_node_run,
)
_BUSY_TIMEOUT_MS = 30000
_FIX_LOOP_TIMEOUT = 1200
_BLOCKED_REASON_FOR_EXCEPTION = (
(work_node_run.CorruptPersistedStateError, "corrupt_persisted_state"),
(work_graph_execute.ClassifierBlockedError, "classifier_blocked"),
(work_node_run.UsageLimitWaitTimeoutError, "usage_limit_wait_timeout"),
(work_graph_orchestrate.ConsistencyCheckFailedError, "prd_consistency_check_failed"),
)
_ROUTABLE_EXCEPTIONS = tuple(exc for exc, _ in _BLOCKED_REASON_FOR_EXCEPTION)
def _blocked_reason_for(exc):
for exc_type, reason in _BLOCKED_REASON_FOR_EXCEPTION:
if isinstance(exc, exc_type):
return reason
raise AssertionError("no blocked_reason registered for " + type(exc).__name__)
def _block_task(conn, task_id, blocked_reason):
task_row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
conn.execute(
"UPDATE task SET status = 'blocked_pending_human', blocked_reason = ?, "
"updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
(blocked_reason, 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_row[0], task_id, json.dumps({"from_status": task_row[1], "to_status": "blocked_pending_human"})),
)
conn.commit()
def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
"""PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
run_fix_loop` (root-cause judge + reroute + fix + reverify, U-017).
Adds 'final_status' to `run_result` and returns it."""
task_id = run_result["task_id"]
verdict = run_result["verdict"]["verdict"]
if verdict == "PASS":
org_loop_route.close_task_done(conn, task_id)
run_result["final_status"] = "done"
return run_result
benchmark_id = run_result["doc_artifact_ids"]["benchmark"]
benchmark_content = e2e_pipeline._read_artifact_content(conn, target_folder_path, benchmark_id)
fix_result = org_level_loop.run_fix_loop(
conn, task_id, target_folder_path,
bug_description="Validator reported {} for goal: {}".format(verdict, goal_text),
context=run_result["verdict"]["raw_response"],
fix_prompt=e2e_pipeline._READY_PROMPT,
verify_prompt=e2e_pipeline._VALIDATOR_PROMPT.format(
goal_text=goal_text, benchmark_content=benchmark_content
),
timeout=_FIX_LOOP_TIMEOUT,
)
run_result["final_status"] = fix_result["final_status"]
return run_result
def run_goal_to_completion(db_path, target_folder_path, goal_text):
"""One Goal, its own sqlite connection, start to a terminal task.status
(done or blocked_pending_human). This is the unit of work each scheduler
worker thread runs.
Creates the task itself (rather than delegating to e2e_pipeline.
run_end_to_end) so task_id is in scope even if e2e_pipeline.
run_task_lifecycle raises partway through -- needed to route
CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
to task.status='blocked_pending_human' (see module docstring's "Second
gap" note). Any OTHER exception still propagates uncaught, same as
before -- run_scheduler_once's own try/except is what keeps one Goal's
unexpected failure from stopping the rest.
"""
conn = schema.connect(db_path)
conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
try:
project_id = conn.execute(
"SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
).fetchone()[0]
task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
try:
result = e2e_pipeline.run_task_lifecycle(conn, task_id, target_folder_path, goal_text)
except _ROUTABLE_EXCEPTIONS as e:
blocked_reason = _blocked_reason_for(e)
_block_task(conn, task_id, blocked_reason)
return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": blocked_reason}
return finish_task_lifecycle(conn, result, target_folder_path, goal_text)
finally:
conn.close()
def run_scheduler_once(db_path, goals, max_workers=None):
"""`goals`: [(target_folder_path, goal_text), ...], one entry per Goal to
dispatch this round. Runs up to `max_workers` concurrently (default:
`org_concurrency_config.max_concurrent_goals`, the same knob
`org_node_lifecycle._check_concurrency` reads -- read fresh here so a
config change takes effect on the next scheduler round without a
restart); goals beyond that queue inside the executor and start as slots
free up, which is what actually makes "raise max_concurrent_goals -> more
Goals genuinely run in parallel" (U-032's Done line) true, rather than
something only exercised by an exception-driven retry loop.
A goal that raises (including `LoopLimitReachedError` once `loop_round`
hits 3 -> `blocked_pending_human`, which IS the expected outcome of that
path, not a scheduler bug) doesn't stop the others -- 04-P's "调度器继续
执行队列中其他任务" applies to a raised exception in one goal exactly the
same as it does to one that's already `blocked_pending_human` going in.
Returns a list of {'goal_text', 'result'} or {'goal_text', 'error'} in
`goals` order.
"""
if max_workers is None:
conn = schema.connect(db_path)
try:
config = settings_config.get_concurrency_config(conn)
finally:
conn.close()
max_workers = config["max_concurrent_goals"] if config else 1
results = [None] * len(goals)
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {
pool.submit(run_goal_to_completion, db_path, target_folder_path, goal_text): i
for i, (target_folder_path, goal_text) in enumerate(goals)
}
for future in concurrent.futures.as_completed(futures):
i = futures[future]
goal_text = goals[i][1]
try:
results[i] = {"goal_text": goal_text, "result": future.result()}
except Exception as e:
results[i] = {"goal_text": goal_text, "error": str(e)}
return results