task_execution.py
9.98 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
191
192
193
194
195
196
197
198
199
"""c_task_execute / c_task_pause (2026-08-03): manual Start/Pause/Stop
controls for v_task_detail, confirmed with the user first:
1. **Soft stop only.** `org_node_trigger.py`/`work_graph_execute.py` both
invoke the real `claude` CLI via blocking `subprocess.run()` -- no
process handle is stored anywhere a "stop" action could reach in and
kill. Pause/Stop let whatever real call is currently in flight finish
naturally; only the NEXT phase is withheld. A genuine hard-kill would
need switching those call sites to `Popen` + tracking PIDs somewhere
reachable by task_id, which is out of scope for this pass.
2. **No new `task.status` enum value.** Paused/stopped tasks reuse
`status='blocked_pending_human'` with `blocked_reason='user_paused'` /
`'user_stopped'` (distinguished in the audit trail, not functionally --
both resume the same way) rather than a 01-M schema change. This reuses
the already-built, already-tested `manual_resume` mechanism
(`org_loop_route._manual_resume`) to restore the pre-paused status.
**Why a manual Start button doesn't contradict the "lights-out" design**:
`c_task_create`'s own contract (03-C) already assumed task creation would
auto-trigger scheduling ("若 max_concurrent_goals 已达上限，任务落 queued
但不触发调度" -- implying it DOES trigger scheduling otherwise) via an
always-on Org Graph scheduler that was never actually wired up as a running
process (`src/scheduler.py` builds the mechanism, nothing spins it up
continuously). `start_task_execution_in_background()` below is that same
auto-trigger, just invoked by an explicit user action instead of firing
automatically the instant a task is created.
"""
import threading
from src import e2e_pipeline, org_node_lifecycle, schema, scheduler
_RESUMABLE_STATUSES = ("queued", "planning_brd_prd_sdd", "coding")
_DOC_ARTIFACT_TYPES = ("brd", "prd", "sdd", "benchmark")
_CODE_ARTIFACT_TYPES = ("code_diff", "test_report")
class TaskExecutionError(Exception):
"""The task isn't in a state this module knows how to (re)start from --
e.g. already running, already done, or blocked for a reason unrelated
to a user-initiated pause/stop."""
class ExecutionPausedError(Exception):
"""Internal control-flow signal, not a failure: a pause/stop was
requested while this background thread was between phases. Caught by
the background runner and treated as a clean stop -- the task is
ALREADY `blocked_pending_human` by construction (that's literally how
the pause was requested), nothing more to record."""
def _check_not_paused(conn, task_id):
status = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
if status == "blocked_pending_human":
raise ExecutionPausedError(task_id)
def _latest_artifact_ids(conn, task_id, artifact_types):
rows = conn.execute(
"SELECT artifact_type, artifact_id FROM artifact WHERE task_id = ? AND artifact_type IN ({}) "
"ORDER BY created_at".format(",".join("?" * len(artifact_types))),
(task_id, *artifact_types),
).fetchall()
return {r[0]: r[1] for r in rows}
def _artifact_content(conn, target_folder_path, task_id, artifact_type):
row = conn.execute(
"SELECT artifact_id FROM artifact WHERE task_id = ? AND artifact_type = ? ORDER BY created_at DESC LIMIT 1",
(task_id, artifact_type),
).fetchone()
if row is None:
raise TaskExecutionError("task {} has no {} artifact to resume from".format(task_id, artifact_type))
return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0])
def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default):
"""Like _artifact_content, but for the 2026-08-06 test_dataset/
test_script nodes -- a task resumed from before this feature existed
(or any other reason those two specific artifacts are missing) should
still be resumable, just without them, not hard-fail."""
row = conn.execute(
"SELECT artifact_id FROM artifact WHERE task_id = ? AND artifact_type = ? ORDER BY created_at DESC LIMIT 1",
(task_id, artifact_type),
).fetchone()
if row is None:
return default
return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0])
def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
"""Same return shape as `e2e_pipeline.run_task_lifecycle` -- continues
from wherever `task.status` indicates real work last completed, instead
of re-running phases that already produced real artifacts. Pausing only
ever happens BETWEEN phases (checked right before each org_node
trigger), so the only statuses a paused task can be resumed from are
'queued' (nothing ran yet), 'planning_brd_prd_sdd' (Planner done, Coder
not yet triggered), and 'coding' (Coder done, Validator not yet
triggered) -- anything else raises `TaskExecutionError` rather than
guessing at a mid-phase resume this module was never designed for."""
status = conn.execute("SELECT status FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
if status not in _RESUMABLE_STATUSES:
raise TaskExecutionError("task {} is status={}, not resumable from here".format(task_id, status))
if status == "queued":
return e2e_pipeline.run_task_lifecycle(conn, task_id, target_folder_path, goal_text)
doc_artifact_ids = _latest_artifact_ids(conn, task_id, _DOC_ARTIFACT_TYPES)
sdd_content = _artifact_content(conn, target_folder_path, task_id, "sdd")
prd_content = _artifact_content(conn, target_folder_path, task_id, "prd")
benchmark_content = _artifact_content(conn, target_folder_path, task_id, "benchmark")
# 2026-08-06 nodes -- see _artifact_content_or_default's own docstring
# for why these two specifically fall back instead of hard-failing.
test_dataset_content = _artifact_content_or_default(
conn, target_folder_path, task_id, "test_dataset", "N/A: not provided"
)
test_script_content = _artifact_content_or_default(
conn, target_folder_path, task_id, "test_script", "N/A: not provided"
)
if status == "planning_brd_prd_sdd":
_check_not_paused(conn, task_id)
org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
code_artifact_ids = e2e_pipeline.implement_code(
conn, task_id, sdd_content, goal_text, target_folder_path,
benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
test_script_content=test_script_content,
)
else: # status == "coding" -- Coder already ran
code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES)
_check_not_paused(conn, task_id)
org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
verdict = e2e_pipeline.validate_implementation_with_own_graph(
conn, task_id, sdd_content, prd_content, goal_text, target_folder_path,
)
return {
"task_id": task_id, "doc_artifact_ids": doc_artifact_ids,
"code_artifact_ids": code_artifact_ids, "verdict": verdict,
}
def _mark_blocked(conn, task_id, exc):
reason = scheduler._blocked_reason_for(exc) if isinstance(exc, scheduler._ROUTABLE_EXCEPTIONS) else "execution_error"
scheduler._block_task(conn, task_id, reason, detail=str(exc))
def start_task_execution_in_background(db_path, task_id, target_folder_path, goal_text):
"""Spawns a daemon thread that runs (or resumes) this task's real
execution to completion. Own sqlite connection per thread -- same
reasoning as `scheduler.run_goal_to_completion`, a shared connection
across threads needs the same `conn_lock` discipline U-027 already
proved is easy to get wrong. On PASS/FAIL, applies the same
`scheduler.finish_task_lifecycle` closing step (`close_task_done` /
`org_level_loop.run_fix_loop`) a scheduled task gets, so a manually
started one reaches the same real terminal state. Returns the Thread
(already started) so callers/tests can `.join()` it if needed."""
def run():
# check_same_thread=False (2026-08-06, F): same reasoning as
# scheduler.run_goal_to_completion's own conn -- this thread's conn
# can reach work_graph_orchestrate.run_graph()'s multi-node
# concurrent layers (G) via finish_task_lifecycle -> run_fix_loop ->
# judge_root_cause_via_work_graph, whose worker threads aren't this
# one.
conn = schema.connect(db_path, check_same_thread=False)
try:
result = run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text)
scheduler.finish_task_lifecycle(conn, result, target_folder_path, goal_text)
except ExecutionPausedError:
pass
except Exception as e:
_mark_blocked(conn, task_id, e)
finally:
conn.close()
thread = threading.Thread(target=run, daemon=True)
thread.start()
return thread
def request_pause(conn, task_id, reason):
"""`reason`: 'user_paused' or 'user_stopped' -- see module docstring
for why these are functionally identical (both a soft stop reusing
blocked_pending_human), kept as distinct strings only so the audit
trail (execution_event / blocked_reason) preserves which button the
user actually clicked. Only valid while a background execution could
plausibly be in flight; raises TaskExecutionError otherwise (nothing
running to pause, or already terminal)."""
row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
if row is None:
raise TaskExecutionError("no such task: " + task_id)
project_id, status = row
if status not in ("planning_brd_prd_sdd", "coding"):
raise TaskExecutionError(
"task {} is status={}, nothing in-flight to pause/stop".format(task_id, status)
)
scheduler._block_task(conn, task_id, reason)
return {"task_id": task_id, "status": "blocked_pending_human", "blocked_reason": reason}