new_goal_dispatch.py
2.92 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
"""Periodic auto-dispatch for new Goals that were created but never
started (2026-08-13, `docs/todo/scheduler-new-goal-dispatch-requirements.md`
FR-1). `scheduler.run_scheduler_once()` exists and works, but nothing in
production ever called it: a `task.status='queued'` row (created via the
API/console, never given to `run_goal_to_completion`) sat there until a
human clicked the console's Start button.
Scope: deliberately NOT `task_reconciliation.py`'s job -- that module
resumes tasks that already started and got abandoned mid-flight (status in
`task_execution._RESUMABLE_STATUSES`, stale by `_DEFAULT_STALE_AFTER_SECONDS`
= 30 min). This module only ever looks at `status='queued'` (a task that has
never been dispatched at all) and does not apply a staleness window --
`docs/done/scheduler-task-reconciliation-design.md` §7 records the decision
to keep the two loops independent, not merged into one "万能调度循环", and
that a 'queued' task should NOT wait out a 30-minute staleness window before
its very first dispatch (the whole point of FR-1 is near-real-time pickup of
brand-new Goals, not eventual recovery of stuck ones)."""
from src import schema, task_execution
def find_queued_tasks(conn):
"""Returns [{"task_id", "target_folder_path", "goal_text"}, ...] for
every `status='queued'` task not already in flight in this process.
Read-only -- kept separate from `dispatch_once()` so candidate selection
is independently testable without spinning up real background threads
(same split as `task_reconciliation.find_stale_resumable_tasks`)."""
rows = conn.execute(
"SELECT t.task_id, p.target_folder_path, t.goal_text "
"FROM task t JOIN project p ON p.project_id = t.project_id "
"WHERE t.status = 'queued'"
).fetchall()
candidates = []
for task_id, target_folder_path, goal_text in rows:
if task_execution.is_in_flight(task_id):
continue
candidates.append({"task_id": task_id, "target_folder_path": target_folder_path, "goal_text": goal_text})
return candidates
def dispatch_once(db_path):
"""One dispatch pass: find candidates, dispatch each via
`task_execution.start_task_execution_in_background()`. Own short-lived
connection for the scan, closed before dispatching -- same pattern as
`task_reconciliation.reconcile_once`. A candidate that fails to dispatch
(e.g. the console's Start button won the race between scan and here) is
skipped, not treated as fatal for the whole pass. Returns the candidate
list (dispatched or not) for callers/tests to inspect."""
conn = schema.connect(db_path)
try:
candidates = find_queued_tasks(conn)
finally:
conn.close()
for c in candidates:
try:
task_execution.start_task_execution_in_background(
db_path, c["task_id"], c["target_folder_path"], c["goal_text"]
)
except Exception:
pass
return candidates