From 921dcbf9c5645fb7d16a5379bb263aaf7c97ec0f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 19:22:13 +0800 Subject: [PATCH] fix(memory): recover App outcomes after refresh replay Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/reward_memory/README.md | 13 + .../reward_memory/codex_app_outcome.py | 284 ++++++++++-------- .../project_lifecycle_refresh_state.py | 14 + loopx/cli_commands/quota_reward_memory.py | 14 +- .../capabilities/reward_memory_app_outcome.ts | 53 ++++ .../control_plane/effect_runtime_handlers.ts | 2 + .../test_reward_memory_codex_app_outcome.py | 120 ++++++++ .../control_plane/test_quota_reward_memory.py | 113 +++++++ .../reward_memory_app_outcome.test.ts | 46 +++ 9 files changed, 523 insertions(+), 136 deletions(-) create mode 100644 loopx/control_plane/capabilities/reward_memory_app_outcome.ts create mode 100644 tests/control_plane/test_quota_reward_memory.py create mode 100644 tests/control_plane_ts/reward_memory_app_outcome.test.ts diff --git a/loopx/capabilities/reward_memory/README.md b/loopx/capabilities/reward_memory/README.md index b60dcee9a7..dbf8f94339 100644 --- a/loopx/capabilities/reward_memory/README.md +++ b/loopx/capabilities/reward_memory/README.md @@ -293,6 +293,19 @@ makes zero provider calls. The App does not need a separate manual `reward-memory ingest-event` command for this lifecycle. DSH currently carries recall context but does not claim this post-settlement ingest boundary. +App candidates have a stable key for the exact Goal, Agent, Todo, Turn and +settlement effect. If the first refresh omitted a reflection or its stage +failed, retry that same `refresh-state` identity with inline reflection JSON. +The original accounting and rollout receipts remain unchanged; spend discovers +the private candidate by its exact key. If spend already committed, the refresh +retry itself can finalize the candidate after reading both original settlement +steps. No second charge or manual ingestion is needed. Candidate processing is +serialized; a changed reflection or destination binding is rejected. The current +Todo validation declaration must match the staged attestation; an explicit +refresh retry reruns validation and cannot reuse an earlier pass to mask a +current failure. Unstamped legacy candidates cannot gain write authority from +old metadata. These receipts establish ingestion and readback, not memory utility. + Dashboard, CLI/status, and Lark projections reuse the same capability owner and public receipt. Dashboard writes only an ignored config pointer and registered Goal-local Agent allowlist through the existing preview/apply/readback diff --git a/loopx/capabilities/reward_memory/codex_app_outcome.py b/loopx/capabilities/reward_memory/codex_app_outcome.py index d4fc3ef34d..f2ca5cf491 100644 --- a/loopx/capabilities/reward_memory/codex_app_outcome.py +++ b/loopx/capabilities/reward_memory/codex_app_outcome.py @@ -1,6 +1,5 @@ from __future__ import annotations -import hashlib import json import os import shlex @@ -9,8 +8,12 @@ from pathlib import Path from typing import Any +from ...file_lock import exclusive_cross_runtime_file_lock from ...history import load_registry from ...materials import find_registry_goal, goal_state_path +from ...paths import resolve_runtime_root +from ...control_plane.effect_runtime import effect_runtime_result +from ...control_plane.reward_memory import reward_memory_goal_policy from ...control_plane.coordination.local_authority import ( read_canonical_todos_if_promoted, @@ -20,6 +23,7 @@ resolve_private_completion_validation_declaration, ) from ...control_plane.todos.completion_validation_projection import ( + completion_validation_declaration_sha256, project_completion_validation_authority, ) from .outcome_lifecycle import ( @@ -27,6 +31,7 @@ _identity_token, _reflection, _reflection_digest, + _validated_reflection_evidence, _write_sidecar, run_configured_turn_outcome_ingest, ) @@ -101,6 +106,25 @@ def _candidate_base( } +def _candidate_binding( + *, registry_path: Path, identity: Mapping[str, Any], digest: str, + declaration: dict[str, Any] | None, previous: Mapping[str, Any] | None, + revalidating: bool, +) -> dict[str, Any]: + goal = find_registry_goal(load_registry(registry_path), str(identity["goal_id"])) + if goal is None: + raise ValueError("Codex App reward memory Goal is unavailable") + return dict(effect_runtime_result("reward_memory.app_outcome.bind", { + "identity": dict(identity), "reflection_digest": digest, + "policy": reward_memory_goal_policy(goal, agent_id=str(identity["agent_id"])), + "validation_declaration_sha256": ( + completion_validation_declaration_sha256(declaration) if declaration else None + ), + "previous": dict(previous) if previous is not None else None, + "revalidating": revalidating, + })) + + def _canonical_todo( *, runtime_root: Path, @@ -253,6 +277,7 @@ def _run_reflection_validator( expected_refs = reflection.get("evidence_refs") if ( not isinstance(validation, Mapping) + or not isinstance(expected_refs, list) or set(validation) - allowed or validation.get("schema_version") != REFLECTION_VALIDATION_SCHEMA_VERSION or validation.get("status") != "validated" @@ -321,12 +346,10 @@ def stage_codex_app_turn_outcome_candidate( ), ) digest = _reflection_digest(reflection_json) - candidate_id = "app:" + hashlib.sha256( - ( - f"{goal_id}\n{agent_id}\n{todo_id}\n{turn_instance_id}\n" - f"{effect_id}\n{digest}" - ).encode("utf-8") - ).hexdigest()[:20] + identity = {"goal_id": goal_id, "agent_id": agent_id, "todo_id": todo_id, + "turn_instance_id": turn_instance_id, "effect_id": effect_id} + key = effect_runtime_result("reward_memory.app_outcome.identity", {"identity": identity}) + candidate_id = str(key["candidate_id"]) declaration = _validation_declaration( registry_path=registry_path, runtime_root=runtime_root, @@ -334,95 +357,73 @@ def stage_codex_app_turn_outcome_candidate( todo_id=todo_id, state_file=state_file, ) - validation_request = { - "schema_version": REFLECTION_VALIDATION_REQUEST_SCHEMA_VERSION, - "goal_id": goal_id, - "agent_id": agent_id, - "todo_id": todo_id, - "turn_instance_id": turn_instance_id, - "effect_id": effect_id, - "reflection": reflection, - "reflection_digest": digest, - } - validation = ( - _run_reflection_validator( - declaration=declaration, - workspace=validation_workspace, - request=validation_request, - reflection=reflection, - reflection_digest=digest, - ) - if declaration is not None - else { - "ok": False, - "status": "unavailable", - "validator_kind": "none", - "summary": "Todo has no caller-declared validation command", - "exit_code": None, - } - ) path = codex_app_outcome_candidate_sidecar_path( - _goal_repo(registry_path, goal_id), - goal_id=goal_id, - agent_id=agent_id, - candidate_id=candidate_id, + _goal_repo(registry_path, goal_id), goal_id=goal_id, + agent_id=agent_id, candidate_id=candidate_id, ) - previous = _load_candidate_sidecar(path) - if previous is not None and previous.get("schema_version") != ( - CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION - ): - raise ValueError("Codex App reward memory candidate sidecar is invalid") - sidecar = { - "schema_version": CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION, - "goal_id": goal_id, - "agent_id": agent_id, - "todo_id": todo_id, - "turn_instance_id": turn_instance_id, - "effect_id": effect_id, - "candidate_id": candidate_id, - "reflection": reflection, - "reflection_digest": digest, - "task_validation": validation, - "observed_at": observed_at, - "raw_content_captured": False, - } - if previous is not None: - for field in ( - "goal_id", - "agent_id", - "todo_id", - "turn_instance_id", - "effect_id", - "candidate_id", - "reflection", - "reflection_digest", - ): - if previous.get(field) != sidecar[field]: - raise ValueError("Codex App reward memory candidate identity mismatch") - previous_validation = previous.get("task_validation") - if ( - isinstance(previous_validation, Mapping) - and previous_validation.get("ok") is True + with exclusive_cross_runtime_file_lock(path, operation="app_outcome_stage"): + previous = _load_candidate_sidecar(path) + if previous is not None and ( + previous.get("schema_version") != CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION + or previous.get("reflection") != reflection ): - validation = dict(previous_validation) - elif validation.get("ok") is True: - _write_sidecar(path, sidecar) - else: + raise ValueError("Codex App reward memory candidate identity mismatch") + binding = _candidate_binding( + registry_path=registry_path, identity=identity, digest=digest, + declaration=declaration, previous=previous, revalidating=True, + ) + validation_request = { + "schema_version": REFLECTION_VALIDATION_REQUEST_SCHEMA_VERSION, + "goal_id": goal_id, + "agent_id": agent_id, + "todo_id": todo_id, + "turn_instance_id": turn_instance_id, + "effect_id": effect_id, + "reflection": reflection, + "reflection_digest": digest, + } + validation = ( + _run_reflection_validator( + declaration=declaration, + workspace=validation_workspace, + request=validation_request, + reflection=reflection, + reflection_digest=digest, + ) + if declaration is not None + else { + "ok": False, + "status": "unavailable", + "validator_kind": "none", + "summary": "Todo has no caller-declared validation command", + "exit_code": None, + } + ) + sidecar = { + "schema_version": CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION, + **binding, + "reflection": reflection, + "task_validation": validation, + "observed_at": previous.get("observed_at", observed_at) if previous else observed_at, + "raw_content_captured": False, + } + # Always keep the current independent result; an older passed invocation + # cannot mask a failed or changed validator on an explicit refresh retry. _write_sidecar(path, sidecar) - validation_bound = validation.get("ok") is True - return _candidate_base( - goal_id=goal_id, - agent_id=agent_id, - status=("validation_bound" if validation_bound else "awaiting_evidence_validation"), - reason_code=( - None if validation_bound else "reflection_not_bound_to_independent_validation" - ), - ) | { - "candidate_id": candidate_id, - "reflection_digest": digest, - "validation_bound": validation_bound, - "sidecar_receipt_reused": previous is not None, - } + validation_bound = validation.get("ok") is True + return _candidate_base( + goal_id=goal_id, + agent_id=agent_id, + status=("validation_bound" if validation_bound else "awaiting_evidence_validation"), + reason_code=( + None if validation_bound else "reflection_not_bound_to_independent_validation" + ), + ) | { + "candidate_id": candidate_id, + "reflection_digest": digest, + "validation_bound": validation_bound, + "sidecar_receipt_reused": previous is not None, + } def run_staged_codex_app_turn_outcome_ingest( @@ -433,58 +434,77 @@ def run_staged_codex_app_turn_outcome_ingest( todo_id: str, turn_instance_id: str, effect_id: str, - candidate_id: str, + candidate_id: str | None, writeback_appended: bool, spend_appended: bool, -) -> dict[str, Any]: + runtime_root: Path | None = None, +) -> dict[str, Any] | None: """Finalize one staged App reflection after exact writeback+spend readback.""" + identity = {"goal_id": goal_id, "agent_id": agent_id, "todo_id": todo_id, + "turn_instance_id": turn_instance_id, "effect_id": effect_id} + key = effect_runtime_result("reward_memory.app_outcome.identity", {"identity": identity}) path = codex_app_outcome_candidate_sidecar_path( _goal_repo(registry_path, goal_id), goal_id=goal_id, agent_id=agent_id, - candidate_id=candidate_id, + candidate_id=str(key["candidate_id"]), ) - value = _load_candidate_sidecar(path) - if value is None or value.get("schema_version") != ( - CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION - ): - raise ValueError("Codex App reward memory candidate sidecar is unavailable") - for field, expected in { - "goal_id": goal_id, - "agent_id": agent_id, - "todo_id": todo_id, - "turn_instance_id": turn_instance_id, - "effect_id": effect_id, - "candidate_id": candidate_id, - }.items(): - if value.get(field) != expected: - raise ValueError("Codex App reward memory candidate identity mismatch") - reflection = value.get("reflection") - if not isinstance(reflection, Mapping): - raise ValueError("Codex App reward memory candidate reflection is invalid") - receipt = run_configured_turn_outcome_ingest( - registry_path=registry_path, - goal_id=goal_id, - agent_id=agent_id, - turn_key=effect_id, - host_result={ - "reward_memory_reflection_json": json.dumps( - reflection, - ensure_ascii=False, - sort_keys=True, - separators=(",", ":"), - ) - }, - settlement_evidence={ + if not path.exists() and candidate_id: + # Legacy metadata is an exact lookup only, never a directory scan. A + # missing binding stamp fails closed below; it cannot acquire authority. + path = codex_app_outcome_candidate_sidecar_path( + _goal_repo(registry_path, goal_id), goal_id=goal_id, + agent_id=agent_id, candidate_id=candidate_id, + ) + if not path.exists() and not candidate_id: + return None + with exclusive_cross_runtime_file_lock(path, operation="app_outcome_finalize"): + value = _load_candidate_sidecar(path) + if value is None or value.get("schema_version") != ( + CODEX_APP_OUTCOME_SIDECAR_SCHEMA_VERSION + ): + raise ValueError("Codex App reward memory candidate sidecar is unavailable") + reflection = value.get("reflection") + if not isinstance(reflection, Mapping): + raise ValueError("Codex App reward memory candidate reflection is invalid") + serialized = json.dumps(reflection, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + declaration = _validation_declaration( + registry_path=registry_path, + runtime_root=runtime_root or resolve_runtime_root(load_registry(registry_path), None, registry_path=registry_path), + goal_id=goal_id, todo_id=todo_id, state_file=Path(), + ) + _candidate_binding( + registry_path=registry_path, identity=identity, + digest=_reflection_digest(serialized), declaration=declaration, + previous=value, revalidating=False, + ) + settlement_evidence = { "schema_version": "turn_post_settlement_evidence_v0", "task_validation": dict(value.get("task_validation") or {}), "writeback": {"ok": writeback_appended, "appended": writeback_appended}, "quota_spend": {"ok": spend_appended, "appended": spend_appended}, - }, - observed_at=str(value.get("observed_at") or "") or None, - ) - return receipt | {"host_wiring": "codex_app_refresh_spend_post_settlement"} + } + if _validated_reflection_evidence( + reflection_json=serialized, reflection=reflection, + settlement_evidence=settlement_evidence, + ) is None: + return _candidate_base( + goal_id=goal_id, agent_id=agent_id, status="evidence_validation_required", + reason_code="reflection_not_bound_to_independent_validation", + ) | {"host_wiring": "codex_app_refresh_spend_post_settlement"} + receipt = run_configured_turn_outcome_ingest( + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + turn_key=effect_id, + host_result={ + "reward_memory_reflection_json": serialized + }, + settlement_evidence=settlement_evidence, + observed_at=str(value.get("observed_at") or "") or None, + ) + return receipt | {"host_wiring": "codex_app_refresh_spend_post_settlement"} def stage_codex_app_turn_outcome_candidate_fail_open( @@ -503,7 +523,7 @@ def stage_codex_app_turn_outcome_candidate_fail_open( def run_staged_codex_app_turn_outcome_ingest_fail_open( **kwargs: Any, -) -> dict[str, Any]: +) -> dict[str, Any] | None: try: return run_staged_codex_app_turn_outcome_ingest(**kwargs) except (OSError, RuntimeError, TypeError, ValueError): diff --git a/loopx/cli_commands/project_lifecycle_refresh_state.py b/loopx/cli_commands/project_lifecycle_refresh_state.py index 382058d1d0..c6de63fdd5 100644 --- a/loopx/cli_commands/project_lifecycle_refresh_state.py +++ b/loopx/cli_commands/project_lifecycle_refresh_state.py @@ -73,6 +73,7 @@ lark_explore_graph_syncer, ) from .quota_reward_memory import ( + attach_reward_memory_ingest_after_spend, reward_memory_candidate_details, stage_reward_memory_outcome_candidate, ) @@ -756,6 +757,19 @@ def handle_refresh_state_command( elif settlement_receipt_repair: payload["receipt_repair_required"] = False payload["receipt_repaired"] = True + candidate = payload.get("reward_memory_outcome_candidate") + if (material_refresh_ready and isinstance(candidate, Mapping) + and candidate.get("validation_bound") is True): + # A late reflection may recover after spend already committed. + # Read the same settlement; never repeat its accounting effect. + attach_reward_memory_ingest_after_spend( + payload, execute=True, registry_path=registry_path, + runtime_root=runtime_root, goal_id=args.goal_id, + agent_id=args.agent_id, todo_id=args.todo_id, + turn_instance_id=args.turn_instance_id, + replan_obligation_id=getattr(args, "replan_obligation_id", None), + goal_ref=goal_ref, + ) if material_refresh_ready and post_writeback_hooks: settlement_identity = ( payload.get("settlement_identity") diff --git a/loopx/cli_commands/quota_reward_memory.py b/loopx/cli_commands/quota_reward_memory.py index c448a7de5c..ef3535b677 100644 --- a/loopx/cli_commands/quota_reward_memory.py +++ b/loopx/cli_commands/quota_reward_memory.py @@ -90,16 +90,19 @@ def attach_reward_memory_ingest_after_spend( registry_path=registry_path, goal_ref=goal_ref, ) - identity = readback.identity.value if readback else None - writeback_event = readback.writeback_event if readback else None + if readback is None: + return + identity = readback.identity.value + writeback_event = readback.writeback_event details_value = ( writeback_event.get("details") if isinstance(writeback_event, Mapping) else None ) details: Mapping[str, Any] = details_value if isinstance(details_value, Mapping) else {} candidate_id = str(details.get("reward_memory_candidate_id") or "").strip() - if not candidate_id or identity is None or not identity.todo_id: + if (identity is None or not identity.todo_id + or readback.writeback.failure is not None or readback.spend.failure is not None): return - payload["reward_memory_ingest"] = ( + result = ( run_staged_codex_app_turn_outcome_ingest_fail_open( registry_path=registry_path, goal_id=identity.goal_id, @@ -108,10 +111,13 @@ def attach_reward_memory_ingest_after_spend( turn_instance_id=identity.turn_instance_id, effect_id=identity.effect_id, candidate_id=candidate_id, + runtime_root=runtime_root, writeback_appended=readback.writeback.failure is None, spend_appended=readback.spend.failure is None, ) ) + if result is not None: + payload["reward_memory_ingest"] = result def attach_reward_memory_recall_after_should_run( diff --git a/loopx/control_plane/capabilities/reward_memory_app_outcome.ts b/loopx/control_plane/capabilities/reward_memory_app_outcome.ts new file mode 100644 index 0000000000..4bb15ab1e9 --- /dev/null +++ b/loopx/control_plane/capabilities/reward_memory_app_outcome.ts @@ -0,0 +1,53 @@ +import { createHash } from "node:crypto"; +import type { JsonObject } from "../effect_program.ts"; +import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; +import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; + +/** Stable lookup does not depend on a successful first stage or log mutation. */ +export function rewardMemoryAppOutcomeIdentity(params: JsonObject): JsonObject { + const identity = requireJsonObject(params.identity, "identity"); + const fields = ["goal_id", "agent_id", "todo_id", "turn_instance_id", "effect_id"] as const; + const values = fields.map(key => requireNonEmptyString(identity[key], key)); + const digest = createHash("sha256").update(JSON.stringify(values)).digest("hex"); + return { ...Object.fromEntries(fields.map((key, index) => [key, values[index]])), + candidate_id: `app:${digest.slice(0, 20)}` }; +} + +/** One private candidate per exact settlement. Reflection content may not replace + * an earlier candidate; configuration and validation remain separate fences. */ +export function bindRewardMemoryAppOutcome(params: JsonObject): JsonObject { + const identity = rewardMemoryAppOutcomeIdentity(params); + const policy = requireJsonObject(params.policy, "policy"); + const agent = String(identity.agent_id); + const automation = jsonObject(policy.automation) ?? {}; + const receipt = jsonObject(jsonObject(policy.enablement_receipts)?.[agent]) ?? {}; + const hash = (value: unknown) => createHash("sha256").update(JSON.stringify(value)).digest("hex"); + const expected: JsonObject = { + ...identity, + reflection_digest: requireNonEmptyString(params.reflection_digest, "reflection_digest"), + binding_fingerprint: `sha256:${hash([ + policy.binding_revision ?? null, policy.config_path ?? "", policy.config_digest ?? "", + policy.enabled === true, policy.experimental === true, + Array.isArray(policy.enabled_agents) && policy.enabled_agents.includes(agent), + automation.automatic_ingest ?? null, automation.automatic_recall ?? null, + receipt.status ?? null, receipt.config_digest ?? null, + receipt.actor_binding_verified ?? null, receipt.writability_verified ?? null, + receipt.exact_readback_verified ?? null, + ])}`, + validation_declaration_sha256: params.validation_declaration_sha256 == null ? null + : requireNonEmptyString(params.validation_declaration_sha256, "validation_declaration_sha256"), + }; + const revalidating = requireBoolean(params.revalidating, "revalidating"); + if (params.previous != null) { + const previous = requireJsonObject(params.previous, "previous"); + for (const [key, value] of Object.entries(expected)) { + // A new exact validator invocation can replace validation, never the + // candidate's identity, reflection or destination binding. + if (revalidating && key === "validation_declaration_sha256") continue; + if (previous[key] !== value) { + throw new EffectRuntimeRequestError(`Codex App reward memory candidate ${key} mismatch`); + } + } + } + return expected; +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 64b388427f..e41c81ab13 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -679,6 +679,8 @@ export function createEffectRuntimeHandlers( ["reward_memory.binding.patch", lazyHandler(() => import("./capabilities/reward_memory_binding.ts"), ({patchRewardMemoryBinding}) => patchRewardMemoryBinding)], ["reward_memory.binding.verify_readback", lazyHandler(() => import("./capabilities/reward_memory_binding.ts"), ({verifyRewardMemoryBindingReadback}) => verifyRewardMemoryBindingReadback)], ["reward_memory.binding.project", lazyHandler(() => import("./capabilities/reward_memory_binding.ts"), ({projectRewardMemoryBindings}) => projectRewardMemoryBindings)], + ["reward_memory.app_outcome.bind", lazyHandler(() => import("./capabilities/reward_memory_app_outcome.ts"), ({bindRewardMemoryAppOutcome}) => bindRewardMemoryAppOutcome)], + ["reward_memory.app_outcome.identity", lazyHandler(() => import("./capabilities/reward_memory_app_outcome.ts"), ({rewardMemoryAppOutcomeIdentity}) => rewardMemoryAppOutcomeIdentity)], [ "manager.return_delivery.normalize_attempt", lazyHandler(() => import("./collaboration/return_delivery.ts"), ({normalizeManagerReturnDeliveryAttempt}) => (params) => normalizeManagerReturnDeliveryAttempt(params.attempt)), diff --git a/tests/capabilities/test_reward_memory_codex_app_outcome.py b/tests/capabilities/test_reward_memory_codex_app_outcome.py index c626f0f180..efd6f5ce6d 100644 --- a/tests/capabilities/test_reward_memory_codex_app_outcome.py +++ b/tests/capabilities/test_reward_memory_codex_app_outcome.py @@ -2,6 +2,8 @@ import json import sys +import time +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any @@ -12,9 +14,21 @@ codex_app_outcome_candidate_sidecar_path, run_staged_codex_app_turn_outcome_ingest, stage_codex_app_turn_outcome_candidate, + run_staged_codex_app_turn_outcome_ingest_fail_open, ) +@pytest.fixture(autouse=True) +def candidate_registry(tmp_path: Path) -> None: + # Direct adapter tests keep a real invoked registry. They do not resolve an + # installed or global Goal when qualifying candidate identity fences. + (tmp_path / "registry.json").write_text(json.dumps({ + "schema_version": 1, "common_runtime_root": str(tmp_path / "runtime"), + "goals": [{"id": "goal", "repo": str(tmp_path), + "coordination": {"registered_agents": ["pilot"]}}], + })) + + def _reflection() -> str: evidence_refs = ["artifact:app-route", "receipt:app-validation"] return json.dumps( @@ -94,6 +108,112 @@ def _validator(path: Path, *, attest: bool) -> list[str]: return [sys.executable, str(path)] +@pytest.fixture +def staged_case(tmp_path, monkeypatch): + argv = _validator(tmp_path / "validate.py", attest=True) + declaration = {"validation_command": None, "validation_command_argv": argv, + "validation_label": "exact outcome", "validation_timeout_seconds": 5} + monkeypatch.setattr(codex_app_outcome, "_validation_declaration", lambda **_: declaration) + stage = dict(registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + goal_id="goal", agent_id="pilot", todo_id="todo_app", turn_instance_id="turn-app", + effect_id="effect:app", state_file=tmp_path / "STATE.md", validation_workspace=tmp_path, + reflection_json=_reflection(), observed_at="2026-10-17T00:00:00Z") + result = stage_codex_app_turn_outcome_candidate(**stage) + path = codex_app_outcome_candidate_sidecar_path(tmp_path, goal_id="goal", agent_id="pilot", candidate_id=result["candidate_id"]) + finalize = {key: stage[key] for key in ("registry_path", "runtime_root", "goal_id", "agent_id", "todo_id", "turn_instance_id", "effect_id")} + finalize.update(candidate_id=None, writeback_appended=True, spend_appended=True) + return stage, finalize, path, declaration + + +@pytest.mark.parametrize("field", ["goal_id", "agent_id", "todo_id", "turn_instance_id", "effect_id", + "candidate_id", "reflection_digest", "binding_fingerprint", "validation_declaration_sha256"]) +def test_corrupted_candidate_never_reaches_provider(staged_case, monkeypatch, field): + _, finalize, path, _ = staged_case + value = json.loads(path.read_text()) + value[field] = "other" + path.write_text(json.dumps(value)) + monkeypatch.setattr(codex_app_outcome, "run_configured_turn_outcome_ingest", lambda **_: pytest.fail("invalid candidate reached provider")) + result = run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize) + assert result["external_writes_performed"] is False + assert result["status"] == "runtime_unavailable" + + +def test_reflection_tampering_and_legacy_unstamped_candidates_are_held(staged_case, monkeypatch): + _, finalize, path, _ = staged_case + original = json.loads(path.read_text()) + monkeypatch.setattr(codex_app_outcome, "run_configured_turn_outcome_ingest", lambda **_: pytest.fail("unbound reflection reached provider")) + changed = json.loads(path.read_text()) + changed["reflection"]["content_summary"] = "A different proposed lesson." + path.write_text(json.dumps(changed)) + assert run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize)["external_writes_performed"] is False + del original["binding_fingerprint"] + path.write_text(json.dumps(original)) + assert run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize)["status"] == "runtime_unavailable" + + +@pytest.mark.parametrize("missing", ["writeback_appended", "spend_appended"]) +def test_candidate_waits_for_both_settlement_steps(staged_case, monkeypatch, missing): + _, finalize, _, _ = staged_case + monkeypatch.setattr(codex_app_outcome, "run_configured_turn_outcome_ingest", lambda **_: pytest.fail("unsettled candidate reached provider")) + result = run_staged_codex_app_turn_outcome_ingest_fail_open(**{**finalize, missing: False}) + assert result["status"] == "evidence_validation_required" + assert result["external_writes_performed"] is False + + +def test_changed_route_and_validator_cannot_reuse_earlier_attestation(staged_case, monkeypatch): + stage, finalize, _, declaration = staged_case + monkeypatch.setattr(codex_app_outcome, "run_configured_turn_outcome_ingest", lambda **_: pytest.fail("drifted candidate reached provider")) + declaration["validation_label"] = "replacement declaration" + assert run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize)["status"] == "runtime_unavailable" + # The same reflection can be independently revalidated against the current + # declaration, but the destination route cannot move with it. + assert stage_codex_app_turn_outcome_candidate(**stage)["validation_bound"] is True + registry = stage["registry_path"] + data = json.loads(registry.read_text()) + data["goals"][0]["control_plane"] = {"reward_memory": {"enabled": True, "experimental": True, "config_path": "replacement.json"}} + registry.write_text(json.dumps(data)) + assert run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize)["status"] == "runtime_unavailable" + with pytest.raises(RuntimeError, match="binding_fingerprint"): + stage_codex_app_turn_outcome_candidate(**stage) + + +def test_failed_revalidation_does_not_reuse_previous_pass(staged_case, monkeypatch): + stage, finalize, path, _ = staged_case + _validator(stage["validation_workspace"] / "validate.py", attest=False) + assert stage_codex_app_turn_outcome_candidate(**stage)["validation_bound"] is False + assert json.loads(path.read_text())["task_validation"]["ok"] is False + monkeypatch.setattr(codex_app_outcome, "run_configured_turn_outcome_ingest", lambda **_: pytest.fail("failed revalidation reached provider")) + assert run_staged_codex_app_turn_outcome_ingest_fail_open(**finalize)["status"] == "evidence_validation_required" + + +def test_concurrent_stages_cannot_replace_the_first_reflection(staged_case, monkeypatch): + stage, _, path, _ = staged_case + path.unlink() # A disposable fixture, before either competing first stage. + calls = [] + original = codex_app_outcome._run_reflection_validator + + def validate(**kwargs): + calls.append(kwargs["reflection_digest"]) + time.sleep(0.05) # Allow the other caller to contend for this candidate. + return original(**kwargs) + + monkeypatch.setattr(codex_app_outcome, "_run_reflection_validator", validate) + changed = json.loads(stage["reflection_json"]) + changed["content_summary"] = "A different competing proposal for the same Turn." + + def attempt(text): + try: + return stage_codex_app_turn_outcome_candidate(**{**stage, "reflection_json": text}) + except ValueError: + return None + + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(attempt, [stage["reflection_json"], json.dumps(changed)])) + assert sum(result is not None for result in results) == 1 + assert len(calls) == 1 + assert json.loads(path.read_text())["reflection_digest"] == calls[0] + + def test_app_refresh_stages_private_candidate_and_spend_finalizes_it( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/control_plane/test_quota_reward_memory.py b/tests/control_plane/test_quota_reward_memory.py new file mode 100644 index 0000000000..0dc9f5239b --- /dev/null +++ b/tests/control_plane/test_quota_reward_memory.py @@ -0,0 +1,113 @@ +"""Native App recovery across immutable refresh and spend receipts.""" +from __future__ import annotations + +import json +import sys +from pathlib import Path +from urllib.parse import quote + +import pytest + +from loopx.rollout_event_log import rollout_event_log_path +from tests.capabilities.test_reward_memory_codex_app_outcome import _reflection +from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TODO_ID, _initialize_fixture_authority, _run_cli, + _spend_run_count, _write_fixture, +) + + +def _prepare(tmp_path: Path): + project, runtime, registry = _write_fixture(tmp_path) + reflection = _reflection() + expected = project / "approved-reflection.json" + expected.write_text(reflection) + validator = project / "validate.py" + validator.write_text( + "import hashlib,json,sys\nfrom pathlib import Path\n" + f"expected=json.loads(Path({str(expected)!r}).read_text())\n" + "text=sys.stdin.read()\n" + "if text:\n" + " r=json.loads(text)\n" + " digest='sha256:'+hashlib.sha256(json.dumps(expected,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).hexdigest()\n" + " assert r['reflection']==expected and r['reflection_digest']==digest\n" + " print(json.dumps({'schema_version':'reward_memory_reflection_validation_v0','status':'validated','reflection_digest':digest,'evidence_refs':expected['evidence_refs']}))\n" + ) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + argv = quote(json.dumps([sys.executable, str(validator)], separators=(",", ":")), safe="") + state.write_text(state.read_text().replace("action_kind=validate -->", f"action_kind=validate validation_command_argv={argv} -->")) + _initialize_fixture_authority(project, runtime) + binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", "turn-app-recovery") + + def cli(*args): + rc, payload = _run_cli(registry, runtime, *args, cwd=project) + assert rc == 0, payload + return payload + + cli("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, *binding) + cli("todo", "complete", "--goal-id", GOAL_ID, *binding, "--claimed-by", AGENT_ID, + "--evidence", "fixture independently validated", "--next-agent-todo", "Continue fixture", "--next-claimed-by", AGENT_ID, "--next-action-kind", "implement") + refresh = ("refresh-state", "--goal-id", GOAL_ID, "--classification", "app_recovery_verified", + "--delivery-batch-scale", "single_surface", "--delivery-outcome", "outcome_progress", + *binding, "--no-global-sync", "--suppress-external-sinks") + spend = ("quota", "spend-slot", "--goal-id", GOAL_ID, "--slots", "1", "--source", "heartbeat", "--execute", *binding) + return runtime, cli, refresh, spend, reflection + + +@pytest.mark.parametrize("after_spend", [False, True]) +def test_native_replay_recovers_failed_stage_without_rewriting_or_respending(tmp_path, after_spend): + runtime, cli, refresh, spend, reflection = _prepare(tmp_path) + first = cli(*refresh, "--reward-memory-reflection-json", "not-inline-json.json") + assert first["appended"] is True + assert first["reward_memory_outcome_candidate"]["status"] == "runtime_unavailable" + rollout = rollout_event_log_path(runtime, GOAL_ID) + original_events = rollout.read_bytes() + original_runs = (runtime / "goals" / GOAL_ID / "runs/index.jsonl").read_bytes() + if after_spend: + assert "reward_memory_ingest" not in cli(*spend) + recovered = cli(*refresh, "--reward-memory-reflection-json", reflection) + assert recovered["idempotent_replay"] is True and recovered["appended"] is False + assert recovered["rollout_event"]["event_id"] == first["rollout_event"]["event_id"] + assert recovered["rollout_event"]["appended"] is False + assert recovered["reward_memory_outcome_candidate"]["status"] == "validation_bound" + finalized = recovered if after_spend else cli(*spend) + # No experiment was enabled in this fixture: the hook is reached, yet the + # existing default-off owner performs no external effect. + assert finalized["reward_memory_ingest"]["status"] == "disabled" + assert finalized["reward_memory_ingest"]["external_writes_performed"] is False + assert rollout.read_bytes().startswith(original_events) + runs = runtime / "goals" / GOAL_ID / "runs/index.jsonl" + assert runs.read_bytes().startswith(original_runs) + assert _spend_run_count(runtime) == 1 + settled_runs = runs.read_bytes() + settled_events = rollout.read_bytes() + repeated = cli(*refresh, "--reward-memory-reflection-json", reflection) + assert repeated["reward_memory_ingest"]["external_writes_performed"] is False + assert cli(*spend)["idempotent_replay"] is True + assert runs.read_bytes() == settled_runs + assert rollout.read_bytes() == settled_events + + +def test_native_late_reflection_can_follow_a_refresh_without_memory(tmp_path): + runtime, cli, refresh, spend, reflection = _prepare(tmp_path) + cli(*refresh) + assert "reward_memory_ingest" not in cli(*spend) + recovered = cli(*refresh, "--reward-memory-reflection-json", reflection) + assert recovered["idempotent_replay"] is True + assert recovered["reward_memory_outcome_candidate"]["validation_bound"] is True + assert recovered["reward_memory_ingest"]["status"] == "disabled" + assert _spend_run_count(runtime) == 1 + + +@pytest.mark.parametrize("replacement", ["malformed-json", "changed_reflection"]) +def test_failed_late_stage_cannot_finalize_an_earlier_candidate(tmp_path, replacement): + runtime, cli, refresh, spend, reflection = _prepare(tmp_path) + cli(*refresh, "--reward-memory-reflection-json", reflection) + cli(*spend) + if replacement == "changed_reflection": + changed = json.loads(reflection) + changed["content_summary"] = "A different proposal for this exact settlement." + replacement = json.dumps(changed) + failed = cli(*refresh, "--reward-memory-reflection-json", replacement) + assert failed["reward_memory_outcome_candidate"]["validation_bound"] is False + assert "reward_memory_ingest" not in failed + assert _spend_run_count(runtime) == 1 diff --git a/tests/control_plane_ts/reward_memory_app_outcome.test.ts b/tests/control_plane_ts/reward_memory_app_outcome.test.ts new file mode 100644 index 0000000000..ecdbe8b966 --- /dev/null +++ b/tests/control_plane_ts/reward_memory_app_outcome.test.ts @@ -0,0 +1,46 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { bindRewardMemoryAppOutcome, rewardMemoryAppOutcomeIdentity } from "../../loopx/control_plane/capabilities/reward_memory_app_outcome.ts"; + +const identity = {goal_id: "goal", agent_id: "writer", todo_id: "todo_a", turn_instance_id: "turn-a", effect_id: "effect:a"}; +const policy = {enabled: true, experimental: true, config_path: "private.json", config_digest: "sha256:a", enabled_agents: ["writer"], + automation: {automatic_ingest: true}, enablement_receipts: {writer: {status: "verified", config_digest: "sha256:a", writability_verified: true, exact_readback_verified: true}}}; +const request = {identity, policy, reflection_digest: "sha256:reflection", validation_declaration_sha256: "validator-a", revalidating: false}; + +test("the lookup key owns one exact settlement, independently of stage success", () => { + const expected = rewardMemoryAppOutcomeIdentity({identity}); + assert.deepEqual(rewardMemoryAppOutcomeIdentity({identity: {...identity}}), expected); + assert.equal(bindRewardMemoryAppOutcome(request).candidate_id, expected.candidate_id); + for (const field of Object.keys(identity)) { + assert.notEqual(rewardMemoryAppOutcomeIdentity({identity: {...identity, [field]: "other"}}).candidate_id, expected.candidate_id); + assert.throws(() => rewardMemoryAppOutcomeIdentity({identity: {...identity, [field]: ""}})); + } +}); + +test("replay rejects identity, reflection, binding and validator drift", () => { + const previous = bindRewardMemoryAppOutcome(request); + assert.deepEqual(bindRewardMemoryAppOutcome({...request, previous}), previous); + for (const field of Object.keys(previous)) { + assert.throws(() => bindRewardMemoryAppOutcome({...request, previous: {...previous, [field]: "other"}}), /mismatch/); + } + for (const changed of [ + {...policy, config_digest: "sha256:replacement"}, + {...policy, config_path: "replacement.json"}, + {...policy, enabled: false}, + {...policy, enabled_agents: []}, + {...policy, automation: {automatic_ingest: false}}, + {...policy, binding_revision: "new-route"}, + {...policy, enablement_receipts: {}}, + ]) assert.throws(() => bindRewardMemoryAppOutcome({...request, policy: changed, previous}), /binding_fingerprint/); + assert.throws(() => bindRewardMemoryAppOutcome({...request, previous, validation_declaration_sha256: "validator-b"}), /validation_declaration/); + assert.doesNotThrow(() => bindRewardMemoryAppOutcome({...request, previous, validation_declaration_sha256: "validator-b", revalidating: true})); + // Revalidation never permits a different reflection or a replacement route. + assert.throws(() => bindRewardMemoryAppOutcome({...request, previous, reflection_digest: "other", revalidating: true}), /reflection_digest/); + assert.throws(() => bindRewardMemoryAppOutcome({...request, previous, policy: {...policy, config_path: "other.json"}, revalidating: true}), /binding_fingerprint/); +}); + +test("other Agents' declarations do not invalidate this candidate", () => { + const previous = bindRewardMemoryAppOutcome(request); + assert.deepEqual(bindRewardMemoryAppOutcome({...request, previous, policy: {...policy, + enabled_agents: ["writer", "reviewer"], enablement_receipts: {...policy.enablement_receipts, reviewer: {status: "changed"}}}}), previous); +});