diff --git a/pbl_evidence/collector.py b/pbl_evidence/collector.py
index 8b51fe2..ca5f019 100644
--- a/pbl_evidence/collector.py
+++ b/pbl_evidence/collector.py
@@ -22,6 +22,7 @@ from .evidence_map import (
     normalize_dt, now_str, first_present, map_event_to_evidence_type,
     build_dedup_key, event_source_id, summarize_payload,
 )
+from .pk import gen_pk
 
 # 运行时事件表候选名：M11b 落 pbl_runtime_event；历史/别名兼容 runtime_event。
 EVENT_TABLE_CANDIDATES = ('pbl_runtime_event', 'runtime_event', 'pbl_event')
@@ -227,6 +228,11 @@ async def collect_evidence_from_events(tenant_id,
             # artifact_id 为 NOT NULL：行为类证据（协作/反思/观测）无强归属产出物时以 0 占位，
             # 不因此丢事件——证据链完整性优先于归属字段（M6 评估侧按 artifact_id>0 过滤）。
             record.setdefault('artifact_id', 0)
+            # DISC.1 方案(a)：主键由应用层显式生成——models/pbl_evidence.json 把 id 定义为
+            # str(32) NOT NULL 且无 default（DDL 渲染 VARCHAR(32) NOT NULL，禁 AUTO_INCREMENT），
+            # DB 侧不生成主键，INSERT 必须自带 id。全模块唯一生成落点就在这里（record 构造处），
+            # _upsert_evidence 内不再调 gen_pk，只做回读赋值供 created_ids 使用。
+            record['id'] = gen_pk('ev')
 
             record = pick_columns(record, tgt_cols)
             stats['items'].append(record)
@@ -255,6 +261,9 @@ def first_present_col(cols, candidates):
 
 async def _upsert_evidence(record, tgt_cols, update_existing=False):
     """幂等写入：预查 + ON DUPLICATE KEY UPDATE 双保险。返回 created/skipped/updated。"""
+    # 主键必须在 record 构造处（collect_evidence_from_events）已由 gen_pk 生成——单一落点约定。
+    # 这里 fail-fast：任何绕过构造处的调用方（漏生成）立刻暴露，而不是把无主键的行交给 DB。
+    assert record.get('id'), 'record 缺少 id：主键必须在 record 构造处由 gen_pk 生成（DISC.1 方案(a)）'
     tid = record['tenant_id']
     sid = record['source_event_id']
     etp = record['evidence_type']