From 886db46b9fada33d581782a921e83d6992814cae Mon Sep 17 00:00:00 2001
From: "agent.develop" <agent.develop@hrs6.local>
Date: Tue, 22 Sep 2026 17:20:20 +0800
Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?=
 =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tests/m5a_live_db_harness.py | 448 +++++++++++++++++++++++++++++++++++
 1 file changed, 448 insertions(+)
 create mode 100644 tests/m5a_live_db_harness.py

diff --git a/tests/m5a_live_db_harness.py b/tests/m5a_live_db_harness.py
new file mode 100644
index 0000000..e5e295f
--- /dev/null
+++ b/tests/m5a_live_db_harness.py
@@ -0,0 +1,448 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""M5a 幂等 / C-3 fail-closed **真实库**验收 harness（任务 u7RBcE_UjcEOkfhw6WRlV 产出）。
+
+与既有 ``tests/test_m5a_idempotency.py``（离线、无 DB、只断 fail-closed 与纯函数）
+的分工：本 harness 是**增量证据**——它在一次性沙箱库里建出
+``pbl_runtime_event`` + ``pbl_evidence`` 两张真实表，跑**同一条采集路径两次**，
+用 MariaDB 的 ``uk_ev_dedup`` 唯一索引与 ``INSERT ... ON DUPLICATE KEY UPDATE``
+实测幂等，并把 DB 原样回显（SHOW INDEX / 1062 Duplicate entry）打进日志。
+
+被测代码是**真源码**（``pbl_evidence/collector.py``），不是复刻逻辑：
+只把最底层数据访问适配层 ``db.q_all/q_one/q_exec/table_exists/table_columns``
+替换成 pymysql 直连沙箱库的等价实现（sqlor 的 ``${name}$`` 占位符 → pymysql
+``%(name)s``），``collector`` 的扫描/映射/去重/幂等写入逻辑一行未改。
+宿主包（sqlor/appPublic/ahserver）在本机未安装，故用 importlib 造假包绕过
+``pbl_evidence/__init__.py``，与离线测试同一手法。
+
+用法::
+
+    python3 tests/m5a_live_db_harness.py            # 跑完自动清理沙箱库
+    python3 tests/m5a_live_db_harness.py --keep     # 保留沙箱库供人工核对
+    PBL_M5A_DB=host:port:user:pwd 覆盖默认 test/test123@127.0.0.1:3306
+
+退出码：0 = 全部断言通过（stdout 末行 ``LIVE DB HARNESS ALL PASS``）；1 = 失败。
+"""
+
+import argparse
+import contextlib
+import importlib
+import io
+import os
+import pathlib
+import re
+import sys
+import traceback
+import types
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent      # modules/pbl_evidence
+PKG_DIR = REPO_ROOT / 'pbl_evidence'
+
+SANDBOX_SCHEMA = os.environ.get('PBL_M5A_SANDBOX', 'pbl_m5a_u7rb')
+_TENANT = 't1'
+_EVENT_ID_A = 'evt_u7rb_0001'
+_EVENT_ID_B = 'evt_u7rb_0002'
+
+# 沙箱 DDL：列集合与 models/pbl_evidence.json 一致（MariaDB 方言，本机实测可建）
+DDL_PBL_EVIDENCE = """
+CREATE TABLE `pbl_evidence` (
+  `id` VARCHAR(32) NOT NULL,
+  `tenant_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `artifact_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `evidence_type` VARCHAR(64) NOT NULL DEFAULT 'observation',
+  `source_event_id` VARCHAR(32) NOT NULL,
+  `session_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `learner_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `blueprint_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `payload_json` LONGTEXT NULL,
+  `occurred_at` DATETIME NOT NULL,
+  `dedup_key` VARCHAR(32) NOT NULL DEFAULT '',
+  `collect_batch_id` VARCHAR(32) NOT NULL DEFAULT '',
+  `created_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00',
+  `updated_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_ev_dedup` (`tenant_id`, `source_event_id`, `evidence_type`),
+  KEY `idx_ev_learner` (`tenant_id`, `learner_id`, `occurred_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
+"""
+
+DDL_PBL_RUNTIME_EVENT = """
+CREATE TABLE `pbl_runtime_event` (
+  `event_id` VARCHAR(64) NOT NULL,
+  `tenant_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `event_type` VARCHAR(64) NOT NULL,
+  `payload_json` LONGTEXT NULL,
+  `occurred_at` DATETIME NOT NULL,
+  `learner_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `session_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `blueprint_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `artifact_id` VARCHAR(32) NOT NULL DEFAULT '0',
+  `created_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00',
+  PRIMARY KEY (`event_id`),
+  KEY `idx_re_time` (`tenant_id`, `occurred_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
+"""
+
+EVENTS_SEED = (
+    (_EVENT_ID_A, 'artifact_submitted',
+     '{"title":"bridge-design","score":88}', '2026-09-22 10:00:00', 'stu_01', 'sess_01', 'bp_01', 'art_01'),
+    (_EVENT_ID_B, 'team_message_sent',
+     '{"channel":"team","chars":120}', '2026-09-22 10:05:00', 'stu_01', 'sess_01', 'bp_01', '0'),
+)
+
+_failures = []
+
+
+def check(name, ok, detail=''):
+    """记录一条断言并原样打印（PASS/FAIL），失败不中断，最后统一 exit 1。"""
+    print('[%s] %s%s' % ('PASS' if ok else 'FAIL', name,
+                         ('  -> ' + detail) if detail else ''))
+    if not ok:
+        _failures.append(name)
+    return ok
+
+
+# ── 被测代码隔离加载（不触发宿主依赖）────────────────────────────────
+def load_collector():
+    if 'pbl_evidence' not in sys.modules:
+        pkg = types.ModuleType('pbl_evidence')
+        pkg.__path__ = [str(PKG_DIR)]
+        pkg.__package__ = 'pbl_evidence'
+        sys.modules['pbl_evidence'] = pkg
+    return importlib.import_module('pbl_evidence.collector')
+
+
+# ── 沙箱库 + db 适配层替身 ───────────────────────────────────────────
+class Sandbox:
+    """一次性沙箱库；把 collector 的数据访问函数换成 pymysql 直连实现。"""
+
+    def __init__(self, conn_kwargs):
+        self.conn_kwargs = dict(conn_kwargs)
+        self.conn = None
+
+    def connect_root(self):
+        import pymysql
+        kw = dict(self.conn_kwargs)
+        kw.pop('database', None)
+        return pymysql.connect(**kw)
+
+    def reset_schema(self):
+        conn = self.connect_root()
+        try:
+            with conn.cursor() as cur:
+                cur.execute('DROP DATABASE IF EXISTS `%s`' % SANDBOX_SCHEMA)
+                cur.execute('CREATE DATABASE `%s` CHARACTER SET utf8mb4 '
+                            'COLLATE utf8mb4_general_ci' % SANDBOX_SCHEMA)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def drop_schema(self):
+        conn = self.connect_root()
+        try:
+            with conn.cursor() as cur:
+                cur.execute('DROP DATABASE IF EXISTS `%s`' % SANDBOX_SCHEMA)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def open(self):
+        import pymysql
+        kw = dict(self.conn_kwargs)
+        kw['database'] = SANDBOX_SCHEMA
+        kw['cursorclass'] = pymysql.cursors.DictCursor
+        kw['autocommit'] = True
+        self.conn = pymysql.connect(**kw)
+
+    def close(self):
+        if self.conn is not None:
+            try:
+                self.conn.close()
+            except Exception:
+                pass
+            self.conn = None
+
+    def ddl(self, stmt):
+        with self.conn.cursor() as cur:
+            cur.execute(stmt)
+
+    def raw(self, sql, args=None):
+        """执行并返回 list[dict]（原样回显用）。"""
+        with self.conn.cursor() as cur:
+            cur.execute(sql, args or {})
+            try:
+                return list(cur.fetchall() or [])
+            except Exception:
+                return []
+
+    # ── collector.db 替身：${name}$ → %(name)s ──
+    @staticmethod
+    def to_pymysql(sql):
+        return re.sub(r'\$\{(\w+)\}\$', r'%(\1)s', sql)
+
+    def q_all(self, sql, params=None):
+        with self.conn.cursor() as cur:
+            cur.execute(self.to_pymysql(sql), dict(params or {}))
+            return [dict(r) for r in (cur.fetchall() or [])]
+
+    def q_one(self, sql, params=None):
+        rows = self.q_all(sql, params)
+        return rows[0] if rows else None
+
+    def q_exec(self, sql, params=None):
+        with self.conn.cursor() as cur:
+            n = cur.execute(self.to_pymysql(sql), dict(params or {}))
+        return {'affected_rows': n}
+
+    def table_exists(self, table):
+        row = self.q_one(
+            'SELECT COUNT(*) AS c FROM information_schema.TABLES '
+            'WHERE TABLE_SCHEMA=${db}$ AND TABLE_NAME=${t}$',
+            {'db': SANDBOX_SCHEMA, 't': table})
+        return bool(row) and int(list(row.values())[0] or 0) > 0
+
+    def table_columns(self, table):
+        rows = self.q_all(
+            'SELECT COLUMN_NAME AS col FROM information_schema.COLUMNS '
+            'WHERE TABLE_SCHEMA=${db}$ AND TABLE_NAME=${t}$',
+            {'db': SANDBOX_SCHEMA, 't': table})
+        return {str(r.get('col')) for r in rows if r.get('col')}
+
+    def get_dbname(self):
+        return SANDBOX_SCHEMA
+
+
+def install_stub(collector, box):
+    """把 collector 模块命名空间里的数据访问函数指向沙箱实现（真源码其余不动）。
+
+    collector 里这些调用都是 ``await``，故同步的 pymysql 实现要包成协程；
+    ``get_dbname`` 在真源码里是同步函数，保持同步。
+    """
+    def _async(fn):
+        async def wrapper(*a, **kw):
+            return fn(*a, **kw)
+        wrapper.__name__ = getattr(fn, '__name__', 'stub')
+        return wrapper
+
+    for attr, fn in (('q_all', box.q_all), ('q_one', box.q_one), ('q_exec', box.q_exec),
+                     ('table_exists', box.table_exists), ('table_columns', box.table_columns),
+                     ('invalidate_cache', lambda t=None: None)):
+        setattr(collector, attr, _async(fn))
+    setattr(collector, 'get_dbname', box.get_dbname)
+
+
+# ── 阶段 1：C-3 fail-closed（事件表缺失）────────────────────────────
+def phase_fail_closed(collector, box):
+    print('\n--- 阶段 1：C-3 fail-closed（pbl_runtime_event 缺失）---')
+    print('沙箱库 %s 内表清单（原样回显）: %s'
+          % (SANDBOX_SCHEMA, box.raw('SHOW TABLES') or '[] （空库）'))
+    resolved = None
+    import asyncio
+    resolved = asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
+        collector.resolve_event_table())
+    print('resolve_event_table() 实测返回: %r' % (resolved,))
+    check('C-3.1 resolve_event_table() 在事件表缺失时返回 None', resolved is None,
+          repr(resolved))
+
+    err = None
+    tb_text = ''
+    import asyncio
+    try:
+        asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
+            collector.collect_evidence_from_events(tenant_id=_TENANT))
+    except collector.CollectError as exc:      # 真源码抛出的类型
+        err = exc
+        tb_text = ''.join(traceback.format_exc())
+    except Exception as exc:                  # noqa: BLE001
+        err = exc
+        tb_text = ''.join(traceback.format_exc())
+
+    print('抛出的异常类型: %s' % (type(err).__name__ if err else 'None（未抛错=静默空成功）'))
+    print('异常消息原文: %s' % (str(err) if err else '<无>'))
+    if tb_text:
+        print('traceback 原文:')
+        for line in tb_text.rstrip().splitlines():
+            print('  | ' + line)
+    check('C-3.2 collect_evidence_from_events(tenant_id="t1") 抛 CollectError',
+          isinstance(err, collector.CollectError), type(err).__name__ if err else 'no raise')
+    msg = str(err or '')
+    check("C-3.3 异常消息含 'M11b'（硬依赖声明，非空成功）", 'M11b' in msg,
+          '消息=%s' % msg[:120])
+    check('C-3.4 未产生任何证据行（fail-closed 不写脏数据）',
+          box.table_exists('pbl_evidence') is False or
+          int((box.raw('SELECT COUNT(*) AS c FROM `pbl_evidence`') or [{'c': 0}])[0]['c']) == 0)
+
+
+# ── 阶段 2：真实双次采集幂等 ────────────────────────────────────────
+def phase_idempotent(collector, box):
+    print('\n--- 阶段 2：同 source_event_id + evidence_type 连续采集两次（真库）---')
+    box.ddl(DDL_PBL_RUNTIME_EVENT)
+    box.ddl(DDL_PBL_EVIDENCE)
+    print('建表后 SHOW TABLES: %s'
+          % [list(r.values())[0] for r in box.raw('SHOW TABLES')])
+    print('uk_ev_dedup 索引定义（SHOW INDEX 原样回显）:')
+    for r in box.raw("SHOW INDEX FROM `pbl_evidence` WHERE Key_name='uk_ev_dedup'"):
+        print('  Seq_in_index={0} Column_name={1} Non_unique={2} Index_name={3}'
+              .format(r.get('Seq_in_index'), r.get('Column_name'),
+                      r.get('Non_unique'), r.get('Key_name')))
+
+    with box.conn.cursor() as cur:
+        for ev in EVENTS_SEED:
+            cur.execute(
+                'INSERT INTO `pbl_runtime_event` (event_id, tenant_id, event_type, '
+                'payload_json, occurred_at, learner_id, session_id, blueprint_id, '
+                'artifact_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)', ev)
+    print('事件表写入 %d 行（pbl_runtime_event）' % len(EVENTS_SEED))
+
+    import asyncio
+
+    def run_collect(label):
+        loop = asyncio.get_event_loop_policy().new_event_loop()
+        try:
+            stats = loop.run_until_complete(
+                collector.collect_evidence_from_events(tenant_id=_TENANT))
+        finally:
+            loop.close()
+        slim = {k: stats[k] for k in ('scanned', 'created', 'skipped', 'updated',
+                                      'ignored', 'failed', 'event_table', 'watermark')
+                if k in stats}
+        print('%s stats = %s' % (label, slim))
+        return stats
+
+    s1 = run_collect('第 1 次采集')
+    rows1 = box.raw('SELECT tenant_id, source_event_id, evidence_type, dedup_key '
+                    'FROM `pbl_evidence` ORDER BY source_event_id, evidence_type')
+    print('第 1 次后表内行（原样回显，%d 行）:' % len(rows1))
+    for r in rows1:
+        print('  %s' % r)
+
+    s2 = run_collect('第 2 次采集（同一批事件重放）')
+    rows2 = box.raw('SELECT tenant_id, source_event_id, evidence_type, dedup_key '
+                    'FROM `pbl_evidence` ORDER BY source_event_id, evidence_type')
+    print('第 2 次后表内行（原样回显，%d 行）:' % len(rows2))
+    for r in rows2:
+        print('  %s' % r)
+
+    check('IDM.1 第 1 次采集 created == 2（两条事件各落一条证据）',
+          s1.get('created') == 2, 'created=%s' % s1.get('created'))
+    check('IDM.2 第 2 次采集 skipped >= 1（命中唯一键走跳过分支）',
+          s2.get('skipped', 0) >= 1, 'skipped=%s' % s2.get('skipped'))
+    check('IDM.3 第 2 次采集 created == 0（重放不新增）',
+          s2.get('created') == 0, 'created=%s' % s2.get('created'))
+    check('IDM.4 两次采集后表内仍只有 2 行（每三元组唯一一行）',
+          len(rows2) == 2 and len(rows1) == 2, 'rows=%d' % len(rows2))
+    # 唯一索引的 DB 级行为：绕过预查，直接重复 INSERT，必须 1062 Duplicate entry
+    dup_err = ''
+    try:
+        with box.conn.cursor() as cur:
+            cur.execute(
+                'INSERT INTO `pbl_evidence` (id, tenant_id, source_event_id, '
+                'evidence_type, occurred_at, dedup_key) '
+                'VALUES (%s,%s,%s,%s,%s,%s)',
+                ('manual_dup_1', _TENANT, _EVENT_ID_A, 'artifact',
+                 '2026-09-22 10:00:00', 'x' * 32))
+    except Exception as exc:                     # noqa: BLE001
+        dup_err = '%s: %s' % (type(exc).__name__, exc)
+    print('直接重复 INSERT 的 DB 原始报错（uk_ev_dedup 真防重）: %s'
+          % (dup_err or '<未报错>'))
+    check('IDM.6 唯一键 uk_ev_dedup 在 DB 层拦截重复三元组（1062 Duplicate entry）',
+          '1062' in dup_err or 'Duplicate entry' in dup_err, dup_err[:160])
+
+    # collector 实际写入语句形态确认（ON DUPLICATE KEY UPDATE 双保险）
+    src = (PKG_DIR / 'collector.py').read_text(encoding='utf-8')
+    check("IDM.7 写入路径含 'ON DUPLICATE KEY UPDATE'（并发/重放第二层防重）",
+          'ON DUPLICATE KEY UPDATE' in src)
+    check('IDM.8 dedup_key 与 uk 三元组同源（md5(tenant|event|type)）',
+          _dedup_matches(rows2))
+    return rows2
+
+
+def _dedup_matches(rows):
+    from hashlib import md5
+    for r in rows:
+        want = md5('%s|%s|%s' % (r['tenant_id'], r['source_event_id'],
+                                 r['evidence_type']).encode('utf-8')).hexdigest()
+        if str(r.get('dedup_key')) != want:
+            return False
+    return True
+
+
+# ── 阶段 3：第三次采集（幂等可重放，补跑一次并对比）─────────────────
+def phase_third_run(collector, box, s2_stats):
+    import asyncio
+    loop = asyncio.get_event_loop_policy().new_event_loop()
+    try:
+        s3 = loop.run_until_complete(
+            collector.collect_evidence_from_events(tenant_id=_TENANT))
+    finally:
+        loop.close()
+    slim3 = {k: s3[k] for k in ('scanned', 'created', 'skipped', 'updated', 'ignored', 'failed')
+             if k in s3}
+    slim2 = {k: s2_stats[k] for k in ('scanned', 'created', 'skipped', 'updated', 'ignored', 'failed')
+             if k in s2_stats}
+    print('第 3 次采集 stats = %s（与第 2 次对比 %s）'
+          % (slim3, '一致' if slim2 == slim3 else '不一致'))
+    rows = box.raw('SELECT COUNT(*) AS c FROM `pbl_evidence`')
+    cnt = int((rows or [{'c': -1}])[0]['c'])
+    print('第 3 次后表内总行数 = %d' % cnt)
+    check('IDM.5 第三次采集统计与第二次逐字段一致', slim2 == slim3, '%s vs %s' % (slim2, slim3))
+    check('IDM.9 三次采集后表内行数仍为 2（无重复行累积）', cnt == 2, 'count=%d' % cnt)
+
+
+def db_kwargs():
+    env = os.environ.get('PBL_M5A_DB')
+    if env:
+        host, port, user, pwd = env.split(':')
+        return dict(host=host, port=int(port), user=user, password=pwd)
+    # 本机测试库凭据来自 projects/pbls/env/test.json 的 db 段（唯一事实源，未新增秘密）
+    return dict(host='127.0.0.1', port=3306, user='test', password='test123',
+                charset='utf8mb4', connect_timeout=5)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument('--keep', action='store_true', help='跑完保留沙箱库')
+    args = ap.parse_args()
+
+    print('M5a live-DB harness | sandbox schema = %s | target = modules/pbl_evidence'
+          % SANDBOX_SCHEMA)
+    try:
+        import pymysql                                    # noqa: F401
+    except Exception as exc:                              # noqa: BLE001
+        print('FAIL 环境缺少 pymysql，无法跑真实库幂等: %r' % (exc,))
+        return 1
+
+    collector = load_collector()
+    box = Sandbox(db_kwargs())
+    box.reset_schema()
+    box.open()
+    try:
+        install_stub(collector, box)
+        phase_fail_closed(collector, box)
+        _rows = phase_idempotent(collector, box)
+        # 重跑一次统计对比（放在 phase 2 之后，避免影响其内部断言顺序）
+        import asyncio
+        loop = asyncio.get_event_loop_policy().new_event_loop()
+        try:
+            s2 = loop.run_until_complete(
+                collector.collect_evidence_from_events(tenant_id=_TENANT))
+        finally:
+            loop.close()
+        phase_third_run(collector, box, s2)
+    finally:
+        box.close()
+        if not args.keep:
+            box.drop_schema()
+            print('沙箱库已清理: DROP DATABASE %s' % SANDBOX_SCHEMA)
+        else:
+            print('--keep：沙箱库 %s 保留' % SANDBOX_SCHEMA)
+
+    if _failures:
+        print('LIVE DB HARNESS FAILED (%d): %s' % (len(_failures), ' | '.join(_failures)))
+        return 1
+    print('LIVE DB HARNESS ALL PASS')
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())