From 57f651fd79496b27717de6edfe647ba2a5037a49 Mon Sep 17 00:00:00 2001
From: "agent.develop" <agent.develop@pipeline.local>
Date: Tue, 22 Sep 2026 22:06:04 +0800
Subject: [PATCH] =?UTF-8?q?fix(pbl=5Fevidence):=20api.py=20=E4=B8=A4?=
 =?UTF-8?q?=E6=9D=A1=E6=8F=92=E5=85=A5=E8=B7=AF=E5=BE=84=E6=98=BE=E5=BC=8F?=
 =?UTF-8?q?=E8=90=BD=E5=9C=B0=E4=B8=BB=E9=94=AE=20gen=5Fpk?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- import 区加 from pbl_evidence.pk import gen_pk（与 db/collector 同风格）
- pbl_artifact_create: pick_columns 前 record['id'] = gen_pk('at')（build_insert 路径）
- pbl_evidence_collect: pick_columns 前 record['id'] = gen_pk('ev')（INSERT INTO pbl_evidence 路径）
- 每条插入路径恰好一次 gen_pk，无兜底分支；ON DUPLICATE KEY UPDATE 列集合排除 id，
  避免竞态命中唯一索引时改写已存在行主键
- 响应 QC#3（上次 api 替换为 no-op）/QC#4（直接改源文件）/QC#6（单一落点）

对应 pk.py 自述契约：gen_pk 被 collector/api 每一条插入路径调用。
---
 pbl_evidence/api.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/pbl_evidence/api.py b/pbl_evidence/api.py
index 201e502..dbe4786 100644
--- a/pbl_evidence/api.py
+++ b/pbl_evidence/api.py
@@ -25,6 +25,7 @@ import hashlib
 
 from pbl_common.api import PblError, actor_id, json_dump, now_str, tenant_id
 
+from pbl_evidence.pk import gen_pk
 from pbl_evidence.db import (
     build_insert, build_where, invalidate_cache, json_loads, pick_columns,
     q_all, q_exec, q_one, table_columns,
@@ -98,6 +99,7 @@ async def pbl_artifact_create(**kw):
         if st:
             record['submitted_at'] = st
 
+    record['id'] = gen_pk('at')
     record = pick_columns(record, cols)
     if 'title' not in record:
         raise PblError('PBL_E_DB', 'pbl_artifact 缺 title 列，表结构与模型不一致')
@@ -273,12 +275,15 @@ async def pbl_evidence_collect(**kw):
         'created_at': now_str(),
         'updated_at': now_str(),
     }
+    record['id'] = gen_pk('ev')
     record = pick_columns(record, cols)
     ins_cols = [c for c in record if c in cols]
     col_sql = ', '.join('`%s`' % c for c in ins_cols)
     val_sql = ', '.join('${%s}$' % c for c in ins_cols)
-    dup_cols = [c for c in ins_cols if c not in ('tenant_id', 'source_event_id',
-                                                 'evidence_type', 'dedup_key', 'created_at')]
+    # 'id' 必须排除：主键在插入前由 pk 模块一次性生成（单一落点），
+    # 命中唯一索引走 ON DUPLICATE 时不能把已存在行的主键改写成本次新生成的值。
+    dup_cols = [c for c in ins_cols if c not in ('id', 'tenant_id', 'source_event_id',
+                                                'evidence_type', 'dedup_key', 'created_at')]
     sql = 'INSERT INTO `pbl_evidence` (%s) VALUES (%s)' % (col_sql, val_sql)
     if dup_cols:
         sql += ' ON DUPLICATE KEY UPDATE ' + ', '.join('`%s`=VALUES(`%s`)' % (c, c) for c in dup_cols)