v_task_detail/v_dashboard: Start/Pause/Stop task execution controls

Users had no way to actually start a queued goal from the console -- task.status just sits at 'queued' forever. Traced the root cause through 03-C: c_task_create's own contract ("若 max_concurrent_goals 已达上限，任务落 queued 但不触发调度") implies task creation auto-triggers scheduling otherwise, and 03-C/04-P repeatedly assume an always-on Org Graph scheduler that watches the queue -- src/scheduler.py builds the mechanism but nothing ever spins it up as a running process, so no "start" button was ever designed because none was supposed to be needed. Design (confirmed with user, both recommended options chosen): - Pause/Stop are a soft stop, not a real subprocess kill -- the current in-flight `claude` call finishes naturally, only the next phase is withheld. Hard-killing would need Popen + PID tracking, out of scope. - No new task.status enum value -- reuses blocked_pending_human with blocked_reason='user_paused'/'user_stopped', riding the existing manual_resume mechanism instead of a schema change. New src/task_execution.py: run_or_resume_task_lifecycle() resumes from whichever phase actually hasn't run yet (queued/planning_brd_prd_sdd/ coding), start_task_execution_in_background() runs it to completion on its own daemon thread + sqlite connection, request_pause() blocks an in-flight task. Extended /route's manual_resume handling so resuming from a user-initiated pause/stop actually re-continues execution in the background, not just flips the status field back. Real gap found while building this: backend_server.create_server() never stored db_path on the server object -- needed for the new background thread to open its own connection to the same db. 22 new tests (14 src/test_task_execution.py + 8 TaskExecutionRouteTest in test_backend_server.py, HTTP wiring mocked around the already-unit-tested background thread mechanics). Verified against a real running server via curl: /execute 409s a non-queued task, /pause 409s a queued task and a double-pause, succeeds on a coding task. src/ suite: 496 tests, all green. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

v_task_detail/v_dashboard: Start/Pause/Stop task execution controls
Users had no way to actually start a queued goal from the console -- task.status just sits at 'queued' forever. Traced the root cause through 03-C: c_task_create's own contract ("若 max_concurrent_goals 已达上限，任务落 queued 但不触发调度") implies task creation auto-triggers scheduling otherwise, and 03-C/04-P repeatedly assume an always-on Org Graph scheduler that watches the queue -- src/scheduler.py builds the mechanism but nothing ever spins it up as a running process, so no "start" button was ever designed because none was supposed to be needed. Design (confirmed with user, both recommended options chosen): - Pause/Stop are a soft stop, not a real subprocess kill -- the current in-flight `claude` call finishes naturally, only the next phase is withheld. Hard-killing would need Popen + PID tracking, out of scope. - No new task.status enum value -- reuses blocked_pending_human with blocked_reason='user_paused'/'user_stopped', riding the existing manual_resume mechanism instead of a schema change. New src/task_execution.py: run_or_resume_task_lifecycle() resumes from whichever phase actually hasn't run yet (queued/planning_brd_prd_sdd/ coding), start_task_execution_in_background() runs it to completion on its own daemon thread + sqlite connection, request_pause() blocks an in-flight task. Extended /route's manual_resume handling so resuming from a user-initiated pause/stop actually re-continues execution in the background, not just flips the status field back. Real gap found while building this: backend_server.create_server() never stored db_path on the server object -- needed for the new background thread to open its own connection to the same db. 22 new tests (14 src/test_task_execution.py + 8 TaskExecutionRouteTest in test_backend_server.py, HTTP wiring mocked around the already-unit-tested background thread mechanics). Verified against a real running server via curl: /execute 409s a non-queued task, /pause 409s a queued task and a double-pause, succeeds on a coding task. src/ suite: 496 tests, all green. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 7c720ab8
Showing 6 changed files with 733 additions and 5 deletions
docs/development-plan.md
frontend/app.js
src/backend_server.py
src/task_execution.py
src/test_backend_server.py
src/test_task_execution.py
@@ -473,3 +473,17 @@ PH0-PH545 荳ｪ Unit牙驛ｨ螳梧縲ゞ-042 逵溷ｮ櫁荳ｾ蜴区ｵ玖ｿ幄｡御ｸｭ譛滄龍
 - **`v_goal_list`亥ｯｹ蟾ｲ譛 `v_task_board`/`v_task_detail` 逧黄螻包ｼ碁撼蜈ｨ譁ｰ鬘ｵ髱｢**壽頗邏｢/遲幃牙ｷｲ譛会ｼ梧眠蠅樊賜蠎擾ｼtask_query.list_tasks()` 蜉 `sort`/`order` 蜿よ焚檎區蜷榊黒譬｡鬪悟蜷埼亟豕ｨ蜈･会ｼ嫣v_task_detail` 譁ｰ蠅"郛冶ｾ醍岼譬"/"蛻髯､莉ｻ蜉｡"御ｻ惠 `status='queued'` 譌ｶ蜃ｺ邇ｰ窶披比ｸ譌ｦ Planner/Coder 蟾ｲ扈丞渕莠手ｿ吩ｸｪ goal_text 莠ｧ蜃ｺ霑悄螳 artifact/work_graph_run梧隼蜀呎蛻髯､蟆ｱ莨夊ｮｩ螳｡隶｡隶ｰ蠖墓欠蜷台ｸ荳ｪ荳榊ｭ伜惠逧岼譬ｼ瑚ｿ呎ｭ｣譏ｯ霑吩ｸｪ莉｣遐∝ｺ謎ｸ逶ｴ蝨ｨ驕ｿ蜈咲噪荳榊庄霑ｽ貅ｯ貍らｧｻ梧園莉･ `task_query.update_task_goal_text()`/`delete_task()` 驛ｽ蝨ｨ髱 `queued` 迥ｶ諤∽ｸ区拠扈晢ｼTaskNotEditableError`粂TTP 409峨
 - **逵溷ｮ樣ｪ瑚ｯ∵慮蜿醍鴫蟷ｶ菫ｮ螟堺ｸ荳ｪ逵 bug**啻delete_task()` 隨ｬ荳迚亥ｯｹ荳荳ｪ逵溷ｮ槫蟒ｺ逧ｼ郁ｵｰ `task_create.create_task()`御ｸ肴弍豬玖ｯ戊蟾ｱ謇句逧 INSERT荏queued` 莉ｻ蜉｡謇ｧ陦 `DELETE FROM task` 譌ｶ檎悄螳樊鴨蜃ｺ `sqlite3.IntegrityError`窶披覗task_create.create_task()` 閾ｪ蟾ｱ蝨ｨ蜷御ｸ荳ｪ莠句苅驥梧ｻ莨壼荳譚｡ `execution_event` 陦鯉ｼ01-M 閾ｪ蟾ｱ逧､夜醗梧ｲ｡譛 `ON DELETE CASCADE`会ｼ悟頭諤戊ｿ吩ｸｪ莉ｻ蜉｡莉取ｲ｡陲ｫ Planner/Coder/Validator 遒ｰ霑ｼ御ｹ溷ｷｲ扈剰蟆第怏荳譚｡閾ｪ蟾ｱ蛻帛ｻｺ譌ｶ逡吩ｸ狗噪莠倶ｻｶ縲り｡･荳贋ｺ酔莠句苅蜀蛻霑呎擅莉ｻ蜉｡閾ｪ蟾ｱ逧 `execution_event`queued` 諤∽ｸ倶ｸ榊庄閭ｽ譛臥悄豁｣逧 artifact/bug_record/work_graph_run梧園莉･霑呎弍蜚ｯ荳蜿ｯ閭ｽ謖｡霍ｯ逧､夜醗峨りｿ呎ｬ｡ bug 譏ｯ豬玖ｯ戊蟾ｱ逧 `_insert_task` 霎勧蜃ｽ謨ｰ域焔蜀 INSERT御ｻ惹ｸ榊 `execution_event`我ｻ取ｲ｡證ｴ髴ｲ霑噪窶披泌黒豬矩夊ｿｸ堺ｻ｣陦ｨ逵溽噪閭ｽ逕ｨ悟刈莠ｸ荳ｪ襍ｰ逵溷ｮ `task_create.create_task()` 霍ｯ蠕噪蝗槫ｽ呈ｵ玖ｯ暮煤豁ｻ縲
 - **鬪瑚ｯ∵婿蠑**啻src/test_project_query.py`12 荳ｪ逕ｨ萓具ｼ峨〜src/test_task_query.py` 譁ｰ蠅樒噪謗貞ｺ/郛冶ｾ/蛻髯､逕ｨ萓具ｼ亥性荳企擇霑呎擅逵溷ｮ櫁ｷｯ蠕屓蠖抵ｼ峨〜src/test_backend_server.py` 譁ｰ蠅樔ｸ､荳ｪ逵溷ｮ HTTP 霍ｯ逕ｱ豬玖ｯ慕ｱｻProjectListAndCrudRouteTest`/`TaskCrudRouteTest`檎悄螳櫁ｵｷ譛榊苅蝎ｨ+逵溷ｮ櫁ｯｷ豎ゑｼ御ｸ mock峨よｵ剰ｧ亥勣謇ｩ螻墓悽谺｡莉堺ｸ榊庄逕ｨ碁｢晏､門ｯｹ逹逵溷ｮ櫁ｿ占｡檎噪譛榊苅蝎ｨ逕ｨ `curl` 襍ｰ莠ｸ驕榊驛ｨ譁ｰ蠅樊磁蜿｣亥性蜑埼擇謠仙芦逧悄蛻髯､ bug 蟆ｱ譏ｯ霑吩ｸ豁･謚灘芦逧ｼ峨474 荳ｪ豬玖ｯ募扈ｿ縲
+
+### 5.1 `v_task_detail`/`v_dashboard` 蠑蟋/證ょ●/蛛懈ｭ｢亥ｯｹ 5 闃ょ粥閭ｽ謇ｩ螻慕噪霑ｽ蜉
+
+逕ｨ謌ｷ螳樣刔菴ｿ逕ｨ譁ｰ蟒ｺ螂ｽ逧 `v_task_detail`/`v_dashboard` 鬘ｵ髱｢譌ｶ蜿醍鴫荳､螟豐｡譛"蠑蟋倶ｻｻ蜉｡"逧蜿｣窶披覗goal` 荳譌ｦ蛻帛ｻｺ形task.status` 蛛懷惠 `queued` 荳榊蜿伜喧梧ｲ｡譛我ｻｻ菴墓潔髓ｮ閭ｽ隶ｩ螳悄豁｣霍題ｵｷ譚･縲
+
+**霑ｽ貅ｯ荳ｺ莉荵 02-V/03-C 隶ｾ隶｡譌ｶ貍乗脂莠ｿ吩ｸｪ謖蛾聴**壻ｸ肴弍 UI 驕玲ｼ擾ｼ梧弍 03-C 閾ｪ蟾ｱ逧 `c_task_create` 螂醍ｺｦ03-C-謗ｧ蛻ｶ荳取磁蜿｣.md` 隨ｬ 42 陦 `side_effects`牙逧弍"闍･ `max_concurrent_goals` 蟾ｲ霎ｾ荳企剞御ｻｻ蜉｡關ｽ `queued` 菴ｸ崎ｧｦ蜿題ｰｺｦ"窶披泌渚霑擂隸ｻ梧ｲ｡蛻ｰ荳企剞譌ｶ**鮟倩ｮ､蟆ｱ蠎碑ｯ･隗ｦ蜿題ｰｺｦ**縲03-C/04-P 驥悟渚螟榊邇ｰ"Org Graph 隹ｺｦ蝎ｨ蟆 `task.assigned_org_node` 謖髄逧 Org Agent 蜚､襍ｷ"霑咏ｱｻ謠剰ｿｰ悟譁∞隶ｾ譛我ｸ荳ｪ蟶ｸ鬩ｻ霑占｡檎噪隹ｺｦ蝎ｨ霑帷ｨ具ｼ御ｻｻ蜉｡荳蛻帛ｻｺ蟆ｱ閾ｪ蜉ｨ陲ｫ螳込襍ｰ謇ｧ陦鯉ｼ悟屏豁､莉取悴隶ｾ隶｡霑ｸ荳ｪ"莠ｺ蟾･轤ｹ蜃ｻ蠑蟋"逧潔髓ｮ窶披03-C 閾ｪ蟾ｱ驛ｽ豐｡隶ｾ諠ｳ霑怙隕∝ｮゆｽ `src/scheduler.py`-032牙宵騾莠"莉取汾荳ｪ迥ｶ諤∵耳霑帛芦蠎"逧**蜃ｽ謨ｰ**御ｻ取悴陲ｫ莉ｻ菴募ｸｸ鬩ｻ霑帷ｨ区戟扈ｭ隹畑郁ｿ吩ｸ轤ｹ蝨ｨ U-032 蠖捺慮逧ｼ蜿題ｿｨ倶ｸｭ荵溷ｷｲ扈剰｢ｫ隶ｰ蠖戊ｿｸ谺｡瑚ｿ呎ｬ｡霑ｽ貅ｯ譏ｯ蜷御ｸ荳ｪ譬ｹ蝗蝨ｨ謗ｧ蛻ｶ蜿ｰ螻る擇逧峩謗･蜷取棡峨ゆｸ我ｸｪ逵溷ｮ櫁荳ｾ髣ｭ邇ｯsupervisor/self_hosting_e2e.py`芽霍鷹夲ｼ碁擒逧弍豬玖ｯ穂ｻ｣遐∵仞蠑剰ｰ畑 `scheduler.run_goal_to_completion`御ｸ肴弍霑吩ｸｪ"蟶ｸ鬩ｻ隹ｺｦ蝎ｨ"逵溽噪蟄伜惠縲
+
+**隶ｾ隶｡蜿冶**井ｸ守畑謌ｷ遑ｮ隶､蜷取牛譚ｿ御ｸ､螟騾我ｺ耳闕宣｡ｹ会ｼ
+1. **霓ｯ蛛懈ｭ｢御ｸ榊★逵溯ｿ帷ｨ kill**啻org_node_trigger.py`/`work_graph_execute.py` 驛ｽ譏ｯ髦ｻ蝪樒噪 `subprocess.run()` 隹 `claude`梧ｲ｡譛我ｻｻ菴募慍譁ｹ菫晏ｭ倩ｿ帷ｨ句唱譟ｻ"蛛懈ｭ｢"謫堺ｽ懷悉譚縲１ause/Stop 蜿ｪ諡ｦ謌ｪ"荳倶ｸ髦ｶ谿ｵ隕∽ｸ崎ｦ∬ｧｦ蜿"梧ｭ｣蝨ｨ霍醍噪驍｣谺｡逵溷ｮ櫁ｰ畑莨夊辟ｶ霍大ｮ後ら悄豁｣遑ｬ譚髴隕∵滑霑吩ｺ幄ｰ畑轤ｹ謐｢謌 `Popen` + 謖 `task_id` 蜿ｯ譟･逧 PID 陦ｨ瑚ｿ呎ｬ｡荳榊★縲
+2. **荳榊刈譁ｰ逧 `task.status` 譫壻ｸｾ蛟ｼ**壽嘯蛛/蛛懈ｭ｢蜷守噪莉ｻ蜉｡螟咲畑蟾ｲ譛臥噪 `status='blocked_pending_human'`碁擒 `blocked_reason='user_paused'`/`'user_stopped'` 蝨ｨ螳｡隶｡霓ｨ霑ｹ驥悟玄蛻ｼ亥粥閭ｽ荳贋ｸ､閠△螟肴婿蠑丞ｮ悟荳譬ｷ会ｼ檎峩謗･蛟溽畑蟾ｲ扈剰ｷ鷹夂噪 `manual_resume` 譛ｺ蛻ｶorg_loop_route._manual_resume`壻ｻ `execution_event` 驥梧怙蜷惹ｸ譚｡ `task_status_changed` 莠倶ｻｶ隸ｻ蜃ｺ證ょ●蜑咲噪迥ｶ諤∝ｹｶ霑伜次会ｼ御ｸ肴隼 01-M schema縲
+
+**螳樒鴫**壽眠蠅 `src/task_execution.py`窶披覗run_or_resume_task_lifecycle()` 謖 `task.status`queued`/`planning_brd_prd_sdd`/`coding`芽ｷｳ霑ｷｲ扈冗悄螳樔ｺｧ蜃ｺ artifact 逧亳谿ｵ悟宵驥肴眠隗ｦ蜿題ｿ俶ｲ｡霍醍噪髦ｶ谿ｵ嫣start_task_execution_in_background()` 襍ｷ荳荳ｪ迢ｬ遶 daemon 郤ｿ遞九∝ｼ閾ｪ蟾ｱ逧 sqlite 霑樊磁亥､咲畑 `scheduler.run_goal_to_completion` 蜷梧ｷ逧"豈冗ｺｿ遞倶ｸ荳ｪ霑樊磁"逅罰芽ｷ大芦蠎包ｼ梧蜉/螟ｱ雍･驛ｽ襍ｰ `scheduler.finish_task_lifecycle` 謾ｶ蟆ｾ瑚ｷ溯｢ｫ隹ｺｦ蝎ｨ豁｣蟶ｸ隹ｺｦ逧ｻｻ蜉｡關ｽ蛻ｰ蜷梧ｷ逧ｻ域ｼ嫣request_pause()` 蜿ｪ蜈∬ｮｸ莉 `planning_brd_prd_sdd`/`coding`育悄逧怏髦ｶ谿ｵ蝨ｨ鬟橸ｼ芽ｽｬ `blocked_pending_human`縲TTP 譁ｰ蠅 `POST /api/tasks/:id/execute`縲〜/pause`縲〜/stop`悟ｹｶ謚 `/route` 逧 `manual_resume` 蛻髪謇ｩ螻墓壼ｦよ棡證ょ●/蛛懈ｭ｢蜑咲噪 `blocked_reason` 譏ｯ `user_paused`/`user_stopped`檎憾諤∽ｸ譌ｦ陲ｫ霑伜次蛻ｰ `queued`/`planning_brd_prd_sdd`/`coding`悟ｰｱ閾ｪ蜉ｨ驥肴眠 spawn 蜷主床謇ｧ陦檎ｺｿ遞具ｼ瑚御ｸ榊宵譏ｯ謚顔憾諤∝ｭ玲ｮｵ鄙ｻ蝗槫悉窶披泌凄蛻咏畑謌ｷ轤ｹ"扈ｧ扈ｭ"莉･荳ｺ莉ｻ蜉｡莨夂ｻｧ扈ｭ霍托ｼ悟ｮ樣刔荳願ｿ倩ｦ∝轤ｹ荳谺｡"蠑蟋"御ｽ馴ｪ御ｸ顔ｭ我ｺ取ｲ｡菫ｮ縲らｳｻ扈溽ｺｧ髦ｻ蝪橸ｼloop_limit_reached`/`budget_exceeded`臥噪 `manual_resume` 陦御ｸｺ荳榊序御ｻ咲┯蜿ｪ鄙ｻ迥ｶ諤√Ａv_task_detail` 謖 `task.status` 荳蛾我ｸ譏ｾ遉ｺ"蠑蟋"/"證ょ●+蛛懈ｭ｢"/"扈ｧ扈ｭ謇ｧ陦"謖蛾聴嫣v_dashboard` 蜿ｪ蜉莠ｸ荳ｪ霓ｻ驥"蠑蟋"井ｸ埼怙隕 Pause/Stop 逧憾諤∝愛譁ｭ峨
+- **螳樒鴫霑ｨ倶ｸｭ蜿醍鴫蟷ｶ菫ｮ螟堺ｸ荳ｪ逵 gap**啻backend_server.create_server()` 譛ｬ霄ｫ莉取悴謚 `db_path` 蟄伜芦 `server` 蟇ｹ雎｡荳岩披比ｹ句燕謇譛芽ｯｷ豎ょ､炊蝎ｨ蜈ｱ逕ｨ `server.conn` 荳譚｡霑樊磁蜊ｳ蜿ｯ瑚ｿ呎ｬ｡譁ｰ蠅樒噪蜷主床謇ｧ陦檎ｺｿ遞矩怙隕∝ｼ閾ｪ蟾ｱ迢ｬ遶狗噪霑樊磁井ｸ崎霍ｨ郤ｿ遞句莠ｫ `server.conn`会ｼ悟書邇ｰ霑吩ｸｪ蟄玲ｮｵ譬ｹ譛ｬ荳榊ｭ伜惠縲り｡･荳 `server.db_path = resolved_db_path`縲
+- **鬪瑚ｯ∵婿蠑**啻src/test_task_execution.py`14 荳ｪ逕ｨ萓具ｼ瑚ｦ尠 resume-from-豈丈ｸｪ蜿ｯ諱｢螟咲憾諤√∵嘯蛛應ｿ｡蜿ｷ逧慮蠎冗ｫ樔ｺ峨〜request_pause` 逧憾諤∵｡鬪鯉ｼ峨〜src/test_backend_server.py` 譁ｰ蠅 `TaskExecutionRouteTest`8 荳ｪ逵溷ｮ HTTP 霍ｯ逕ｱ逕ｨ萓具ｼ形start_task_execution_in_background` 逕ｨ mock 髫皮ｦｻ窶披泌ｮｼ夂悄逧ｼ郤ｿ遞九∫悄逧ｯｹ `:memory:` 豬玖ｯ募ｺ灘ｼ荳譚｡迢ｬ遶玖ｿ樊磁莉手瑚ｿ槫芦蜿ｦ荳荳ｪ遨ｺ蠎難ｼ瑚ｿ咏ｱｻ譖ｴ蠎募ｱら噪譌ｶ蠎乗惻蛻ｶ蟾ｲ扈丞惠 `test_task_execution.py` 驥檎畑逵溷ｮ樔ｸｴ譌ｶ譁ｻｶ蠎馴ｪ瑚ｯ∬ｿｼ粂TTP 螻ょ宵鬪瑚ｯ∵磁郤ｿ夊ｰ畑蜿よ焚縲∫憾諤∫√〜manual_resume` 譏ｯ蜷ｦ謖 `blocked_reason` 豁｣遑ｮ蜀ｳ螳夊ｦ∽ｸ崎ｦ∬蜉ｨ扈ｭ霍托ｼ峨ょｯｹ逹逵溷ｮ櫁ｿ占｡檎噪譛榊苅蝎ｨ逕ｨ `curl` 襍ｰ莠ｸ驕 `/execute`磯撼 `queued` 莉ｻ蜉｡霑泌屓 409 `TASK_NOT_QUEUED`峨〜/pause`queued` 莉ｻ蜉｡霑泌屓 409 `TASK_NOT_PAUSABLE`嫣coding` 莉ｻ蜉｡謌仙粥霓ｬ `blocked_pending_human`/`user_paused`幃螟 pause 霑泌屓 409会ｼ御ｻ･蜿 `frontend/app.js` 螳樣刔荳句書逧潔髓ｮ HTML縲Ａsrc/` 蜈ｨ驥 496 荳ｪ豬玖ｯ募扈ｿ域眠蠅 22 荳ｪ14 + 8峨
@@ -598,12 +598,30 @@ function mountDashboard() {
       quotaBanner.style.display = "none";
       return;
     }
+    // v_dashboard Start affordance (2026-08-03): the full Pause/Stop set
+    // only makes sense on v_task_detail (they need the task's own status
+    // context to pick which one to show) -- the dashboard only ever needs
+    // "queued and nothing has started yet", so it gets just Start.
+    const startRowHtml = active.status !== "queued" ? "" : (
+      '<div class="row"><span></span><span>'
+      + '<button class="btn btn-primary" id="db_start_task" type="button">开始（execute）</button>'
+      + "</span></div>"
+    );
     taskSummary.innerHTML = (
       '<div class="row"><span>Goal</span><span>' + escapeHtml(active.goal_text) + "</span></div>"
       + '<div class="row"><span>Loop Round</span><span>' + active.loop_round + "</span></div>"
       + '<div class="row"><span>Status</span><span><a href="#/tasks/' + active.task_id + '">'
       + escapeHtml(active.status) + "</a></span></div>"
+      + startRowHtml
     );
+    const dbStartBtn = document.getElementById("db_start_task");
+    if (dbStartBtn) {
+      dbStartBtn.addEventListener("click", async () => {
+        dbStartBtn.disabled = true;
+        await fetch("/api/tasks/" + active.task_id + "/execute", { method: "POST" });
+        refreshTasksAndOrgGraph();
+      });
+    }
  
     const detailResp = await fetch("/api/tasks/" + active.task_id);
     const detail = await detailResp.json();
@@ -1031,7 +1049,27 @@ function mountTaskDetail(taskId) {
       return '<div class="org-node' + (activeSet.has(node) ? " active" : "") + '">' + node + "</div>" + arrow;
     }).join("");
  
-    const alertHtml = task.status !== "blocked_pending_human" ? "" : (
+    // v_task_detail Start/Pause/Stop (2026-08-03): a task only ever has one
+    // real thing "running" against it at a time -- exactly one of these
+    // three blocks is shown, keyed off task.status, matching the states
+    // task_execution.py itself understands (see its module docstring for
+    // why Pause/Stop are a soft stop reusing blocked_pending_human rather
+    // than a genuine subprocess kill / new status enum value).
+    const isUserPauseStop = task.status === "blocked_pending_human"
+      && (task.blocked_reason === "user_paused" || task.blocked_reason === "user_stopped");
+    const execControlsHtml = task.status === "queued" ? (
+      '<div class="btn-row" style="margin:0 0 16px;">'
+      + '<button class="btn btn-primary" id="btn_start_task" type="button">开始（execute）</button></div>'
+    ) : (task.status === "planning_brd_prd_sdd" || task.status === "coding") ? (
+      '<div class="btn-row" style="margin:0 0 16px;">'
+      + '<button class="btn" id="btn_pause_task" type="button">暂停（pause）</button>'
+      + '<button class="btn btn-danger" id="btn_stop_task" type="button">停止（stop）</button></div>'
+    ) : "";
+
+    const alertHtml = task.status !== "blocked_pending_human" ? "" : isUserPauseStop ? (
+      '<div class="alert-banner">' + (task.blocked_reason === "user_paused" ? "⏸ 已暂停" : "⏹ 已停止")
+      + '<button class="btn btn-primary" id="btn_manual_resume" type="button">继续执行</button></div>'
+    ) : (
       '<div class="alert-banner">⚠ 已阻塞（' + escapeHtml(task.blocked_reason || "") + '），需要人工干预'
       + '<button class="btn btn-danger" id="btn_manual_resume" type="button">人工干预（manual_resume）</button></div>'
     );
@@ -1083,6 +1121,7 @@ function mountTaskDetail(taskId) {
     content.innerHTML = (
       "<h1>" + escapeHtml(task.goal_text) + "</h1>"
       + goalCrudHtml
+      + execControlsHtml
       + alertHtml
       + '<div class="meta-row">'
       + "<div><b>STATUS</b>" + escapeHtml(task.status) + "</div>"
@@ -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) {
@@ -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=&quot;127.0.0.1&quot;, 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
+"""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}
@@ -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"
+"""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()