Reuse verified spend receipts in combined settlement

docs/reference/protocols/turn-envelope-v0.md

@@ -263,6 +263,8 @@ and delivery postconditions it follows the canonical owed quota command for the
 same Goal, Agent, Todo/obligation and Turn, then verifies receipts. It collects no
 guard, changes no binding and performs no Todo completion. Failed validation,
 unsatisfied checkpoints, dry runs and no-spend closeout cannot debit quota.
+For a fresh append, the parent can reuse the child's authoritative readback
+when the original identity, prior receipts and new spend event all match.
 An ambiguous response is reconciled through the original receipts; exact retry
 cannot debit twice. Ordinary refresh without the option retains its behavior.
 When `--delivery-workspace-path` names the producing worktree, the combined
@@ -302,6 +304,7 @@ writeback 读取，避免重复交付；当前准入、选择与停止事实仍
 Todo 或 obligation/Turn 的 canonical 待结算命令并验证回执；不重新 guard、换绑或
 完成 Todo。验证失败、检查点未满足、dry-run 与无扣额收尾不能扣额；响应不确定时
 按原回执恢复，精确重试不会重复扣额。不传此选项保持原 refresh 行为。
+首次追加扣额成功且原身份、前序回执和新扣额事件全部匹配时，父进程可以复用子进程的权威回读。
 指定 `--delivery-workspace-path` 后，组合操作也在该交付工作区执行待结算命令，
 继续使用原调用方的 CLI 实现。原工作区和结算校验保持；目录缺失或不兼容时保留
 原待结算命令以便恢复。

loopx/cli_commands/refresh_settlement.py

@@ -13,6 +13,9 @@
 from pathlib import Path
 from typing import Any
 
+from ..control_plane.effect_program import (
+    SETTLEMENT_RECEIPT_SCHEMA_VERSION, SettlementStepKind, decode_settlement_receipt_payload,
+)
 from ..control_plane.quota.settlement import (
     SettlementIdentity, attach_settlement_progress, read_heartbeat_settlement, settlement_result_payload,
 )
@@ -46,6 +49,53 @@ def settle_requested_refresh(payload, args, registry_path, goal_ref):
             payload.update(view)
 
 
+def _has_verified_fresh_spend(
+    refresh: Mapping[str, Any], child: Mapping[str, Any], identity: Mapping[str, Any],
+) -> bool:
+    # The pinned child already read the authoritative receipt chain after
+    # append. Match it to the parent's own writeback readback before reuse.
+    progress = refresh["settlement_progress"]
+    source = progress["quota_spend_source"]
+    effect_id = identity["effect_id"]
+    if (progress.get("state") != "spend_required"
+            or child.get("ok") is not True or child.get("appended") is not True
+            or child.get("idempotent_replay") is not False or child.get("transaction_repaired") is not False
+            or child.get("dry_run") is not False or child.get("mode") != "spend-slot"
+            or child.get("settlement_identity") != identity or child.get("source") != source
+            or child.get("effect_id") != f"{effect_id}#{SettlementStepKind.QUOTA_SPEND.value}"
+            or child.get("settlement_owed") is not None or child.get("result_delivery_error") is not None
+            or child.get("settlement_progress") != {
+                "schema_version": "quota_settlement_progress_v0", "state": "settled",
+                "next_step": None, "quota_spend_source": source,
+            }):
+        return False
+    before, after = refresh.get("settlement_result"), child.get("settlement_result")
+    for result in (before, after):
+        if (not isinstance(result, dict) or set(result) != {"ok", "receipts", "failure"}
+                or result["ok"] is not True or result["failure"] is not None
+                or not isinstance(result["receipts"], list)):
+            return False
+    receipts = after["receipts"]
+    if (len(receipts) != 3 or receipts[:-1] != before["receipts"]
+            or any(not isinstance(row, dict) or row.get("schema_version") != SETTLEMENT_RECEIPT_SCHEMA_VERSION
+                   for row in receipts)):
+        return False
+    event = child.get("rollout_event")
+    if (not isinstance(event, dict) or event.get("event_kind") != "quota_spend"
+            or not isinstance(event.get("event_id"), str) or not event["event_id"]):
+        return False
+    try:
+        checked = tuple(decode_settlement_receipt_payload(row) for row in receipts)
+    except RuntimeError:
+        return False
+    return (
+        tuple(row.step_kind for row in checked)
+        == (SettlementStepKind.VALIDATION, SettlementStepKind.DURABLE_WRITEBACK, SettlementStepKind.QUOTA_SPEND)
+        and all(row.effect_id == effect_id and row.status == "committed" and row.source_ref for row in checked)
+        and checked[-1].source_ref == f"rollout_event:{event['event_id']}"
+    )
+
+
 def settle_refreshed_turn(
     payload: dict[str, Any], *, registry_path: Path, runtime_root: Path,
     goal_id: str, agent_id: str, todo_id: str | None,
@@ -124,6 +174,11 @@ def settle_refreshed_turn(
         "ok", "appended", "idempotent_replay", "json_path", "rollout_event",
         "reward_memory_ingest", "error", "error_code",
     ) if key in spent}
+    if error is None and not settled_replay and _has_verified_fresh_spend(payload, spent, expected):
+        payload["settlement_progress"] = spent["settlement_progress"]
+        payload["settlement_result"] = spent["settlement_result"]
+        payload.pop("settlement_owed", None)
+        return
     try:
         readback = read_heartbeat_settlement(
             runtime_root, registry_path=registry_path, goal_ref=goal_ref,

tests/control_plane/test_refresh_settle_turn.py

@@ -414,7 +414,8 @@ def test_execution_evidence_uses_explicit_finding_binding_in_settlement(tmp_path
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_owed_command_identity_and_lost_response_recover_without_another_debit(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("response", ["complete", "lost", "missing_receipt", "wrong_turn", "wrong_writeback", "wrong_event"])
+def test_owed_command_checks_child_receipts_or_recovers_without_another_debit(tmp_path, monkeypatch, provider, response):
     from loopx.cli_commands import refresh_settlement
 
     isolate_sqlite_runtime(tmp_path, monkeypatch)
@@ -432,22 +433,46 @@ def settle(payload):
             runtime_root=runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID,
             replan_obligation_id=None, turn_instance_id=TURN_ID, goal_ref=None)
     real_run = subprocess.run
+    real_read = refresh_settlement.read_heartbeat_settlement
     calls = []
-    def lose_response(command, **kwargs):
+    readbacks = []
+    def observe_readback(*args, **kwargs):
+        readbacks.append(kwargs["turn_instance_id"])
+        return real_read(*args, **kwargs)
+    def observe_child(command, **kwargs):
         calls.append(command)
-        real_run(command, **kwargs)
-        raise OSError("Synthetic response loss after the real commitment")
-    monkeypatch.setattr(refresh_settlement.subprocess, "run", lose_response)
+        result = real_run(command, **kwargs)
+        assert result.returncode == 0
+        if response == "lost":
+            raise OSError("Synthetic response loss after the real commitment")
+        child = json.loads(result.stdout)
+        receipts = child["settlement_result"]["receipts"]
+        if response == "missing_receipt":
+            receipts.pop()
+        elif response == "wrong_turn":
+            child["settlement_identity"]["turn_instance_id"] = "different-turn"
+        elif response == "wrong_writeback":
+            receipts[1]["source_ref"] = "rollout_event:foreign-writeback"
+        elif response == "wrong_event":
+            child["rollout_event"]["event_id"] = "different-event"
+        return subprocess.CompletedProcess(command, result.returncode, json.dumps(child), result.stderr)
+    monkeypatch.setattr(refresh_settlement, "read_heartbeat_settlement", observe_readback)
+    monkeypatch.setattr(refresh_settlement.subprocess, "run", observe_child)
     wrong = copy.deepcopy(original)
     wrong["settlement_owed"]["command"] = wrong["settlement_owed"]["command"].replace(TURN_ID, "different-turn")
     settle(wrong)
-    assert not wrong["ok"] and not calls and _spend_run_count(runtime) == 0
+    assert not wrong["ok"] and not calls and readbacks == [TURN_ID] and _spend_run_count(runtime) == 0
+    readbacks.clear()
     result = copy.deepcopy(original)
     settle(result)
     assert result["ok"] and result["settlement_progress"]["state"] == "settled", result
-    assert "Synthetic response loss" in result["quota_spend"]["transport_warning"]
+    assert readbacks == ([] if response == "complete" else [TURN_ID])
+    if response == "lost":
+        assert "Synthetic response loss" in result["quota_spend"]["transport_warning"]
     assert len(calls) == 1 and _spend_run_count(runtime) == 1
     monkeypatch.setattr(refresh_settlement.subprocess, "run", real_run)
     replay = copy.deepcopy(original)
     settle(replay)
     assert replay["ok"] and replay["quota_spend"]["idempotent_replay"] and _spend_run_count(runtime) == 1
+    assert readbacks == ([TURN_ID] if response == "complete" else [TURN_ID, TURN_ID])
+    assert result["settlement_result"] == replay["settlement_result"]