From e76fa0fc473f831377a0a832d6177deac15df403 Mon Sep 17 00:00:00 2001 From: ymq Date: Sat, 8 Aug 2026 11:15:03 +0800 Subject: [PATCH] fix: PM _claim_task match_role=False to review all roles --- pipeline_service/agent_loop.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 42d6caf..ea30a95 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -150,7 +150,7 @@ def _resolve_workspace(workspace_dir): return workspace_dir -async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted'): +async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted', match_role: bool = True): """原子认领一条符合角色的任务。返回 task record 或 None。 认领策略： @@ -158,11 +158,17 @@ async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted'): 且排除有步骤记录的 DAG 任务（那些归 executor）。 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态 3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。 + + match_role=False 时不过滤角色（PM审核所有角色的review任务）。 """ role = _normalize_role(role) + if match_role: + where_role = "AND (role=${role}$ OR role='')" + else: + where_role = "" recs = await sor.sqlExe( "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " - "WHERE tenant_id=${tid}$ AND state=${state}$ AND (role=${role}$ OR role='') " + "WHERE tenant_id=${tid}$ AND state=${state}$ " + where_role + " " "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) " "ORDER BY created_at ASC LIMIT 1", {"tid": tenant_id, "state": state, "role": role}) @@ -412,7 +418,7 @@ async def pm_review_run(project_id: str, agent_id: str = None, model_name: str = db = _get_db() async with db.sqlorContext("pipeline") as sor: # 认领 review 状态的任务（不限角色，PM审核所有角色产出） - task = await _claim_task(sor, project_id, 'pm', state=TASK_REVIEW) + task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False) if not task: return {"status": "idle", "message": "没有待审核任务"}