From be108100c49c8b9f991b15623590e188ebb1903f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 23 Oct 2026 04:13:34 +0800 Subject: [PATCH] fix(memory): accept bound selection receipts in turn recall Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/agent_turn_recall/README.md | 7 + loopx/capabilities/agent_turn_recall/cli.py | 82 +++++++--- tests/capabilities/test_agent_turn_recall.py | 101 ++++++++++++ .../test_selected_turn_recall_cli.py | 153 ++++++++++++++++++ 4 files changed, 323 insertions(+), 20 deletions(-) create mode 100644 tests/control_plane/test_selected_turn_recall_cli.py diff --git a/loopx/capabilities/agent_turn_recall/README.md b/loopx/capabilities/agent_turn_recall/README.md index 55543d7678..9de6ca2bdc 100644 --- a/loopx/capabilities/agent_turn_recall/README.md +++ b/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. diff --git a/loopx/capabilities/agent_turn_recall/cli.py b/loopx/capabilities/agent_turn_recall/cli.py index b3c106ad43..8941c85cdd 100644 --- a/loopx/capabilities/agent_turn_recall/cli.py +++ b/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( diff --git a/tests/capabilities/test_agent_turn_recall.py b/tests/capabilities/test_agent_turn_recall.py index ca8d2ccc4e..9e8a3cd9c1 100644 --- a/tests/capabilities/test_agent_turn_recall.py +++ b/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"] diff --git a/tests/control_plane/test_selected_turn_recall_cli.py b/tests/control_plane/test_selected_turn_recall_cli.py new file mode 100644 index 0000000000..c361e45571 --- /dev/null +++ b/tests/control_plane/test_selected_turn_recall_cli.py @@ -0,0 +1,153 @@ +"""Native selection packets reach recall without changing their bound work.""" + +from __future__ import annotations + +import hashlib +import json +import sys +from pathlib import Path + +import pytest + +from loopx.capabilities.agent_turn_recall.runtime import agent_turn_recall_receipt_path +from loopx.capabilities.reward_memory.experiment import load_reward_memory_experiment_config +from tests.capabilities.test_agent_turn_recall import active_record, raw_config +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, + ALTERNATIVE_TODO_ID, + GOAL_ID, + _configure_selectable_alternative, + _heartbeat_receipt_count, + _initialize_fixture_authority, + _run_cli, + _spend_run_count, + _write_fixture, +) + + +def _enable_fixture_memory(project: Path, registry: Path) -> Path: + """Use the real provider adapter with a credential-free subprocess fixture.""" + binary = project / "memory-provider" + binary.write_text( + f"#!{sys.executable}\n" + "import json,sys\nfrom pathlib import Path\n" + "root=Path(__file__).parent\nargs=sys.argv[1:]\nactor=None\n" + "if args[0]=='--actor-peer-id': actor,args=args[1],args[2:]\n" + "op=args[0]\n" + "if op=='--version': print('OpenViking 0.4.23')\n" + "elif op=='version': print('Server: 0.5.0')\n" + "elif op=='status': print('{}')\n" + "elif op=='search':\n" + " with (root/'search-count').open('a') as f: f.write('search\\n')\n" + " scope=args[args.index('-u')+1]\n" + " assert actor==scope.split('/peers/')[1].split('/')[0]\n" + " print(json.dumps({'resources':[{'uri':scope+'/guidance.json','level':2}]}))\n" + "elif op=='read': print(json.dumps({'content':(root/'guidance.json').read_text()}))\n" + "else: raise SystemExit(2)\n", + encoding="utf-8", + ) + binary.chmod(0o700) + config = raw_config(goal_id=GOAL_ID, agent_id=AGENT_ID, peer_ref=f"agent:{AGENT_ID}") + config["project_provider_binding"]["provider_binary"] = str(binary) + relative = ".loopx/config/reward-memory/recall.json" + path = project / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(config), encoding="utf-8") + normalized = load_reward_memory_experiment_config(project=project, config_path=relative) + (project / "guidance.json").write_text( + active_record(normalized, expires_at="2099-01-01T00:00:00Z"), encoding="utf-8" + ) + digest = "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() + data = json.loads(registry.read_text()) + data["goals"][0]["control_plane"] = {"reward_memory": { + "enabled": True, "experimental": True, "config_path": relative, + "config_digest": digest, "enabled_agents": [AGENT_ID], + "enablement_receipts": {AGENT_ID: { + "schema_version": "reward_memory_enablement_receipt_v0", "status": "verified", + "goal_id": GOAL_ID, "agent_id": AGENT_ID, "config_digest": digest, + "provider_id": "openviking", "isolation_mode": "goal_scoped_agent_private", + "actor_binding_verified": True, "writability_verified": True, + "exact_readback_verified": True, "probe_count": 1, "write_count": 1, + "external_writes_performed": True, "observed_at": "2026-01-01T00:00:00Z", + }}, + }} + registry.write_text(json.dumps(data), encoding="utf-8") + return project / "search-count" + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_native_selected_packet_recall_and_rejection_boundaries( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, +) -> None: + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + _initialize_fixture_authority(project, runtime, provider=provider) + turn = "turn-selected-memory" + guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--turn-instance-id", turn) + rc, first = _run_cli(registry, runtime, *guard) + assert rc == 0, first + assert first["heartbeat_receipt"]["settlement_binding_owed"] is True + rc, selected = _run_cli(registry, runtime, *guard, "--todo-id", ALTERNATIVE_TODO_ID) + assert rc == 0, selected + assert selected["heartbeat_receipt"]["status"] == "upgraded" + # Bind the synthetic provider after selection so quota has not recalled yet. + searches = _enable_fixture_memory(project, registry) + packet = project / "selected.json" + packet.write_text(json.dumps(selected), encoding="utf-8") + recall = ("agent-turn-recall", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", turn, "--quota-decision-json", str(packet)) + rc, preview = _run_cli(registry, runtime, *recall) + assert rc == 0, json.dumps(preview) + assert preview["status"] == "preview" and not searches.exists() + assert preview["context"]["situation"]["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + rc, applied = _run_cli(registry, runtime, *recall, "--execute") + assert rc == 0, applied + assert applied["status"] == "applied" and applied["provider_call_count"] == 1 + receipt = agent_turn_recall_receipt_path(project, goal_id=GOAL_ID, agent_id=AGENT_ID) + persisted = receipt.read_bytes() + rc, replay = _run_cli(registry, runtime, *recall, "--execute") + assert rc == 0 and replay["status"] == "deduplicated", replay + assert replay["context"] == applied["context"] and replay["provider_call_count"] == 0 + assert receipt.read_bytes() == persisted + + for field in ("goal_id", "agent_identity", "turn", "todo", "effect", "status", "binding"): + invalid = json.loads(json.dumps(selected)) + if field == "goal_id": + invalid[field] = "other-goal" + elif field == "agent_identity": + invalid[field]["agent_id"] = "other-agent" + elif field == "turn": + invalid["heartbeat_receipt"]["turn_instance_id"] = "other-turn" + elif field == "todo": + invalid["heartbeat_receipt"]["settlement_identity"]["todo_id"] = "todo_other" + elif field == "effect": + invalid["heartbeat_receipt"]["settlement_identity"]["effect_id"] = "unrelated-effect" + elif field == "status": + invalid["heartbeat_receipt"]["status"] = "write_failed" + else: + del invalid["heartbeat_receipt"]["settlement_identity"] + packet.write_text(json.dumps(invalid), encoding="utf-8") + rc, refused = _run_cli(registry, runtime, *recall, "--execute") + assert rc == 2 and refused["status"] == "invalid_request", refused + assert refused["provider_call_count"] == 0 and receipt.read_bytes() == persisted + + selected["decision_freshness_warning"] = {"rebase_required_count": 1} + packet.write_text(json.dumps(selected), encoding="utf-8") + rc, conflict = _run_cli(registry, runtime, *recall, "--execute") + assert rc == 0 and conflict["status"] == "not_available", conflict + assert conflict["provider_call_count"] == 0 and conflict["context"]["guidance"] == [] + assert conflict["context"]["situation"]["conflict_state"] == "unresolved" + assert searches.read_text().splitlines() == ["search"] + assert _heartbeat_receipt_count(runtime, turn) == 2 + assert _spend_run_count(runtime) == 0 + + data = json.loads(registry.read_text()) + data["goals"][0]["control_plane"]["reward_memory"]["enabled"] = False + registry.write_text(json.dumps(data), encoding="utf-8") + rc, disabled = _run_cli(registry, runtime, *recall, "--execute") + assert rc == 0 and disabled["provider_call_count"] == 0, disabled + assert disabled["status"] == "disabled" + assert searches.read_text().splitlines() == ["search"]