Add exact state observation modes

runs/frontier-fidelity-envelope-v1/run_qwen235_fixed_pd_state_replay.py

@@ -37,6 +37,7 @@ ALLOWED_FLAG_CHANGES = {
     "--metrics_config_store_frontier_stage_batch_ledger",
     "--no-metrics_config_keep_individual_batch_metrics",
     "--metrics_config_keep_individual_batch_metrics",
+    "--metrics_config_enable_op_level_tracing",
 }
 
 
@@ -104,13 +105,18 @@ def _semantic_options(command: list[str]) -> dict[str, tuple[str, ...]]:
 
 
 def transform_command(
-    command: list[str], *, metrics_root: Path, run_id: str
+    command: list[str], *, metrics_root: Path, run_id: str, op_trace: bool = False
 ) -> list[str]:
     result = replace_option(
         command, "--metrics_config_output_dir", str(metrics_root.resolve())
     )
     result = replace_option(result, "--metrics_config_run_id", run_id)
     result = enable_state_outputs(result)
+    op_trace_flag = "--metrics_config_enable_op_level_tracing"
+    if op_trace:
+        if op_trace_flag in result:
+            raise ValueError(f"frozen command already enables {op_trace_flag}")
+        result.append(op_trace_flag)
 
     before = _semantic_options(command)
     after = _semantic_options(result)
@@ -130,6 +136,8 @@ def transform_command(
     }
     if not required <= changed:
         raise ValueError(f"required controlled changes missing: {sorted(required - changed)}")
+    if op_trace and op_trace_flag not in changed:
+        raise ValueError("op-trace observation flag was not audited as a command change")
     return result
 
 
@@ -158,6 +166,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--config", action="append", choices=CONFIGS)
     parser.add_argument("--timeout-seconds", type=float, default=1200)
     parser.add_argument("--resume", action="store_true")
+    parser.add_argument("--op-trace", action="store_true")
     return parser.parse_args()
 
 
@@ -170,6 +179,7 @@ def run_cell(
     python_deps: Path,
     timeout_seconds: float,
     resume: bool,
+    op_trace: bool,
     q30: Any,
 ) -> dict[str, Any]:
     base_run = base_sim_root / "runs" / config / "eval"
@@ -192,6 +202,7 @@ def run_cell(
             == sha256_file(base_command_path)
             and previous.get("inputs", {}).get("base_result_sha256")
             == sha256_file(base_result_path)
+            and bool(previous.get("op_trace_enabled")) == op_trace
         ):
             return previous
     if run_root.exists() and any(run_root.iterdir()):
@@ -202,6 +213,7 @@ def run_cell(
         base_command,
         metrics_root=run_root / "frontier_metrics",
         run_id=f"qwen235_fixed_pd_state_{config}",
+        op_trace=op_trace,
     )
     atomic_json(run_root / "command.json", command)
     manifest = {
@@ -218,7 +230,7 @@ def run_cell(
             "metrics run id",
             "full Frontier stage/batch ledger enabled",
             "individual batch metrics enabled",
-        ],
+        ] + (["op-level tracing enabled"] if op_trace else []),
         "command_sha256": sha256_file(run_root / "command.json"),
         "frontier": {
             "source": str(frontier_source),
@@ -232,6 +244,7 @@ def run_cell(
             "NVIDIA_VISIBLE_DEVICES": "void",
             "FRONTIER_LOG_LEVEL": "WARNING",
         },
+        "op_trace_enabled": op_trace,
     }
     atomic_json(run_root / "run_manifest.json", manifest)
 
@@ -281,6 +294,13 @@ def run_cell(
         ledger_path=paths["ledger"],
     )
     atomic_json(run_root / "common-state.json", state)
+    if op_trace:
+        op_trace_matches = sorted(run_root.glob("frontier_metrics/**/op_traces.jsonl"))
+        if len(op_trace_matches) != 1:
+            raise ValueError(
+                f"expected one op trace for {config}, found {len(op_trace_matches)}"
+            )
+        paths["op_trace"] = op_trace_matches[0]
     result = {
         "schema": "qwen235-fixed-pd-state-replay-result-v1",
         "status": "PASS",
@@ -299,6 +319,7 @@ def run_cell(
         },
         "common_state": state,
         "collective_fallback_evidence": fallback_evidence,
+        "op_trace_enabled": op_trace,
     }
     atomic_json(result_path, result)
     return result
@@ -320,6 +341,7 @@ def main() -> None:
             python_deps=args.python_deps,
             timeout_seconds=args.timeout_seconds,
             resume=args.resume,
+            op_trace=args.op_trace,
             q30=q30,
         )
         results.append(result)