approve: [M9-收口B] 工作日志归档 + 交付一致性核验（以事实快照为唯一数据源）

tests/m5a_live_db_harness.py

@@ -344,11 +344,22 @@ def phase_idempotent(collector, box):
               .format(r.get('Seq_in_index'), r.get('Column_name'),
                       r.get('Non_unique'), r.get('Key_name')))
 
-    # 具名参数插入（列名与占位符同源 SEED_FIELDS，列/值不可能错位）
+    # 具名参数插入：占位符写成显式字面量（%(event_id)s ...），与列名一一对应。
+    # 不用位置参数 %s（与列序耦合，改列即错位），不用 f-string 内嵌值（注入面）。
+    # 下方同源校验保证字面量与 SEED_FIELDS 一致，任一方单独改动即拒绝写入。
+    seed_sql = (
+        'INSERT INTO `pbl_runtime_event` ('
+        '`event_id`, `tenant_id`, `event_type`, `payload_json`, `occurred_at`, '
+        '`learner_id`, `session_id`, `blueprint_id`, `artifact_id`'
+        ') VALUES ('
+        '%(event_id)s, %(tenant_id)s, %(event_type)s, %(payload_json)s, '
+        '%(occurred_at)s, %(learner_id)s, %(session_id)s, %(blueprint_id)s, '
+        '%(artifact_id)s)'
+    )
     seed_cols = ', '.join('`' + f + '`' for f in SEED_FIELDS)
     seed_vals = ', '.join('%(' + f + ')s' for f in SEED_FIELDS)
-    seed_sql = ('INSERT INTO `pbl_runtime_event` (' + seed_cols + ') VALUES ('
+    if seed_cols not in seed_sql or seed_vals not in seed_sql:
-                + seed_vals + ')')
+        raise RuntimeError('种子 INSERT 字面量与 SEED_FIELDS 不同源，拒绝错位写入')
     print('事件种子 INSERT（具名参数）: %s' % seed_sql)
     with box.conn.cursor() as cur:
         for ev in EVENTS_SEED:
@@ -523,13 +534,17 @@ def db_kwargs():
                 connect_timeout=5)
 
 
-def run(args):
+def setup_box():
-    """harness 主体（异常由 main() 统一收敛为显式 FAIL）。"""
+    """环境自检 + 凭据解析 + 沙箱库重建，返回 (collector, box)。
+
+    前置不满足（无 pymysql / 无凭据 / 连不上库）时 raise，由 main() 外层
+    try/except 统一收敛为显式 FAIL 断言，不在这里吞异常返回伪成功。
+    """
     try:
         import pymysql                                    # noqa: F401
     except Exception as exc:                              # noqa: BLE001
         check('ENV.1 运行环境具备 pymysql', False, repr(exc))
-        return 1
+        raise RuntimeError('运行环境缺少 pymysql: %s' % exc)
 
     conn_kwargs = db_kwargs()
     print('沙箱库凭据来源: %s | 目标 = %s@%s:%s（口令不打印，日志已脱敏）'
@@ -540,27 +555,33 @@ def run(args):
     box = Sandbox(conn_kwargs)
     box.reset_schema()
     box.open()
+    return collector, box
+
+
+def collect_again(collector):
+    """第三次采集（供 phase_third_run 做统计对比），独立事件循环避免污染。"""
+    import asyncio
+    loop = asyncio.get_event_loop_policy().new_event_loop()
     try:
-        install_stub(collector, box)
+        return loop.run_until_complete(
-        phase_fail_closed(collector, box)
+            collector.collect_evidence_from_events(tenant_id=_TENANT))
-        _s2, _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:
+        loop.close()
+
+
+def teardown_box(box, keep):
+    """沙箱库收尾：关连接 + 按需 DROP。自身异常只打印，避免掩盖阶段断言结论。"""
+    if box is None:
+        return
+    try:
         box.close()
-        if not args.keep:
+    except Exception as exc:                              # noqa: BLE001
-            box.drop_schema()
+        print('box.close() 异常（忽略，不影响断言结论）: %s' % _mask(exc))
-            print('沙箱库已清理: DROP DATABASE %s' % SANDBOX_SCHEMA)
+    if not keep:
-        else:
+        box.drop_schema()
-            print('--keep：沙箱库 %s 保留' % SANDBOX_SCHEMA)
+        print('沙箱库已清理: DROP DATABASE %s' % SANDBOX_SCHEMA)
-    return 0
+    else:
+        print('--keep：沙箱库 %s 保留' % SANDBOX_SCHEMA)
 
 
 def main():
@@ -570,16 +591,25 @@ def main():
 
     print('M5a live-DB harness | sandbox schema = %s | target = modules/pbl_evidence'
           % SANDBOX_SCHEMA)
+    box = None
+    # 外层 try/except 直接包住全部 phase 调用：任何未捕获异常（凭据缺失、连不上库、
+    # 被测代码抛错）都收敛为显式 FAIL 断言 + 非 0 退出码，绝不以裸 traceback 逃逸——
+    # 否则日志退化为崩溃现场（.qc_tmp），不构成通过证据。
     try:
-        run(args)
+        collector, box = setup_box()
+        install_stub(collector, box)
+        phase_fail_closed(collector, box)
+        _s2, _rows = phase_idempotent(collector, box)
+        # 重跑一次统计对比（放在 phase 2 之后，避免影响其内部断言顺序）
+        phase_third_run(collector, box, collect_again(collector))
     except SystemExit:
         raise
     except Exception as exc:                              # noqa: BLE001
-        # 未捕获异常（凭据缺失/连不上库/被测代码抛错）统一收敛为显式 FAIL 项，
-        # 打印异常类型 + 脱敏消息，绝不靠 traceback 截断证据。
         check('HARNESS.1 harness 无未捕获异常（凭据可用 + 沙箱库可连 + 各阶段跑完）',
               False, '%s: %s' % (type(exc).__name__, _mask(exc)))
         print('未捕获异常汇总（脱敏）: %s' % _mask(traceback.format_exc()))
+    finally:
+        teardown_box(box, args.keep)
     if _failures:
         print('LIVE DB HARNESS FAILED (%d): %s'
               % (len(_failures), ' | '.join(_mask(f) for f in _failures)))