fix(memory): consume committed selection receipts in turn recall

loopx/capabilities/agent_turn_recall/README.md

@@ -70,6 +70,13 @@ keeps turn admission bounded and ensures the recall query uses the same selected
 Todo and interaction state that the host is following. `-` may be used to read
 the packet from stdin when the host owns a safe pipeline.
 
+An explicit selection or requalification may return an `upgraded` committed
+receipt. Recall accepts that packet with its exact settlement identity, checking
+the Goal, Agent, Turn, selected Todo or replan obligation, and canonical effect
+identity. A missing or inconsistent binding is rejected before provider access.
+Legacy `committed` and `replayed` packets without a settlement binding remain
+valid; all existing conflict, freshness, scope and deduplication checks still apply.
+
 The result carries a private `context.guidance` list for agent reasoning. It is
 not action authority. The agent must still obey the current interaction
 contract, capability gates, write scopes, and user gates.

loopx/capabilities/agent_turn_recall/cli.py

@@ -10,6 +10,10 @@
 
 from ...history import load_registry
 from ...materials import find_registry_goal, goal_repo
+from ...control_plane.quota.effect_program import SettlementIdentity
+from ...control_plane.work_items.autonomous_replan_obligation import (
+    replan_obligation_id_from_packet,
+)
 from ..reward_memory.experiment import resolve_reward_memory_experiment
 from .core import (
     AGENT_TURN_RECALL_SCHEMA_VERSION,
@@ -51,6 +55,22 @@ def _render(payload: dict[str, object]) -> str:
     return "\n".join(lines) + "\n"
 
 
+def _runtime_unavailable_payload(goal_id: str, agent_id: str) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "schema_version": AGENT_TURN_RECALL_SCHEMA_VERSION,
+        "status": "runtime_unavailable",
+        "reason_code": "automatic_recall_runtime_failed",
+        "goal_id": goal_id,
+        "agent_id": agent_id,
+        "provider_call_count": 0,
+        "grants_new_action_authority": False,
+        "quota_spend_performed": False,
+        "external_writes_performed": False,
+        "suppress_external_sinks": True,
+    }
+
+
 def register_agent_turn_recall_commands(
     subparsers: argparse._SubParsersAction[argparse.ArgumentParser],
     add_subcommand_format: Callable[[argparse.ArgumentParser], None],
@@ -132,8 +152,37 @@ def _validate_quota_identity(
         raise ValueError(
             "quota decision turn_instance_id does not match --turn-instance-id"
         )
-    if receipt.get("status") not in {"committed", "replayed"}:
+    if receipt.get("status") not in {"committed", "replayed", "upgraded"}:
         raise ValueError("quota decision heartbeat receipt is not committed")
+    binding = receipt.get("settlement_identity")
+    if binding is None and receipt.get("status") != "upgraded":
+        return
+    try:
+        identity = SettlementIdentity.from_runtime_payload(binding)
+    except RuntimeError:
+        raise ValueError("quota decision settlement identity is invalid") from None
+    if (
+        identity.goal_id != goal_id
+        or identity.agent_id != agent_id
+        or identity.turn_instance_id != turn_instance_id
+    ):
+        raise ValueError("quota decision settlement identity does not match this turn")
+    if identity.todo_id:
+        if identity.todo_id != _mapping(quota_decision.get("selected_todo")).get("todo_id"):
+            raise ValueError("quota decision settlement identity does not match selected Todo")
+    elif identity.replan_obligation_id != replan_obligation_id_from_packet(
+        quota_decision.get("autonomous_replan_obligation")
+    ):
+        raise ValueError("quota decision settlement identity does not match replan obligation")
+    expected = SettlementIdentity(
+        goal_id=goal_id,
+        agent_id=agent_id,
+        todo_id=identity.todo_id,
+        turn_instance_id=turn_instance_id,
+        replan_obligation_id=identity.replan_obligation_id,
+    )
+    if identity.effect_id != expected.effect_id:
+        raise ValueError("quota decision settlement effect identity is invalid")
 
 
 def _load_receipt(path: Path) -> dict[str, Any] | None:
@@ -207,12 +256,17 @@ def handle_agent_turn_recall_command(
         else:
             identity = _identity_scope(config)
             quota_decision = _quota_decision(args.quota_decision_json)
-            _validate_quota_identity(
-                quota_decision,
-                goal_id=args.goal_id,
-                agent_id=args.agent_id,
-                turn_instance_id=args.turn_instance_id,
-            )
+            try:
+                _validate_quota_identity(
+                    quota_decision,
+                    goal_id=args.goal_id,
+                    agent_id=args.agent_id,
+                    turn_instance_id=args.turn_instance_id,
+                )
+            except RuntimeError:
+                payload = _runtime_unavailable_payload(args.goal_id, args.agent_id)
+                print_payload(payload, output_format(args), _render)
+                return 2
             situation = build_agent_turn_situation(
                 quota_decision,
                 goal_id=args.goal_id,
@@ -250,19 +304,7 @@ def handle_agent_turn_recall_command(
                 except RuntimeError:
                     # This failure precedes the provider; do not fabricate a
                     # zero-call receipt for errors after recall has begun.
-                    payload = {
-                        "ok": False,
-                        "schema_version": AGENT_TURN_RECALL_SCHEMA_VERSION,
-                        "status": "runtime_unavailable",
-                        "reason_code": "automatic_recall_runtime_failed",
-                        "goal_id": args.goal_id,
-                        "agent_id": args.agent_id,
-                        "provider_call_count": 0,
-                        "grants_new_action_authority": False,
-                        "quota_spend_performed": False,
-                        "external_writes_performed": False,
-                        "suppress_external_sinks": True,
-                    }
+                    payload = _runtime_unavailable_payload(args.goal_id, args.agent_id)
                     print_payload(payload, output_format(args), _render)
                     return 2
                 payload = run_agent_turn_recall(

tests/capabilities/test_agent_turn_recall.py

@@ -43,6 +43,7 @@
     load_reward_memory_experiment_config,
     resolve_reward_memory_surface_config,
 )
+from loopx.control_plane.quota.effect_program import SettlementIdentity
 
 
 SURFACE = "agent_workflow.turn_admission"
@@ -436,6 +437,75 @@ def test_configured_session_scope_is_canonical() -> None:
         _resolve_session_ref(identity, "session:other")
 
 
+def bound_quota_decision(*, status: str = "upgraded") -> dict[str, Any]:
+    turn = "turn-selected-recall"
+    return quota_decision() | {
+        "goal_id": "goal",
+        "agent_identity": {"agent_id": "pilot"},
+        "heartbeat_receipt": {
+            "turn_instance_id": turn,
+            "status": status,
+            "settlement_identity": SettlementIdentity(
+                goal_id="goal", agent_id="pilot", todo_id="todo_123",
+                turn_instance_id=turn,
+            ).as_dict(),
+        },
+    }
+
+
+@pytest.mark.parametrize("status", ["committed", "replayed", "upgraded"])
+def test_bound_quota_receipt_matches_selected_work(status: str) -> None:
+    _validate_quota_identity(
+        bound_quota_decision(status=status), goal_id="goal", agent_id="pilot",
+        turn_instance_id="turn-selected-recall",
+    )
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [("goal_id", "other"), ("agent_id", "other"),
+     ("turn_instance_id", "other"), ("todo_id", "todo_other"),
+     ("effect_id", "unrelated-effect"), ("schema_version", "unknown")],
+)
+def test_selected_quota_receipt_rejects_inconsistent_binding(field: str, value: str) -> None:
+    decision = bound_quota_decision()
+    decision["heartbeat_receipt"]["settlement_identity"][field] = value
+    with pytest.raises(ValueError, match="settlement"):
+        _validate_quota_identity(
+            decision, goal_id="goal", agent_id="pilot",
+            turn_instance_id="turn-selected-recall",
+        )
+
+
+def test_upgraded_quota_receipt_requires_committed_binding() -> None:
+    decision = bound_quota_decision()
+    del decision["heartbeat_receipt"]["settlement_identity"]
+    with pytest.raises(ValueError, match="settlement identity"):
+        _validate_quota_identity(
+            decision, goal_id="goal", agent_id="pilot",
+            turn_instance_id="turn-selected-recall",
+        )
+
+
+def test_upgraded_replan_receipt_uses_its_own_binding() -> None:
+    decision = bound_quota_decision()
+    decision["heartbeat_receipt"]["settlement_identity"] = SettlementIdentity(
+        goal_id="goal", agent_id="pilot", todo_id=None,
+        turn_instance_id="turn-selected-recall", replan_obligation_id="replan-0123456789abcdef",
+    ).as_dict()
+    decision["autonomous_replan_obligation"] = {"obligation_id": "replan-0123456789abcdef"}
+    _validate_quota_identity(
+        decision, goal_id="goal", agent_id="pilot",
+        turn_instance_id="turn-selected-recall",
+    )
+    decision["autonomous_replan_obligation"]["obligation_id"] = "replan-fedcba9876543210"
+    with pytest.raises(ValueError, match="replan obligation"):
+        _validate_quota_identity(
+            decision, goal_id="goal", agent_id="pilot",
+            turn_instance_id="turn-selected-recall",
+        )
+
+
 def test_same_turn_receipt_round_trip_reuses_private_context(tmp_path: Path) -> None:
     receipt_path = tmp_path / "pilot.json"
     receipt = {
@@ -613,6 +683,37 @@ def unavailable(operation: str, _params: Any) -> Any:
     assert not agent_turn_recall_receipt_path(tmp_path, goal_id="goal", agent_id="pilot").exists()
 
 
+def test_selected_receipt_identity_transport_failure_stops_before_provider(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from loopx.control_plane import effect_program
+
+    config = normalized_config(tmp_path)
+    decision = bound_quota_decision()
+    monkeypatch.setattr(recall_cli, "resolve_reward_memory_experiment", lambda **_kwargs: ({"status": "available"}, config))
+    monkeypatch.setattr(recall_cli, "_goal_repo", lambda *_args: tmp_path)
+    monkeypatch.setattr(recall_cli, "_quota_decision", lambda *_args: decision)
+
+    def unavailable(*_args: Any, **_kwargs: Any) -> Any:
+        raise RuntimeError("private runtime startup detail")
+
+    monkeypatch.setattr(effect_program, "effect_runtime_result", unavailable)
+    effect_program._settlement_identity_full.cache_clear()
+    monkeypatch.setattr(recall_cli, "run_agent_turn_recall", lambda *_args, **_kwargs: pytest.fail("provider boundary reached"))
+    payloads: list[dict[str, Any]] = []
+    code = recall_cli.handle_agent_turn_recall_command(
+        argparse.Namespace(command="agent-turn-recall", goal_id="goal", agent_id="pilot",
+                           turn_instance_id="turn-selected-recall", quota_decision_json="unused",
+                           session_ref=None, force_refresh=False, execute=True),
+        registry_path=tmp_path / "registry.json", output_format=lambda *_args: "json",
+        print_payload=lambda payload, *_args: payloads.append(payload),
+    )
+    assert code == 2 and payloads[0]["status"] == "runtime_unavailable"
+    assert payloads[0]["provider_call_count"] == 0
+    assert "private runtime" not in json.dumps(payloads[0])
+    assert not agent_turn_recall_receipt_path(tmp_path, goal_id="goal", agent_id="pilot").exists()
+
+
 def test_expiry_survives_candidate_review_and_activation(tmp_path: Path) -> None:
     config = normalized_config(tmp_path)
     corpus = resolve_reward_memory_surface_config(config, SURFACE)["corpus"]