Spaces:
Running
Running
File size: 6,446 Bytes
2527a7b 0229f82 2527a7b 0eff9d3 2527a7b 0229f82 2527a7b 7535283 0229f82 7535283 0229f82 2527a7b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | """Run the 100-case renamed, extensionless real-AI content-recognition matrix."""
from __future__ import annotations
import argparse
import json
import os
import secrets
import shutil
import subprocess
import tempfile
import time
from pathlib import Path
from smoke_content_ai import assets, command, launch, receipt_after, stop
DETECTIONS = Path("/var/lib/lumi-eggcracker/detections")
def percentile(values: list[float], point: int) -> float:
if not values:
raise RuntimeError("empty latency distribution")
ordered = sorted(values)
return ordered[min(len(ordered) - 1, (len(ordered) * point + 99) // 100 - 1)]
def main() -> int:
if os.geteuid() != 0:
raise SystemExit("content matrix must run as root")
parser = argparse.ArgumentParser()
parser.add_argument("--assets-manifest", required=True, type=Path)
parser.add_argument("--user", required=True)
parser.add_argument("--repetitions", type=int, default=100)
parser.add_argument("--output", required=True, type=Path)
args = parser.parse_args()
if (
args.repetitions != 100
or args.output.exists()
or args.output.is_symlink()
or not args.output.parent.is_dir()
):
raise SystemExit("output must be new and repetitions must equal 100")
results: dict[str, object] = {"canary_survivals": 0, "kills": 0, "result": "FAIL"}
try:
runner, model, _manifest = assets(args.assets_manifest)
starts: list[float] = []
first_seen: list[float] = []
qualification_delay: list[float] = []
qualification: list[float] = []
empties: list[float] = []
with tempfile.TemporaryDirectory(prefix="lumi-content-matrix-", dir="/tmp") as raw:
root = Path(raw)
os.chmod(root, 0o755)
disguised_runner = root / secrets.token_hex(12)
disguised_model = root / secrets.token_hex(12)
wrapper = root / f"{secrets.token_hex(8)}.py"
shutil.copyfile(runner, disguised_runner)
os.chmod(disguised_runner, 0o755)
try:
os.link(model, disguised_model)
except OSError:
shutil.copyfile(model, disguised_model)
wrapper.write_text(
"import os,sys\nos.execv(sys.argv[1], sys.argv[1:])\n", encoding="utf-8"
)
argv = command(disguised_runner, disguised_model)
if any(item.endswith(".gguf") for item in argv):
raise RuntimeError("matrix invocation accidentally exposes a model suffix")
for _ in range(args.repetitions):
canary = subprocess.Popen(["/bin/sleep", "30"], start_new_session=True)
process: subprocess.Popen[bytes] | None = None
try:
before = set(DETECTIONS.glob("*.json"))
started = time.monotonic_ns()
process = launch(args.user, wrapper, argv, root / "out", runner.parent)
receipt = receipt_after(before)
stop(process)
process = None
if (
receipt.get("detector", {}).get("profile") != "content.gguf-llama"
or receipt.get("detector", {}).get("detection_path") != "CONTENT"
or canary.poll() is not None
or receipt.get("containment", {}).get("surviving_pids")
):
raise RuntimeError("content containment or canary proof failed")
starts.append(
(receipt["containment"]["first_stop_monotonic_ns"] - started) / 1_000_000
)
observation = receipt.get("detector", {}).get("observation", {})
if observation.get("first_seen_monotonic_ns") is not None:
first_seen.append(
(observation["first_seen_monotonic_ns"] - started) / 1_000_000
)
if (
observation.get("qualified_monotonic_ns") is not None
and observation.get("first_seen_monotonic_ns") is not None
):
qualification_delay.append(
(
observation["qualified_monotonic_ns"]
- observation["first_seen_monotonic_ns"]
)
/ 1_000_000
)
qualification.append(
float(receipt["containment"]["qualification_to_first_stop_ms"])
)
empties.append(float(receipt["containment"]["trigger_to_empty_ms"]))
results["kills"] = int(results["kills"]) + 1
results["canary_survivals"] = int(results["canary_survivals"]) + 1
finally:
stop(process)
stop(canary)
results["latency_ms"] = {
"process_start_to_first_seen_p95": percentile(first_seen, 95),
"first_seen_to_qualified_p95": percentile(qualification_delay, 95),
"process_start_to_first_stop_p95": percentile(starts, 95),
"qualification_to_first_stop_p95": percentile(qualification, 95),
"trigger_to_empty_p95": percentile(empties, 95),
}
# Process start-to-qualification is retained as diagnostic evidence:
# a real model can spend seconds loading before its exact runtime ELF
# identity is observable. The release gate is deterministic response
# after qualification and the authoritative cgroup-empty proof.
if percentile(qualification, 95) >= 100 or percentile(empties, 95) >= 500:
raise RuntimeError("content latency gate failed")
results["result"] = "PASS"
args.output.write_text(json.dumps(results, sort_keys=True) + "\n", encoding="utf-8")
return 0
except (OSError, RuntimeError, TypeError, json.JSONDecodeError) as error:
results["error"] = str(error)
if not args.output.exists():
args.output.write_text(json.dumps(results, sort_keys=True) + "\n", encoding="utf-8")
raise SystemExit(f"content matrix failed: {error}") from error
if __name__ == "__main__":
raise SystemExit(main())
|