Add dry run gate for long telemetry pilot

runs/intervention-response-v2/pilot_controller.py

@@ -62,6 +62,100 @@ def configure(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
     }
 
 
+def validate_inputs(args: argparse.Namespace, manifest: dict[str, Any]) -> None:
+    if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
+        raise RuntimeError("unexpected phase-aware pilot manifest schema")
+    if manifest.get("status") != "PASS":
+        raise RuntimeError("phase-aware pilot manifest did not pass preflight")
+    failed_invariants = [
+        name
+        for name, passed in manifest.get("sanity", {}).get("invariants", {}).items()
+        if not passed
+    ]
+    if failed_invariants:
+        raise RuntimeError(f"phase-aware pilot invariants failed: {failed_invariants}")
+
+    required = {
+        "manifest": args.manifest,
+        "aituner_root": args.aituner_root,
+        "vllm_source": args.vllm_source,
+        "venv_python": args.venv / "bin/python",
+        "venv_vllm": args.venv / "bin/vllm",
+        "model": args.model,
+        "client": args.client,
+        "burnin_study": Path(manifest["burnin"]["study"]),
+    }
+    for replicate, repetition in manifest["repetitions"].items():
+        required[f"rep{replicate}_study"] = Path(repetition["study"])
+        required[f"rep{replicate}_trace"] = Path(
+            repetition["merged_trace"]["path"]
+        )
+    missing = {name: str(path) for name, path in required.items() if not path.exists()}
+    if missing:
+        raise RuntimeError(f"phase-aware pilot input paths missing: {missing}")
+
+
+def dry_run_plan(args: argparse.Namespace, manifest: dict[str, Any]) -> dict[str, Any]:
+    sessions = []
+    for index, session in enumerate(manifest["sessions"]):
+        cell = f"tp4_mns{int(session['mns'])}"
+        entry = {"cell": cell, "gpus": (0, 1, 2, 3), "port": 8950 + index}
+        repetition = manifest["repetitions"][str(session["replicate"])]
+        session_root = args.run_root / "sessions" / str(session["session"])
+        high = repetition["selections"]["high"]
+        commands = {
+            "server": base.server_command(cell, entry["gpus"], entry["port"]),
+            "warmup": client_command(
+                entry,
+                study=repetition["study"],
+                anchor=float(high["anchor"]),
+                output=session_root / "warmup",
+                warmup=True,
+            ),
+            "burnin": client_command(
+                entry,
+                study=manifest["burnin"]["study"],
+                anchor=float(manifest["burnin"]["anchor"]),
+                output=session_root / "burnin",
+                warmup=False,
+            ),
+        }
+        for level in repetition["load_order"]:
+            selection = repetition["selections"][level]
+            commands[level] = client_command(
+                entry,
+                study=repetition["study"],
+                anchor=float(selection["anchor"]),
+                output=session_root / level,
+                warmup=False,
+            )
+        sessions.append(
+            {
+                "session": session["session"],
+                "replicate": int(session["replicate"]),
+                "mns": int(session["mns"]),
+                "port": entry["port"],
+                "load_order": repetition["load_order"],
+                "remaining_projection_h20_hours": remaining_projection(
+                    len(manifest["sessions"]), index
+                ),
+                "commands": {
+                    role: shlex.join(command) for role, command in commands.items()
+                },
+            }
+        )
+    return {
+        "schema": "intervention-response-phase-aware-pilot-dry-run-v2",
+        "status": "PASS",
+        "manifest": str(args.manifest),
+        "run_root": str(args.run_root),
+        "session_count": len(sessions),
+        "projected_h20_hours": remaining_projection(len(sessions), 0),
+        "hard_cap_h20_hours": float(manifest["budget"]["hard_cap_h20_hours"]),
+        "sessions": sessions,
+    }
+
+
 def load_state(path: Path, hard_cap: float) -> dict[str, Any]:
     if path.exists():
         return json.loads(path.read_text(encoding="utf-8"))
@@ -423,21 +517,22 @@ def parser() -> argparse.ArgumentParser:
     result.add_argument("--venv", type=Path, required=True)
     result.add_argument("--model", type=Path, required=True)
     result.add_argument("--client", type=Path, required=True)
+    result.add_argument("--dry-run", action="store_true")
     return result
 
 
 def main() -> None:
     args = parser().parse_args()
     manifest = json.loads(args.manifest.read_text(encoding="utf-8"))
-    if manifest.get("schema") != "intervention-response-phase-aware-pilot-manifest-v2":
-        raise RuntimeError("unexpected phase-aware pilot manifest schema")
-    if manifest["status"] != "PASS":
-        raise RuntimeError("phase-aware pilot manifest did not pass preflight")
+    validate_inputs(args, manifest)
+    configure(args, manifest)
+    if args.dry_run:
+        print(json.dumps(dry_run_plan(args, manifest), indent=2, sort_keys=True))
+        return
     args.run_root.mkdir(parents=True, exist_ok=True)
     copied_manifest = args.run_root / "pilot-manifest.json"
     if not copied_manifest.exists():
         atomic_json(copied_manifest, manifest)
-    configure(args, manifest)
     state_path = args.run_root / "controller-state.json"
     state = load_state(state_path, base.GPU_LIMIT)
     state["status"] = "running"

runs/intervention-response-v2/prepare_pilot.py

@@ -114,18 +114,34 @@ def resolve_role_trace(base: dict[str, Any], role: str) -> Path:
     return trace
 
 
+def private_request_id(
+    *, source_sha256: str, line_number: int, original_id: str
+) -> str:
+    payload = f"{source_sha256}:{line_number}:{original_id}".encode()
+    return f"phase-v2-{hashlib.sha256(payload).hexdigest()}"
+
+
 def merge_role_traces(sources: tuple[Path, Path], target: Path) -> dict[str, Any]:
     target.parent.mkdir(parents=True, exist_ok=True)
     temporary = target.with_suffix(target.suffix + ".tmp")
     rows = 0
     with temporary.open("w", encoding="utf-8") as output:
         for source in sources:
+            source_digest = sha256_file(source)
             with source.open(encoding="utf-8") as input_file:
-                for line in input_file:
+                for line_number, line in enumerate(input_file, start=1):
                     if not line.strip():
                         continue
-                    json.loads(line)
-                    output.write(line if line.endswith("\n") else line + "\n")
+                    row = json.loads(line)
+                    original_id = str(
+                        row.get("request_id") or row.get("id") or line_number
+                    )
+                    row["request_id"] = private_request_id(
+                        source_sha256=source_digest,
+                        line_number=line_number,
+                        original_id=original_id,
+                    )
+                    output.write(json.dumps(row, ensure_ascii=False) + "\n")
                     rows += 1
     os.replace(temporary, target)
     return {
@@ -135,6 +151,7 @@ def merge_role_traces(sources: tuple[Path, Path], target: Path) -> dict[str, Any
         "rows": rows,
         "sources": [str(source) for source in sources],
         "source_sha256": [sha256_file(source) for source in sources],
+        "request_id_scheme": "sha256(source_sha256:line_number:original_id)",
     }
 
 

runs/intervention-response-v2/test_analysis.py

@@ -85,9 +85,38 @@ def main() -> None:
     assert record["offered_req_s_per_gpu"] == 0.25
     assert len(prepare.SESSION_ORDER) == 6
     assert {mns for _replicate, mns in prepare.SESSION_ORDER} == {16, 64}
+    first_id = prepare.private_request_id(
+        source_sha256="a" * 64, line_number=1, original_id="1"
+    )
+    assert first_id == prepare.private_request_id(
+        source_sha256="a" * 64, line_number=1, original_id="1"
+    )
+    assert first_id != prepare.private_request_id(
+        source_sha256="b" * 64, line_number=1, original_id="1"
+    )
     controller = load_controller_module()
     assert math.isclose(controller.remaining_projection(6, 0), 7.7)
     assert math.isclose(controller.remaining_projection(6, 5), 1.45)
+    parsed = controller.parser().parse_args(
+        [
+            "--manifest",
+            "/tmp/manifest.json",
+            "--run-root",
+            "/tmp/run",
+            "--aituner-root",
+            "/tmp/aituner",
+            "--vllm-source",
+            "/tmp/vllm",
+            "--venv",
+            "/tmp/venv",
+            "--model",
+            "/tmp/model",
+            "--client",
+            "/tmp/client.py",
+            "--dry-run",
+        ]
+    )
+    assert parsed.dry_run is True
     pilot_analysis = load_pilot_analysis_module()
     stable = pilot_analysis.stable_adjacent_features(
         [