task_reconciliation.py
5.1 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
"""Periodic recovery for tasks abandoned mid-flight (2026-08-06, real
incident: a task sat in status='validating' for 15+ hours with zero
`execution_event` activity and no live process working on it, because
whatever one-off call had driven it that far had already exited and
nothing in production polls for orphaned resumable tasks).
Scope: FR-2 only from `docs/todo/scheduler-production-polling-loop-
requirements.md` (in-flight task reconciliation) -- deliberately NOT FR-1
(new-goal auto-dispatch via `scheduler.run_scheduler_once()`), which is a
separate, larger, deliberately-not-bundled decision (see that doc's
non-goals/risks: "改变系统自主权边界"). This module only ever resumes a
task that already exists and was already in progress; it never creates a
new task or starts work nobody asked for.
A task only becomes a reconciliation candidate if ALL of:
1. `task.status` is in `task_execution._RESUMABLE_STATUSES`, and
2. it has no execution happening in THIS process right now
(`task_execution.is_in_flight()` -- covers the common case where the
console's Start button and this loop share the same guard, so an
actively-running task is never double-dispatched), and
3. its most recent `execution_event` is older than `stale_after_seconds`
(or it has none at all, in which case `task.created_at` is used) --
deliberately NOT `task.updated_at`, which does not change while a
phase is genuinely in progress (`implement_code`/
`validate_implementation_with_own_graph` never touch it mid-phase --
only the `task_status_changed` transitions do). Using
`task.updated_at` was the real blind spot during this incident's own
diagnosis: it looked "stale" for reasons unrelated to whether the task
was actually still being worked on.
`_DEFAULT_STALE_AFTER_SECONDS` (1800s = 30 min) is comfortably above the
longest normal single-node blocking call in this codebase (Validator's
own work graph, `timeout=1200s`) so a node that's still genuinely inside
one blocking `claude -p` call is never misdiagnosed as abandoned -- see
the requirements doc's FR-3/non-goal #4 for why this doesn't need to be
more precise than that for v1.
"""
import time
from datetime import datetime, timezone
from src import schema, task_execution
_DEFAULT_STALE_AFTER_SECONDS = 1800
def _to_epoch(sqlite_timestamp):
# Same parsing as work_node_run._parse_sqlite_timestamp: sqlite's
# CURRENT_TIMESTAMP / *_at defaults are UTC 'YYYY-MM-DD HH:MM:SS' with
# no timezone marker.
return datetime.strptime(sqlite_timestamp, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc).timestamp()
def find_stale_resumable_tasks(conn, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
"""Returns [{"task_id", "target_folder_path", "goal_text"}, ...] for
every task that's a genuine reconciliation candidate. Read-only --
does not dispatch anything, kept separate from `reconcile_once()` so
candidate selection is independently testable without spinning up
real background threads."""
cutoff = now() - stale_after_seconds
placeholders = ",".join("?" * len(task_execution._RESUMABLE_STATUSES))
rows = conn.execute(
"SELECT t.task_id, p.target_folder_path, t.goal_text, t.created_at, "
"(SELECT MAX(e.created_at) FROM execution_event e WHERE e.task_id = t.task_id) AS last_event_at "
"FROM task t JOIN project p ON p.project_id = t.project_id "
"WHERE t.status IN ({})".format(placeholders),
task_execution._RESUMABLE_STATUSES,
).fetchall()
candidates = []
for task_id, target_folder_path, goal_text, created_at, last_event_at in rows:
if task_execution.is_in_flight(task_id):
continue
last_activity = _to_epoch(last_event_at or created_at)
if last_activity >= cutoff:
continue
candidates.append({"task_id": task_id, "target_folder_path": target_folder_path, "goal_text": goal_text})
return candidates
def reconcile_once(db_path, stale_after_seconds=_DEFAULT_STALE_AFTER_SECONDS, now=time.time):
"""One reconciliation pass: find candidates, dispatch each via
`task_execution.start_task_execution_in_background()`. Uses its own
short-lived connection for the scan, closed before dispatching -- the
background threads each open their own (same pattern as
`scheduler.py`). Returns the candidate list (dispatched or not) for
callers/tests to inspect.
A candidate that fails to dispatch (e.g. `TaskExecutionError` because
something else claimed it between the scan and here -- a real race,
not a bug) is skipped, not treated as a fatal error for the whole
pass: 04-P's "调度器继续执行队列中其他任务" applies here the same way
it does in `scheduler.run_scheduler_once`."""
conn = schema.connect(db_path)
try:
candidates = find_stale_resumable_tasks(conn, stale_after_seconds, now)
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