yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: PM _claim_task match_role=False to review all roles

Browse Source
This commit is contained in:
ymq 2026-08-08 11:15:03 +08:00
parent afe076ec5b
commit e76fa0fc47
1 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
pipeline_service/agent_loop.py
View File

@ -150,7 +150,7 @@ def _resolve_workspace(workspace_dir):
return 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。 """原子认领一条符合角色的任务。返回 task record 或 None。
认领策略： 认领策略：
@ -158,11 +158,17 @@ async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted'):
且排除有步骤记录的 DAG 任务（那些归 executor）。 且排除有步骤记录的 DAG 任务（那些归 executor）。
2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态
3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。 3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。
match_role=False 时不过滤角色（PM审核所有角色的review任务）。
""" """
role = _normalize_role(role) role = _normalize_role(role)
if match_role:
where_role = "AND (role=${role}$ OR role='')"
else:
where_role = ""
recs = await sor.sqlExe( recs = await sor.sqlExe(
"SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " "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) " "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) "
"ORDER BY created_at ASC LIMIT 1", "ORDER BY created_at ASC LIMIT 1",
{"tid": tenant_id, "state": state, "role": role}) {"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() db = _get_db()
async with db.sqlorContext("pipeline") as sor: async with db.sqlorContext("pipeline") as sor:
# 认领 review 状态的任务（不限角色，PM审核所有角色产出） # 认领 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: if not task:
return {"status": "idle", "message": "没有待审核任务"} return {"status": "idle", "message": "没有待审核任务"}