test(cyborg): reproduce the public CAGE-2 protocol

docs/index.md

@@ -29,6 +29,7 @@ semantic and protocol authority.
 - [CybORG/CAGE-2 provisioner and backend-manifest guardrails](decisions/cyborg-cage2-provisioner-manifest-guardrails.md)
 - [CybORG conformance-composition guardrails](decisions/cyborg-conformance-guardrails.md)
 - [CybORG researcher run-and-evidence command guardrails](decisions/cyborg-researcher-command-guardrails.md)
+- [CybORG/CAGE-2 protocol-reproduction guardrails](decisions/cyborg-cage2-protocol-reproduction-guardrails.md)
 - [CybORG/CAGE-2 downstream environment-pack guardrails](decisions/cyborg-cage2-example-pack-guardrails.md)
 - [NASim researcher run-and-evidence command guardrails](decisions/nasim-researcher-command-guardrails.md)
 - [CyberBattleSim qualification guardrails](decisions/cyberbattlesim-qualification-guardrails.md)

docs/researcher-command.md

@@ -109,6 +109,71 @@ records, bounded diagnostics, summaries, machine/software inventory, and the
 final relative inventory. Study mode also retains a validated RAES collection
 record; it adds no scientific or equivalence claim.
 
+## Reproduce the public CAGE-2 evaluation protocol
+
+The frozen issue-22 path is separate from the two-seed example above. It
+predeclares and executes the complete 3 trial-length × 3 Red-policy × 1,000
+episode matrix, using one study-scoped Python random stream initialized with
+seed 153. It compares the observed RAES-mediated cumulative Blue reward against
+the published CCS Sleeper table and retains every scheduled slot, terminal
+disposition, confidence-interval input, and evidence join:
+
+```shell
+raes-adapters reproduce \
+  --phase declare \
+  --source-root <raes-adapters-checkout> \
+  --output cage2-reproduction-declaration
+
+raes-adapters reproduce \
+  --phase run \
+  --source-root <raes-adapters-checkout> \
+  --output cage2-study-output
+```
+
+The run phase requires the separately installed qualified CybORG source and
+does not fetch it. The declaration freezes source disagreements as losses: the
+checked-in evaluator says 100 episodes and binds no seed, whereas the public
+validation description says 1,000 episodes and `random.seed(153)`. The observed
+Blue implementation is a public Sleep-policy behavioral reconstruction; it is
+not represented as the unavailable submitted-agent artifact.
+
+Each valid run contains compressed, deterministic JSON with the complete
+referenced RAES evidence records, derived measures, objective and proposition
+results, and archival run record. Compression changes storage only. Every file
+and transitive inventory is SHA-256 sealed, every compressed record is parsed
+and model-validated offline, and no native observations, state, reward vectors,
+logs, or random state are published.
+
+Within a condition, the source-native session is retained and reset so the
+single seeded stream continues exactly as declared. Portable RAES snapshots are
+not retained across those resets: each scheduled slot starts a fresh archival
+run and cannot inherit another slot's action or evidence history.
+
+Recompute the aggregates and six separately-subjected evidence tiers in an
+installed environment that has no CybORG source and needs no network access:
+
+```shell
+raes-adapters reproduce \
+  --phase verify \
+  --bundle <exported-study-output> \
+  --output cage2-recomputed
+```
+
+The result supports an exact, bounded, or failed outcome-reproduction finding
+according to the predeclared method. It deliberately does not infer state or
+observation equivalence, backend identity, deterministic replay, conformance,
+or submitted-agent identity from score similarity. Those limits make the
+export usable as honest apparatus and outcome-reproduction evidence rather
+than a proof-of-concept smoke run. The adapter does not prescribe where a
+research project retains or publishes that export.
+
+Because the public submitted CCS Sleeper artifact is unavailable, a matching
+result is specifically a behavioral-baseline outcome reproduction: it shows
+that the RAES-mediated Scenario 2 apparatus reproduces the published sleeping
+baseline distribution within the frozen rule. It is not a replication of an
+unavailable submitted implementation, and `tiers.json` keeps that distinction
+in the strongest supported claim.
+
 ## Exit status
 
 | Exit | Meaning |

mkdocs.yml

@@ -50,6 +50,7 @@ nav:
       - CybORG reward, objective, and outcome projection guardrails: decisions/cyborg-evaluation-projection-guardrails.md
       - CybORG conformance-composition guardrails: decisions/cyborg-conformance-guardrails.md
       - CybORG researcher run-and-evidence command guardrails: decisions/cyborg-researcher-command-guardrails.md
+      - CybORG/CAGE-2 protocol-reproduction guardrails: decisions/cyborg-cage2-protocol-reproduction-guardrails.md
       - CybORG/CAGE-2 downstream environment-pack guardrails: decisions/cyborg-cage2-example-pack-guardrails.md
       - NASim researcher run-and-evidence command guardrails: decisions/nasim-researcher-command-guardrails.md
       - CyberBattleSim qualification guardrails: decisions/cyberbattlesim-qualification-guardrails.md

src/raes_adapters/cli.py

@@ -79,6 +79,7 @@
     run_cyborg_conformance_suite,
     verify_selected_cyborg_source,
 )
+from raes_adapters.cyborg import reproduction as cyborg_reproduction
 from raes_adapters.cyborg import researcher as cyborg_researcher
 from raes_adapters.nasim import load_qualification as load_nasim_qualification
 from raes_adapters.nasim import researcher as nasim_researcher
@@ -159,6 +160,8 @@ def _single_participant_experiment_bindings_match(
 EXIT_RUNTIME = 5
 EXIT_ARTIFACT = 6
 EXIT_INTERNAL = 70
+_OUTPUT_UNAVAILABLE_CODE = "researcher.output.unavailable"
+_OUTPUT_UNAVAILABLE_MESSAGE = "output root is unavailable"
 _INVENTORY_NAME = "inventory.json"
 _RUNTIME_FAILURE_CODE = "researcher.runtime.failure"
 _CONTROLS_INVALID_CODE = "researcher.validation.controls-invalid"
@@ -317,6 +320,12 @@ def _parser() -> _Parser:
     run_parser.add_argument("--suite", choices=("pr", "full"), default="pr")
     run_parser.add_argument("--output", type=_relative_output, required=True)
     _add_admission_arguments(run_parser)
+
+    reproduce_parser = commands.add_parser("reproduce")
+    reproduce_parser.add_argument("--phase", choices=("declare", "run", "verify"), required=True)
+    reproduce_parser.add_argument("--output", type=_relative_output, required=True)
+    reproduce_parser.add_argument("--source-root", type=Path)
+    reproduce_parser.add_argument("--bundle", type=Path)
     return parser
 
 
@@ -1391,7 +1400,7 @@ def _run_conformance(adapter: _BackendAdapter, args: argparse.Namespace) -> int:
         output = _reserve_output(args.output)
     except _OutputFailure:
         raise _CommandFailure(
-            EXIT_OUTPUT, "researcher.output.unavailable", "output root is unavailable"
+            EXIT_OUTPUT, _OUTPUT_UNAVAILABLE_CODE, _OUTPUT_UNAVAILABLE_MESSAGE
         ) from None
     try:
         conformance_args: dict[str, object] = {
@@ -1468,7 +1477,7 @@ def _native_environment(
         output = _reserve_output(args.output)
     except _OutputFailure:
         raise _CommandFailure(
-            EXIT_OUTPUT, "researcher.output.unavailable", "output root is unavailable"
+            EXIT_OUTPUT, _OUTPUT_UNAVAILABLE_CODE, _OUTPUT_UNAVAILABLE_MESSAGE
         ) from None
     return admitted, output
 
@@ -1719,13 +1728,90 @@ def _conformance_controls_absent(args: argparse.Namespace) -> bool:
 def _dispatch(args: argparse.Namespace) -> int:
     """Dispatch one parsed command through its closed execution path."""
 
-    if args.command == "inspect":
+    if args.command == "reproduce":
+        result = _reproduce(args)
+    elif args.command == "inspect":
         print(json.dumps(_adapter(args).inspection_payload(), sort_keys=True))
-        return 0
-    adapter = _adapter(args)
-    if args.command == "validate":
-        return _validated_admission(adapter, args)
-    return _run_command(adapter, args)
+        result = 0
+    else:
+        adapter = _adapter(args)
+        result = (
+            _validated_admission(adapter, args)
+            if args.command == "validate"
+            else _run_command(adapter, args)
+        )
+    return result
+
+
+def _reproduce(args: argparse.Namespace) -> int:
+    """Dispatch the frozen CAGE-2 study and offline recomputation paths."""
+
+    source_root = args.source_root
+    bundle = args.bundle
+    if args.phase in {"declare", "run"}:
+        if source_root is None or bundle is not None:
+            raise _CommandFailure(
+                EXIT_VALIDATION,
+                _CONTROLS_INVALID_CODE,
+                _CONTROLS_INVALID_MESSAGE,
+            )
+    elif bundle is None or source_root is not None:
+        raise _CommandFailure(
+            EXIT_VALIDATION,
+            _CONTROLS_INVALID_CODE,
+            _CONTROLS_INVALID_MESSAGE,
+        )
+    invocation_root = Path.cwd().resolve()
+    output = (invocation_root / args.output).resolve()
+    try:
+        relative_output = output.relative_to(invocation_root)
+    except ValueError:
+        relative_output = None
+    if relative_output is None or not relative_output.parts:
+        raise _CommandFailure(
+            EXIT_OUTPUT,
+            _OUTPUT_UNAVAILABLE_CODE,
+            _OUTPUT_UNAVAILABLE_MESSAGE,
+        )
+    result: dict[str, object]
+    try:
+        if args.phase == "declare":
+            cyborg_reproduction.write_declaration(source_root, output)
+            result = {
+                "disposition": "declared",
+                "frozen_revision": cyborg_reproduction.FROZEN_SELECTION.frozen_revision,
+                "inventory": _INVENTORY_NAME,
+            }
+        elif args.phase == "run":
+            cyborg_reproduction.run_full_study(source_root, output)
+            verified = cyborg_reproduction.verify_bundle(output)
+            result = {"disposition": "completed", "inventory": _INVENTORY_NAME, **verified}
+        else:
+            cyborg_reproduction.recompute_bundle(bundle, output)
+            result = {
+                "disposition": "verified",
+                "inventory": _INVENTORY_NAME,
+            }
+    except FileExistsError:
+        raise _CommandFailure(
+            EXIT_OUTPUT,
+            _OUTPUT_UNAVAILABLE_CODE,
+            _OUTPUT_UNAVAILABLE_MESSAGE,
+        ) from None
+    except ValueError:
+        raise _CommandFailure(
+            EXIT_VALIDATION,
+            "researcher.validation.reproduction-invalid",
+            "reproduction evidence is invalid",
+        ) from None
+    except Exception:
+        raise _CommandFailure(
+            EXIT_RUNTIME,
+            _RUNTIME_FAILURE_CODE,
+            "reproduction execution failed",
+        ) from None
+    print(json.dumps(result, sort_keys=True))
+    return 0
 
 
 def _run_command(adapter: _BackendAdapter, args: argparse.Namespace) -> int:

src/raes_adapters/cyborg/driver.py

@@ -227,7 +227,7 @@ def _add_action_cost(
     if not math.isclose(red_difference, 0.0, rel_tol=0.0, abs_tol=1e-9):
         raise ValueError
     action_cost = totals[_BLUE] - component_totals[_BLUE]
-    if action_cost > 0.0:
+    if action_cost > 0.0 and not math.isclose(action_cost, 0.0, rel_tol=0.0, abs_tol=1e-9):
         raise ValueError
     if external_address == _RESTORE and not math.isclose(
         action_cost, -1.0, rel_tol=0.0, abs_tol=1e-9
@@ -363,6 +363,31 @@ def __init__(self, *, expected_version: str) -> None:
         self._runtime_workspace: tempfile.TemporaryDirectory[str] | None = None
         self._cyborg_type: _NativeCyborgType | None = None
         self._random_states: dict[int, object] = {}
+        self._ordered_stream_state: object | None = None
+
+    def begin_ordered_stream(self, seed: int) -> None:
+        """Initialize one study-scoped Python random stream without global effects."""
+
+        if type(seed) is not int or not 0 <= seed <= 0xFFFFFFFF:
+            raise ValueError("ordered stream seed is outside the supported range")
+        self._ordered_stream_state = random.Random(seed).getstate()
+
+    def ordered_stream_checkpoint(self) -> object:
+        """Return the current opaque stream checkpoint for bounded orchestration."""
+
+        if self._ordered_stream_state is None:
+            raise RuntimeError("ordered stream is not initialized")
+        return self._ordered_stream_state
+
+    def restore_ordered_stream(self, checkpoint: object) -> None:
+        """Restore a checkpoint previously returned by this driver."""
+
+        validator = random.Random()
+        try:
+            validator.setstate(cast(tuple[object, ...], checkpoint))
+        except Exception:
+            raise ValueError("ordered stream checkpoint is invalid") from None
+        self._ordered_stream_state = checkpoint
 
     def construct(
         self,
@@ -418,6 +443,8 @@ def _construct(
                 try:
                     if seed is not None:
                         random.seed(seed)
+                    elif self._ordered_stream_state is not None:
+                        random.setstate(cast(tuple[object, ...], self._ordered_stream_state))
                     native = (
                         cyborg_type(str(scenario_path), "sim")
                         if agents is None
@@ -441,11 +468,14 @@ def _construct(
     def cleanup(self, handle: object) -> bool:
         """Shut down an owned CybORG backend without inspecting native output."""
 
+        retained_state = self._random_states.get(id(handle))
         try:
             cast(_NativeCyborg, handle).shutdown()
         except Exception:
             return False
         self._random_states.pop(id(handle), None)
+        if retained_state is not None and self._ordered_stream_state is not None:
+            self._ordered_stream_state = retained_state
         return True
 
     def reset(self, handle: object, *, seed: int | None) -> bool:
@@ -457,6 +487,11 @@ def reset(self, handle: object, *, seed: int | None) -> bool:
                 if seed is not None:
                     random.seed(seed)
                     cast(_NativeCyborg, handle).set_seed(seed)
+                else:
+                    stream_state = self._random_states.get(id(handle))
+                    if stream_state is None:
+                        raise ValueError
+                    random.setstate(cast(tuple[object, ...], stream_state))
                 cast(_NativeCyborg, handle).reset()
                 self._random_states[id(handle)] = random.getstate()
             except Exception: