scheduler.py
5.75 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
"""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".
"""
import concurrent.futures
from src import e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config
_BUSY_TIMEOUT_MS = 30000
_FIX_LOOP_TIMEOUT = 1200
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."""
conn = schema.connect(db_path)
conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
try:
result = e2e_pipeline.run_end_to_end(conn, target_folder_path, goal_text)
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