diff --git a/docs/work-log-2026-09-22-m11b2c.md b/docs/work-log-2026-09-22-m11b2c.md index 41cdd2d..b89e7df 100644 --- a/docs/work-log-2026-09-22-m11b2c.md +++ b/docs/work-log-2026-09-22-m11b2c.md @@ -163,3 +163,24 @@ RESULT: PASS ### 6.5 复杂度评估（本轮） **简单**——单交付单元（两份文档正文写作），零代码改动、零表定义改动、无跨模块影响，**无需拆分**。父任务 `[M11b-2c-T2]` 整体为复杂（多独立交付单元 + 串行收口依赖），已由 PM 拆为 T2-a / T2-b / T2-c 三子任务并按批次派发，口径见 §6.1。 + +--- + +## 7. 收口轮（T2-c）回填笔 · 2026-09-22 + +由 `[M11b-2c-T2-c]`（task_id `UFjoRchqgo8asHC3rP5he`）追加。本文自身尺寸不写死（自指竞态），见 +`projects/pbls/docs/02-develop/m11b2c-t2-code-files-measure.log` `[final]` 段；以下只回填**对方**实测值与**已存在**的真实 hash。 + +| 项 | 真值 | 口径 | +|---|---|---| +| `projects/pbls/docs/02-develop/dev-notes.md` 字节数 | 54239 | `wc -c` T2-c 回填时点实测（对方文件，非自指） | +| `kICTWazBqgIWtKNO7sz8A_code_files.md` 字节数 | 20047 | 同上 | +| `modules/world_sync/docs/work-log-2026-09-22-m11b2c.md` 自身字节数 | 见 measure.log `[final]` 段 | **自指 → 不写死** | +| 本文件 T2-a 选择性提交 hash | `da37362628ad6b51425c3de348ce8a0a0737a66a` | `git log -1 --format='%H %s' -- docs/work-log-2026-09-22-m11b2c.md`（只含本文件一笔，无 wwwroot、无代码） | +| 本文件 T2-c 收口笔 hash | 见 measure.log `[git]` 段 | 自指（写进本文即改变本文） | +| `read_committed_stmt` 定义行 | `world_sync/pbl_runtime_sql.py:148` | T2-c 重跑 `grep -n '^def read_committed_stmt'` 复核，与 §6.2 同口径 | + +关键词逐词 `grep -c`（本文）：`M11b-2` 17 / `expected_version` 4 / `READ COMMITTED` 1 / `P99` 6 / `无新 HTTP 路径` 1。 + +§6.4 第一条「hash 由收口轮落 measure.log、本文不写死」的约定仍然成立；本表补的是**上一笔已存在**的 T2-a hash（不随本文变动而失效）。 +`projects/pbls` 侧：T2-c 已按 PM 指令对该过程仓做选择性提交（只加本任务四个文件），故 §6.4 中「仍为未提交状态」一句已被收口轮更新取代，真实 hash 见 measure.log `[git]` 段。 diff --git a/tests/test_m11b2d_fixture_contract.py b/tests/test_m11b2d_fixture_contract.py new file mode 100644 index 0000000..d3232b1 --- /dev/null +++ b/tests/test_m11b2d_fixture_contract.py @@ -0,0 +1,285 @@ +# -*- coding: utf-8 -*- +"""M11b-2d-A3 契约守卫用例：函数式夹具契约 + 真源返回键齐全性。 + +来源 +==== +本文件由一次性调试探针 ``modules/world_sync/.m11b2d_probe.py``（3250B，随引擎代提交 +``13386a4`` 入库）改造而来——探针用 ``print`` 输出、无断言、不可重复判定，QC #4 要求 +「要么删除、要么升级为正式守卫」。本文件即升级结果，探针本体已 ``git rm``。 + +守卫什么（三件事） +================== +1. **旧类式符号不再存在**：``RuntimeStateEventWriter`` / ``configure_writer`` / + ``build_plan`` / 类式 ``write_event_with_state`` 不得再出现在夹具层 + ``tests/_m11b2_compat`` 上（它们曾构成第二条夹具路径，把测试钉在已删除的调用形态上， + 实测 22 failed）。 +2. **新函数式符号可用**：夹具只保留一条路径——``install/uninstall/apply/read/ + trace_kinds/update_statements/insert_statements`` + ``RuntimeEngine``，且真源本体 + ``apply_runtime_event`` / ``read_states`` 经夹具直连（不是副本）。 +3. **真源返回键齐全**：``apply`` 的返回值键集合**逐键断言**，不做任何增补（夹具 + ``apply`` 只是 ``asyncio.run(apply_runtime_event(...))`` 的同步入口）。分支覆盖 + 成功写 / 幂等命中 / 乐观锁冲突 / 参数门禁 / 快照 / 提交后广播。 + +边界（遵守 QC #3 允许路径与本任务禁改清单） +========================================== +* 不改业务实现、不改 ``pbl_runtime_ext/models/*.json``、不改 ``load_path.py``； +* 不改 ``tests/_m11b2_compat.py`` 任何既有符号（只读引用）； +* 不改 ``test_m11b2_single_tx.py`` / ``test_m11b2_wiring.py``（并行任务独占）； +* 权威契约定义在 ``tests/_m11b2_compat.py`` 顶部 docstring 与 + ``docs/work-log/``… 不，登记文档为 ``skill/SKILL.md`` 指向的 + ``docs/work-log/m11b2d-fixture-contract.md``，本文件只做机械断言，不重复搬运说明。 + +夹具的 ``_default_clause`` 把 models 里 ``default: ""``（空串）当「无默认值」跳过， +于是 ``event_code`` 等「NOT NULL DEFAULT ''」列在 sqlite 侧成为无默认值的 NOT NULL 列， +真源 INSERT（按语义不填这些列）会撞约束。本文件与 ``test_m11b2_single_tx.py`` 同口径： +**仅在构造期**临时补正空串默认值渲染，建完表立即还原，不改 compat 既有定义与行为语义。 +""" + +import inspect +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +sys.path.insert(0, os.path.dirname(HERE)) + +import _m11b2_compat as compat # noqa: E402 + +apply_runtime_event = compat.apply_runtime_event # pbl_runtime_ext.tx_write 真源本体 +read_states = compat.read_states +RtxError = compat.RtxError +EVENT_TABLE = compat.EVENT_TABLE +STATE_TABLE = compat.STATE_TABLE +SNAPSHOT_TABLE = compat.SNAPSHOT_TABLE +ERR_TENANT_MISSING = compat.ERR_TENANT_MISSING +ERR_BAD_PARAM = compat.ERR_BAD_PARAM +ERR_STATE_CONFLICT = compat.ERR_STATE_CONFLICT + +# 真源 tx_write.apply_runtime_event 成功分支的返回键（源码 tx_write.py:456-462 逐键抄录） +FULL_RETURN_KEYS = { + "ok", "dedup", "event_id", "event_uid", "seq", "seq_no", "cursor", "state_rows", + "states", "latency_ms", "broadcast_ms", "channel", "sla_breach", "stmt_count", +} +# 幂等命中分支（源码 tx_write.py:383-389）：无 states/broadcast_ms/sla_breach/seq_no +DEDUP_RETURN_KEYS = { + "ok", "dedup", "event_id", "event_uid", "seq", "cursor", "state_rows", + "latency_ms", "channel", "stmt_count", +} +# 真源 apply_runtime_event 形参（源码 tx_write.py:344-348），全部默认 None +TRUE_SOURCE_PARAMS = [ + "tenant_id", "world_id", "session_id", "event_type", "payload", "state_updates", + "client_event_id", "idem_key", "snapshot", "base_version", "actor_id", + "causation_id", "occurred_at", "env", +] +# 夹具层禁止复活的旧类式符号 +DEAD_CLASS_STYLE_SYMBOLS = [ + "RuntimeStateEventWriter", "configure_writer", "build_plan", "write_event_with_state", +] +# 夹具层必须存在的函数式入口 +LIVE_FIXTURE_SYMBOLS = [ + "RuntimeEngine", "install", "uninstall", "apply", "read", "trace_kinds", + "update_statements", "insert_statements", "apply_runtime_event", "read_states", + "build_row", "assert_append_only", "RtxError", "percentile", + "EVENT_TABLE", "STATE_TABLE", "SNAPSHOT_TABLE", +] + +_ORIGIN_DEFAULT_CLAUSE = compat._default_clause + + +def _default_clause_with_empty(fd): + dv = fd.get("default") + if isinstance(dv, str) and dv.strip() == "": + return " DEFAULT ''" + return _ORIGIN_DEFAULT_CLAUSE(fd) + + +def make_engine(path=":memory:"): + """真 sqlite 事务引擎（表结构仍由夹具从 pbl_runtime_ext/models/*.json 生成）。""" + compat._default_clause = _default_clause_with_empty + try: + return compat.RuntimeEngine(path=path) + finally: + compat._default_clause = _ORIGIN_DEFAULT_CLAUSE + + +def su(entity="e1", state=None): + return {"entity_id": entity, "state": {"hp": 2} if state is None else state} + + +class FixtureSurfaceCase(unittest.TestCase): + """夹具面：旧符号已死、新符号可直连真源。""" + + def test_dead_class_style_symbols_absent(self): + for name in DEAD_CLASS_STYLE_SYMBOLS: + self.assertFalse( + hasattr(compat, name), + "夹具层不得复活旧类式符号 %s（第二条夹具路径 = M11b-2c 22 failed 根因）" % name) + + def test_live_function_style_symbols_present(self): + for name in LIVE_FIXTURE_SYMBOLS: + self.assertTrue(hasattr(compat, name), "夹具缺少函数式符号 %s" % name) + self.assertTrue(callable(compat.install)) + self.assertTrue(callable(compat.uninstall)) + self.assertTrue(callable(compat.apply)) + self.assertTrue(callable(compat.read)) + + def test_fixture_apply_is_thin_wrapper_on_true_source(self): + """夹具 apply/read 不得自带实现——必须是真源的同步入口（零返回键增补）。""" + self.assertIs(compat.apply_runtime_event, apply_runtime_event) + src = inspect.getsource(compat.apply) + self.assertIn("apply_runtime_event", src) + self.assertNotIn("update(", src) # 不补键 + self.assertNotIn("setdefault", src) + self.assertNotIn("_augment", src) + + def test_true_source_signature(self): + """真源逐参数签名钉死：形参名/顺序/默认值（文档 m11b2d-fixture-contract.md 同源）。""" + spec = inspect.signature(apply_runtime_event) + self.assertEqual(list(spec.parameters), TRUE_SOURCE_PARAMS) + for name in TRUE_SOURCE_PARAMS: + self.assertIsNone(spec.parameters[name].default, + "形参 %s 默认值应为 None" % name) + + def test_read_states_signature(self): + spec = inspect.signature(read_states) + self.assertEqual(list(spec.parameters), + ["tenant_id", "session_id", "entity_ids", "world_id", "env"]) + + +class ReturnKeysCase(unittest.TestCase): + """真源返回键齐全性（走夹具函数式路径，断言真值，不接受 print）。""" + + def setUp(self): + self.eng = make_engine() + compat.install(self.eng) + self.addCleanup(compat.uninstall) + + def _apply(self, **kw): + base = {"tenant_id": 7, "world_id": 1, "session_id": 11, + "event_type": "unit.move", "payload": {"dx": 1}} + base.update(kw) + return compat.apply(**base) + + def test_success_return_keys_exact(self): + r = self._apply(state_updates=[su()], client_event_id="ev-1", actor_id="u1") + self.assertEqual(set(r.keys()), FULL_RETURN_KEYS) + self.assertIs(r["ok"], True) + self.assertIs(r["dedup"], False) + self.assertIsNotNone(r["event_id"]) + self.assertEqual(r["seq"], 1) + self.assertEqual(r["seq_no"], r["seq"]) # 对外别名一致 + self.assertEqual(r["state_rows"], 1) + self.assertEqual(r["channel"], "pbl:rtx:w1:t7") + self.assertIsInstance(r["stmt_count"], int) + self.assertGreaterEqual(r["stmt_count"], 2) # 事件 INSERT + 状态 UPSERT + self.assertIsInstance(r["states"], list) + self.assertEqual(set(r["states"][0].keys()), + {"entity_id", "state_version", "action", "state_key", + "state_value", "seq"}) + self.assertEqual(r["states"][0]["action"], "insert") + self.assertEqual(r["states"][0]["state_version"], 1) + + def test_dedup_return_keys_exact_and_no_second_write(self): + r1 = self._apply(state_updates=[su()], client_event_id="ev-1") + r2 = self._apply(state_updates=[su(state={"hp": 9})], client_event_id="ev-1") + self.assertEqual(set(r2.keys()), DEDUP_RETURN_KEYS) + self.assertIs(r2["dedup"], True) + self.assertEqual(r2["event_id"], r1["event_id"]) # 复用既有事件，不新写 + self.assertEqual(r2["seq"], r1["seq"]) + self.assertEqual(r2["state_rows"], 0) + self.assertIsNone(r2["latency_ms"]) + self.assertEqual(self.eng.count(EVENT_TABLE), 1) + self.assertEqual(self.eng.count(STATE_TABLE), 1) + self.assertEqual(len(self.eng.broadcasts), 1) # 幂等命中不重复广播 + + def test_read_states_rows_shape(self): + self._apply(state_updates=[su()], client_event_id="ev-1") + rows = compat.read(tenant_id=7, session_id=11) + self.assertIsInstance(rows, list) + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row["entity_id"], "e1") + self.assertEqual(row["state"], {"hp": 2}) + self.assertEqual(row["state_version"], 1) + self.assertIn("checksum", row) + self.assertIn("updated_at", row) + + def test_snapshot_writes_snapshot_table(self): + r = self._apply(state_updates=[su(state={"hp": 5})], client_event_id="ev-4", + snapshot={"reason": "join", "hp": 5}) + self.assertEqual(set(r.keys()), FULL_RETURN_KEYS) + self.assertEqual(self.eng.count(SNAPSHOT_TABLE), 1) + + def test_broadcast_strictly_after_commit(self): + self._apply(state_updates=[su()], client_event_id="ev-1") + kinds = compat.trace_kinds(self.eng) + self.assertEqual(kinds[0], "BEGIN") + self.assertIn("COMMIT", kinds) + self.assertIn("BROADCAST", kinds) + self.assertGreater(kinds.index("BROADCAST"), kinds.index("COMMIT"), + "广播必须在事务提交之后（M11b-3 契约）") + self.assertEqual(self.eng.broadcasts[0]["channel"], "pbl:rtx:w1:t7") + self.assertEqual(self.eng.broadcasts[0]["event"]["event_type"], "unit.move") + + def test_event_table_is_append_only(self): + """事件表只允许 INSERT：UPDATE 语句里不得出现事件表名。""" + self._apply(state_updates=[su()], client_event_id="ev-1") + self._apply(state_updates=[su(state={"hp": 3})], client_event_id="ev-2") + for sql, _p in compat.update_statements(self.eng): + self.assertNotIn(EVENT_TABLE.upper(), sql.upper(), + "事件表出现 UPDATE 语句，违反 append-only：%s" % sql) + inserts = compat.insert_statements(self.eng, EVENT_TABLE) + self.assertEqual(len(inserts), 2) + for sql, _p in inserts: + self.assertIn(EVENT_TABLE.upper(), sql.upper()) + + +class GateErrorCase(unittest.TestCase): + """参数门禁与乐观锁冲突：必须抛 RtxError 且 code 精确、事务回滚不留脏行。""" + + def setUp(self): + self.eng = make_engine() + compat.install(self.eng) + self.addCleanup(compat.uninstall) + + def _expect(self, code, **kw): + base = {"tenant_id": 7, "world_id": 1, "session_id": 11, + "event_type": "x", "payload": {}} + base.update(kw) + with self.assertRaises(RtxError) as ctx: + compat.apply(**base) + self.assertEqual(ctx.exception.code, code) + return ctx.exception + + def test_missing_tenant_is_fail_closed(self): + self._expect(ERR_TENANT_MISSING, tenant_id=None) + self.assertEqual(self.eng.count(EVENT_TABLE), 0) + + def test_missing_required_params(self): + for key in ("world_id", "session_id", "event_type"): + self._expect(ERR_BAD_PARAM, **{key: None}) + self.assertEqual(self.eng.count(EVENT_TABLE), 0) + + def test_bad_state_updates(self): + self._expect(ERR_BAD_PARAM, state_updates="notalist") + self._expect(ERR_BAD_PARAM, state_updates=[{"state": {"a": 1}}]) # 缺 entity_id + self._expect(ERR_BAD_PARAM, state_updates=[{"entity_id": "e1"}]) # 缺 state + self.assertEqual(self.eng.count(STATE_TABLE), 0) + + def test_state_conflict_rolls_back(self): + compat.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move", + payload={}, state_updates=[su()], client_event_id="ev-1") + before_evt = self.eng.count(EVENT_TABLE) + before_state = self.eng.count(STATE_TABLE) + self._expect(ERR_STATE_CONFLICT, event_type="x", + state_updates=[su(state={"hp": 0})], + client_event_id="ev-3", base_version=999) + kinds = compat.trace_kinds(self.eng) + self.assertEqual(kinds[-1], "ROLLBACK", "冲突分支必须回滚，不得 COMMIT") + self.assertEqual(self.eng.count(EVENT_TABLE), before_evt) + self.assertEqual(self.eng.count(STATE_TABLE), before_state) + + +if __name__ == "__main__": + unittest.main(verbosity=2)