diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py
index 39faa9a..d489909 100644
--- a/pipeline_service/agent_loop.py
+++ b/pipeline_service/agent_loop.py
@@ -328,7 +328,7 @@ __REPO_STATE__
 
 # ── Agent 核心逻辑 ──
 
-async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True):
+async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True, set_state='running'):
     role = _normalize_role(role)
     where_role = "AND (role=${role}$ OR role='')" if match_role else ""
     recs = await sor.sqlExe(
@@ -343,13 +343,15 @@ async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True):
     task_id = task.id
     from appPublic.uniqueID import getID
     claim_token = getID()
+    # claimed_by IS NULL 保证原子认领（并发 poller / start_agents 不会双重认领）；
+    # updated_at=NOW() 作为心跳，供 stale 回收判断。
     await sor.sqlExe(
-        "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ "
-        "WHERE id=${tid}$ AND state=${state}$",
-        {"cb": claim_token, "tid": task_id, "state": state})
+        "UPDATE pipeline_tasks SET state=${setstate}$, claimed_by=${cb}$, updated_at=NOW() "
+        "WHERE id=${tid}$ AND state=${state}$ AND claimed_by IS NULL",
+        {"setstate": set_state, "cb": claim_token, "tid": task_id, "state": state})
     check = await sor.sqlExe(
-        "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$",
-        {"tid": task_id, "cb": claim_token})
+        "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state=${setstate}$ AND claimed_by=${cb}$",
+        {"tid": task_id, "setstate": set_state, "cb": claim_token})
     if not check:
         logger.info(f"claim lost race: task={task_id}")
         return None
@@ -650,6 +652,12 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
 
         # ── Tool Loop（原生 function calling）──
         for turn in range(30):
+            # 心跳：标记任务仍在执行，供 stale 回收判断（进程崩溃后任务不再被 touch 即被回收）。
+            # 必须 COMMIT 让心跳对其他连接可见，否则 poller 看不到心跳会误判为僵尸。
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET updated_at=NOW() WHERE id=${tid}$ AND state='running'",
+                {"tid": task_id})
+            await sor.sqlExe("COMMIT", {})
             try:
                 resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4)
             except Exception as e:
@@ -786,7 +794,7 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
 async def pm_review_run(project_id, agent_id=None, model_name=None):
     db = _get_db()
     async with db.sqlorContext("pipeline") as sor:
-        task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False)
+        task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False, set_state='review')
         if not task:
             return {"status": "idle", "message": "没有待审核任务"}
 
@@ -826,6 +834,11 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
 
         decision = None
         for turn in range(5):
+            # 心跳：PM 审核期间持续标记，进程崩溃后由 stale 回收重置；COMMIT 使其对其他连接可见
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET updated_at=NOW() WHERE id=${tid}$ AND state='review'",
+                {"tid": task_id})
+            await sor.sqlExe("COMMIT", {})
             try:
                 raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3)
             except Exception as e:
@@ -870,7 +883,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
                 "quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm",
             })
             await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id})
-            await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
+            await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
             next_role = await _get_next_role(task_role)
             if next_role:
                 next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
@@ -886,7 +899,7 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
             return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}
 
         else:
-            await sor.sqlExe("UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id})
+            await sor.sqlExe("UPDATE pipeline_tasks SET state='completed', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id})
             return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
 
 
diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py
index f814066..f95ec9e 100644
--- a/pipeline_service/agent_loop_v2.py
+++ b/pipeline_service/agent_loop_v2.py
@@ -930,19 +930,33 @@ class AgentExecutor:
 
         # 先精确匹配，再前缀匹配兜底（兼容截断 ID）
         recs = await sor.sqlExe(
-            "SELECT id FROM pipeline_agent_questions WHERE id=${qid}$", {"qid": qid})
+            "SELECT id, task_id FROM pipeline_agent_questions WHERE id=${qid}$", {"qid": qid})
         if not recs and len(qid) >= 6:
             recs = await sor.sqlExe(
-                "SELECT id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1",
+                "SELECT id, task_id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1",
                 {"prefix": qid + "%"})
         if not recs:
             return f"问题不存在: {qid}"
-        qid = getattr(recs[0], "id", qid)
+        full_qid = getattr(recs[0], "id", qid)
+        task_id = getattr(recs[0], "task_id", "") or ""
 
         await sor.sqlExe(
-            "UPDATE pipeline_agent_questions SET answer=${a}$, status='answered' WHERE id=${qid}$",
-            {"a": answer, "qid": qid})
-        return "OK: 已回答"
+            "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', "
+            "answered_by='main_agent', status='answered' WHERE id=${qid}$",
+            {"a": answer, "qid": full_qid})
+
+        # 恢复任务：waiting → submitted，并清 claimed_by。
+        # 否则任务永久卡在 waiting（poll 器要求 state='submitted' AND claimed_by IS NULL 才会重新认领）。
+        resumed = False
+        if task_id:
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL "
+                "WHERE id=${tid}$ AND state='waiting'",
+                {"tid": task_id})
+            resumed = True
+
+        suffix = "，任务已恢复执行" if resumed else ""
+        return "OK: 已回答" + suffix
 
     async def _t_add_repo(self, sor, p, pid):
         if not pid:
diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 359cccd..73f7da1 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -549,6 +549,12 @@ def load_pipeline_service():
             while True:
                 try:
                     async with poll_db.sqlorContext("pipeline") as sor:
+                        # 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时 10 分钟未更新）。
+                        # 否则任务永久卡 running，后续任务链断裂。
+                        await sor.sqlExe(
+                            "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() "
+                            "WHERE state='running' AND pipeline_id='role_task' "
+                            "AND updated_at < (NOW() - INTERVAL 10 MINUTE)", {})
                         recs = await sor.sqlExe(
                             "SELECT id, tenant_id, role FROM pipeline_tasks "
                             "WHERE state='submitted' AND pipeline_id='role_task' "
@@ -591,6 +597,11 @@ def load_pipeline_service():
             while True:
                 try:
                     async with pm_db.sqlorContext("pipeline") as sor:
+                        # 回收僵尸 review 任务：PM 审核进程崩溃后 claimed_by 残留，重新放回审核队列。
+                        await sor.sqlExe(
+                            "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() "
+                            "WHERE state='review' AND claimed_by IS NOT NULL "
+                            "AND updated_at < (NOW() - INTERVAL 10 MINUTE)", {})
                         recs = await sor.sqlExe(
                             "SELECT id, tenant_id, role FROM pipeline_tasks "
                             "WHERE state='review' AND claimed_by IS NULL "
diff --git a/pipeline_service/questions.py b/pipeline_service/questions.py
index 611cf0a..1739351 100644
--- a/pipeline_service/questions.py
+++ b/pipeline_service/questions.py
@@ -50,7 +50,10 @@ async def agent_ask(tenant_id: str, task_id: str, from_role: str, question: str,
             "status": Q_PENDING,
         })
         if task_id:
-            await sor.U('pipeline_tasks', {"id": task_id, "state": "waiting"})
+            # 置 waiting 同时清 claimed_by，否则任务僵尸（poll器要求 claimed_by IS NULL 才会重新认领）
+            await sor.sqlExe(
+                "UPDATE pipeline_tasks SET state='waiting', claimed_by=NULL WHERE id=${tid}$",
+                {"tid": task_id})
         logger.info(f"agent_ask: task={task_id} role={from_role} qid={qid}")
         return qid
 
@@ -89,7 +92,10 @@ async def answer_question(question_id: str, answer: str, answered_by: str = "",
                 t = trecs[0]
                 tstate = t.state if hasattr(t, 'state') else t['state']
                 if tstate == "waiting":
-                    await sor.U('pipeline_tasks', {"id": task_id, "state": "submitted"})
+                    # 恢复 submitted 同时清 claimed_by，否则 poll 器永不重新认领该任务
+                    await sor.sqlExe(
+                        "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$",
+                        {"tid": task_id})
                     resumed = True
 
         logger.info(f"answer_question: qid={question_id} task={task_id} source={answer_source} resumed={resumed}")