diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index f95ec9e..0c8de8d 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -856,6 +856,31 @@ class AgentExecutor: for sr in (submitted_rows or []): lines.append(f" - [{getattr(sr, 'role', '?')}] {getattr(sr, 'title', '')} (id={getattr(sr, 'id', '')})") + # 运行中任务明细:claimed_by + 心跳陈旧度,帮助 agent 直接识别僵尸任务 + # (进程崩溃/协程挂起后心跳不再更新,超时即疑似僵尸,会被 poller 自动回收重跑) + if rc: + running_rows = await sor.sqlExe( + "SELECT id, title, role, claimed_by, " + "TIMESTAMPDIFF(MINUTE, updated_at, NOW()) AS mins " + "FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='running' " + "ORDER BY updated_at ASC LIMIT 10", + {"pid": pid}) + lines.append("运行中任务:") + for rr in (running_rows or []): + rrole = getattr(rr, 'role', '?') + rtitle = getattr(rr, 'title', '') + rid = getattr(rr, 'id', '') + rcb = getattr(rr, 'claimed_by', '') or '' + try: + mins = int(getattr(rr, 'mins', 0) or 0) + except (TypeError, ValueError): + mins = 0 + if mins >= 10: + flag = f"⚠️心跳超时{mins}分钟(疑似僵尸,将被回收重跑)" + else: + flag = f"已运行{mins}分钟" + lines.append(f" - [{rrole}] {rtitle} (id={rid}) | {flag} | claimed_by={rcb[:8]}") + return "\n".join(lines) async def _t_list_deliverables(self, sor, p, pid):