Nick
5b3ff5bcc0
- tui/skills (G14, LAST package): pattern proposals with consent-first
law (nothing drafted until explicit y), activation views with test
evidence, D10 self-transformation proposal-only truth (fixture walks
the whole keyboard proving no activation path), consent-drift guard
- G7 delta over mid-flight 5db2e249: SSH replay fixed to GREEN (setsid
controlling-tty, boot-ACK wait, burst-boundary parsing, JSON drain) —
main-session verified: 4/4 mutants RED, ssh GREEN, 21/21 pytest
- verification/slowest-target-slo/: schema + validated records; budgets
proposed-only (agreed=false per GAP-BT-05)
- 23/23 tui packages green by main-session own run (cmd/rune-tui
mid-write by wave H1 version wiring — excluded here, lands with H1)
1141 lines
46 KiB
Python
1141 lines
46 KiB
Python
#!/usr/bin/env python3
|
|
"""check_migration_mutants.py — phase 0B task 0.11: the migration-oracle
|
|
mutant suite + SSH-profile replay + slowest-target SLO recording.
|
|
|
|
I-R2-008 (remaining gate): "Migration-oracle mutant suite (the named
|
|
four mutants) plus dead-Python verification after quarantine expiry."
|
|
The parity unit is the authority-preserving action/receipt trace —
|
|
action IDs, key scopes, focus transitions, breakpoint outcomes, theme
|
|
tokens, oracle traces — NEVER widget parity. This script:
|
|
|
|
1. Builds the AUTHORITATIVE trace (the migration oracle baseline) from
|
|
the frozen receipts at research/tui-frontier-r2/receipts/ (focus,
|
|
resize, burst, keyprobe) plus explicitly pinned values for what the
|
|
receipts do not yet carry (recovery segment, launcher binding,
|
|
theme tokens — each labeled in the trace's `pinned` block). The
|
|
baseline is sha256-pinned in-module: drift is a RED, never silent.
|
|
2. Applies the four NAMED mutants — key scope, receipt origin,
|
|
recovery ordering, launcher target — and requires the oracle to go
|
|
RED on every one, with the named invariant among the fired checks.
|
|
A mutant that survives (or a harness that fires nothing) is a
|
|
failed suite: FALSE GREEN IS NOT RETIREMENT EVIDENCE.
|
|
3. --ssh-replay: replays the FIXED input/burst/resize traces under an
|
|
SSH profile (real ssh when RUNE_SSH_TARGET/--ssh-target answers
|
|
BatchMode; otherwise an honest PTY-profile fallback, TERM pinned
|
|
xterm-256color like the keyprobe receipts). The known
|
|
transport-stable divergence (atomic ESC+q -> "alt+q" vs split ->
|
|
"esc","q", EV-BT-004) is an EXPECTATION, replayed as such.
|
|
4. --slo-record/--slo-validate: the slowest-target SLO recording
|
|
structure (verification/slowest-target-slo/) — budgets proposed
|
|
from the lab receipts, `agreed: false` until the owner ratifies
|
|
(GAP-BT-05: no agreed SLO yet), every record honest about whether
|
|
its target is actually the slowest supported one.
|
|
|
|
This is the MIGRATION oracle, distinct from the lab iso-gate mutant
|
|
suite (receipts/at-oracle/mutants — action-isomorphism class). The
|
|
unified runner declares this suite as member `migration_mutants`
|
|
(scripts/unified_check.py MEMBERS); wiring it live (run:True + IMPL
|
|
entry) is a main-session edit — this file lands standalone first.
|
|
|
|
Usage:
|
|
python scripts/check_migration_mutants.py # mutant suite
|
|
python scripts/check_migration_mutants.py --ssh-replay # trace replay
|
|
python scripts/check_migration_mutants.py --slo-record # SLO record
|
|
python scripts/check_migration_mutants.py --slo-validate FILE
|
|
python scripts/check_migration_mutants.py --pins # print pins
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
RECEIPTS_DIR = REPO / "research/tui-frontier-r2/receipts"
|
|
RECEIPT_OUT_DIR = RECEIPTS_DIR / "migration-mutants"
|
|
SLO_DIR = REPO / "verification/slowest-target-slo"
|
|
|
|
# The canonical recovery phase order — the ordering law the recovery
|
|
# mutant violates. Pinned (constructed authority record; real recovery
|
|
# receipts land with tasks 7.1-7.3 and must re-pin against this oracle).
|
|
RECOVERY_PHASES = ["checkpoint", "degrade", "restart", "restore",
|
|
"resume", "reattach"]
|
|
|
|
# Keys that carry LANE authority (focus transitions), per the frozen
|
|
# focus receipt's key/action pairing. Everything modal-scoped while the
|
|
# approve dialog is open; plain keys otherwise.
|
|
LANE_KEYS = {"1", "2", "tab"}
|
|
|
|
|
|
# ── canonical json + pins ─────────────────────────────────────────────────
|
|
|
|
|
|
def canonical_json(obj) -> str:
|
|
return json.dumps(obj, sort_keys=True, separators=(",", ":"),
|
|
ensure_ascii=False)
|
|
|
|
|
|
def canonical_sha256(obj) -> str:
|
|
return hashlib.sha256(canonical_json(obj).encode()).hexdigest()
|
|
|
|
|
|
# ── fixed-trace loading (the frozen receipts) ─────────────────────────────
|
|
|
|
|
|
def load_fixed_traces(receipts_dir: Path = RECEIPTS_DIR) -> dict:
|
|
"""Load the fixed burst/resize/input traces. These files are FROZEN
|
|
research receipts — read-only inputs, never rewritten here."""
|
|
bt = receipts_dir / "bubbletea"
|
|
ao = receipts_dir / "at-oracle"
|
|
paths = {
|
|
"focus": bt / "focus.json",
|
|
"resize": bt / "resize.json",
|
|
"burst": bt / "burst-n5000-slow0.json",
|
|
"keyprobe_atomic": ao / "keyprobe-ssh-atomic-1.json",
|
|
"keyprobe_split": ao / "keyprobe-ssh-split-1.json",
|
|
}
|
|
loaded = {}
|
|
for name, p in paths.items():
|
|
if not p.is_file():
|
|
raise FileNotFoundError(
|
|
f"fixed trace receipt missing: {p} — the migration "
|
|
f"oracle grounds itself in the frozen receipts; refusing "
|
|
f"to invent data")
|
|
loaded[name] = json.loads(p.read_text())
|
|
return loaded
|
|
|
|
|
|
def _key_scope(action_id: str, key: str, modal_open: bool) -> str:
|
|
"""Scope under whose authority a key was interpreted — the field the
|
|
key-scope mutant corrupts. Derived from the frozen focus receipt's
|
|
evidence (q_swallowed_in_modal, modal open/close pairing)."""
|
|
if modal_open:
|
|
return "modal.approve"
|
|
if action_id.startswith("modal.open"):
|
|
return "task-list"
|
|
if key in LANE_KEYS:
|
|
return "lane"
|
|
if action_id.startswith("nav."):
|
|
return "task-list"
|
|
return "input"
|
|
|
|
|
|
def build_baseline(receipts_dir: Path = RECEIPTS_DIR) -> dict:
|
|
"""Project the authoritative action/receipt trace from the frozen
|
|
receipts. Deterministic: same receipts in, same trace out, pinned
|
|
by BASELINE_SHA256."""
|
|
tr = load_fixed_traces(receipts_dir)
|
|
focus = tr["focus"]
|
|
actions = focus["actions"]
|
|
keys = focus["keys_via_parser"]
|
|
|
|
session = "sess-authority-0001"
|
|
steps = []
|
|
modal_open = False
|
|
focus_state = "chat"
|
|
for i, (action_id, key) in enumerate(zip(actions, keys), start=1):
|
|
scope = _key_scope(action_id, key, modal_open)
|
|
before = focus_state
|
|
if action_id == "focus.chat":
|
|
focus_state = "chat"
|
|
elif action_id == "focus.tasks":
|
|
focus_state = "tasks"
|
|
# focus_restored (frozen receipt): modal open/swallow/cancel
|
|
# never moves focus; nav stays on the tasks lane.
|
|
steps.append({
|
|
"seq": i,
|
|
"kind": "action",
|
|
"action_id": action_id,
|
|
"key": key,
|
|
"key_scope": scope,
|
|
"focus_before": before,
|
|
"focus_after": focus_state,
|
|
"breakpoint_outcome": "none",
|
|
"receipt": {
|
|
"id": f"rcpt-{action_id or 'noop'}-seq{i}",
|
|
"origin_session": session,
|
|
"origin_kind": "live",
|
|
},
|
|
})
|
|
if action_id == "modal.open.approve":
|
|
modal_open = True
|
|
elif action_id == "modal.cancel.approve":
|
|
modal_open = False
|
|
|
|
# Recovery segment: pinned constructed authority record (the frozen
|
|
# receipts carry no failure/recovery trace yet — tasks 7.1-7.3
|
|
# record the real one; this segment pins the ORDERING LAW now).
|
|
snapshot_id = "snap-authority-0001"
|
|
checkpoint_seq = len(steps) + 1
|
|
for j, phase in enumerate(RECOVERY_PHASES, start=checkpoint_seq):
|
|
step = {
|
|
"seq": j,
|
|
"kind": "recovery",
|
|
"phase": phase,
|
|
"breakpoint_outcome": "none",
|
|
"receipt": {
|
|
"id": f"rcpt-recovery-{phase}-seq{j}",
|
|
"origin_session": session,
|
|
"origin_kind": "live",
|
|
},
|
|
}
|
|
if phase == "checkpoint":
|
|
step["snapshot_id"] = snapshot_id
|
|
step["task_id"] = "task-2"
|
|
if phase == "restore":
|
|
step["snapshot_id"] = snapshot_id
|
|
if phase == "resume":
|
|
step["resumes_seq"] = checkpoint_seq
|
|
if phase == "reattach":
|
|
step["term"] = "xterm-256color"
|
|
steps.append(step)
|
|
|
|
launcher = {
|
|
"kind": "native",
|
|
"executable": "tui/cmd/rune-tui",
|
|
"digest": "sha256:" + hashlib.sha256(
|
|
b"qualified-native-launcher-of-record:v1:tui/cmd/rune-tui"
|
|
).hexdigest(),
|
|
"attestation": "scripts/build_paired.sh+build_attestation.verify_build_pairing",
|
|
}
|
|
return {
|
|
"trace_id": "migration-authority-trace-1",
|
|
"session": session,
|
|
"parity_unit": "authority-preserving action/receipt trace",
|
|
"launcher": launcher,
|
|
"theme_tokens": ["accent=medium_purple", "surface=panel",
|
|
"text=default", "muted=dim"],
|
|
"steps": steps,
|
|
"pinned": {
|
|
"recovery_segment": "constructed authority record; real "
|
|
"recovery receipts land with tasks 7.1-7.3",
|
|
"theme_tokens": "pinned token list (visual-system parity "
|
|
"unit); receipts do not yet carry tokens",
|
|
"source_receipts": {
|
|
name: canonical_sha256(doc)
|
|
for name, doc in tr.items()
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
# Pins — filled after the builder froze (update DELIBERATELY, never
|
|
# silently: `--pins` prints current values for re-pinning).
|
|
BASELINE_SHA256 = "0792967580af18f8ca15aa144566f5cd1d27f1eacf06a04c3391b1da2671dbdf"
|
|
LAUNCHER_PIN = "e0ec3949081e131778f30a81d89aa5d123dd6e6339570e3c9add9844e7c43e8b"
|
|
|
|
|
|
# ── the migration oracle ──────────────────────────────────────────────────
|
|
|
|
|
|
def check_trace(candidate: dict, baseline: dict) -> dict:
|
|
"""Judge a candidate native trace against the authoritative trace.
|
|
|
|
GREEN only when every authority-preserving invariant holds; RED
|
|
otherwise, with `fired_checks` naming the catching invariants (the
|
|
receipt convention from the lab mutant suite). The oracle compares
|
|
action/receipt traces, never pixels or widget counts.
|
|
"""
|
|
fired: list[str] = []
|
|
details: list[str] = []
|
|
|
|
def fire(name: str, why: str) -> None:
|
|
if name not in fired:
|
|
fired.append(name)
|
|
details.append(f"{name}: {why}")
|
|
|
|
# -- envelope -----------------------------------------------------------
|
|
for field in ("trace_id", "session", "parity_unit"):
|
|
if candidate.get(field) != baseline.get(field):
|
|
fire("trace_envelope_exact",
|
|
f"{field} {candidate.get(field)!r} != {baseline.get(field)!r}")
|
|
|
|
# -- action ids (order + values) ---------------------------------------
|
|
c_actions = [(s["seq"], s["action_id"]) for s in candidate.get("steps", [])
|
|
if s.get("kind") == "action"]
|
|
b_actions = [(s["seq"], s["action_id"]) for s in baseline["steps"]
|
|
if s["kind"] == "action"]
|
|
if c_actions != b_actions:
|
|
fire("action_ids_exact", "action id sequence diverged")
|
|
|
|
# -- key scopes: every key under the same authority ---------------------
|
|
b_by_seq = {s["seq"]: s for s in baseline["steps"]}
|
|
for s in candidate.get("steps", []):
|
|
ref = b_by_seq.get(s.get("seq"))
|
|
if ref is None:
|
|
continue
|
|
if s.get("key") != ref.get("key") or \
|
|
s.get("key_scope") != ref.get("key_scope"):
|
|
fire("key_scopes_exact",
|
|
f"seq {s.get('seq')}: key/scope "
|
|
f"({s.get('key')!r},{s.get('key_scope')!r}) != "
|
|
f"({ref.get('key')!r},{ref.get('key_scope')!r})")
|
|
|
|
# -- receipts stay bound to their origin --------------------------------
|
|
for s in candidate.get("steps", []):
|
|
ref = b_by_seq.get(s.get("seq"))
|
|
if ref is None:
|
|
continue
|
|
rcpt = s.get("receipt") or {}
|
|
rref = ref.get("receipt") or {}
|
|
expected_id = f"rcpt-{s.get('action_id') or 'noop'}-seq{s.get('seq')}" \
|
|
if s.get("kind") == "action" else rref.get("id")
|
|
if (rcpt.get("origin_session") != candidate.get("session")
|
|
or rcpt.get("origin_session") != rref.get("origin_session")
|
|
or rcpt.get("origin_kind") != rref.get("origin_kind")
|
|
or rcpt.get("id") != rref.get("id")
|
|
or rcpt.get("id") != expected_id):
|
|
fire("receipts_origin_bound",
|
|
f"seq {s.get('seq')}: receipt origin/id not bound "
|
|
f"({rcpt.get('origin_session')!r}, {rcpt.get('id')!r})")
|
|
|
|
# -- recovery ordering ---------------------------------------------------
|
|
c_phases = [s.get("phase") for s in candidate.get("steps", [])
|
|
if s.get("kind") == "recovery"]
|
|
if c_phases != RECOVERY_PHASES:
|
|
fire("recovery_ordering_valid",
|
|
f"phase order {c_phases} != canonical {RECOVERY_PHASES}")
|
|
else:
|
|
checkpoints: dict[str, int] = {}
|
|
for s in candidate.get("steps", []):
|
|
if s.get("kind") == "recovery":
|
|
if s.get("phase") == "checkpoint":
|
|
checkpoints[s.get("snapshot_id")] = s["seq"]
|
|
if s.get("phase") == "restore" and \
|
|
checkpoints.get(s.get("snapshot_id")) is None:
|
|
fire("recovery_ordering_valid",
|
|
f"restore seq {s['seq']} references unknown/"
|
|
f"later snapshot {s.get('snapshot_id')!r}")
|
|
if s.get("phase") == "resume" and \
|
|
s.get("resumes_seq") not in checkpoints.values():
|
|
fire("recovery_ordering_valid",
|
|
f"resume seq {s['seq']} resumes from non-checkpoint "
|
|
f"{s.get('resumes_seq')!r}")
|
|
|
|
# -- launcher target ------------------------------------------------------
|
|
launcher = candidate.get("launcher") or {}
|
|
if launcher != baseline["launcher"]:
|
|
fire("launcher_target_attested",
|
|
f"launcher binding != qualified native binding of record "
|
|
f"({launcher.get('kind')!r}, {launcher.get('executable')!r})")
|
|
launcher_text = canonical_json(launcher).lower()
|
|
if launcher.get("kind") != "native" or "python" in launcher_text:
|
|
# exact-token semantics: a fallback python launcher must fire even
|
|
# if someone re-pins the binding (NEG-003/NEG-004 lesson)
|
|
fire("launcher_target_attested",
|
|
"launcher is not the attested native target (python token "
|
|
"or non-native kind in the launch chain)")
|
|
|
|
# -- focus transitions + breakpoints + theme tokens -----------------------
|
|
for s in candidate.get("steps", []):
|
|
ref = b_by_seq.get(s.get("seq"))
|
|
if ref is None:
|
|
continue
|
|
if (s.get("focus_before"), s.get("focus_after")) != \
|
|
(ref.get("focus_before"), ref.get("focus_after")):
|
|
fire("focus_transitions_exact",
|
|
f"seq {s.get('seq')}: focus transition diverged")
|
|
if s.get("breakpoint_outcome") != ref.get("breakpoint_outcome"):
|
|
fire("breakpoint_outcomes_exact",
|
|
f"seq {s.get('seq')}: breakpoint outcome diverged")
|
|
if candidate.get("theme_tokens") != baseline["theme_tokens"]:
|
|
fire("theme_tokens_exact", "theme token set diverged")
|
|
|
|
return {"verdict": "GREEN" if not fired else "RED",
|
|
"fired_checks": fired,
|
|
"details": details}
|
|
|
|
|
|
def trace_diff(a: dict, b: dict, prefix: str = "") -> list[str]:
|
|
"""Leaf paths where two traces differ — keeps mutants surgical."""
|
|
diffs: list[str] = []
|
|
if isinstance(a, dict) and isinstance(b, dict):
|
|
for k in sorted(set(a) | set(b)):
|
|
if k not in a or k not in b:
|
|
diffs.append(f"{prefix}{k}")
|
|
else:
|
|
diffs += trace_diff(a[k], b[k], f"{prefix}{k}.")
|
|
elif isinstance(a, list) and isinstance(b, list):
|
|
if len(a) != len(b):
|
|
diffs.append(f"{prefix[:-1]}")
|
|
for i, (x, y) in enumerate(zip(a, b)):
|
|
diffs += trace_diff(x, y, f"{prefix}[{i}].")
|
|
elif a != b:
|
|
diffs.append(prefix[:-1] if prefix else "<root>")
|
|
return diffs
|
|
|
|
|
|
# ── the four named mutants ────────────────────────────────────────────────
|
|
|
|
|
|
def mut_key_scope(trace: dict) -> dict:
|
|
"""KEY SCOPE: a key claims authority it did not have — the modal-
|
|
swallowed 'q' (frozen receipt: q_swallowed_in_modal) is recorded as
|
|
if interpreted under plain input scope. Classic escaped-modal
|
|
regression: the same byte sequence acting under the wrong authority.
|
|
"""
|
|
for s in trace["steps"]:
|
|
if s.get("kind") == "action" and s.get("action_id") == "modal.swallow":
|
|
s["key_scope"] = "input"
|
|
break
|
|
return trace
|
|
|
|
|
|
def mut_receipt_origin(trace: dict) -> dict:
|
|
"""RECEIPT ORIGIN: one receipt claims it was minted by a different
|
|
session — the forged-provenance regression (a receipt attesting
|
|
work it did not witness; provenance would be lying)."""
|
|
for s in trace["steps"]:
|
|
if s.get("kind") == "action":
|
|
s["receipt"]["origin_session"] = "sess-other-9999"
|
|
break
|
|
return trace
|
|
|
|
|
|
def mut_recovery_order(trace: dict) -> dict:
|
|
"""RECOVERY ORDERING: resume executes before restore — the crash-
|
|
recovery regression (resuming from state that was never restored,
|
|
the classic lost-update on restart)."""
|
|
steps = trace["steps"]
|
|
idx = {s.get("phase"): i for i, s in enumerate(steps)
|
|
if s.get("kind") == "recovery"}
|
|
i_restore, i_resume = idx["restore"], idx["resume"]
|
|
steps[i_restore], steps[i_resume] = steps[i_resume], steps[i_restore]
|
|
return trace
|
|
|
|
|
|
def mut_launcher_target(trace: dict) -> dict:
|
|
"""LAUNCHER TARGET: the trace is produced by a python fallback
|
|
launcher instead of the attested native executable — the
|
|
silent-Python-fallback regression I-R2-008 exists to kill (a green
|
|
harness that invokes Python after quarantine is not evidence)."""
|
|
trace["launcher"] = {
|
|
"kind": "python-fallback",
|
|
"executable": "python3 src/rune/rune_tui.py",
|
|
"digest": "sha256:unattested",
|
|
"attestation": "none",
|
|
}
|
|
return trace
|
|
|
|
|
|
MUTANTS = {
|
|
"mut-key-scope": mut_key_scope,
|
|
"mut-receipt-origin": mut_receipt_origin,
|
|
"mut-recovery-order": mut_recovery_order,
|
|
"mut-launcher-target": mut_launcher_target,
|
|
}
|
|
|
|
MUTANT_EXPECTED_CHECK = {
|
|
"mut-key-scope": "key_scopes_exact",
|
|
"mut-receipt-origin": "receipts_origin_bound",
|
|
"mut-recovery-order": "recovery_ordering_valid",
|
|
"mut-launcher-target": "launcher_target_attested",
|
|
}
|
|
|
|
|
|
def run_mutants(baseline: dict, mutants: dict = MUTANTS) -> list[dict]:
|
|
"""Run each mutant against the oracle. A mutant SURVIVING (green) or
|
|
firing without its named invariant is a failed harness row."""
|
|
rows = []
|
|
for name, fn in mutants.items():
|
|
expected = MUTANT_EXPECTED_CHECK.get(name, "__unknown__")
|
|
cand = fn(copy.deepcopy(baseline))
|
|
verdict = check_trace(cand, baseline)
|
|
rows.append({
|
|
"mutant": name,
|
|
"oracle_verdict": verdict["verdict"],
|
|
"fired_checks": verdict["fired_checks"],
|
|
"expected_check": expected,
|
|
"fired_expected_check": expected in verdict["fired_checks"],
|
|
"caught": verdict["verdict"] == "RED"
|
|
and expected in verdict["fired_checks"],
|
|
})
|
|
return rows
|
|
|
|
|
|
def run_suite(receipts_dir: Path = RECEIPTS_DIR,
|
|
receipt_path: Path | None = None) -> dict:
|
|
"""The suite: baseline GREEN (no false positive) + 4/4 named mutants
|
|
RED with their named invariants. Receipt written to the research
|
|
receipts tree (mirrors the at-oracle mutant-suite convention)."""
|
|
baseline = build_baseline(receipts_dir)
|
|
drift = canonical_sha256(baseline) != BASELINE_SHA256
|
|
base_verdict = check_trace(baseline, baseline)
|
|
rows = run_mutants(baseline)
|
|
summary = {
|
|
"test": "migration_oracle_mutant_suite",
|
|
"harness": "scripts/check_migration_mutants.py (migration "
|
|
"oracle: authority-preserving action/receipt trace "
|
|
"parity — NOT the lab iso gate)",
|
|
"binding": "tasks.md 0.11; migration-and-retirement 'Deliberate "
|
|
"mutants'; I-R2-008 remaining gate",
|
|
"baseline": {
|
|
"sha256": canonical_sha256(baseline),
|
|
"pin_holds": not drift,
|
|
"oracle_verdict_on_self": base_verdict["verdict"],
|
|
"source_receipts": baseline["pinned"]["source_receipts"],
|
|
},
|
|
"mutants": rows,
|
|
"expectation": "baseline pin holds + oracle GREEN on baseline + "
|
|
"every named mutant RED with its named invariant",
|
|
"baseline_green": base_verdict["verdict"] == "GREEN" and not drift,
|
|
"all_mutants_red": bool(rows)
|
|
and all(r["caught"] for r in rows),
|
|
}
|
|
summary["verdict"] = "GREEN" if (summary["baseline_green"]
|
|
and summary["all_mutants_red"]) else "RED"
|
|
out = receipt_path or (RECEIPT_OUT_DIR / "mutant-summary.json")
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n")
|
|
summary["_receipt_path"] = str(out)
|
|
return summary
|
|
|
|
|
|
# ── SSH-profile replay harness (fixed burst/resize/input traces) ──────────
|
|
#
|
|
# I-R2-006 remaining gate: "SSH channel full replay". The fixed traces
|
|
# live in the frozen receipts; this harness replays them under the SSH
|
|
# profile. Transport honesty: real ssh (BatchMode) when --ssh-target /
|
|
# RUNE_SSH_TARGET answers, else a PTY carrying the SSH profile the
|
|
# keyprobe receipts recorded (TERM=xterm-256color, tty stdin) — the
|
|
# receipt always names which transport actually ran.
|
|
|
|
KEY_NAME_BYTES = {"tab": b"\t", "alt+q": b"\x1bq"}
|
|
ESC_TIMEOUT_S = 0.05 # the vendored parser's pinned EscTimeout
|
|
SPLIT_GAP_S = 0.20 # > EscTimeout: split delivery, EV-BT-004
|
|
PROBE_SENTINEL = b"\x04\x04"
|
|
PROBE_MARKER = b"@@PROBE_JSON@@"
|
|
|
|
PROBE_SRC = r'''
|
|
import fcntl, json, os, select, signal, struct, sys, termios, time, tty
|
|
fd = sys.stdin.fileno()
|
|
tty.setraw(fd)
|
|
events, sizes = [], []
|
|
def _winch(sig, frm):
|
|
ws = fcntl.ioctl(fd, termios.TIOCGWINSZ, b"\0" * 8)
|
|
r, c = struct.unpack("HH", ws[:4])
|
|
sizes.append([time.time_ns(), r, c])
|
|
os.write(fd, ("\x11ACKR%dC%d\r" % (r, c)).encode())
|
|
signal.signal(signal.SIGWINCH, _winch)
|
|
ws0 = fcntl.ioctl(fd, termios.TIOCGWINSZ, b"\0" * 8)
|
|
r0, c0 = struct.unpack("HH", ws0[:4])
|
|
sizes.append([time.time_ns(), r0, c0])
|
|
os.write(fd, ("\x11ACKR%dC%d\r" % (r0, c0)).encode())
|
|
deadline = time.time() + float(sys.argv[1] if len(sys.argv) > 1 else 20)
|
|
while time.time() < deadline:
|
|
r, _, _ = select.select([fd], [], [], 0.05)
|
|
if not r:
|
|
continue
|
|
b = os.read(fd, 4096)
|
|
if PROBE_SENTINEL_VIEW in b:
|
|
break
|
|
events.append([time.time_ns(), b.hex()])
|
|
os.write(fd, b"\x06")
|
|
out = {"term": os.environ.get("TERM"), "initial_size": [r0, c0],
|
|
"events": events, "sizes": sizes,
|
|
"stdin_is_tty": os.isatty(fd)}
|
|
os.write(sys.stdout.fileno(), b"\n@@PROBE_JSON@@"
|
|
+ json.dumps(out).encode() + b"\n")
|
|
'''.replace("PROBE_SENTINEL_VIEW", repr(PROBE_SENTINEL))
|
|
|
|
# Session preamble: SIGWINCH reaches the probe only when the slave is
|
|
# its CONTROLLING terminal, which requires setsid-then-open (the
|
|
# pexpect pattern). PTY spawn uses this; over ssh the remote pty
|
|
# already provides one.
|
|
SESSION_PREAMBLE = r'''
|
|
import os, sys
|
|
os.setsid()
|
|
if len(sys.argv) > 2:
|
|
_f = os.open(sys.argv[2], os.O_RDWR)
|
|
os.dup2(_f, 0); os.dup2(_f, 1); os.dup2(_f, 2)
|
|
if _f > 2:
|
|
os.close(_f)
|
|
'''
|
|
|
|
|
|
def key_bytes(name: str) -> bytes:
|
|
if name in KEY_NAME_BYTES:
|
|
return KEY_NAME_BYTES[name]
|
|
return name.encode()
|
|
|
|
|
|
def parse_key_events(events: list) -> list[str]:
|
|
"""Reference parser for the FIXED trace alphabet with the pinned
|
|
50ms EscTimeout: ESC + next byte inside the window is one key
|
|
(alt+q); ESC alone past it is 'esc'. This is the harness's own
|
|
instrument (keyprobe convention) — it judges transport behavior,
|
|
it is not the product parser."""
|
|
keys: list[str] = []
|
|
flat: list[tuple[int, bytes]] = [(t, bytes.fromhex(h)) for t, h in events]
|
|
i = 0
|
|
while i < len(flat):
|
|
t, chunk = flat[i]
|
|
j = 0
|
|
while j < len(chunk):
|
|
b = chunk[j:j + 1]
|
|
if b == b"\x1b":
|
|
nxt_t, nxt_b = (flat[i + 1] if i + 1 < len(flat) else (0, b""))
|
|
same_chunk = j + 1 < len(chunk)
|
|
if same_chunk:
|
|
keys.append("alt+" + chunk[j + 1:j + 2].decode())
|
|
j += 2
|
|
continue
|
|
if nxt_b and (nxt_t - t) / 1e9 <= ESC_TIMEOUT_S:
|
|
keys.append("alt+" + nxt_b[:1].decode())
|
|
i += 1
|
|
j = len(chunk)
|
|
continue
|
|
keys.append("esc")
|
|
elif b == b"\t":
|
|
keys.append("tab")
|
|
else:
|
|
keys.append(b.decode())
|
|
j += 1
|
|
i += 1
|
|
return keys
|
|
|
|
|
|
def validate_frozen_records(traces: dict) -> list[str]:
|
|
"""Record-check the frozen burst/resize/focus receipts against their
|
|
own claims (determinism, no loss, no order violations, budgets
|
|
present). A corrupted receipt input is a RED replay input."""
|
|
problems = []
|
|
burst, resize, focus = (traces["burst"], traces["resize"],
|
|
traces["focus"])
|
|
# burst carries no `deterministic` field (that is a resize/focus
|
|
# claim) — check only the claims the receipt actually makes
|
|
if burst.get("order_violations") != 0 \
|
|
or burst.get("missing_seqs_sample") is not None \
|
|
or burst.get("last_seq_applied") != 4999 \
|
|
or not burst.get("final_contains_last") \
|
|
or burst.get("err", "") != "":
|
|
problems.append("frozen burst receipt fails its own claims")
|
|
if not resize.get("deterministic") or not resize.get("focus_preserved") \
|
|
or not resize.get("width_within_80") \
|
|
or resize.get("final_80x24_sha_r1") != resize.get("final_80x24_sha_r2"):
|
|
problems.append("frozen resize receipt fails its own claims")
|
|
if not focus.get("actions_expected") or not focus.get("keys_match_script") \
|
|
or not focus.get("q_swallowed_in_modal"):
|
|
problems.append("frozen focus receipt fails its own claims")
|
|
return problems
|
|
|
|
|
|
class _PtyTransport:
|
|
"""SSH-profile PTY: TERM=xterm-256color (what the keyprobe receipts
|
|
recorded), raw probe child on the slave, 80x24 initial winsize."""
|
|
|
|
name = "pty-ssh-profile"
|
|
|
|
def __init__(self):
|
|
import pty
|
|
import fcntl
|
|
import struct
|
|
import termios
|
|
self.master, slave = pty.openpty()
|
|
fcntl.ioctl(self.master, termios.TIOCSWINSZ,
|
|
struct.pack("HHHH", 24, 80, 0, 0))
|
|
slave_path = os.ttyname(slave)
|
|
os.close(slave) # the probe opens it itself AFTER setsid
|
|
env = dict(os.environ, TERM="xterm-256color")
|
|
self.proc = subprocess.Popen(
|
|
[sys.executable, "-s", "-c", SESSION_PREAMBLE + PROBE_SRC,
|
|
"20", slave_path],
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, env=env, close_fds=True)
|
|
|
|
def wait_ready(self, duration_s: float = 5.0) -> bool:
|
|
"""Wait for the probe's initial size ACK — input written to the
|
|
master before the child opens the slave is dropped by the
|
|
kernel, so the harness must not feed the script early. Reads on
|
|
a master whose slave is not open yet raise EIO: that is
|
|
'not ready', not an error."""
|
|
import select
|
|
deadline = time.monotonic() + duration_s
|
|
while time.monotonic() < deadline:
|
|
r, _, _ = select.select([self.master], [], [], 0.05)
|
|
if not r:
|
|
continue
|
|
try:
|
|
chunk = os.read(self.master, 65536)
|
|
except OSError:
|
|
continue
|
|
if b"ACKR" in chunk:
|
|
time.sleep(0.05) # let setraw settle before feeding
|
|
return True
|
|
return False
|
|
|
|
def write(self, data: bytes) -> None:
|
|
os.write(self.master, data)
|
|
|
|
def resize(self, rows: int, cols: int) -> None:
|
|
import fcntl
|
|
import struct
|
|
import termios
|
|
fcntl.ioctl(self.master, termios.TIOCSWINSZ,
|
|
struct.pack("HHHH", rows, cols, 0, 0))
|
|
|
|
def read_output(self, duration_s: float, done=None) -> bytes:
|
|
import select
|
|
buf = b""
|
|
deadline = time.monotonic() + duration_s
|
|
while time.monotonic() < deadline:
|
|
r, _, _ = select.select([self.master], [], [], 0.05)
|
|
if r:
|
|
try:
|
|
buf += os.read(self.master, 65536)
|
|
except OSError:
|
|
break
|
|
if done is not None:
|
|
if done(buf):
|
|
break
|
|
elif PROBE_MARKER in buf:
|
|
# marker seen; keep draining until the JSON parses whole
|
|
if _extract_probe_json(buf) is not None:
|
|
break
|
|
return buf
|
|
|
|
def close(self):
|
|
try:
|
|
self.proc.terminate()
|
|
self.proc.wait(timeout=5)
|
|
except Exception:
|
|
self.proc.kill()
|
|
os.close(self.master)
|
|
|
|
|
|
class _SshTransport:
|
|
"""Real SSH transport: same probe source, remote python, forced tty
|
|
(-tt) so raw-mode/ioctl/SIGWINCH behave as over a real channel."""
|
|
|
|
def __init__(self, target: str):
|
|
self.name = f"ssh:{target}"
|
|
self.proc = subprocess.Popen(
|
|
["ssh", "-tt", "-o", "BatchMode=yes",
|
|
"-o", "ConnectTimeout=5", target,
|
|
"python3", "-s", "-c", PROBE_SRC, "20"],
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL)
|
|
|
|
def write(self, data: bytes) -> None:
|
|
self.proc.stdin.write(data)
|
|
self.proc.stdin.flush()
|
|
|
|
def resize(self, rows: int, cols: int) -> None:
|
|
# SIGWINCH on a remote pty is driven by the remote terminal
|
|
# emulator; over the ssh channel we send the xterm CSI resize
|
|
# escape — best-effort; the replay records what the probe
|
|
# actually observed.
|
|
self.write(f"\x1b[8;{rows};{cols}t".encode())
|
|
|
|
def read_output(self, duration_s: float) -> bytes:
|
|
import selectors
|
|
sel = selectors.DefaultSelector()
|
|
sel.register(self.proc.stdout, selectors.EVENT_READ)
|
|
buf = b""
|
|
deadline = time.monotonic() + duration_s
|
|
while time.monotonic() < deadline:
|
|
for key, _ in sel.select(0.05):
|
|
chunk = key.fileobj.read1(65536)
|
|
if not chunk:
|
|
return buf
|
|
buf += chunk
|
|
if PROBE_MARKER in buf:
|
|
return buf
|
|
return buf
|
|
|
|
def close(self):
|
|
try:
|
|
self.proc.kill()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _extract_probe_json(output: bytes) -> dict | None:
|
|
idx = output.find(PROBE_MARKER)
|
|
if idx < 0:
|
|
return None
|
|
tail = output[idx + len(PROBE_MARKER):].strip()
|
|
try:
|
|
return json.loads(tail)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
|
|
def ssh_replay(receipts_dir: Path = RECEIPTS_DIR,
|
|
ssh_target: str | None = None,
|
|
burst_tokens: int = 300,
|
|
receipt_path: Path | None = None) -> dict:
|
|
"""Replay the FIXED input/burst/resize traces under the SSH profile.
|
|
|
|
Transport: real ssh when a BatchMode target answers (the I-R2-006
|
|
'SSH channel full replay'), else a PTY carrying the recorded SSH
|
|
profile (TERM=xterm-256color, tty stdin) — the receipt names which
|
|
actually ran; the fallback is never presented as the real channel.
|
|
"""
|
|
traces = load_fixed_traces(receipts_dir)
|
|
frozen_problems = validate_frozen_records(traces)
|
|
|
|
transport = None
|
|
fallback_reason = ""
|
|
target = ssh_target or os.environ.get("RUNE_SSH_TARGET", "")
|
|
if target:
|
|
probe = subprocess.run(
|
|
["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5",
|
|
target, "echo ok"], capture_output=True, timeout=15)
|
|
if probe.returncode == 0:
|
|
transport = _SshTransport(target)
|
|
else:
|
|
fallback_reason = (f"ssh target {target!r} unreachable "
|
|
f"(rc={probe.returncode})")
|
|
if transport is None:
|
|
if target and not fallback_reason:
|
|
fallback_reason = "no ssh target configured"
|
|
elif not target:
|
|
fallback_reason = ("no ssh target configured "
|
|
"(RUNE_SSH_TARGET/--ssh-target)")
|
|
transport = _PtyTransport()
|
|
|
|
problems: list[str] = list(frozen_problems)
|
|
ready = getattr(transport, "wait_ready", lambda **k: True)()
|
|
if not ready:
|
|
problems.append("probe never reported ready (initial ACK "
|
|
"missing) — input replay would be dishonest")
|
|
try:
|
|
# -- input trace: the frozen key script + the split variant ------
|
|
input_script: list[tuple[str, bytes, float]] = []
|
|
for name in traces["focus"]["keys_via_parser"]:
|
|
input_script.append((name, key_bytes(name), 0.02))
|
|
input_script.append(("esc;split", b"\x1b", SPLIT_GAP_S))
|
|
input_script.append(("q;split", b"q", 0.02))
|
|
|
|
writes: list[tuple[str, int]] = [] # name, write-time ns
|
|
for name, data, gap in input_script:
|
|
writes.append((name, time.monotonic_ns()))
|
|
transport.write(data)
|
|
time.sleep(gap)
|
|
|
|
# -- burst trace: sequenced tokens, chunked -----------------------
|
|
tokens = [f"s{i:05d}\n".encode() for i in range(burst_tokens)]
|
|
burst_t0 = time.monotonic_ns()
|
|
for k in range(0, len(tokens), 10):
|
|
transport.write(b"".join(tokens[k:k + 10]))
|
|
time.sleep(0.001)
|
|
burst_write_ns = time.monotonic_ns() - burst_t0
|
|
|
|
# -- resize trace: 80x24 -> 140x30 -> 80x24 -----------------------
|
|
time.sleep(0.1)
|
|
transport.resize(30, 140)
|
|
time.sleep(0.2)
|
|
transport.resize(24, 80)
|
|
time.sleep(0.2)
|
|
|
|
transport.write(PROBE_SENTINEL)
|
|
output = transport.read_output(10.0)
|
|
finally:
|
|
transport.close()
|
|
|
|
probe_data = _extract_probe_json(output)
|
|
if probe_data is None:
|
|
problems.append("probe produced no JSON (marker missing/truncated)")
|
|
probe_data = {"events": [], "sizes": [], "term": None,
|
|
"initial_size": None, "stdin_is_tty": None}
|
|
|
|
# input: parse with the 50ms rule, compare to frozen expectations.
|
|
# Only events BEFORE the first burst token are the input trace.
|
|
burst_start = next((i for i, (_, h) in enumerate(probe_data["events"])
|
|
if b"s00000" in bytes.fromhex(h)), None)
|
|
if burst_start is None:
|
|
problems.append("no burst tokens observed in probe events")
|
|
burst_start = len(probe_data["events"])
|
|
parsed = parse_key_events(probe_data["events"][:burst_start])
|
|
expected_keys = list(traces["focus"]["keys_via_parser"]) + ["esc", "q"]
|
|
frozen_atomic = traces["keyprobe_atomic"]["parsed_keys"]
|
|
frozen_split = traces["keyprobe_split"]["parsed_keys"]
|
|
input_ok = parsed == expected_keys
|
|
if not input_ok:
|
|
problems.append(f"input parse diverged: {parsed} != {expected_keys}")
|
|
|
|
# burst: reassemble the byte stream, extract tokens, order+loss
|
|
stream = b"".join(bytes.fromhex(h) for _, h in probe_data["events"])
|
|
import re
|
|
seen = re.findall(rb"s(\d{5})\n", stream)
|
|
seqs = [int(x) for x in seen]
|
|
missing = sorted(set(range(burst_tokens)) - set(seqs))
|
|
ordered = seqs == sorted(seqs)
|
|
dupes = len(seqs) != len(set(seqs))
|
|
if missing or not ordered or dupes:
|
|
problems.append(f"burst replay loss/order violation "
|
|
f"(missing={len(missing)}, ordered={ordered}, "
|
|
f"dupes={dupes})")
|
|
|
|
# resize: probe must have observed initial + both resizes
|
|
observed = [(r, c) for _, r, c in probe_data["sizes"]]
|
|
expected_sizes = [(24, 80), (30, 140), (24, 80)]
|
|
resize_ok = observed == expected_sizes
|
|
if not resize_ok:
|
|
problems.append(f"resize observations {observed} != {expected_sizes}")
|
|
|
|
receipt = {
|
|
"test": "ssh_profile_replay_fixed_traces",
|
|
"harness": "scripts/check_migration_mutants.py --ssh-replay",
|
|
"binding": "tasks.md 0.11 + I-R2-006 remaining gate (SSH channel "
|
|
"full replay); traces frozen at "
|
|
"research/tui-frontier-r2/receipts/",
|
|
"transport": {
|
|
"actual": transport.name,
|
|
"ssh_requested": bool(target),
|
|
"fallback_reason": fallback_reason,
|
|
"term": probe_data.get("term"),
|
|
"stdin_is_tty": probe_data.get("stdin_is_tty"),
|
|
"esc_timeout_s": ESC_TIMEOUT_S,
|
|
},
|
|
"traces": {
|
|
"input": {
|
|
"replayed": [w[0] for w in writes],
|
|
"parsed_keys": parsed,
|
|
"expected_keys": expected_keys,
|
|
"match": input_ok,
|
|
"frozen_atomic_expectation": frozen_atomic,
|
|
"frozen_split_expectation": frozen_split,
|
|
"ev_bt_004_divergence_reproduced":
|
|
parsed[-2:] == ["esc", "q"],
|
|
},
|
|
"burst": {
|
|
"tokens_sent": burst_tokens,
|
|
"tokens_received": len(set(seqs)),
|
|
"missing": len(missing),
|
|
"ordered": ordered,
|
|
"write_phase_ms": round(burst_write_ns / 1e6, 1),
|
|
},
|
|
"resize": {
|
|
"observed_sizes": observed,
|
|
"expected_sizes": expected_sizes,
|
|
"match": resize_ok,
|
|
},
|
|
},
|
|
"frozen_record_problems": frozen_problems,
|
|
"problems": problems,
|
|
"verdict": "GREEN" if not problems else "RED",
|
|
}
|
|
out = receipt_path or (RECEIPT_OUT_DIR / "ssh-replay.json")
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n")
|
|
receipt["_receipt_path"] = str(out)
|
|
return receipt
|
|
|
|
|
|
# ── slowest-target SLO recording structure ────────────────────────────────
|
|
#
|
|
# tasks.md 0.11: "slowest-supported-target run against the agreed SLO."
|
|
# GAP-BT-05: no agreed SLO exists yet — so the budgets below are
|
|
# PROPOSED from the lab receipts (agreed:false until the owner
|
|
# ratifies), and every record states honestly whether its target is the
|
|
# slowest SUPPORTED target or just the machine that ran it.
|
|
|
|
PROPOSED_BUDGETS = {
|
|
"key_p99_ms": 5.0, # lab: 0.481 (burst-n5000 key_send_to_update p99)
|
|
"msg_p99_ms": 5.0, # lab: 0.488
|
|
"updates_per_s_min": 1000, # lab: 7041 stream updates/s
|
|
"order_violations_max": 0, # lab: 0
|
|
"missing_seqs_max": 0, # lab: null
|
|
"bytes_per_frame_avg_max": 4096, # lab: 49
|
|
}
|
|
|
|
SLO_RECORD_REQUIRED = [
|
|
"schema_version", "ran_at", "runner", "target", "is_slowest_supported_target",
|
|
"transport", "run", "budgets", "verdict", "artifacts",
|
|
]
|
|
|
|
|
|
def build_slo_record(replay_receipt: dict, *,
|
|
slowest_target: bool = False) -> dict:
|
|
cpu = ""
|
|
try:
|
|
cpu = (Path("/proc/cpuinfo").read_text().split("model name")[1]
|
|
.split("\n")[0].split(":")[1].strip())
|
|
except Exception:
|
|
pass
|
|
ram_gb = None
|
|
try:
|
|
for line in Path("/proc/meminfo").read_text().splitlines():
|
|
if line.startswith("MemTotal"):
|
|
ram_gb = round(int(line.split()[1]) / 1e6, 1)
|
|
except Exception:
|
|
pass
|
|
burst = replay_receipt["traces"]["burst"]
|
|
measured = {
|
|
"updates_per_s": round(burst["tokens_received"]
|
|
/ max(burst["write_phase_ms"] / 1000, 1e-9), 1),
|
|
"order_violations": 0 if burst["ordered"] else 1,
|
|
"missing_seqs": burst["missing"],
|
|
}
|
|
verdict = "GREEN" if (
|
|
replay_receipt["verdict"] == "GREEN"
|
|
and measured["order_violations"] <= PROPOSED_BUDGETS["order_violations_max"]
|
|
and measured["missing_seqs"] <= PROPOSED_BUDGETS["missing_seqs_max"]
|
|
and measured["updates_per_s"] >= PROPOSED_BUDGETS["updates_per_s_min"]
|
|
) else "RED"
|
|
return {
|
|
"schema_version": 1,
|
|
"ran_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"runner": "scripts/check_migration_mutants.py --slo-record",
|
|
"target": {
|
|
"name": "lab-desktop",
|
|
"cpu": cpu,
|
|
"ram_gb": ram_gb,
|
|
"note": "Ryzen 9 5950X-class lab machine — NOT the slowest "
|
|
"supported target (GAP-BT-05); the slowest-target "
|
|
"run gets its own record when that hardware runs "
|
|
"this harness",
|
|
},
|
|
"is_slowest_supported_target": slowest_target,
|
|
"transport": replay_receipt["transport"],
|
|
"run": {
|
|
"replay_receipt": replay_receipt["_receipt_path"],
|
|
"replay_verdict": replay_receipt["verdict"],
|
|
"input_match": replay_receipt["traces"]["input"]["match"],
|
|
"resize_match": replay_receipt["traces"]["resize"]["match"],
|
|
"measured": measured,
|
|
},
|
|
"budgets": {
|
|
"proposed": PROPOSED_BUDGETS,
|
|
"agreed": False,
|
|
"agreement_note": "no agreed SLO yet (GAP-BT-05); budgets "
|
|
"proposed from lab receipts "
|
|
"burst-n5000-slow0.json",
|
|
},
|
|
"verdict": verdict,
|
|
"artifacts": [replay_receipt["_receipt_path"]],
|
|
}
|
|
|
|
|
|
def slo_validate(record: dict) -> list[str]:
|
|
"""Structural + arithmetic validation of an SLO record. Used by the
|
|
record writer itself and by tests/CI so a malformed or lying record
|
|
cannot stand in for the slowest-target run."""
|
|
errs = []
|
|
for field in SLO_RECORD_REQUIRED:
|
|
if field not in record:
|
|
errs.append(f"missing field: {field}")
|
|
if errs:
|
|
return errs
|
|
if record["schema_version"] != 1:
|
|
errs.append(f"unknown schema_version {record['schema_version']}")
|
|
m = record["run"]["measured"]
|
|
b = record["budgets"]["proposed"]
|
|
if m["order_violations"] > b["order_violations_max"] \
|
|
or m["missing_seqs"] > b["missing_seqs_max"] \
|
|
or m["updates_per_s"] < b["updates_per_s_min"]:
|
|
computed_green = False
|
|
else:
|
|
computed_green = True
|
|
if computed_green != (record["verdict"] == "GREEN"):
|
|
errs.append(f"verdict {record['verdict']!r} disagrees with "
|
|
f"measured-vs-budget arithmetic")
|
|
if record["is_slowest_supported_target"] and not record["budgets"]["agreed"]:
|
|
errs.append("a slowest-target claim requires an AGREED SLO — "
|
|
"proposed budgets cannot ratify it")
|
|
return errs
|
|
|
|
|
|
def slo_record(slowest_target: bool = False) -> dict:
|
|
replay = ssh_replay()
|
|
record = build_slo_record(replay, slowest_target=slowest_target)
|
|
errs = slo_validate(record)
|
|
if errs:
|
|
record["validation_errors"] = errs
|
|
SLO_DIR.mkdir(parents=True, exist_ok=True)
|
|
rec_dir = SLO_DIR / "records"
|
|
rec_dir.mkdir(exist_ok=True)
|
|
slug = record["target"]["name"].replace("/", "-")
|
|
out = rec_dir / f"{record['ran_at'].replace(':', '')}-{slug}.json"
|
|
out.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n")
|
|
record["_record_path"] = str(out)
|
|
record["_valid"] = not errs
|
|
return record
|
|
|
|
|
|
# ── main ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
|
|
ap.add_argument("--ssh-replay", action="store_true",
|
|
help="replay the fixed input/burst/resize traces "
|
|
"under the SSH profile")
|
|
ap.add_argument("--ssh-target", default="",
|
|
help="ssh target for the real-channel replay "
|
|
"(default RUNE_SSH_TARGET env)")
|
|
ap.add_argument("--slo-record", action="store_true",
|
|
help="run the replay and record an SLO run record")
|
|
ap.add_argument("--slowest-target", action="store_true",
|
|
help="label the SLO record as the slowest supported "
|
|
"target (operator asserts constrained hardware)")
|
|
ap.add_argument("--slo-validate", metavar="FILE",
|
|
help="validate an existing SLO record file")
|
|
ap.add_argument("--pins", action="store_true",
|
|
help="print current baseline/launcher pins")
|
|
args = ap.parse_args(argv)
|
|
|
|
if args.pins:
|
|
baseline = build_baseline()
|
|
print(f"BASELINE_SHA256 = {canonical_sha256(baseline)!r}")
|
|
print(f"LAUNCHER_PIN = {canonical_sha256(baseline['launcher'])!r}")
|
|
print(f"in-module pin holds: "
|
|
f"{canonical_sha256(baseline) == BASELINE_SHA256}")
|
|
return 0
|
|
|
|
if args.slo_validate:
|
|
doc = json.loads(Path(args.slo_validate).read_text())
|
|
errs = slo_validate(doc)
|
|
for e in errs:
|
|
print(f"INVALID: {e}")
|
|
print(f"{'VALID' if not errs else 'INVALID'}: {args.slo_validate}")
|
|
return 0 if not errs else 1
|
|
|
|
if args.slo_record:
|
|
rec = slo_record(slowest_target=args.slowest_target)
|
|
print(f"SLO record verdict={rec['verdict']} "
|
|
f"slowest_target={rec['is_slowest_supported_target']} "
|
|
f"valid={rec['_valid']} -> {rec['_record_path']}")
|
|
return 0 if rec["verdict"] == "GREEN" and rec["_valid"] else 1
|
|
|
|
if args.ssh_replay:
|
|
r = ssh_replay(ssh_target=args.ssh_target or None)
|
|
print(f"transport={r['transport']['actual']} "
|
|
f"(fallback: {r['transport']['fallback_reason'] or 'none'})")
|
|
print(f"input match={r['traces']['input']['match']} "
|
|
f"burst missing={r['traces']['burst']['missing']} "
|
|
f"ordered={r['traces']['burst']['ordered']} "
|
|
f"resize match={r['traces']['resize']['match']}")
|
|
for p in r["problems"]:
|
|
print(f"PROBLEM: {p}")
|
|
print(f"VERDICT: {r['verdict']} -> {r['_receipt_path']}")
|
|
return 0 if r["verdict"] == "GREEN" else 1
|
|
|
|
summary = run_suite()
|
|
for row in summary["mutants"]:
|
|
print(f" [{row['oracle_verdict']:<5}] {row['mutant']} "
|
|
f"fired={','.join(row['fired_checks'])} "
|
|
f"expected-fired={row['fired_expected_check']}")
|
|
print(f"baseline_green={summary['baseline_green']} "
|
|
f"all_mutants_red={summary['all_mutants_red']} "
|
|
f"verdict={summary['verdict']} -> {summary['_receipt_path']}")
|
|
return 0 if summary["verdict"] == "GREEN" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|
|
|
|
|
|
|
|
|
|
|
|
|