Verify Spark Windows recovery after server-started cancellation

benchmarks/edge_harness/experiments/audit_spark_windows_restart.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Audit native-Windows Spark cancellation/restart reports and server logs."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+from pathlib import Path
+
+
+MODEL_SHA256 = "902bde2522394954ac17821b3e5fd0df02defbc6944f122253f2580acf0503f4"
+SERVER_SHA256 = "9ffc5919acb4cb43c7be5f3053b59014cce70a87d3a801fcad49c4f459984f52"
+TASK_START = re.compile(r"slot launch_slot_: id\s+\d+ \| task \d+ \| processing task")
+
+
+def sha256(path: Path) -> str:
+    with path.open("rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def clean_ledger(value: dict) -> bool:
+    return (value["reserved"] == {"host_ram": 0}
+            and value["owners"] == [] and value["quarantined"] == [])
+
+
+def audit(root: Path, route: str) -> dict:
+    report_path = root / f"{route}_v4_report.json"
+    first_log = root / f"{route}_v4_server.log"
+    second_log = root / f"{route}_v4_server_restart.log"
+    run = json.loads(report_path.read_text(encoding="utf-8"))
+    before = run["execution_plan"]
+    after = run["restart_check"]["fresh_execution_plan"]
+    abort = run["abort_check"]
+    restart = run["restart_check"]
+    if run["status"] != "passed" or run["device"] != ("cpu" if route == "cpu" else "Vulkan1"):
+        raise ValueError(f"{route}: runtime outcome or device differs")
+    if any(plan["model_sha256"] != MODEL_SHA256 or plan["server_sha256"] != SERVER_SHA256
+           for plan in (before, after)):
+        raise ValueError(f"{route}: model or server binary changed between sessions")
+    expected_offload = ["0", "29"] if route == "cpu" else ["29", "29"]
+    expected_buffers = [] if route == "cpu" else ["Vulkan1"]
+    if any(plan["offloaded_layers"] != expected_offload
+           or plan["model_buffer_devices"] != expected_buffers for plan in (before, after)):
+        raise ValueError(f"{route}: execution placement differs from the intended hardware")
+    if route == "cpu" and any(plan["all_model_buffers"] != ["CPU_Mapped", "CPU_REPACK"]
+                              for plan in (before, after)):
+        raise ValueError("CPU route had a non-CPU model buffer")
+    if route == "radeon" and any(plan["expected_device_name"] != "AMD Radeon(TM) 890M Graphics"
+                                 for plan in (before, after)):
+        raise ValueError("Radeon route did not pin the 890M by name")
+    if ([row["answer"] for row in run["checks"]] != ["Paris", "1511"]
+            or not all(row["correct"] for row in run["checks"] + run["measured"])
+            or restart["fresh_request"]["answer"] != "Paris"
+            or not restart["fresh_request"]["correct"]):
+        raise ValueError(f"{route}: named complete-request checks failed")
+    if (abort["abort_case"] != "inventory_120"
+            or abort["server_tasks_after_abort_request_start"]
+            <= abort["server_tasks_before_abort_request"]
+            or abort["stale_output"] or not abort["worker_exited"]
+            or restart["late_output"]):
+        raise ValueError(f"{route}: cancellation was not verified after server task start")
+    if not all(clean_ledger(value) for value in (
+        abort["ledger_after_abort"], restart["first_runtime_ledger_after_shutdown"],
+        run["ledger_after_shutdown"],
+    )):
+        raise ValueError(f"{route}: reservation or quarantine remained")
+    if (before["worker_generation"] == after["worker_generation"]
+            or restart["fresh_request"]["stage_event"]["worker_generation"]
+            != after["worker_generation"]):
+        raise ValueError(f"{route}: resumed request did not use a fresh worker")
+    first_starts = len(TASK_START.findall(first_log.read_text(encoding="utf-8", errors="replace")))
+    second_starts = len(TASK_START.findall(second_log.read_text(encoding="utf-8", errors="replace")))
+    if first_starts < abort["server_tasks_after_abort_request_start"] or second_starts < 1:
+        raise ValueError(f"{route}: server logs do not contain the recorded tasks")
+    return {
+        "status": "passed",
+        "device": run["device"],
+        "old_generation": before["worker_generation"],
+        "new_generation": after["worker_generation"],
+        "first_server_task_starts": first_starts,
+        "second_server_task_starts": second_starts,
+        "started_before_cancel": True,
+        "fresh_answer": restart["fresh_request"]["answer"],
+        "ledger_cleared": True,
+        "files": {p.name: {"bytes": p.stat().st_size, "sha256": sha256(p)}
+                  for p in (report_path, first_log, second_log)},
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--evidence-dir", required=True, type=Path)
+    args = parser.parse_args()
+    result = {
+        "scope": "one in-flight cancel and fresh-stage restart per native Windows Spark placement",
+        "checkpoint": "Spark-X2.5-1.7B Q4_K_M",
+        "model_sha256": MODEL_SHA256,
+        "llama_server_sha256": SERVER_SHA256,
+        "routes": {route: audit(args.evidence_dir, route) for route in ("cpu", "radeon")},
+        "limits": [
+            "A fresh StageRuntime is created after cancellation; no same-session state migration is claimed.",
+            "The backend returns complete requests rather than incremental token events.",
+            "This is one restart per placement, not concurrent or sustained recovery qualification.",
+        ],
+    }
+    path = args.evidence_dir / "audit_report.json"
+    path.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
+    print(json.dumps({route: row["status"] for route, row in result["routes"].items()}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

benchmarks/edge_harness/experiments/probe_omni_llamacpp_spark.py

@@ -11,6 +11,7 @@ import argparse
 import asyncio
 import json
 import math
+import re
 import time
 from pathlib import Path
 from types import SimpleNamespace
@@ -38,9 +39,13 @@ async def main() -> None:
     parser.add_argument("--reserve-gib", type=int, default=4)
     parser.add_argument("--abort-check", action="store_true")
     parser.add_argument("--context-check", action="store_true")
+    parser.add_argument("--restart-check", action="store_true",
+                        help="After in-flight cancellation, start a fresh stage and verify a request")
     args = parser.parse_args()
     if args.warmups < 0 or args.repeats <= 0 or args.capacity_gib < args.reserve_gib or args.reserve_gib < 4:
         parser.error("invalid serial profile or explicit memory budget")
+    if args.restart_check and not args.abort_check:
+        parser.error("--restart-check requires --abort-check")
 
     from vllm_omni.config.stage_config import (
         DeployConfig,
@@ -158,22 +163,89 @@ async def main() -> None:
                 raise RuntimeError("llama.cpp accepted a prompt beyond the declared context")
             report["context_check"]["recovery"] = await request_one(*cases[0])
         if args.abort_check:
+            def started_server_tasks() -> int:
+                log = args.server_log.read_text(encoding="utf-8", errors="replace")
+                return len(re.findall(
+                    r"slot launch_slot_: id\s+\d+ \| task \d+ \| processing task",
+                    log,
+                ))
+
+            tasks_before_abort = started_server_tasks()
             request_id = f"abort-{time.monotonic_ns()}"
             await pool.submit_initial(
                 request_id, state,
-                {"text": "Count from one upward and continue until the token limit.", "max_tokens": 96},
+                {"text": inventory, "max_tokens": 96},
             )
-            await asyncio.sleep(0.01)
+            deadline = time.monotonic() + 10
+            while started_server_tasks() <= tasks_before_abort:
+                if time.monotonic() >= deadline:
+                    raise TimeoutError("abort request did not reach the owned llama.cpp server")
+                await asyncio.sleep(0.02)
+            premature = pool.poll_graph_output(0)
+            if premature is not None:
+                premature.release_stage_buffers()
+                raise RuntimeError("abort request completed before in-flight cancellation")
             await pool.abort_requests([request_id])
             await asyncio.sleep(0.1)
             stale = pool.poll_graph_output(0)
             report["abort_check"] = {
+                "abort_case": "inventory_120",
+                "server_tasks_before_abort_request": tasks_before_abort,
+                "server_tasks_after_abort_request_start": started_server_tasks(),
                 "stale_output": stale is not None,
                 "worker_exited": pool.stage_client._proc.poll() is not None,
                 "ledger_after_abort": runtime.resource_ledger.snapshot(),
             }
+            if stale is not None:
+                stale.release_stage_buffers()
             if stale is not None or not report["abort_check"]["worker_exited"]:
                 raise RuntimeError("llama.cpp cancel did not drain worker and stale output")
+            if args.restart_check:
+                old_generation = report["execution_plan"]["worker_generation"]
+                runtime.shutdown()
+                released = runtime.resource_ledger.snapshot()
+                if released["reserved"]["host_ram"] or released["quarantined"]:
+                    raise RuntimeError("cancelled stage retained a host-RAM reservation")
+                restart_log = args.server_log.with_name(
+                    args.server_log.stem + "_restart" + args.server_log.suffix
+                )
+                restart_deploy = DeployConfig(
+                    async_chunk=False,
+                    stages=[StageDeployConfig(
+                        stage_id=0,
+                        backend={**backend, "log_file": str(restart_log)},
+                        resource_budget=deploy.stages[0].resource_budget,
+                    )],
+                )
+                restart_configs = [
+                    stage.to_omegaconf()
+                    for stage in merge_pipeline_deploy(pipeline, restart_deploy)
+                ]
+                runtime = StageRuntime(
+                    restart_configs, "local-spark-gguf-restart", "",
+                    stage_init_timeout=120, async_chunk=False,
+                )
+                restarted_at = time.perf_counter()
+                runtime.initialize()
+                fresh_startup_s = time.perf_counter() - restarted_at
+                pool = runtime.stage_pools[0]
+                state = SimpleNamespace(sampling_params_list=[None])
+                new_plan = pool.stage_client.execution_plan
+                if new_plan["worker_generation"] == old_generation:
+                    raise RuntimeError("restart reused the cancelled worker generation")
+                restarted = await request_one(*cases[0])
+                late = pool.poll_graph_output(0)
+                report["restart_check"] = {
+                    "first_runtime_ledger_after_shutdown": released,
+                    "fresh_startup_s": fresh_startup_s,
+                    "fresh_server_log": str(restart_log),
+                    "fresh_execution_plan": new_plan,
+                    "fresh_request": restarted,
+                    "late_output": late is not None,
+                }
+                if late is not None:
+                    late.release_stage_buffers()
+                    raise RuntimeError("restart emitted an extra output")
         report["status"] = "passed"
     except BaseException as exc:
         report["status"] = "failed"