fix: 任务链卡死根治 — 心跳+stale回收+claimed_by僵尸锁清除
- questions.py: agent_ask/answer_question 置 waiting/submitted 时清 claimed_by，否则 poll 器永不重新认领 - agent_loop_v2.py: _t_answer_question 回答后恢复任务 waiting→submitted（此前只标记问题已答，任务永久卡 waiting） - agent_loop.py: _claim_task 加 claimed_by IS NULL 原子认领 + updated_at 心跳；role/pm 循环心跳 COMMIT；pm 认领保持 review 态 - init.py: role poller 回收僵尸 running 任务、pm poller 回收僵尸 review 任务（心跳超时 10 分钟）
This commit is contained in:
ymq
parent
d2f3caf8a7
commit
bc5ce2f06d
@ -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 "项目完成"}
|
||||
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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 "
|
||||
|
||||
@ -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}")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user