diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index d489909..1076e39 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -661,7 +661,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None): try: resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4) except Exception as e: - await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) + err_msg = f"{type(e).__name__}: {str(e)[:400]}" + await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id}) logger.error(f"role_agent_run llm failed: task={task_id} err={e}") return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} @@ -842,7 +843,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): try: raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3) except Exception as e: - await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) + err_msg = f"{type(e).__name__}: {str(e)[:400]}" + await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id}) return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} act = _parse_agent_action(raw) @@ -903,6 +905,78 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} +# ── 失败任务处理（第二层：PM/cockpit 判定重跑，最多3次，超限报故障给用户）── + +_RETRY_HINTS = ( + "timeout", "timed out", "connection", "connect", "network", "unreachable", + "rate limit", "429", "500", "502", "503", "504", "reset", "broken pipe", + "eof", "temporar", "busy", "overloaded", +) + + +def _classify_failure(last_error: str) -> str: + """按失败原因分类：retry(瞬时，可重跑) / fault(永久，报故障)。 + + 空错误通常是 asyncio.TimeoutError（str 为空），视为瞬时可重跑。 + """ + err = (last_error or "").strip().lower() + if not err: + return "retry" + for h in _RETRY_HINTS: + if h in err: + return "retry" + return "fault" + + +async def handle_failed_task(task_id: str, project_id: str) -> dict: + """处理一个失败任务：判定重跑或报故障。由 failed poller 周期调用。 + + - retry_count < 3 且失败原因判定为瞬时 → 重跑(retry_count+1, 回 submitted) + - retry_count >= 3 或判定为永久错误 → 报故障(建问题通知用户，任务置 waiting) + """ + db = _get_db() + async with db.sqlorContext("pipeline") as sor: + recs = await sor.sqlExe( + "SELECT id, title, role, retry_count, last_error FROM pipeline_tasks " + "WHERE id=${tid}$ AND state='failed'", + {"tid": task_id}) + if not recs: + return {"status": "idle", "task_id": task_id} + t = recs[0] + title = getattr(t, "title", "") or "" + role = getattr(t, "role", "") or "" + try: + retry_count = int(getattr(t, "retry_count", 0) or 0) + except (TypeError, ValueError): + retry_count = 0 + last_error = getattr(t, "last_error", "") or "" + + # 已重试 3 次仍未成功 → 报故障 + if retry_count >= 3: + decision = "fault" + else: + decision = _classify_failure(last_error) + + if decision == "retry": + await sor.sqlExe( + "UPDATE pipeline_tasks SET retry_count=retry_count+1, state='submitted', " + "claimed_by=NULL, last_error=NULL, updated_at=NOW() " + "WHERE id=${tid}$ AND state='failed'", + {"tid": task_id}) + logger.info("failed task retry: task=%s role=%s attempt=%d", task_id, role, retry_count + 1) + return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1} + + # 报故障：建问题通知用户，任务置 waiting 等待人工介入 + from .questions import agent_ask + reporter = "cockpit" if role == "pm" else "pm" + qid = await agent_ask( + project_id, task_id, reporter, + f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}", + context={"fault": True, "last_error": last_error, "retry_count": retry_count}) + logger.info("failed task fault: task=%s role=%s qid=%s", task_id, role, qid) + return {"status": "fault", "task_id": task_id, "question_id": qid} + + async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10): results = [] for _ in range(max_iterations): diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 8c81e8b..9f49193 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -38,7 +38,7 @@ from .questions import ( get_question, list_questions, get_task_qna, ) from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop -from .agent_loop import pm_review_run, pm_review_loop +from .agent_loop import pm_review_run, pm_review_loop, handle_failed_task MODULE_NAME = "pipeline_service" MODULE_VERSION = "3.4.0" @@ -632,6 +632,46 @@ def load_pipeline_service(): add_startup(_pm_poller) + # Failed task poller: 失败任务自动重跑(最多3次)，超限报故障给用户 + async def _failed_poller(app): + from sqlor.dbpools import DBPools as _DBP3 + fd_db = _DBP3() + _fd_dispatched = set() + + async def _fd_poll_loop(): + while True: + try: + async with fd_db.sqlorContext("pipeline") as sor: + recs = await sor.sqlExe( + "SELECT id, tenant_id FROM pipeline_tasks " + "WHERE state='failed' AND pipeline_id='role_task' " + "ORDER BY created_at ASC LIMIT 10", + {}) + for rec in (recs or []): + tid = getattr(rec, 'id', '') + pid = getattr(rec, 'tenant_id', '') + if tid and pid and tid not in _fd_dispatched: + _fd_dispatched.add(tid) + debug(f"failed_poller handling task={tid}") + + async def _fd_dispatch(tid, pid): + try: + await handle_failed_task(tid, pid) + except Exception as e: + debug(f"failed_poller dispatch error: task={tid} err={e}") + finally: + _fd_dispatched.discard(tid) + + asyncio.ensure_future(_fd_dispatch(tid, pid)) + except Exception as e: + debug(f"failed_poller error: {e}") + await asyncio.sleep(60) + + asyncio.create_task(_fd_poll_loop()) + debug("failed poller started") + + add_startup(_failed_poller) + # v2 DDL auto-init: create tables and columns at startup async def _init_v2_tables(app): from sqlor.dbpools import DBPools as _DBP2 @@ -686,6 +726,17 @@ def load_pipeline_service(): _debug("sd_org_settings.agent_config column added") except Exception: pass + # failed 任务重试计数 + 失败原因 + try: + await sor.sqlExe("ALTER TABLE pipeline_tasks ADD COLUMN retry_count int NOT NULL DEFAULT 0", {}) + _debug("pipeline_tasks.retry_count column added") + except Exception: + pass + try: + await sor.sqlExe("ALTER TABLE pipeline_tasks ADD COLUMN last_error text", {}) + _debug("pipeline_tasks.last_error column added") + except Exception: + pass except Exception as e: _debug(f"v2 DDL init: {e}")