From b2d10a6dcd295cf46398c52313d525a546227333 Mon Sep 17 00:00:00 2001
From: song <22676124+songoow@users.noreply.github.com>
Date: Sun, 20 Sep 2026 02:34:49 -0400
Subject: [PATCH 1/4] test(chat): characterize missing actions and persisted
 Lark routing modes

Signed-off-by: song <22676124+songoow@users.noreply.github.com>
---
 .../test_lark_goal_topic_connections.py       | 69 +++++++++++++++++++
 tests/test_chat_server_cors.py                | 33 +++++++++
 2 files changed, 102 insertions(+)

diff --git a/tests/extensions/test_lark_goal_topic_connections.py b/tests/extensions/test_lark_goal_topic_connections.py
index 70d24bb28c..2add32f6ff 100644
--- a/tests/extensions/test_lark_goal_topic_connections.py
+++ b/tests/extensions/test_lark_goal_topic_connections.py
@@ -2122,6 +2122,75 @@ def _prep_goal_channel_target(root: Path) -> Path:
     return target_path
 
 
+@pytest.mark.parametrize(
+    ("routing", "expected"),
+    [
+        ({}, ("addressed_only", "direct_session", "topic_reply")),
+        (
+            {"incoming_mode": "all"},
+            ("configured_chat_all", "direct_session", "topic_reply"),
+        ),
+        (
+            {
+                "incoming_mode": "all",
+                "capture_scope": " ADDRESSED_ONLY ",
+                "ingress_mode": " SESSION_QUEUE ",
+                "reply_mode": " TOPIC_REPLY ",
+            },
+            ("addressed_only", "session_queue", "topic_reply"),
+        ),
+        ({"capture_scope": "invalid"}, None),
+        ({"ingress_mode": "async-inbox"}, None),
+        ({"reply_mode": "invalid"}, None),
+    ],
+)
+def test_connection_readback_and_event_route_share_persisted_mode_rules(
+    tmp_path: Path,
+    routing: dict[str, str],
+    expected: tuple[str, str, str] | None,
+) -> None:
+    target_path = _prep_goal_channel_target(tmp_path)
+    binding_path = tmp_path / "binding.json"
+    payload = _legacy_v0_binding_payload("om_topic_alpha", "agent-alpha")
+    payload["bindings"]["goal-alpha"]["routing"] = routing
+    write_goal_channel_binding(binding_path, payload)
+    before = binding_path.read_bytes()
+    rows = list_lark_connections(
+        registry=_registry(tmp_path),
+        target_path=target_path,
+        binding_paths={"goal-alpha": binding_path},
+        runner=_runner({}),
+    )
+    decision = decide_lark_topic_event(
+        target_payload=read_goal_channel_targets(target_path),
+        binding_payloads={"goal-alpha": read_goal_channel_binding(binding_path)},
+        event={
+            "chat_id": CHAT_ID,
+            "root_id": "om_topic_alpha",
+            "message_id": "om_incoming",
+            "content": "@mew bot hello",
+        },
+    )
+    assert len(rows) == 1
+    if expected is None:
+        assert rows[0]["reply_ready"] is False
+        assert rows[0]["health_error_code"] == "invalid_routing_state"
+        assert decision == {
+            "matched": False,
+            "reason": "invalid_routing_state",
+            "route": None,
+        }
+    else:
+        assert rows[0]["reply_ready"] is True
+        assert decision["matched"] is True
+        for key, value in zip(
+            ("capture_scope", "ingress_mode", "reply_mode"), expected
+        ):
+            assert rows[0][key] == value
+            assert decision["route"][key] == value
+    assert binding_path.read_bytes() == before
+
+
 def test_reconnect_after_upgrade_reuses_legacy_topic_root_without_resend(
     tmp_path: Path,
 ) -> None:
diff --git a/tests/test_chat_server_cors.py b/tests/test_chat_server_cors.py
index 92b9f02fa3..7521619925 100644
--- a/tests/test_chat_server_cors.py
+++ b/tests/test_chat_server_cors.py
@@ -213,6 +213,39 @@ def test_chat_action_context_cannot_persist_or_emit_overflowed_float(
         server.server_close()
 
 
+@pytest.mark.parametrize(
+    "action", ["snapshot", "apply", "cancel", "regenerate", "reject", "defer"]
+)
+def test_missing_action_returns_the_same_http_error(
+    tmp_path: Path, action: str
+) -> None:
+    server, thread = _start_server()
+    server.action_store = ChatActionStore(tmp_path / "actions")
+    server.action_service = ChatActionService(
+        store=server.action_store, registry_path=tmp_path / "registry.json"
+    )
+    try:
+        response = _request(
+            server.server_address[1],
+            method="GET" if action == "snapshot" else "POST",
+            origin=None,
+            path="/api/actions/missing"
+            + ("" if action == "snapshot" else f"/{action}"),
+            body=None if action == "snapshot" else b"{}",
+        )
+        assert response.status == 404
+        assert json.loads(response.read()) == {
+            "ok": False,
+            "error": "typed Chat action proposal was not found",
+            "error_code": "action_not_found",
+        }
+        assert server.action_store.list() == []
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+        server.server_close()
+
+
 def test_chat_status_forwards_valid_goal_activation_scope(monkeypatch) -> None:
     calls: list[dict[str, object]] = []
 

From d231971ec01b04263bd6921a047318f1b87a1a85 Mon Sep 17 00:00:00 2001
From: song <22676124+songoow@users.noreply.github.com>
Date: Sun, 20 Sep 2026 02:34:49 -0400
Subject: [PATCH 2/4] refactor(chat): deduplicate action errors and Lark
 routing defaults

Signed-off-by: song <22676124+songoow@users.noreply.github.com>
---
 loopx/chat_server.py                          | 31 +++++-------
 .../extensions/lark/goal_topic_connections.py | 49 ++-----------------
 loopx/extensions/lark/goal_topic_routing.py   | 31 ++++++++++++
 3 files changed, 45 insertions(+), 66 deletions(-)

diff --git a/loopx/chat_server.py b/loopx/chat_server.py
index de9c8c7d33..c0f3ebc762 100644
--- a/loopx/chat_server.py
+++ b/loopx/chat_server.py
@@ -1055,6 +1055,13 @@ def _action_preview(self) -> None:
             status=201,
         )
 
+    def _action_not_found(self) -> None:
+        self._send_error(
+            "typed Chat action proposal was not found",
+            status=404,
+            error_code="action_not_found",
+        )
+
     def _action_snapshot(self, proposal_id: str) -> None:
         try:
             proposal = self.server.action_service.load(proposal_id)
@@ -1062,11 +1069,7 @@ def _action_snapshot(self, proposal_id: str) -> None:
             self._send_error(str(exc), status=400, error_code="invalid_proposal_id")
             return
         if proposal is None:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         self._send_json(
             {
@@ -1107,11 +1110,7 @@ def _action_cancel(self, proposal_id: str) -> None:
                 raise ValueError("action cancel request must be empty")
             proposal = self.server.action_service.cancel(proposal_id)
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1144,11 +1143,7 @@ def _action_transition(self, proposal_id: str, transition: str) -> None:
             else:
                 raise ValueError("unsupported action transition")
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1190,11 +1185,7 @@ def _action_apply(self, proposal_id: str) -> None:
             )
             return
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
diff --git a/loopx/extensions/lark/goal_topic_connections.py b/loopx/extensions/lark/goal_topic_connections.py
index 55d7bd3932..48dc33e006 100644
--- a/loopx/extensions/lark/goal_topic_connections.py
+++ b/loopx/extensions/lark/goal_topic_connections.py
@@ -92,6 +92,7 @@
     IngressMode,
     ReplyMode,
     _routing_value,
+    _connection_routing_modes,
     decide_lark_topic_route_event,
 )
 from .presentation.kanban import (
@@ -1035,29 +1036,7 @@ def list_lark_connections(
             )
             connector_status: dict[str, Any] | None = None
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 raw_connector = binding.get("connector")
                 if raw_connector is not None:
                     if not isinstance(raw_connector, Mapping):
@@ -1216,29 +1195,7 @@ def decide_lark_topic_event(
                 else {}
             )
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 connector = binding.get("connector")
                 if connector is not None:
                     if not isinstance(connector, Mapping):
diff --git a/loopx/extensions/lark/goal_topic_routing.py b/loopx/extensions/lark/goal_topic_routing.py
index 81b8e0cccf..09caab8c50 100644
--- a/loopx/extensions/lark/goal_topic_routing.py
+++ b/loopx/extensions/lark/goal_topic_routing.py
@@ -47,6 +47,37 @@ def _routing_value(
         raise ValueError(f"{field} must be one of: {allowed}") from exc
 
 
+def _connection_routing_modes(
+    routing: Mapping[str, Any],
+) -> tuple[str, str, str]:
+    """Normalize persisted modes for both connection readback and event routing."""
+
+    capture_scope = _routing_value(
+        CaptureScope,
+        routing.get("capture_scope")
+        or (
+            "configured_chat_all"
+            if routing.get("incoming_mode") == "all"
+            else "addressed_only"
+        ),
+        default=CaptureScope.ADDRESSED_ONLY.value,
+        field="capture_scope",
+    )
+    ingress_mode = _routing_value(
+        IngressMode,
+        routing.get("ingress_mode"),
+        default=IngressMode.DIRECT_SESSION.value,
+        field="ingress_mode",
+    )
+    reply_mode = _routing_value(
+        ReplyMode,
+        routing.get("reply_mode"),
+        default=ReplyMode.TOPIC_REPLY.value,
+        field="reply_mode",
+    )
+    return capture_scope, ingress_mode, reply_mode
+
+
 def _normalize_mention_name(name: str) -> str:
     cleaned = str(name or "").strip()
     if cleaned.startswith("@"):

From 66861dabf68256f613be59cb0f508385d442624f Mon Sep 17 00:00:00 2001
From: song <22676124+songoow@users.noreply.github.com>
Date: Sun, 20 Sep 2026 02:40:31 -0400
Subject: [PATCH 3/4] test(semantics): witness settlement and receipt source
 boundaries

Execute the Turn reducer and receipt readback directly and through Python. Separate internal producers, external decoder admission and retained compatibility; record remaining producer gaps without widening F1/F2. Refs #4447.

Signed-off-by: song <22676124+songoow@users.noreply.github.com>
---
 ...20-settlement-receipt-source-boundaries.md |  52 ++
 ...tlement-receipt-source-boundaries.zh-CN.md |  43 ++
 scripts/settlement_receipt_source_witness.mts |  19 +
 ...st_settlement_receipt_source_boundaries.py | 550 ++++++++++++++++++
 4 files changed, 664 insertions(+)
 create mode 100644 docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.md
 create mode 100644 docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.zh-CN.md
 create mode 100644 scripts/settlement_receipt_source_witness.mts
 create mode 100644 tests/architecture/test_settlement_receipt_source_boundaries.py

diff --git a/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.md b/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.md
new file mode 100644
index 0000000000..aaacae22e2
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.md
@@ -0,0 +1,52 @@
+# Executed settlement and receipt source boundaries
+
+Measured from `36134771355f05c9bcc5657ce739bb86426383dc`, for #4447 Stage 2a.
+The scope revision in #4789 remains a proposal. This entry changes neither RFC
+acceptance nor the F1/F2 production domain: **7/26**, with these four entries
+still outside it. Source treatment and scanner enrollment are separate claims.
+
+- **Settlement envelopes (`settlement_step_kind`, `settlement_failure_kind`).**
+  `turn_driver/settlement.ts::reduceTurnSettlementTransaction` calls the shared
+  `effect_program.ts` builders; its `result.receipts[].step_kind` and
+  `result.failure.{kind,step_kind}` cross `effect_runtime_result` into
+  `effect_program.py::decode_settlement_result_payload`. The witness executes
+  the real reducer directly and through the bridge: committed replay produces
+  all four steps without dispatch/checkpoint; identity/prefix rejection,
+  provider refusal, terminal refusal and unknown prepared outcome produce
+  eight failure kinds. Real file readback adds `writeback_missing`.
+- **External input is a separate obligation.** The live
+  `settlement.bind_gate` handler runs `settlementResultInput` before its builder.
+  Injected envelopes exercise admission of four step and eleven failure values,
+  plus unknown/null/numeric rejection in each step/failure slot through both TS
+  and Python decoders. These are input witnesses, not eleven producer witnesses.
+  `permission_denied` has task-lease producers in `task_lease_acquire.ts` and
+  `task_lease_lifecycle.ts`, but this batch does not execute them. `cancelled`
+  remains decoder-admitted without an identified producing branch; it is not
+  newly classified as compatibility-only. Both production obligations stay open.
+- **Receipt phases (`receipt_bound_monitor_phase`, `receipt_bound_replay_phase`).**
+  `quota/settlement_readback.ts::readQuotaSettlement` derives facts from isolated
+  synthetic receipt files and calls `quota/settlement_phase.ts`; Python
+  `read_heartbeat_settlement` decodes the results. Exact monitor commit, absent
+  or wrong commit, completion/writeback/spend prefixes, repeated readback and
+  conflicting identity exercise actual callers. Monitor emits `poll_due` or
+  `settled` even without spend; `settlement_pending` remains accepted by the
+  Python work-lane consumer only as compatibility evidence. The legacy monitor
+  effect id and terminal-to-replay adapter are retained. Removing them requires
+  separate historical-reader/caller migration evidence. Replay's three values
+  are produced; autonomous-replan binding is additionally exercised through the
+  phase bridge, not a full replan receipt transaction.
+
+Run `uv run --extra test python -m pytest tests/architecture/test_settlement_receipt_source_boundaries.py`:
+**62 passed** using Python 3.12.3 and qualified Node 22.22.3. With the existing
+binding witness, settlement-driver and quota-settlement tests: **142 passed**.
+The two native TS settlement/readback suites: **58 passed**, no skips.
+Docs governance, focused lint/type checks and semantic drift smoke pass; the
+latter still reports F1/F2 **7/26** and 19 unverified cross-runtime entries.
+The thin `scripts/settlement_receipt_source_witness.mts` imports shipped owners;
+it introduces no runtime rule, registry metadata or global vocabulary.
+
+Limits: synthetic readback is not proof of durable writers, all call sites, live
+CLI/backend qualification or F6 history compatibility. The existing budget-text
+failure classifier is characterized, not repaired. No production refactor or
+frontend/Lark/CLI change is included; the next owner action is evidence review
+and the named missing producer witnesses, not automatic source closure.
diff --git a/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.zh-CN.md b/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.zh-CN.md
new file mode 100644
index 0000000000..ac8725d381
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/semantic-vocabulary-convergence-v0/2026-09-20-settlement-receipt-source-boundaries.zh-CN.md
@@ -0,0 +1,43 @@
+# Settlement 与 receipt 的可执行来源边界
+
+基线 `36134771355f05c9bcc5657ce739bb86426383dc`，对应 #4447 Stage 2a。
+#4789 的范围修订仍是提案。本条不改变 RFC 验收或 F1/F2 生产域：仍为
+**7/26**，本组四项仍在域外。来源处理与生产扫描纳入是不同结论。
+
+- **Settlement envelope（`settlement_step_kind`、`settlement_failure_kind`）。**
+  `turn_driver/settlement.ts::reduceTurnSettlementTransaction` 调用共享
+  `effect_program.ts` builder；`result.receipts[].step_kind` 与
+  `result.failure.{kind,step_kind}` 经 `effect_runtime_result` 到达
+  `effect_program.py::decode_settlement_result_payload`。Witness 直接执行真实
+  reducer 并经 bridge 复核：已提交重放产生四种 step，不再 dispatch/checkpoint；
+  identity/prefix 拒绝、provider 拒绝、terminal 拒绝及 prepared outcome unknown
+  产生八种 failure。真实文件读回另证明 `writeback_missing`。
+- **外部输入是独立义务。** 实际 `settlement.bind_gate` handler 先执行
+  `settlementResultInput`。注入 envelope 覆盖四种 step、十一种 failure 的接纳，
+  以及各 step/failure 槽位对 unknown/null/数字的 TS 与 Python 解码拒绝。这是
+  输入证据，不是十一种生产证据。`permission_denied` 在
+  `task_lease_acquire.ts`、`task_lease_lifecycle.ts` 有生产点，本组尚未执行；
+  `cancelled` 仅证明解码接纳，尚未识别生产分支，不将其新归类为 compatibility-only。
+  两项生产义务继续保留。
+- **Receipt phase（`receipt_bound_monitor_phase`、`receipt_bound_replay_phase`）。**
+  `quota/settlement_readback.ts::readQuotaSettlement` 从隔离合成 receipt 文件提取
+  事实，调用 `quota/settlement_phase.ts`，Python `read_heartbeat_settlement`
+  解码结果。精确 monitor commit、缺失或错误 commit、completion/writeback/spend
+  前缀、重复读回及冲突 identity 均经过实际调用点。Monitor 无需 spend 即产生
+  `poll_due` 或 `settled`；`settlement_pending` 仅以 Python work-lane 消费者的
+  兼容接纳证明保留。旧 monitor effect id 与 terminal-to-replay adapter 继续保留；
+  删除需要独立的历史 reader/caller 迁移证据。Replay 三种值均有产生证据；
+  autonomous-replan binding 另经 phase bridge 验证，未执行完整 replan receipt 交易。
+
+运行 `uv run --extra test python -m pytest tests/architecture/test_settlement_receipt_source_boundaries.py`：
+Python 3.12.3、合格 Node 22.22.3 下 **62 passed**；连同已有 binding witness、
+settlement-driver、quota-settlement 测试共 **142 passed**。两组原生 TS
+settlement/readback 测试 **58 passed**，无跳过。文档治理、焦点 lint/type 检查和
+语义漂移 smoke 通过；后者仍报告 F1/F2 **7/26**、19 项跨运行时未验证。薄脚本
+`scripts/settlement_receipt_source_witness.mts` 导入现有 owner，不新增 runtime
+规则、registry 元数据或全局词表。
+
+边界：合成读回不证明 durable writer、所有调用点、真实 CLI/backend 资格或 F6
+历史兼容。现有 budget 文本分类只被刻画，未被修复。未做生产重构，也未改变
+frontend/Lark/CLI；后续由相应 owner 评审证据并补已点名的生产 witness，不自动
+宣告来源全部关闭。
diff --git a/scripts/settlement_receipt_source_witness.mts b/scripts/settlement_receipt_source_witness.mts
new file mode 100644
index 0000000000..de620023c4
--- /dev/null
+++ b/scripts/settlement_receipt_source_witness.mts
@@ -0,0 +1,19 @@
+// Bounded source witness: execute shipped owners, without defining vocabulary
+// values, classifying sources, or enrolling them in the production scanner.
+import { readFileSync } from "node:fs";
+import { reduceTurnSettlementTransaction } from "../loopx/control_plane/turn_driver/settlement.ts";
+import { readQuotaSettlement } from "../loopx/control_plane/quota/settlement_readback.ts";
+
+const request = JSON.parse(readFileSync(0, "utf8"));
+let result: unknown;
+switch (request.operation) {
+  case "turn":
+    result = reduceTurnSettlementTransaction(request.input);
+    break;
+  case "readback":
+    result = await readQuotaSettlement(request.input);
+    break;
+  default:
+    throw new Error("unsupported settlement source witness operation");
+}
+process.stdout.write(JSON.stringify(result));
diff --git a/tests/architecture/test_settlement_receipt_source_boundaries.py b/tests/architecture/test_settlement_receipt_source_boundaries.py
new file mode 100644
index 0000000000..1554afeda8
--- /dev/null
+++ b/tests/architecture/test_settlement_receipt_source_boundaries.py
@@ -0,0 +1,550 @@
+"""Executed sources, wire inputs and compatibility are different evidence.
+
+The Turn reducer creates step/failure values from journal/provider facts; the
+shared envelope also accepts them as external inputs. Receipt phases instead
+come from readback facts. These witnesses do not enroll any of these four
+vocabularies in F1/F2 or establish whole-program producer completeness.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from itertools import product
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.effect_program import (
+    SettlementIdentity,
+    decode_settlement_result_payload,
+    receipt_bound_monitor_phase,
+    receipt_bound_replay_phase,
+    receipt_bound_terminal_phase,
+)
+from loopx.control_plane.effect_runtime import (
+    EffectRuntimeRejected,
+    effect_runtime_result,
+)
+from loopx.control_plane.quota.settlement import read_heartbeat_settlement
+from loopx.control_plane.turn_driver.settlement import execute_turn_driver_settlement
+from loopx.control_plane.work_items.work_lane import (
+    preserve_heartbeat_receipt_bound_work_lane,
+)
+
+ROOT = Path(__file__).resolve().parents[2]
+PHASES = [
+    "host_execute",
+    "typed_result",
+    "validation",
+    "durable_writeback",
+    "quota_spend",
+]
+COMMITTED = {"ok": True, "appended": True}
+IDENTITY_INPUT = dict(
+    goal_id="witness-goal",
+    agent_id="witness-agent",
+    todo_id="todo_witness",
+    turn_instance_id="witness-turn",
+)
+
+
+@pytest.fixture(scope="module")
+def identity():
+    return SettlementIdentity(**IDENTITY_INPUT).as_dict()
+
+
+def _direct(operation, payload):
+    completed = subprocess.run(
+        [
+            "node",
+            "--no-warnings",
+            "--experimental-strip-types",
+            str(ROOT / "scripts/settlement_receipt_source_witness.mts"),
+        ],
+        input=json.dumps({"operation": operation, "input": payload}),
+        capture_output=True,
+        text=True,
+        cwd=ROOT,
+        timeout=30,
+    )
+    assert completed.returncode == 0, completed.stderr
+    return json.loads(completed.stdout)
+
+
+def _turn(identity, **overrides):
+    return {
+        "schema_version": "loopx_turn_settlement_transaction_v0",
+        "transaction_plan": {"settlement_plan": {"identity": identity}},
+        "transaction_phases": PHASES,
+        "completed_phases": PHASES,
+        "committed_effect_id": identity["effect_id"],
+        "writeback_payload": COMMITTED,
+        "quota_spend_payload": COMMITTED,
+        "terminal_closeout_required": False,
+        "terminal_closeout_payload": None,
+        "failed_provider_attempt": None,
+        "effect_attempts": {},
+        "provider_observations": {},
+        **overrides,
+    }
+
+
+# These are causal inputs, not a loop over declared enum members.
+@pytest.mark.parametrize(
+    "overrides,expected",
+    [
+        (
+            {"transaction_plan": {"settlement_plan": {"identity": {}}}},
+            "invalid_identity",
+        ),
+        ({"completed_phases": PHASES[:2]}, "receipt_missing"),
+        ({"committed_effect_id": "another-effect"}, "identity_mismatch"),
+        (
+            {
+                "completed_phases": PHASES[:3],
+                "writeback_payload": None,
+                "quota_spend_payload": None,
+                "failed_provider_attempt": {
+                    "step_kind": "durable_writeback",
+                    "payload": {"ok": False, "reason": "refused"},
+                },
+            },
+            "writeback_rejected",
+        ),
+        (
+            {
+                "completed_phases": PHASES[:4],
+                "quota_spend_payload": None,
+                "failed_provider_attempt": {
+                    "step_kind": "quota_spend",
+                    "payload": {"ok": False, "reason": "refused"},
+                },
+            },
+            "quota_spend_rejected",
+        ),
+        (
+            {
+                "completed_phases": PHASES[:4],
+                "quota_spend_payload": None,
+                "failed_provider_attempt": {
+                    "step_kind": "quota_spend",
+                    "payload": {"ok": False, "reason": "budget exhausted"},
+                },
+            },
+            "budget_rejected",
+        ),
+        (
+            {
+                "terminal_closeout_required": True,
+                "turn_result_kind": "validated_progress",
+            },
+            "terminal_closeout_rejected",
+        ),
+    ],
+)
+def test_turn_owner_produces_failures_and_bridge_decodes_them(
+    identity, overrides, expected
+):
+    request = _turn(identity, **overrides)
+    direct = _direct("turn", request)
+    bridged = effect_runtime_result("turn.settlement.reduce", request)
+    assert bridged == direct
+    assert direct["decision"] == "failed"
+    assert direct["provider_effects"] == []
+    _, _, failure = decode_settlement_result_payload(bridged["result"])
+    assert failure.kind.value == expected
+
+
+def test_prepared_unknown_is_not_retry_or_success(identity):
+    request = _turn(
+        identity,
+        completed_phases=PHASES[:3],
+        writeback_payload=None,
+        quota_spend_payload=None,
+        effect_attempts={
+            "durable_writeback": {
+                "status": "prepared",
+                "effect_ref": identity["effect_id"] + "#durable_writeback",
+            }
+        },
+        provider_observations={
+            "durable_writeback": {"kind": "unknown", "reason": "unavailable"}
+        },
+    )
+    direct = _direct("turn", request)
+    assert effect_runtime_result("turn.settlement.reduce", request) == direct
+    assert direct["decision"] == "failed"
+    assert direct["provider_effects"] == []
+    assert direct["result"]["failure"]["kind"] == "effect_outcome_unknown"
+
+
+@pytest.mark.parametrize("terminal", [False, True])
+def test_committed_replay_constructs_receipts_without_executing_again(
+    identity, terminal
+):
+    overrides = dict(terminal_closeout_required=terminal)
+    if terminal:
+        overrides.update(
+            turn_result_kind="validated_completion",
+            terminal_closeout_payload={
+                **COMMITTED,
+                "completion": {
+                    "todo_id": identity["todo_id"],
+                    "continuation": "no_followup",
+                },
+            },
+        )
+    request = _turn(identity, **overrides)
+    direct = _direct("turn", request)
+    assert effect_runtime_result("turn.settlement.reduce", request) == direct
+    expected = ["validation", "durable_writeback", "quota_spend"]
+    if terminal:
+        expected.append("terminal_closeout")
+    assert direct["decision"] == "complete"
+    assert direct["provider_effects"] == []
+    assert [r["step_kind"] for r in direct["result"]["receipts"]] == expected
+    assert {r["effect_id"] for r in direct["result"]["receipts"]} == {
+        identity["effect_id"]
+    }
+
+    def forbidden(*args):
+        pytest.fail("settled replay must not dispatch or checkpoint providers")
+
+    result = execute_turn_driver_settlement(
+        request["transaction_plan"],
+        transaction_phases=PHASES,
+        completed_phases=PHASES,
+        committed_effect_id=identity["effect_id"],
+        writeback_payload=COMMITTED,
+        quota_spend_payload=COMMITTED,
+        writeback=forbidden,
+        spend=forbidden,
+        checkpoint=forbidden,
+        terminal_closeout=forbidden,
+        terminal_checkpoint=forbidden,
+        terminal_closeout_payload=request["terminal_closeout_payload"],
+        terminal_closeout_required=terminal,
+        turn_result_kind=request.get("turn_result_kind"),
+    )
+    assert result.failure is None
+    assert [r.step_kind.value for r in result.receipts] == expected
+
+
+# Explicit injected envelopes prove decoder admission only, never production.
+@pytest.mark.parametrize(
+    "kind",
+    [
+        "invalid_identity",
+        "receipt_missing",
+        "identity_mismatch",
+        "writeback_missing",
+        "writeback_rejected",
+        "quota_spend_rejected",
+        "terminal_closeout_rejected",
+        "cancelled",
+        "permission_denied",
+        "budget_rejected",
+        "effect_outcome_unknown",
+    ],
+)
+def test_external_failure_envelope_survives_ts_and_python_decoders(kind):
+    envelope = {
+        "value": None,
+        "receipts": [],
+        "failure": {
+            "kind": kind,
+            "step_kind": "validation",
+            "reason": "external refusal",
+        },
+    }
+    result = effect_runtime_result("settlement.bind_gate", {"result": envelope})
+    assert result == {"execute": False, "result": envelope}
+    _, _, failure = decode_settlement_result_payload(result["result"])
+    assert failure.kind.value == kind
+
+
+@pytest.mark.parametrize(
+    "step", ["validation", "durable_writeback", "quota_spend", "terminal_closeout"]
+)
+def test_external_receipt_step_survives_both_decoders(identity, step):
+    envelope = {
+        "value": {},
+        "failure": None,
+        "receipts": [
+            {
+                "step_kind": step,
+                "status": "committed",
+                "effect_id": identity["effect_id"],
+            }
+        ],
+    }
+    result = effect_runtime_result("settlement.bind_gate", {"result": envelope})
+    assert result == {"execute": True, "result": envelope}
+    _, receipts, _ = decode_settlement_result_payload(result["result"])
+    assert receipts[0].step_kind.value == step
+
+
+@pytest.mark.parametrize("slot", ["receipt_step", "failure_step", "failure_kind"])
+@pytest.mark.parametrize("bad", ["unknown", None, 3])
+def test_external_unknowns_are_rejected_by_each_decoder(slot, bad):
+    envelope = {"value": None, "receipts": [], "failure": None}
+    if slot == "receipt_step":
+        envelope["receipts"] = [
+            {"step_kind": bad, "status": "committed", "effect_id": "effect"}
+        ]
+    else:
+        envelope["failure"] = {
+            "kind": "cancelled",
+            "step_kind": "validation",
+            "reason": "refused",
+        }
+        envelope["failure"]["kind" if slot == "failure_kind" else "step_kind"] = bad
+    with pytest.raises(EffectRuntimeRejected, match="settlement|must be"):
+        effect_runtime_result("settlement.bind_gate", {"result": envelope})
+    with pytest.raises(RuntimeError, match="shape mismatch"):
+        decode_settlement_result_payload(envelope)
+
+
+def _readback_fixture(
+    root,
+    identity,
+    *,
+    completion=False,
+    writeback=False,
+    spend=False,
+    monitor_effect=None,
+    material=False,
+    guard_effect=None,
+):
+    goal = root / "goals" / identity["goal_id"]
+    (goal / "runs").mkdir(parents=True)
+    common = {"goal_id": identity["goal_id"], "agent_id": identity["agent_id"]}
+
+    def event(kind, **details):
+        return {
+            **common,
+            "schema_version": "loopx_rollout_event_v0",
+            "event_id": kind,
+            "event_kind": kind,
+            "run_id": identity["turn_instance_id"],
+            "details": {"settlement_effect_id": identity["effect_id"], **details},
+        }
+
+    events = [
+        event(
+            "quota_should_run",
+            todo_id=identity["todo_id"],
+            settlement_effect_id=guard_effect or identity["effect_id"],
+        )
+    ]
+    runs = []
+    for present, event_kind, classification in [
+        (writeback, "refresh_state", "state_refreshed"),
+        (spend, "quota_spend", "quota_slot_spent"),
+    ]:
+        if present:
+            events.append(event(event_kind))
+            runs.append(
+                {
+                    **identity,
+                    "classification": classification,
+                    "settlement_identity": identity,
+                    "delivery_outcome": "outcome_progress",
+                }
+            )
+    if completion:
+        events.append(event("todo_complete", no_followup=True))
+    # A row without the exact native commit must not close a monitor Turn.
+    runs.append(
+        {
+            **identity,
+            "classification": "quota_monitor_poll",
+            "material_change": material,
+            "quota_monitor_poll_commit": {"effect_id": monitor_effect},
+        }
+    )
+    for path, rows in [
+        (goal / "rollout-event-log.jsonl", events),
+        (goal / "runs/index.jsonl", runs),
+    ]:
+        path.write_text("".join(json.dumps(row) + "\n" for row in rows))
+    return {
+        "schema_version": "loopx_quota_settlement_readback_request_v0",
+        "runtime_root": str(root),
+        **IDENTITY_INPUT,
+        "replan_obligation_id": None,
+        "infer_turn_instance_id": False,
+        "allow_unbound_binding": False,
+    }
+
+
+@pytest.mark.parametrize(
+    "completion,writeback,spend,expected",
+    [
+        (False, False, False, "open"),
+        (False, True, True, "open"),
+        (True, False, False, "settlement_pending"),
+        (True, True, False, "settlement_pending"),
+        (True, True, True, "settled"),
+    ],
+)
+def test_replay_phase_comes_from_real_receipt_readback(
+    tmp_path, identity, completion, writeback, spend, expected
+):
+    request = _readback_fixture(
+        tmp_path, identity, completion=completion, writeback=writeback, spend=spend
+    )
+    direct = _direct("readback", request)
+    assert effect_runtime_result("quota.settlement.read", request) == direct
+    readback = read_heartbeat_settlement(tmp_path, **IDENTITY_INPUT)
+    assert readback.replay_phase.value == direct["replay_phase"] == expected
+    assert read_heartbeat_settlement(tmp_path, **IDENTITY_INPUT) == readback
+    if not writeback:
+        assert readback.writeback.failure.kind.value == "writeback_missing"
+
+
+@pytest.mark.parametrize(
+    "suffix,expected",
+    [
+        (None, "poll_due"),
+        (":todo:another", "poll_due"),
+        (":todo:todo_witness", "settled"),
+        ("", "settled"),
+    ],
+)
+@pytest.mark.parametrize("material", [False, True])
+def test_monitor_requires_exact_commit_and_accepts_legacy_commit_id(
+    tmp_path, identity, suffix, expected, material
+):
+    effect = (
+        None
+        if suffix is None
+        else "quota-monitor-poll:witness-goal:witness-agent:witness-turn" + suffix
+    )
+    request = _readback_fixture(
+        tmp_path, identity, monitor_effect=effect, material=material
+    )
+    direct = _direct("readback", request)
+    readback = read_heartbeat_settlement(tmp_path, **IDENTITY_INPUT)
+    assert readback.monitor_phase.value == direct["monitor_phase"] == expected
+    assert readback.spend.failure is not None  # monitor closeout requires no spend
+
+
+def test_readback_rejects_other_effect_before_projecting_phases(tmp_path, identity):
+    request = _readback_fixture(
+        tmp_path,
+        identity,
+        completion=True,
+        writeback=True,
+        spend=True,
+        guard_effect="other-effect",
+    )
+    direct = _direct("readback", request)
+    readback = read_heartbeat_settlement(tmp_path, **IDENTITY_INPUT)
+    assert (
+        readback.identity.failure.kind.value
+        == direct["identity"]["result"]["failure"]["kind"]
+        == "identity_mismatch"
+    )
+    assert readback.monitor_phase is None and readback.replay_phase is None
+
+
+def test_monitor_builder_never_emits_compatibility_pending():
+    for poll, material, writeback, spend in product([False, True], repeat=4):
+        phase = receipt_bound_monitor_phase(
+            poll_present=poll,
+            material_change=material,
+            durable_writeback_present=writeback,
+            quota_spend_present=spend,
+        )
+        assert phase.value == ("settled" if poll else "poll_due")
+
+
+@pytest.mark.parametrize(
+    "phase,obligation",
+    [
+        ("poll_due", "attempt_due_monitor"),
+        ("settlement_pending", "settle_receipt_bound_monitor"),
+        ("settled", "finish_settled_receipt_bound_monitor_turn"),
+    ],
+)
+def test_monitor_consumer_retains_compatibility_pending(phase, obligation):
+    projected = preserve_heartbeat_receipt_bound_work_lane(
+        {},
+        selected_todo={
+            "todo_id": "todo_witness",
+            "selection_binding": "heartbeat_receipt",
+            "task_class": "continuous_monitor",
+            "receipt_bound_monitor_phase": phase,
+        },
+    )
+    assert projected["obligation"] == obligation
+
+
+def test_monitor_consumer_rejects_unknown_phase():
+    with pytest.raises(ValueError, match="explicit monitor phase"):
+        preserve_heartbeat_receipt_bound_work_lane(
+            {},
+            selected_todo={
+                "todo_id": "todo_witness",
+                "selection_binding": "heartbeat_receipt",
+                "task_class": "continuous_monitor",
+                "receipt_bound_monitor_phase": "unknown",
+            },
+        )
+
+
+@pytest.mark.parametrize(
+    "binding,completion,writeback,spend,expected",
+    [
+        ("todo", False, True, True, "open"),
+        ("unbound", False, True, True, "open"),
+        ("autonomous_replan", False, False, True, "open"),
+        ("autonomous_replan", False, True, False, "settlement_pending"),
+        ("autonomous_replan", False, True, True, "settled"),
+    ],
+)
+def test_replay_builder_obeys_binding_not_only_receipt_presence(
+    binding, completion, writeback, spend, expected
+):
+    assert (
+        receipt_bound_replay_phase(
+            binding_kind=binding,
+            completion_receipt_present=completion,
+            durable_writeback_present=writeback,
+            quota_spend_present=spend,
+        ).value
+        == expected
+    )
+
+
+def test_replay_bridge_rejects_unknown_binding():
+    with pytest.raises(EffectRuntimeRejected, match="unsupported settlement binding"):
+        receipt_bound_replay_phase(
+            binding_kind="unknown",
+            completion_receipt_present=True,
+            durable_writeback_present=True,
+            quota_spend_present=True,
+        )
+
+
+@pytest.mark.parametrize(
+    "completion,writeback,spend,expected",
+    [
+        (False, True, True, "open"),
+        (True, False, False, "settlement_pending"),
+        (True, True, True, "settled"),
+    ],
+)
+def test_terminal_compatibility_adapter_keeps_replay_semantics(
+    completion, writeback, spend, expected
+):
+    assert (
+        receipt_bound_terminal_phase(
+            terminal_closeout_present=completion,
+            durable_writeback_present=writeback,
+            quota_spend_present=spend,
+        ).value
+        == expected
+    )

From 277f43ab282653a534c383a046e40fe68e70c061 Mon Sep 17 00:00:00 2001
From: song <22676124+songoow@users.noreply.github.com>
Date: Sun, 20 Sep 2026 04:07:53 -0400
Subject: [PATCH 4/4] test(lark): freeze the context clock in dated runtime
 fixtures

Signed-off-by: song <22676124+songoow@users.noreply.github.com>
---
 .../extensions/test_lark_goal_topic_runtime.py  | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/tests/extensions/test_lark_goal_topic_runtime.py b/tests/extensions/test_lark_goal_topic_runtime.py
index a5e664726e..2367bf870f 100644
--- a/tests/extensions/test_lark_goal_topic_runtime.py
+++ b/tests/extensions/test_lark_goal_topic_runtime.py
@@ -5,6 +5,7 @@
 import subprocess
 import threading
 from collections.abc import Mapping
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -283,6 +284,21 @@ def test_mention_uses_existing_inbox_reply_and_ack_path(tmp_path: Path) -> None:
     assert projection["processed_count"] == 1
 
 
+@pytest.fixture
+def manager_context_clock(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep the dated fixture within retention without disabling compaction."""
+    from loopx.extensions.lark import manager_context
+
+    class FixtureDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            instant = datetime(2026, 9, 13, 6, 1, tzinfo=UTC)
+            return instant.astimezone(tz) if tz is not None else instant.replace(tzinfo=None)
+
+    monkeypatch.setattr(manager_context, "datetime", FixtureDatetime)
+
+
+@pytest.mark.usefixtures("manager_context_clock")
 def test_manager_captures_unaddressed_context_without_granting_turn_authority(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -467,6 +483,7 @@ def decision(**options: Any) -> dict[str, Any]:
     assert answer_calls == []
 
 
+@pytest.mark.usefixtures("manager_context_clock")
 def test_manager_authorized_turn_quietly_recovers_history_as_context(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: