From 6f72e777dab3ac5f9945f1d0a5237da930fb1cc5 Mon Sep 17 00:00:00 2001
From: ymq <ymq@opencomputing.net>
Date: Tue, 1 Sep 2026 09:26:46 +0800
Subject: [PATCH] =?UTF-8?q?fix(agent):=20=E8=A7=92=E8=89=B2=20agent=20?=
 =?UTF-8?q?=E5=AE=9E=E4=BE=8B=20id=E2=80=94=E2=80=94created=5Fby=20?=
 =?UTF-8?q?=E5=AD=98=E7=9C=9F=E5=AE=9E=20id=20=E4=B8=8D=E5=86=8D=E6=8B=BC?=
 =?UTF-8?q?=E5=AD=97=E7=AC=A6=E4=B8=B2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- pipeline_agent_instances 表(项目×角色→稳定实例 ag.xxxx), 幂等注册
- agent/pm/qc 三 poller 派单改用实例 id(原 poller-{role} 拼接, 33字符撞32列宽→1406)
- DDL: 新表 + pipeline_deliverables.created_by 加宽64兜底
---
 pipeline_service/agent_instance.py | 73 ++++++++++++++++++++++++++++++
 pipeline_service/init.py           | 12 +++--
 2 files changed, 82 insertions(+), 3 deletions(-)
 create mode 100644 pipeline_service/agent_instance.py

diff --git a/pipeline_service/agent_instance.py b/pipeline_service/agent_instance.py
new file mode 100644
index 0000000..16da23f
--- /dev/null
+++ b/pipeline_service/agent_instance.py
@@ -0,0 +1,73 @@
+# -*- coding: utf-8 -*-
+"""角色 agent 实例注册（2026-09-01）：项目 × 角色 → 稳定实例 id。
+
+背景：poller 派单时 agent_id 是 `poller-{role}` 拼出来的字符串——
+1. 不是身份，审计/落库查不到「哪个 agent 实例干的」；
+2. `poller-agent.bid_writer_technical` 达 33 字符，超过
+   pipeline_deliverables.created_by varchar(32) → 1406 连环失败，
+   技术章节交付件全部落不了库（商务写者恰好 32 字符没事）。
+
+规则（角色规范名约定：处理方 = role + agentid）：
+- 每个项目 × 角色第一次派单时注册一个实例，id 形如 `ag.{12位}`（15 字符，
+  任何 32 宽字段都放得下），此后复用；
+- `created_by`/审计 `agent_id` 一律存实例 id；
+- pm/qc poller 同走此解析（角色 agent.pm / agent.qc）。
+"""
+
+import logging
+
+logger = logging.getLogger("pipeline.agent_instance")
+
+
+async def resolve_agent_instance(project_id, role):
+    """解析（必要时注册）项目 × 角色的 agent 实例 id。幂等。
+
+    失败兜底返回 `agent.{role}`（规范角色名，≤32 字符），保证调用方不炸。
+    """
+    role = (role or "").strip()
+    if not role:
+        return ""
+    fallback = ("agent." + role) if not role.startswith("agent.") else role
+    try:
+        from sqlor.dbpools import DBPools
+        from appPublic.uniqueID import getID
+        db = DBPools()
+        async with db.sqlorContext("pipeline") as sor:
+            recs = await sor.sqlExe(
+                "SELECT agent_id FROM pipeline_agent_instances "
+                "WHERE project_id=${p}$ AND role=${r}$ AND status='active' LIMIT 1",
+                {"p": project_id or "", "r": role})
+            await sor.sqlExe("COMMIT", {})
+            if recs:
+                aid = getattr(recs[0], "agent_id", "") or ""
+                if aid:
+                    return aid
+            aid = "ag." + getID()[:12]
+            await sor.C("pipeline_agent_instances", {
+                "id": getID(),
+                "project_id": project_id or "",
+                "role": role,
+                "agent_id": aid,
+                "status": "active",
+            })
+            await sor.sqlExe("COMMIT", {})
+            logger.info("agent instance registered: project=%s role=%s id=%s",
+                        project_id, role, aid)
+            return aid
+    except Exception as e:
+        # 并发注册撞唯一键 → 重查一次；仍失败用规范角色名兜底
+        try:
+            from sqlor.dbpools import DBPools
+            db = DBPools()
+            async with db.sqlorContext("pipeline") as sor:
+                recs = await sor.sqlExe(
+                    "SELECT agent_id FROM pipeline_agent_instances "
+                    "WHERE project_id=${p}$ AND role=${r}$ AND status='active' LIMIT 1",
+                    {"p": project_id or "", "r": role})
+                await sor.sqlExe("COMMIT", {})
+                if recs and (getattr(recs[0], "agent_id", "") or ""):
+                    return recs[0].agent_id
+        except Exception:
+            pass
+        logger.warning("resolve_agent_instance fallback role=%s err=%s", role, str(e)[:120])
+        return fallback[:32]
diff --git a/pipeline_service/init.py b/pipeline_service/init.py
index 4b51621..a9a0409 100644
--- a/pipeline_service/init.py
+++ b/pipeline_service/init.py
@@ -777,7 +777,9 @@ def load_pipeline_service():
 
                     async def _dispatch(tid, pid, r):
                         try:
-                            await role_agent_loop(pid, r, agent_id=f"poller-{r}")
+                            from .agent_instance import resolve_agent_instance
+                            _aid = await resolve_agent_instance(pid, r)
+                            await role_agent_loop(pid, r, agent_id=_aid)
                         except Exception as e:
                             debug(f"agent_poller dispatch error: task={tid} err={e}")
                         finally:
@@ -861,7 +863,9 @@ def load_pipeline_service():
 
                     async def _pm_dispatch(tid, pid):
                         try:
-                            await pm_review_run(pid, agent_id="pm-poller")
+                            from .agent_instance import resolve_agent_instance
+                            _aid = await resolve_agent_instance(pid, "agent.pm")
+                            await pm_review_run(pid, agent_id=_aid)
                         except Exception as e:
                             debug(f"pm_poller dispatch error: task={tid} err={e}")
                         finally:
@@ -912,7 +916,9 @@ def load_pipeline_service():
 
                         async def _qc_dispatch(tid, pid):
                             try:
-                                await qc_review_run(pid, agent_id="qc-poller")
+                                from .agent_instance import resolve_agent_instance
+                                _aid = await resolve_agent_instance(pid, "agent.qc")
+                                await qc_review_run(pid, agent_id=_aid)
                             except Exception as e:
                                 debug(f"qc_poller dispatch error: task={tid} err={e}")
                             finally: