From 8c6e49992cc85ca7a72c4d5778ff9ca792d45b1b Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Tue, 18 Aug 2026 17:33:17 +0800
Subject: [PATCH] =?UTF-8?q?feat(worker):=20=E8=AE=A4=E9=A2=86=E5=8A=A0?=
 =?UTF-8?q?=E5=90=8C=E9=A1=B9=E7=9B=AE=E4=B8=B2=E8=A1=8C(NOT=20EXISTS=20?=
 =?UTF-8?q?=E6=97=A0=E5=90=8C=20tenant=20=E6=B4=BB=E8=B7=83=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1)=EF=BC=8C=E9=81=BF=E5=85=8D=E5=A4=9A=20worker=20?=
 =?UTF-8?q?=E5=B9=B6=E5=8F=91=E5=86=99=E5=90=8C=E4=B8=80=20workspace?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pipeline_service/agent_loop.py | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py
index 5c3d145..43fff54 100644
--- a/pipeline_service/agent_loop.py
+++ b/pipeline_service/agent_loop.py
@@ -525,16 +525,26 @@ async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True,
     from appPublic.uniqueID import getID
     claim_token = getID()
     # claimed_by IS NULL 保证原子认领（并发 poller / start_agents 不会双重认领）；
+    # NOT EXISTS 保证同项目(tenant_id)串行：同一项目同一时刻只允许一个任务处于
+    # 活跃状态（running / review·已认领 / qc_review·已认领），避免多 worker 并发写
+    # 同一 workspace。跨项目完全并发。derived table 包裹规避 MySQL 1093（同表子查询）。
     # updated_at=NOW() 作为心跳，供 stale 回收判断。
     await sor.sqlExe(
         "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})
+        "WHERE id=${tid}$ AND state=${state}$ AND claimed_by IS NULL "
+        "AND NOT EXISTS ("
+        "  SELECT 1 FROM (SELECT 1 FROM pipeline_tasks t2 "
+        "    WHERE t2.tenant_id=${tidx}$ AND t2.claimed_by IS NOT NULL "
+        "    AND t2.state IN ('running','review','qc_review')) AS z"
+        ")",
+        {"setstate": set_state, "cb": claim_token, "tid": task_id,
+         "state": state, "tidx": tenant_id})
     check = await sor.sqlExe(
         "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}")
+        # 并发抢输（task 已被别人认领）或同项目已有活跃任务被串行阻塞，均属正常，下轮再试
+        logger.info(f"claim skipped: task={task_id}")
         return None
     return task