diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index 0a9caa2d1e..2d4d1caf9b 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/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 实现。原工作区和结算校验保持;目录缺失或不兼容时保留 原待结算命令以便恢复。 diff --git a/loopx/cli_commands/refresh_settlement.py b/loopx/cli_commands/refresh_settlement.py index e9cbd28847..ec52549a7e 100644 --- a/loopx/cli_commands/refresh_settlement.py +++ b/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, diff --git a/tests/control_plane/test_refresh_settle_turn.py b/tests/control_plane/test_refresh_settle_turn.py index 7c5c362c42..50962b9440 100644 --- a/tests/control_plane/test_refresh_settle_turn.py +++ b/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"]