test: exercise four-model live PD cancellation and worker faults

opt_prime/examples/pd/serving_matrix/COVERAGE.md

@@ -41,4 +41,8 @@ Four models,48producer runs,31/32 EOS comparisons at6026–6428input tokens. Bot
 
 All four short-input paths pass14normal/concurrent/reuse responses each:56total,64remote prefills including cancellation requests,0local fallbacks. Requests trigger actual RNGD subprocess computation; no KV replay. Each model passes explicit cancellation, socket disconnection and subsequent reuse. Gemma/Qwen usek3; static B1 KONI usesk1 with#236. SIGPIPE-safe HTTP is#214/#215. These are B1 hardware requests with queued client concurrency, not hardware batch4. Native parity includes shared wrong answers.
 
-Width2048 live checks now pass for all four models:56responses and64remote prefills. Short+medium total112responses/128prefills, with no local fallbacks. Width8192 live checks also pass all four models:56responses/64remote prefills at6026–6428input tokens. Combined short+medium+long:168responses/192remote prefills,0fallback. Pending-prefill hardware cancellation/worker-failure recovery, cross-host RNGD-to-GPU live PD, hybrid prefix/host restore, applicable hardware batch>1, genuine producer/transfer overlap and matched native-vs-PD latency/throughput remain unverified. GPU-to-GPU live HTTP is tested separately under#240: default BF16 concurrent35/48; explicit uncached/chunk256 producer controls#241 recover48/48concurrent and12/12native/PD serial outputs. Gemma numerical#233 remains unresolved. Overall goal remains active.
+Width2048 live checks now pass for all four models:56responses and64remote prefills. Short+medium total112responses/128prefills, with no local fallbacks. Width8192 live checks also pass all four models:56responses/64remote prefills at6026–6428input tokens. Combined short+medium+long:168responses/192remote prefills,0fallback. Readiness during terminal producer failure/automatic service restart, cross-host RNGD-to-GPU live PD, hybrid prefix/host restore, applicable hardware batch>1, genuine producer/transfer overlap and matched native-vs-PD latency/throughput remain unverified. GPU-to-GPU live HTTP is tested separately under#240: default BF16 concurrent35/48; explicit uncached/chunk256 producer controls#241 recover48/48concurrent and12/12native/PD serial outputs. Gemma numerical#233 remains unresolved. Overall goal remains active.
+
+## Pending-prefill hardware faults (#244)
+
+All four models pass width2048 native-pending cancellation499, owned worker termination with two queued/in-flight500responses plus subsequent500, shared-memory unlink and explicit replacement parity.64/64normal outputs,88remote admission attempts including failures,0fallback. Error/cleanup0.4125–2.2284s excludes replacement initialization. Native-entry markers plus unfinished Futures confirm fault timing; no artificial gate/replay. CPU adapters50passed+5subtests. This is explicit replacement at an idle boundary, not automatic production restart; readiness while terminally unavailable remains to audit. See live-pd-fault244 record/evidence244.

opt_prime/examples/pd/serving_matrix/README.md

@@ -178,3 +178,10 @@ Short-input producer/full-logit parity and real shared-memory requests are verif
 `check_live_http.py` runs actual RNGD prefill and ATOM HTTP decode from token fixtures (it does not load saved KV). Pass `--batching` with the reviewed ATOM scheduler, `--http-module` with the #214 SIGPIPE-fixed server, the model/artifact cache, `--worker-python`, factory/width and fixture/baseline JSON. It checks two serial passes, four queued clients, cancel/disconnect and reuse. Factories: `prefill:Prefill` for short Qwen3, `long_width_variant:make_prefill` for the explicit experimental long Qwen variant, `gemma_prefill:Prefill`, and `koni_prefill:Prefill` for partitioned KONI widths divisible by512. KONI uses `--kv-blocks 1` with #236; other measured artifacts use3. Explicit head_dim is preserved; Llama's missing field is inferred from hidden size/attention heads.
 
 All four short-input live HTTP paths pass56normal/reuse response checks and64remote prefills including cancellations. Native-output parity does not imply answer quality. Width2048 live validation also passes all four models; width8192 live queue is pending; hardware batch remains1 and no matched performance benefit is established. See live-http234 record.
+
+### Native pending-prefill fault checks
+
+`check_live_http.py --prefill-markers "$MARKERS" --engine-factory observer_factory:make_prefill`
+adds cancellation and owned-worker termination while a native call is pending, followed by explicit replacement at an idle scheduler boundary. Set `PREFILL_OBSERVER_FACTORY` to the validated underlying adapter. `--pd-module` selects a trusted local parent worker implementation (the hardware run used #239); `--http-module` and `--batching` retain their existing roles.
+
+The observer writes atomic start/done records without delaying native execution or replaying KV. Successful normal outputs, cancelled requests and failed remote admissions are counted separately. The mode is a destructive test of **its own child process**, not a production auto-restart mechanism. Exact four-model commands and retained failure evidence are in `experiments/2026-10-24-live-pd-fault244.md` and `evidence244/`.

opt_prime/examples/pd/serving_matrix/STATUS.md

@@ -67,3 +67,5 @@ All four models:56/56 normal/concurrent/reuse responses equal native ATOM,64RNGD
 Medium live HTTP:1526–1643tokens,56/56 exact outputs and64remote/0local; short+medium112responses/128remote. Same cancellation/disconnect reuse checks pass. Long live queue active; full goal remains open. See live-http-mid234.
 
 Long live HTTP follow-up: all four width8192 routes pass56/56responses and64remote/0local prefills, cancellation/disconnect reuse included. Short+medium+long168/168responses/192remote. All queue stages exited0. HardwareB1, nativeATOM equality only; no quality, performance or overlap claim. See live-http-long234 record/evidence234.
+
+Four-model pending-prefill fault checks#244 pass:64normal output checks/88remote attempts, cancellation499, three worker-death500responses per model, slab unlink and explicit replacement reuse.50adapter tests+5subtests pass. Initial environment ABI failure preserved. Readiness, automatic restart, matched serving performance and true overlap remain distinct work.

opt_prime/examples/pd/serving_matrix/check_live_http.py

@@ -7,6 +7,7 @@
 import sys
 import threading
 import time
+import urllib.error
 import urllib.request
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -24,8 +25,10 @@
 p.add_argument("--http-module", required=True)
 p.add_argument("--baseline", required=True)
 p.add_argument("--output", required=True)
+p.add_argument("--pd-module", help="trusted local subprocess worker implementation")
+p.add_argument("--prefill-markers", help="enable pending-prefill fault checks using observer_factory")
 a = p.parse_args()
-for name, path in [
+for name, path in ([ ("opt_prime.accel.pd.pd_batching", a.pd_module) ] if a.pd_module else []) + [
     ("opt_prime.accel.platforms.atom.batching", a.batching),
     ("opt_prime.accel.serve.llmd", a.http_module),
 ]:
@@ -61,27 +64,31 @@
 worker = None
 owner = thread = server = http = None
 rows = []
+faults = {}
 try:
     model = a.model
     engine.start(model)
     c = engine.model.config.get_text_config()
-    worker = SubprocessPrefillWorker(
-        a.worker_python,
-        [
-            "--model",
-            model,
-            "--width",
-            str(a.width),
-            "--engine-factory",
-            a.engine_factory,
-        ],
-        slots=1,
-        layers=c.num_hidden_layers,
-        kv_heads=c.num_key_value_heads,
-        head_dim=getattr(c, "head_dim", None) or c.hidden_size // c.num_attention_heads,
-        max_context=a.width,
-        env=dict(os.environ),
-    )
+    def new_worker():
+        return SubprocessPrefillWorker(
+            a.worker_python,
+            [
+                "--model",
+                model,
+                "--width",
+                str(a.width),
+                "--engine-factory",
+                a.engine_factory,
+            ],
+            slots=1,
+            layers=c.num_hidden_layers,
+            kv_heads=c.num_key_value_heads,
+            head_dim=getattr(c, "head_dim", None) or c.hidden_size // c.num_attention_heads,
+            max_context=a.width,
+            env=dict(os.environ, **({"PREFILL_OBSERVER_DIR": a.prefill_markers} if a.prefill_markers else {})),
+        )
+
+    worker = new_worker()
     tok = AutoTokenizer.from_pretrained(model, local_files_only=True)
     b = engine.batcher(prefix_cache=False, prefill_worker=worker, local_prefill=False)
     service = BatchedService(
@@ -164,9 +171,88 @@ def complete(r, rid):
             time.sleep(0.01)
         assert mode not in b.by_id
         complete(cases[0], "after-" + mode)
+    extra_prefills = 0
+    if a.prefill_markers:
+        from multiprocessing import shared_memory
+
+        def wait_for(predicate, description, timeout=30):
+            end = time.monotonic() + timeout
+            while time.monotonic() < end:
+                value = predicate()
+                if value:
+                    return value
+                time.sleep(0.002)
+            raise TimeoutError(description)
+
+        def started(rid):
+            marker = Path(a.prefill_markers) / f"{worker._proc.pid}.json"
+            if not marker.exists():
+                return None
+            event = json.loads(marker.read_text())
+            with b._lock:
+                r = b.by_id.get(rid)
+                pending = r is not None and r.pending is not None and not r.pending.done()
+            return event if pending and event["phase"] == "start" else None
+
+        def outcome(rid):
+            body = {"model": model, "prompt": cases[-1]["ids"], "max_tokens": 64}
+            try:
+                with request("/v1/completions", body, rid) as response:
+                    return {"status": response.status, "body": json.load(response)}
+            except urllib.error.HTTPError as error:
+                return {"status": error.code, "body": json.loads(error.read())}
+
+        with ThreadPoolExecutor(max_workers=2) as clients:
+            pending = clients.submit(outcome, "cancel-during-prefill")
+            event = wait_for(lambda: started("cancel-during-prefill"), "native prefill entry")
+            with request("/v1/cancel", {}, "cancel-during-prefill") as cancelled:
+                assert cancelled.status == 202
+            cancelled = pending.result(timeout=30)
+            assert cancelled["status"] == 499, cancelled
+            wait_for(lambda: not b.by_id, "cancelled lane drain")
+            complete(cases[0], "after-pending-cancel")
+            faults["pending_cancel"] = {"native_entry": event, "outcome": cancelled, "reuse": True}
+            extra_prefills += 1
+
+            failed = clients.submit(outcome, "kill-during-prefill")
+            event = wait_for(lambda: started("kill-during-prefill"), "native prefill entry before kill")
+            queued = clients.submit(outcome, "queued-at-kill")
+            wait_for(lambda: "queued-at-kill" in b.by_id, "second request admission")
+            assert started("kill-during-prefill"), "native call completed before termination"
+            before = time.perf_counter()
+            worker._proc.kill()                       # only this experiment's owned child
+            results = [failed.result(timeout=30), queued.result(timeout=30)]
+            assert all(r["status"] == 500 for r in results), results
+            wait_for(lambda: not b.by_id, "failed request drain")
+            dead = outcome("submit-after-worker-death")
+            assert dead["status"] == 500, dead
+            wait_for(lambda: not b.by_id, "dead-worker rejection drain")
+            assert not b.poisoned and sorted(b.free) == [0]
+            name = worker._shm.name
+            worker.close()
+            try:
+                leaked = shared_memory.SharedMemory(name=name)
+            except FileNotFoundError:
+                pass
+            else:
+                leaked.close()
+                raise AssertionError("dead worker slab still exists")
+            faults["worker_death"] = {"native_entry": event, "outcomes": results,
+                                      "later_request": dead, "failure_s": time.perf_counter() - before,
+                                      "slab_unlinked": True}
+            extra_prefills += 3
+            worker = new_worker()
+            # Explicit replacement at a verified idle boundary, not automatic retry/restart.
+            assert not b.by_id and sorted(b.free) == [0]
+            b.worker = worker
+            complete(cases[0], "after-worker-replacement")
+            faults["worker_death"]["replacement_reuse"] = True
+
     assert b.stats_pd["local"] == 0 and sorted(b.free) == [0]
-    assert b.stats_pd["remote"] == len(rows) + 2
+    assert b.stats_pd["remote"] == len(rows) + 2 + extra_prefills
     result = {
+        "passed": True,
+        "faults": faults,
         "worker": "live_rngd_subprocess_prefill",
         "rows": rows,
         "stats_pd": b.stats_pd,
@@ -180,6 +266,10 @@ def complete(r, rid):
         ),
         flush=True,
     )
+except Exception as error:
+    Path(a.output).write_text(json.dumps({"passed": False, "error": repr(error), "rows": rows,
+                                        "faults": faults}, ensure_ascii=False, indent=2) + "\n")
+    raise
 finally:
     if owner:
         owner.submit("stop")

opt_prime/examples/pd/serving_matrix/evidence244/environment-failure.json

@@ -0,0 +1,9 @@
+{
+  "passed": false,
+  "phase": "before engine initialization",
+  "cause": "Container-global Torch selected instead of established Torch2.11.0+cpu; ATOM extension ABI undefined symbol",
+  "initial_exit": 1,
+  "fix": "Select existing package base with process-local PYTHONUSERBASE; no installation or HOME modification",
+  "verified_torch": "2.11.0+cpu",
+  "verified_transformers": "5.8.1"
+}

opt_prime/examples/pd/serving_matrix/evidence244/events.jsonl

@@ -0,0 +1,3 @@
+{"id": "11a8f7bf084d4e33b17ea4cb18fd82e4", "ts": "2026-10-24T00:05:18+09:00", "task": "Live pending-prefill fault recovery (aicomp #244)", "result": "Prepared native-call observer and live HTTP cancellation/termination/replacement checks; launched owned RNGD+ATOM Qwen0.6B width2048 run35484 using subprocess369c8ff. No fault success claimed yet.", "next": "Monitor same process and verify pending native entry, bounded errors, slab cleanup and replacement output parity; preserve failure evidence.", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {"width": "2048", "hardware_batch": "1"}, "artifacts": [], "experiments": ["experiments/2026-10-24-live-pd-fault244.md"], "labels": ["experiment", "rngd"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"}
+{"id": "9c5a3488784d400d8f179c382b8e0e29", "ts": "2026-10-24T00:10:47+09:00", "task": "Live pending-prefill fault recovery (aicomp #244)", "result": "Qwen0.6B real RNGD->ATOM pending cancel499/reuse pass; owned worker kill yields3bounded500responses,slab unlinked and replacement parity pass(0.4125s failure+cleanup).16successful outputs/22remote attempts. Environment failure retained;3model queue57029 active.", "next": "Collect remaining Qwen4B/Gemma/KONI fault results and publish instrumentation/evidence; no automatic-restart claim.", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {"width": "2048", "hardware_batch": "1"}, "artifacts": [{"path": "analysis/serving-matrix-201/live244-qwen06b-faults.json", "size": 13654, "sha256": "8bd4bfe2eac28439"}], "experiments": ["experiments/2026-10-24-live-pd-fault244.md"], "labels": ["experiment", "rngd"], "review": false, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"}
+{"id": "f4346de8bd5c4af084f29e6f65d979ac", "ts": "2026-10-24T00:17:06+09:00", "task": "Live pending-prefill fault recovery (aicomp #244)", "result": "All4models pass real RNGD->ATOM pending-cancel499,worker-death3x500,slab unlink and explicit replacement parity.64/64normal outputs,88remote attempts/0fallback. Error+cleanup0.4125–2.2284s excluding restart.50adapter tests+5subtests pass; all jobs terminal.", "next": "Publish fault instrumentation/evidence; audit dead-producer readiness, then matched performance and compute-transfer overlap.", "state": "active", "done": false, "blocker": "", "metrics": {}, "params": {"width": "2048", "hardware_batch": "1"}, "artifacts": [{"path": "analysis/serving-matrix-201/live244-summary.json", "size": 1418, "sha256": "81c8de87ffdba800"}], "experiments": ["experiments/2026-10-24-live-pd-fault244.md"], "labels": ["experiment", "rngd"], "review": true, "todos": [], "note": "", "agent": "codex", "host": "xeon-max", "branch": "main", "commit": "351d39d-dirty"}