'
+ "
STATUS" + escapeHtml(task.status) + "
"
@@ -1126,6 +1165,31 @@ function mountTaskDetail(taskId) {
});
}
+ const startBtn = document.getElementById("btn_start_task");
+ if (startBtn) {
+ startBtn.addEventListener("click", async () => {
+ startBtn.disabled = true;
+ await fetch("/api/tasks/" + taskId + "/execute", { method: "POST" });
+ refresh();
+ });
+ }
+ const pauseBtn = document.getElementById("btn_pause_task");
+ if (pauseBtn) {
+ pauseBtn.addEventListener("click", async () => {
+ pauseBtn.disabled = true;
+ await fetch("/api/tasks/" + taskId + "/pause", { method: "POST" });
+ refresh();
+ });
+ }
+ const stopBtn = document.getElementById("btn_stop_task");
+ if (stopBtn) {
+ stopBtn.addEventListener("click", async () => {
+ stopBtn.disabled = true;
+ await fetch("/api/tasks/" + taskId + "/stop", { method: "POST" });
+ refresh();
+ });
+ }
+
const editBtn = document.getElementById("btn_edit_goal_text");
const deleteBtn = document.getElementById("btn_delete_goal");
if (editBtn) {
diff --git a/src/backend_server.py b/src/backend_server.py
index 341690e..dd9f948 100644
--- a/src/backend_server.py
+++ b/src/backend_server.py
@@ -39,7 +39,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from src import (
artifact_query, budget_guardrail, bug_record, deployment_query, execution_log,
org_loop_route, paths, project_query, project_scaffold, schema, settings_config,
- skill_query, skill_update, task_create, task_query, ws_subscribe,
+ skill_query, skill_update, task_create, task_execution, task_query, ws_subscribe,
)
_FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")
@@ -228,6 +228,15 @@ class _Handler(BaseHTTPRequestHandler):
elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/route"):
task_id = self.path[len(_TASKS_PREFIX):-len("/route")]
self._handle_task_route(task_id)
+ elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/execute"):
+ task_id = self.path[len(_TASKS_PREFIX):-len("/execute")]
+ self._handle_task_execute(task_id)
+ elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/pause"):
+ task_id = self.path[len(_TASKS_PREFIX):-len("/pause")]
+ self._handle_task_pause(task_id, "user_paused")
+ elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/stop"):
+ task_id = self.path[len(_TASKS_PREFIX):-len("/stop")]
+ self._handle_task_pause(task_id, "user_stopped")
elif self.path.startswith(_SKILLS_PREFIX) and self.path.endswith("/rollback"):
skill_id = self.path[len(_SKILLS_PREFIX):-len("/rollback")]
self._handle_skill_rollback(skill_id)
@@ -457,18 +466,80 @@ class _Handler(BaseHTTPRequestHandler):
def _handle_task_route(self, task_id):
"""c_org_loop_route (03-C §3.11) -- the underlying route_org_loop()
was built in U-016 with no HTTP route until now; v_task_detail's
- click_retry (manual_resume) is what actually needs it wired up."""
+ click_retry (manual_resume) is what actually needs it wired up.
+
+ **2026-08-03**: if the task was blocked via a user-initiated
+ pause/stop (task_execution.request_pause, blocked_reason in
+ ('user_paused', 'user_stopped')), a successful manual_resume also
+ spawns a background thread to actually continue real execution --
+ matching what the ORIGINAL blocked_pending_human/manual_resume
+ design already implies (04-P: "调度器继续执行队列中其他任务" --
+ resuming should let the scheduler keep going, not just flip a
+ status flag and stop). System-triggered blocks (loop_limit_reached,
+ budget_exceeded) keep their existing behavior unchanged -- only
+ restoring status, no auto-continuation -- that's a separate,
+ already-tested concern this change doesn't touch."""
payload, error = self._read_json_body()
if error:
self._send_json(400, error)
return
+ decided_route = payload.get("decided_route")
+ pre_resume_reason = None
+ if decided_route == "manual_resume":
+ row = self.server.conn.execute(
+ "SELECT blocked_reason FROM task WHERE task_id = ?", (task_id,)
+ ).fetchone()
+ pre_resume_reason = row[0] if row else None
try:
with self.server.conn_lock:
- result = org_loop_route.route_org_loop(self.server.conn, task_id, payload.get("decided_route"))
+ result = org_loop_route.route_org_loop(self.server.conn, task_id, decided_route)
except org_loop_route.LoopLimitReachedError as e:
self._send_json(409, {"error": "LOOP_LIMIT_REACHED", "detail": str(e)})
+ return
except (org_loop_route.TaskNotBlockedError, org_loop_route.TaskNotValidatingError, ValueError) as e:
self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
+ return
+ if pre_resume_reason in ("user_paused", "user_stopped") and result.get("new_status") in (
+ "queued", "planning_brd_prd_sdd", "coding",
+ ):
+ task_row = self.server.conn.execute(
+ "SELECT project_id, goal_text FROM task WHERE task_id = ?", (task_id,)
+ ).fetchone()
+ target_folder_path = self.server.conn.execute(
+ "SELECT target_folder_path FROM project WHERE project_id = ?", (task_row[0],)
+ ).fetchone()[0]
+ task_execution.start_task_execution_in_background(
+ self.server.db_path, task_id, target_folder_path, task_row[1],
+ )
+ self._send_json(200, result)
+
+ def _handle_task_execute(self, task_id):
+ """c_task_execute (2026-08-03): the manual "Start" trigger --
+ confirmed with the user this is genuinely missing (03-C's own
+ c_task_create contract already assumed task creation would
+ auto-trigger scheduling via an always-on scheduler that was never
+ actually wired up as a running process, see task_execution.py's
+ module docstring)."""
+ row = self.server.conn.execute("SELECT project_id, goal_text, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
+ if row is None:
+ self._send_json(404, {"error": "NOT_FOUND"})
+ return
+ project_id, goal_text, status = row
+ if status != "queued":
+ self._send_json(409, {"error": "TASK_NOT_QUEUED", "detail": "task is status={}, not queued".format(status)})
+ return
+ target_folder_path = self.server.conn.execute(
+ "SELECT target_folder_path FROM project WHERE project_id = ?", (project_id,)
+ ).fetchone()[0]
+ task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text)
+ self._send_json(200, {"task_id": task_id, "started": True})
+
+ def _handle_task_pause(self, task_id, reason):
+ try:
+ with self.server.conn_lock:
+ result = task_execution.request_pause(self.server.conn, task_id, reason)
+ except task_execution.TaskExecutionError as e:
+ self._send_json(409, {"error": "TASK_NOT_PAUSABLE", "detail": str(e)})
else:
self._send_json(200, result)
@@ -818,10 +889,12 @@ def create_server(host="127.0.0.1", port=0, db_path=None, skill_repo_path=None):
write an actual commit into whatever repo this process happens to be
running from. Never exposed to HTTP clients -- only settable here, at
process/test-construction time."""
- conn = schema.connect(db_path if db_path is not None else paths.db_path(), check_same_thread=False)
+ resolved_db_path = db_path if db_path is not None else paths.db_path()
+ conn = schema.connect(resolved_db_path, check_same_thread=False)
_ensure_schema(conn)
server = _Server((host, port), _Handler)
server.conn = conn
+ server.db_path = resolved_db_path
server.conn_lock = threading.Lock()
server.skill_repo_path = skill_repo_path if skill_repo_path is not None else skill_query._REPO_ROOT
return server
diff --git a/src/task_execution.py b/src/task_execution.py
new file mode 100644
index 0000000..25ead76
--- /dev/null
+++ b/src/task_execution.py
@@ -0,0 +1,164 @@
+"""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 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")
+ benchmark_content = _artifact_content(conn, target_folder_path, task_id, "benchmark")
+
+ 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)
+ 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(task_id, benchmark_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)
+
+
+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():
+ conn = schema.connect(db_path)
+ 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}
diff --git a/src/test_backend_server.py b/src/test_backend_server.py
index 3756b17..3a18c79 100644
--- a/src/test_backend_server.py
+++ b/src/test_backend_server.py
@@ -716,6 +716,121 @@ class TaskQueryAndRouteRouteTest(unittest.TestCase):
self.assertEqual(body[0]["current_usage"], 10)
+class TaskExecutionRouteTest(unittest.TestCase):
+ """v_task_detail Start/Pause/Stop (2026-08-03): c_task_execute,
+ c_task_pause -- task_execution.py's own logic (resume-from-status,
+ pause mid-flight) is already covered by test_task_execution.py, this
+ only exercises the HTTP wiring. `start_task_execution_in_background`
+ is mocked throughout: it spawns a real background thread that opens
+ its OWN sqlite connection via `schema.connect(db_path)`, which for a
+ ':memory:' test db would silently create a second, disconnected
+ database -- exactly the kind of lower-level mechanic the codebase's
+ own convention (see other route test classes) says belongs in
+ test_task_execution.py, not re-tested at the HTTP layer."""
+
+ def setUp(self):
+ self.server = backend_server.create_server(db_path=":memory:")
+ self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+ self.thread.start()
+ self.server.conn.execute(
+ "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+ "VALUES ('proj-1', 'P1', '/tmp/proj-1', 1, 'active')"
+ )
+ self.server.conn.execute(
+ "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+ "VALUES ('task-1', 'proj-1', 'the goal', 'goal', 'queued', 'planner')"
+ )
+ self.server.conn.commit()
+
+ def tearDown(self):
+ self.server.shutdown()
+ self.server.server_close()
+ self.thread.join(timeout=5)
+
+ def _post(self, path, payload=None):
+ host, port = self.server.server_address
+ conn = http.client.HTTPConnection(host, port, timeout=5)
+ body = json.dumps(payload).encode("utf-8") if payload is not None else b""
+ conn.request("POST", path, body=body)
+ resp = conn.getresponse()
+ raw = resp.read()
+ body = json.loads(raw) if raw else None
+ conn.close()
+ return resp.status, body
+
+ def test_execute_starts_queued_task(self):
+ with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
+ status, body = self._post("/api/tasks/task-1/execute")
+ self.assertEqual(status, 200)
+ self.assertEqual(body, {"task_id": "task-1", "started": True})
+ mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")
+
+ def test_execute_nonexistent_task_returns_404(self):
+ status, body = self._post("/api/tasks/no-such-task/execute")
+ self.assertEqual(status, 404)
+ self.assertEqual(body["error"], "NOT_FOUND")
+
+ def test_execute_non_queued_task_returns_409(self):
+ self.server.conn.execute("UPDATE task SET status = 'coding' WHERE task_id = 'task-1'")
+ self.server.conn.commit()
+ status, body = self._post("/api/tasks/task-1/execute")
+ self.assertEqual(status, 409)
+ self.assertEqual(body["error"], "TASK_NOT_QUEUED")
+
+ def test_pause_blocks_in_flight_task(self):
+ self.server.conn.execute("UPDATE task SET status = 'coding' WHERE task_id = 'task-1'")
+ self.server.conn.commit()
+ status, body = self._post("/api/tasks/task-1/pause")
+ self.assertEqual(status, 200)
+ self.assertEqual(body, {"task_id": "task-1", "status": "blocked_pending_human", "blocked_reason": "user_paused"})
+ row = self.server.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
+ self.assertEqual(tuple(row), ("blocked_pending_human", "user_paused"))
+
+ def test_stop_blocks_in_flight_task_with_distinct_reason(self):
+ self.server.conn.execute("UPDATE task SET status = 'planning_brd_prd_sdd' WHERE task_id = 'task-1'")
+ self.server.conn.commit()
+ status, body = self._post("/api/tasks/task-1/stop")
+ self.assertEqual(status, 200)
+ self.assertEqual(body["blocked_reason"], "user_stopped")
+
+ def test_pause_on_queued_task_returns_409(self):
+ status, body = self._post("/api/tasks/task-1/pause")
+ self.assertEqual(status, 409)
+ self.assertEqual(body["error"], "TASK_NOT_PAUSABLE")
+
+ def test_manual_resume_after_user_pause_auto_continues(self):
+ self.server.conn.execute(
+ "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'user_paused' "
+ "WHERE task_id = 'task-1'"
+ )
+ self.server.conn.execute(
+ "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+ "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
+ "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\"}')"
+ )
+ self.server.conn.commit()
+ with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
+ status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
+ self.assertEqual(status, 200)
+ mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")
+
+ def test_manual_resume_after_system_block_does_not_auto_continue(self):
+ self.server.conn.execute(
+ "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'loop_limit_reached' "
+ "WHERE task_id = 'task-1'"
+ )
+ self.server.conn.execute(
+ "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+ "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
+ "'{\"from_status\": \"queued\", \"to_status\": \"blocked_pending_human\"}')"
+ )
+ self.server.conn.commit()
+ with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
+ status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
+ self.assertEqual(status, 200)
+ mock_start.assert_not_called()
+
+
def _ws_handshake_request(path, key):
return (
"GET {} HTTP/1.1\r\n"
diff --git a/src/test_task_execution.py b/src/test_task_execution.py
new file mode 100644
index 0000000..c7d377c
--- /dev/null
+++ b/src/test_task_execution.py
@@ -0,0 +1,298 @@
+"""c_task_execute / c_task_pause tests (2026-08-03): mocked at the
+subprocess boundary (CLAUDE.md Testing constraint), same pattern as
+test_e2e_pipeline.py -- this file tests task_execution.py's own resume/
+pause wiring, not e2e_pipeline's phase mechanics again.
+
+Run: python3 -m unittest src.test_task_execution -v
+"""
+
+import os
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from src import e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, task_execution, work_graph_execute
+
+
+def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
+ assert prompt == e2e_pipeline._READY_PROMPT
+ return "Ready."
+
+
+class TaskExecutionTestBase(unittest.TestCase):
+ def setUp(self):
+ self.tmp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.tmp.cleanup)
+ subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
+ self.conn = schema.connect(":memory:")
+ schema.create_schema(self.conn)
+ project = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
+ self.project_id = project["project_id"]
+
+ def tearDown(self):
+ self.conn.close()
+
+ def _insert_task(self, task_id, status, assigned_org_node="planner", loop_round=0):
+ self.conn.execute(
+ "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node, loop_round) "
+ "VALUES (?, ?, 'a goal', 'goal', ?, ?, ?)",
+ (task_id, self.project_id, status, assigned_org_node, loop_round),
+ )
+ self.conn.commit()
+
+ def _write_artifact(self, task_id, artifact_type, content):
+ run_id = "run-" + artifact_type
+ self.conn.execute(
+ "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+ "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id),
+ )
+ rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
+ abs_path = os.path.join(self.tmp.name, rel_path)
+ os.makedirs(os.path.dirname(abs_path), exist_ok=True)
+ with open(abs_path, "w") as f:
+ f.write(content)
+ subprocess.run(["git", "add", "-A"], cwd=self.tmp.name, check=True, capture_output=True)
+ subprocess.run(
+ ["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-m", "x"],
+ cwd=self.tmp.name, check=True, capture_output=True,
+ )
+ sha = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=self.tmp.name, capture_output=True, text=True
+ ).stdout.strip()
+ self.conn.execute(
+ "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
+ "git_commit_sha, validation_status) VALUES (?, ?, ?, ?, ?, ?, 'passed')",
+ ("art-" + artifact_type, artifact_type, task_id, run_id, rel_path, sha),
+ )
+ self.conn.commit()
+
+
+class ResumeFromCodingTest(TaskExecutionTestBase):
+ """Coder already ran (real code_diff/test_report artifacts exist) --
+ resuming from 'coding' must skip straight to Validator, not redo
+ Planner or Coder."""
+
+ def test_skips_to_validator_only(self):
+ self._insert_task("task-1", status="coding", assigned_org_node="coder")
+ self._write_artifact("task-1", "sdd", "# SDD content")
+ self._write_artifact("task-1", "benchmark", "# Benchmark content")
+ self._write_artifact("task-1", "code_diff", "diff content")
+ self._write_artifact("task-1", "test_report", "report content")
+
+ with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger:
+ result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+
+ self.assertEqual(result["verdict"]["verdict"], "PASS")
+ self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff")
+ # only the validator's real call happened -- no coder call
+ mock_trigger.assert_called_once()
+ self.assertEqual(mock_trigger.call_args[0][0], "validator")
+ row = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()
+ self.assertEqual(row[0], "validating")
+
+
+class ResumeFromPlanningTest(TaskExecutionTestBase):
+ """Planner already ran (real doc artifacts exist) -- resuming from
+ 'planning_brd_prd_sdd' must run Coder then Validator, not redo Planner."""
+
+ def test_runs_coder_then_validator(self):
+ self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner")
+ # trigger_org_node_full("coder", ...) requires ALL FOUR doc artifacts
+ # validation_status='passed' (org_node_lifecycle._REQUIRED_ARTIFACTS),
+ # not just the two this module's own resume logic reads content from.
+ self._write_artifact("task-1", "brd", "# BRD content")
+ self._write_artifact("task-1", "prd", "# PRD content")
+ self._write_artifact("task-1", "sdd", "# SDD content")
+ self._write_artifact("task-1", "benchmark", "# Benchmark content")
+
+ with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+ patch.object(work_graph_execute, "write_workflow_script"), \
+ patch.object(work_graph_execute, "execute_work_graph", return_value={"response": "done, tests pass"}), \
+ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger:
+ result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+
+ self.assertEqual(result["verdict"]["verdict"], "PASS")
+ self.assertIn("code_diff", result["code_artifact_ids"])
+ mock_trigger.assert_called_once()
+ self.assertEqual(mock_trigger.call_args[0][0], "validator")
+
+
+class ResumeUnresumableStatusTest(TaskExecutionTestBase):
+ def test_done_task_raises(self):
+ self._insert_task("task-1", status="done")
+ with self.assertRaises(task_execution.TaskExecutionError):
+ task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+
+ def test_blocked_task_raises(self):
+ self._insert_task("task-1", status="blocked_pending_human")
+ with self.assertRaises(task_execution.TaskExecutionError):
+ task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+
+
+class PauseMidResumeTest(TaskExecutionTestBase):
+ """If task.status flips to blocked_pending_human WHILE this thread is
+ between phases (checked right before each org_node trigger),
+ ExecutionPausedError must stop cleanly without triggering the next
+ phase."""
+
+ def test_pause_before_validator_stops_cleanly(self):
+ self._insert_task("task-1", status="coding", assigned_org_node="coder")
+ self._write_artifact("task-1", "sdd", "# SDD content")
+ self._write_artifact("task-1", "benchmark", "# Benchmark content")
+ self._write_artifact("task-1", "code_diff", "diff content")
+ self._write_artifact("task-1", "test_report", "report content")
+
+ real_lookup = task_execution._latest_artifact_ids
+
+ def lookup_then_pause(conn, task_id, artifact_types):
+ # simulates a concurrent request_pause() landing right after
+ # this (real, unmocked) lookup runs -- the last step before
+ # _check_not_paused() is reached in the real code path.
+ result = real_lookup(conn, task_id, artifact_types)
+ conn.execute(
+ "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'user_paused' "
+ "WHERE task_id = 'task-1'"
+ )
+ conn.commit()
+ return result
+
+ with patch.object(task_execution, "_latest_artifact_ids", side_effect=lookup_then_pause), \
+ patch.object(org_node_trigger, "trigger_org_node") as mock_trigger:
+ with self.assertRaises(task_execution.ExecutionPausedError):
+ task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
+ mock_trigger.assert_not_called()
+
+
+class CheckNotPausedTest(TaskExecutionTestBase):
+ def test_raises_when_blocked(self):
+ self._insert_task("task-1", status="blocked_pending_human")
+ with self.assertRaises(task_execution.ExecutionPausedError):
+ task_execution._check_not_paused(self.conn, "task-1")
+
+ def test_does_not_raise_when_not_blocked(self):
+ self._insert_task("task-1", status="coding")
+ task_execution._check_not_paused(self.conn, "task-1") # no raise
+
+
+class RequestPauseTest(TaskExecutionTestBase):
+ def test_pauses_a_coding_task(self):
+ self._insert_task("task-1", status="coding")
+ result = task_execution.request_pause(self.conn, "task-1", "user_paused")
+ self.assertEqual(result["status"], "blocked_pending_human")
+ row = self.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
+ self.assertEqual(row, ("blocked_pending_human", "user_paused"))
+
+ def test_stops_a_planning_task(self):
+ self._insert_task("task-1", status="planning_brd_prd_sdd")
+ result = task_execution.request_pause(self.conn, "task-1", "user_stopped")
+ self.assertEqual(result["blocked_reason"], "user_stopped")
+
+ def test_cannot_pause_a_queued_task(self):
+ self._insert_task("task-1", status="queued")
+ with self.assertRaises(task_execution.TaskExecutionError):
+ task_execution.request_pause(self.conn, "task-1", "user_paused")
+
+ def test_cannot_pause_a_done_task(self):
+ self._insert_task("task-1", status="done")
+ with self.assertRaises(task_execution.TaskExecutionError):
+ task_execution.request_pause(self.conn, "task-1", "user_paused")
+
+ def test_nonexistent_task_raises(self):
+ with self.assertRaises(task_execution.TaskExecutionError):
+ task_execution.request_pause(self.conn, "no-such-task", "user_paused")
+
+
+class StartTaskExecutionInBackgroundTest(unittest.TestCase):
+ """Real threading (own connection per thread, same as scheduler.py) --
+ needs a real file db_path, not ':memory:' (a fresh ':memory:' connection
+ per thread would be a completely separate, empty database)."""
+
+ def setUp(self):
+ self.repo = tempfile.TemporaryDirectory()
+ self.addCleanup(self.repo.cleanup)
+ subprocess.run(["git", "init"], cwd=self.repo.name, check=True, capture_output=True)
+ self.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+ self.tmp_db.close()
+ self.db_path = self.tmp_db.name
+ self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
+
+ conn = schema.connect(self.db_path)
+ schema.create_schema(conn)
+ project = project_scaffold.create_project(conn, self.repo.name, init_git_if_missing=False)
+ self.project_id = project["project_id"]
+ conn.execute(
+ "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+ "VALUES ('task-1', ?, 'a goal', 'goal', 'coding', 'coder')", (self.project_id,)
+ )
+ conn.commit()
+ conn.close()
+
+ def _write_artifact(self, task_id, artifact_type, content):
+ conn = schema.connect(self.db_path)
+ run_id = "run-" + artifact_type
+ conn.execute(
+ "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+ "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id),
+ )
+ rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
+ abs_path = os.path.join(self.repo.name, rel_path)
+ os.makedirs(os.path.dirname(abs_path), exist_ok=True)
+ with open(abs_path, "w") as f:
+ f.write(content)
+ subprocess.run(["git", "add", "-A"], cwd=self.repo.name, check=True, capture_output=True)
+ subprocess.run(
+ ["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-m", "x"],
+ cwd=self.repo.name, check=True, capture_output=True,
+ )
+ sha = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=self.repo.name, capture_output=True, text=True
+ ).stdout.strip()
+ conn.execute(
+ "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
+ "git_commit_sha, validation_status) VALUES (?, ?, ?, ?, ?, ?, 'passed')",
+ ("art-" + artifact_type, artifact_type, task_id, run_id, rel_path, sha),
+ )
+ conn.commit()
+ conn.close()
+
+ def test_real_thread_completes_and_closes_task(self):
+ self._write_artifact("task-1", "sdd", "# SDD")
+ self._write_artifact("task-1", "benchmark", "# Benchmark")
+ self._write_artifact("task-1", "code_diff", "diff")
+ self._write_artifact("task-1", "test_report", "report")
+
+ with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"):
+ thread = task_execution.start_task_execution_in_background(
+ self.db_path, "task-1", self.repo.name, "a goal",
+ )
+ thread.join(timeout=10)
+
+ conn = schema.connect(self.db_path)
+ status = conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
+ conn.close()
+ self.assertEqual(status, "done")
+
+ def test_generic_failure_marks_blocked(self):
+ self._write_artifact("task-1", "sdd", "# SDD")
+ self._write_artifact("task-1", "benchmark", "# Benchmark")
+ self._write_artifact("task-1", "code_diff", "diff")
+ self._write_artifact("task-1", "test_report", "report")
+
+ with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+ patch.object(org_node_trigger, "trigger_org_node", side_effect=RuntimeError("boom")):
+ thread = task_execution.start_task_execution_in_background(
+ self.db_path, "task-1", self.repo.name, "a goal",
+ )
+ thread.join(timeout=10)
+
+ conn = schema.connect(self.db_path)
+ row = conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
+ conn.close()
+ self.assertEqual(row, ("blocked_pending_human", "execution_error"))
+
+
+if __name__ == "__main__":
+ unittest.main()