fix(benchmark): reconcile native terminal evidence before releasing slots

benchmark/edgebench/README.md

@@ -855,6 +855,48 @@ Omit the flag on the next run to disable it. No active experiment is changed by
 installing this code. Inspect wake receipts and the saved decisions before
 claiming adoption; score countability still requires the usual integrity review.
 
+## Close a managed native attempt
+
+For a native-feedback run with an existing experiment-board row and a fresh
+concurrency admission, pass `--managed-run-json PRIVATE_BINDING.json` to the
+EdgeBench runner. Keep the binding owner-only and outside the solver workspace.
+It contains `goal_id`, the absolute controller `project` path, the original
+`admission_receipt` from `loopx benchmark concurrency-admit --execute --format json`,
+and the board `identity` (`benchmark_id`, `study_id`, `case_id`, `run_id`, `arm_id`,
+`arm_role`, `attempt`). Match `case_id` to `--task` and `run_id` to `--run-id`.
+Use the receipt that created the slot and a unique run ID across the Goal.
+
+The runner retains `managed-run.json` beside the runtime receipt. Before the
+solver starts, the receipt captures the binding and source pins. At return, it
+binds the original solver execution record when termination is confirmed. The
+runner publishes `managed-terminal.json` after validating those receipts and
+the native grading drain for the same run. The controller holds the registered
+Goal and project access.
+
+If grading drains later, the native grading owner must first retain its
+complete, drained `grading_status.json` and the matching history in the same
+attempt. To validate that evidence or retry a proof write, run the following
+once on the host that owns the attempt. Omit `--execute` to preview.
+
+```sh
+python -m benchmark.edgebench.managed_run --trial "$ATTEMPT_DIR" --execute
+```
+
+The replay reads retained evidence. Pending, unknown, or mismatched native
+evidence leaves the reservation active. After a terminal receipt write fails,
+replay can use the runner's completed result when it agrees with the original
+launch pins, the exact native solver record, and the drained grading history.
+Use the proof from the exact attempt on the controller and retain the command's
+JSON receipt with its binding.
+
+```sh
+loopx benchmark terminal-reconcile --goal-id "$GOAL_ID" --project "$PROJECT" \
+  --request-json "$ATTEMPT_DIR/managed-terminal.json" --execute --format json
+```
+
+The [toolkit closeout contract](../../loopx/capabilities/benchmark_toolkit/README.md#concurrency-envelope)
+defines readback, release and replay. Omit the flag on new ordinary trials.
+
 ## Sharing terminal results
 
 The [Portfolio ten-arm retrospective report](studies/portfolio-ten-arm-20261006/README.md)

benchmark/edgebench/run.py

@@ -120,6 +120,8 @@ def main(argv=None):
     parser.add_argument("--worker", choices=PROFILES, required=True)
     parser.add_argument("--feedback", choices=FEEDBACK_MODES, default="native",
                         help="New-run default: native public feedback; blind/best-only are explicit controls")
+    parser.add_argument("--managed-run-json", type=Path,
+                        help="Opt-in private binding for native terminal proof, consumed by the controller")
     parser.add_argument("--turn-envelope", action="store_true",
                         help="Opt-in short heartbeat context with same-invocation full captures")
     parser.add_argument("--context-delivery", choices=("off", "on"),
@@ -133,7 +135,8 @@ def main(argv=None):
     parser.add_argument("--native-goal-continue", action="store_true",
                         help="Opt-in: when a native Goal declares completion with budget left, start "
                              "continuation Goals until the budget edge (native-goal worker only)")
-    parser.add_argument("--model", required=True)
+    # Preserve the existing unique abbreviation after adding --managed-run-json.
+    parser.add_argument("--model", "--m", required=True)
     parser.add_argument("--effort", choices=REASONING_EFFORTS, required=True)
     parser.add_argument("--provider-rate-limit-retry-seconds", type=int, default=0,
                         help="Host API gateway HTTP 429 retry window for new trials; "
@@ -194,6 +197,19 @@ def main(argv=None):
     if args.native_child_limit is not None and (
             args.native_child_limit < 1 or args.feedback != "native"):
         parser.error("--native-child-limit requires a positive limit and native feedback")
+    managed_binding = None
+    if args.managed_run_json is not None:
+        if args.feedback != "native":
+            parser.error("--managed-run-json requires native feedback")
+        from benchmark.edgebench.managed_run import (
+            persist_managed_run, publish_managed_terminal, read_managed_run, retain_managed_solver,
+        )
+        try:
+            managed_binding = read_managed_run(args.managed_run_json, run_id=args.run_id, task_id=args.task)
+        except OSError:
+            parser.error("Could not read the private managed run binding")
+        except ValueError as error:
+            parser.error(str(error))
     api_base_url = os.environ.get("OPENAI_BASE_URL") or None
     api_key = os.environ.get("OPENAI_API_KEY") if api_base_url else None
     if api_base_url and not api_key:
@@ -274,6 +290,9 @@ def main(argv=None):
     # One directory is one attempt: never reuse native registration or overwrite
     # source/profile evidence after an ambiguous launch.
     trial = args.log_dir / "runs" / args.run_id / args.task
+    if (managed_binding is not None
+            and trial.resolve() != args.log_dir.resolve() / "runs" / args.run_id / args.task):
+        parser.error("Managed run requires its exact physical attempt directory")
     if args.restore_checkpoint is None:
         trial.mkdir(parents=True, exist_ok=False)
     else:
@@ -299,6 +318,8 @@ def main(argv=None):
     task = make_task_spec(task_file, load_benchmark(args.tasks_dir))
     if args.native_child_limit is not None and task.game_mode:
         parser.error("--native-child-limit requires a non-game task with workspace instructions")
+    if managed_binding is not None and task.task_id != args.task:
+        parser.error("Managed run task specification differs from the selected task")
     research_sidecar = args.feedback == "native" and not task.game_mode
     if research_sidecar:
         from benchmark.runtime.research_artifacts import validate_research_scope, with_research_capture
@@ -507,7 +528,11 @@ def main(argv=None):
         **({"preserve_failed_worker": True} if args.preserve_failed_worker else {}),
     }
     receipt_path = trial / "runtime-receipt.json"
-    receipt_path.write_text(json.dumps(receipt, indent=2))
+    managed_launch = None
+    if managed_binding is not None:
+        managed_launch = persist_managed_run(trial, managed_binding, receipt=receipt, game_mode=task.game_mode)
+    else:
+        receipt_path.write_text(json.dumps(receipt, indent=2))
     try:
         result, was_interrupted, elapsed = _observe_run(lambda: run_agent(
             task_spec=task, agent=agent, config=config, backend=backend,
@@ -555,7 +580,19 @@ def main(argv=None):
                                task=task.task_id, run_id=args.run_id,
                                model=args.model, effort=args.effort, feedback=args.feedback,
                                native_goal_outcome=native.get("native_goal_outcome"))
-    receipt_path.write_text(json.dumps(receipt, indent=2))
+    if managed_launch is None:
+        receipt_path.write_text(json.dumps(receipt, indent=2))
+    else:
+        try:
+            solver_retained = retain_managed_solver(trial, managed_binding, managed_launch, receipt=receipt, result=result)
+        except (OSError, TypeError, ValueError) as error:
+            logger.warning("Managed solver evidence unavailable (%s)", type(error).__name__)
+        else:
+            try:
+                if not solver_retained or not publish_managed_terminal(trial, execute=True):
+                    logger.warning("Managed terminal proof withheld: native solver termination or grading drain unconfirmed")
+            except (OSError, TypeError, ValueError) as error:
+                logger.warning("Managed terminal proof unavailable (%s)", type(error).__name__)
     print(json.dumps(receipt))
     return 0 if status == "terminal" else 130 if status == "cancelled" else 1
 

loopx/capabilities/benchmark_toolkit/README.md

@@ -929,6 +929,58 @@ transition, and a bounded periodic cadence, pass exact-job receipt and runner-ow
 `runtime-observation`. Apply its typed terminal or runner-invalid transition before
 releasing that reservation, then backfill the reported gap.
 
+Managed providers can deliver a completed closeout through one host operation:
+
+```bash
+loopx benchmark terminal-reconcile \
+  --goal-id <registered-goal-id> \
+  --request-json managed-terminal.json \
+  --execute --format json
+```
+
+Without `--execute` this previews the transition. The request contains the
+original executed, newly-created `admission_receipt` (not an idempotent
+admission lookup), the existing board `identity`
+(`benchmark_id`, `study_id`, `case_id`, `run_id`, `arm_id`, `arm_role`, `attempt`),
+and the provider's `terminal` proof: the same identity, terminal `status`,
+`observed_at`, and explicit `solver_terminal=true` and `grading_drained=true`.
+This is trusted controller input, not solver input. The provider must qualify
+runtime continuity against its retained launch and native event evidence before
+publishing it. A running solver, pending
+grading, an unknown observation, missing board/ledger, mismatched admission, or
+a conflicting terminal status is rejected without releasing a reservation.
+The managed path requires a unique run ID across the Goal's board; it rejects
+the same reservation appearing in multiple studies rather than guessing its owner.
+Every launch or replacement must use a fresh Goal-wide run ID. Deliberately
+reusing both an old run ID and its admission timestamp is outside this managed
+contract; the legacy reservation format cannot distinguish that duplicate tuple.
+Forward the original whole provider proof. Do not assemble its admission and
+terminal pieces from separate runs. The shared CLI validates its bound state
+transition; it does not independently reread provider files or attest the SDK.
+
+The TypeScript policy checks the exact admission generation (`admitted_at` as
+well as run/case/role). The existing local ledger transports lock the concurrency
+envelope and board in that order, recheck the policy's input snapshot, write and
+read back the terminal row, then release and read back that original slot.
+An intervening write requires retrying the original proof; it does not authorize
+a replacement admission. A process interruption after the board write can be
+replayed. Same-terminal replay preserves later scores, insight and observation
+time, and an already released run is idempotent. This is process-crash recovery,
+not a new power-loss durability guarantee. New state transitions use the actual
+host reconciliation time, strictly after the current board observation and no
+earlier than provider termination; replay keeps the persisted timestamp. A stale
+running shard can therefore be merged again without an equal-time ambiguity.
+
+This command never stops workers, deletes artifacts, restarts an experiment,
+changes duration allowances or upgrades score qualification. Keep accepted
+grading alive until it drains. The provider/controller owns bounded retry after
+an unavailable writer; no scheduler or polling loop is installed. A terminal
+board row alone is not a release receipt. Read `board_readback_verified` and
+`slot_readback_verified`; new admission still needs its own resource checks.
+Standalone board upserts and explicit `concurrency-release` retain their
+existing behavior. The managed command additionally requires a registered Goal
+and cannot use an unregistered project override.
+
 For hosts where parallel jobs can exhaust temporary storage, memory, process
 capacity, file descriptors, persistent storage, or provider capacity, enable
 `--require-resource-headroom-receipt`. Each new admission must then include a

loopx/capabilities/benchmark_toolkit/concurrency_envelope.py

@@ -640,6 +640,45 @@ def admit_benchmark_case(
         return result
 
 
+def _release_preview(
+    current: Mapping[str, Any] | None, *, compact_run_id: str, timestamp: str,
+) -> tuple[dict[str, Any], dict[str, Any] | None]:
+    if current is None:
+        return (
+            {
+                "ok": True,
+                "released": False,
+                "idempotent": True,
+                "write_performed": False,
+                "reason_codes": ["concurrency_envelope_not_configured"],
+                "run_id": compact_run_id,
+            },
+            None,
+        )
+    normalized = normalize_benchmark_concurrency_envelope(current)
+    remaining = [
+        item
+        for item in normalized["active_runs"]
+        if item["run_id"] != compact_run_id
+    ]
+    released = len(remaining) != len(normalized["active_runs"])
+    envelope = normalize_benchmark_concurrency_envelope(
+        {**normalized, "updated_at": timestamp, "active_runs": remaining}
+    )
+    return (
+        {
+            "ok": True,
+            "released": released,
+            "idempotent": not released,
+            "write_performed": False,
+            "reason_codes": [] if released else ["run_not_active"],
+            "run_id": compact_run_id,
+            "status": build_benchmark_concurrency_status(envelope),
+        },
+        envelope if released else None,
+    )
+
+
 def release_benchmark_case(
     path: str | Path,
     *,
@@ -652,52 +691,20 @@ def release_benchmark_case(
     compact_run_id = _token(run_id, field="run_id")
     timestamp = _timestamp(released_at or _now_iso(), field="released_at")
 
-    def build(
-        current: Mapping[str, Any] | None,
-    ) -> tuple[dict[str, Any], dict[str, Any] | None]:
-        if current is None:
-            return (
-                {
-                    "ok": True,
-                    "released": False,
-                    "idempotent": True,
-                    "write_performed": False,
-                    "reason_codes": ["concurrency_envelope_not_configured"],
-                    "run_id": compact_run_id,
-                },
-                None,
-            )
-        normalized = normalize_benchmark_concurrency_envelope(current)
-        remaining = [
-            item
-            for item in normalized["active_runs"]
-            if item["run_id"] != compact_run_id
-        ]
-        released = len(remaining) != len(normalized["active_runs"])
-        envelope = normalize_benchmark_concurrency_envelope(
-            {**normalized, "updated_at": timestamp, "active_runs": remaining}
-        )
-        return (
-            {
-                "ok": True,
-                "released": released,
-                "idempotent": not released,
-                "write_performed": False,
-                "reason_codes": [] if released else ["run_not_active"],
-                "run_id": compact_run_id,
-                "status": build_benchmark_concurrency_status(envelope),
-            },
-            envelope if released else None,
-        )
-
     if not execute:
-        result, _ = build(read_benchmark_concurrency_envelope(target))
+        result, _ = _release_preview(
+            read_benchmark_concurrency_envelope(target),
+            compact_run_id=compact_run_id, timestamp=timestamp,
+        )
         result["dry_run"] = True
         return result
     with exclusive_file_lock(
         target, agent_id=agent_id, operation="benchmark_concurrency_release"
     ):
-        result, envelope = build(read_benchmark_concurrency_envelope(target))
+        result, envelope = _release_preview(
+            read_benchmark_concurrency_envelope(target),
+            compact_run_id=compact_run_id, timestamp=timestamp,
+        )
         result["dry_run"] = False
         if envelope is not None:
             atomic_write_json(target, envelope)