eval/run_eval.py

#!/usr/bin/env python3
"""oracle7-deepspace evaluation. Seeds: DEV 0-7 (built against), GATE SUITE 200-203 (the craft's own validation
suite), HELD-OUT 100-111 (never looked at during development; used here once). Writes results/eval.json."""
import copy
import json
import os
import random
import sys
import time
import tracemalloc

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))
from deepspace.fdir import FDIR, Facts, candidates, HYPOTHESES  # noqa: E402
from deepspace.report import SelfReport  # noqa: E402
from deepspace.runtime import run, scenario, score  # noqa: E402
from deepspace.selfmaint import Rulebook, propose  # noqa: E402
from deepspace.sim import FAULTS  # noqa: E402

HELD = range(100, 112)
DEV = range(0, 8)
OUT = {}


class Threshold:
    """classic conservative FDIR: any confirmed limit/consistency violation -> safe mode"""
    def __init__(self):
        self.facts, self.done = Facts(), False

    def decide(self, tm):
        facts = self.facts.update(tm)
        if not self.done and [f for f in facts if f[0] != "transient_spike"]:
            self.done = True
            return {"action": "safe_mode", "arg": {}, "t": tm["t"], "facts": facts}
        return None


class Random:
    def __init__(self, seed):
        self.facts, self.done, self.r = Facts(), False, random.Random(seed)

    def decide(self, tm):
        facts = self.facts.update(tm)
        conf = [f for f in facts if f[0] in HYPOTHESES]
        if not self.done and conf:
            self.done = True
            hyp = HYPOTHESES[conf[0][0]](conf[0])[0]
            a, arg = self.r.choice([c for c in candidates(hyp, tm)])
            return {"action": a, "arg": arg, "t": tm["t"], "facts": facts}
        return None


def evaluate(make, label, seeds=HELD):
    per = {}
    tick_ms, lat = [], []
    twin = 0
    for fault in FAULTS:
        rows = []
        for seed in seeds:
            p = scenario(fault, seed)
            agent = make(seed)
            t0 = time.perf_counter()
            tr = run(p, agent.decide)
            tick_ms.append((time.perf_counter() - t0) * 1000 / 320)
            twin += getattr(agent, "twin_runs", 0)
            s = score(p, tr)
            rows.append(s)
            if s["latency_min"] is not None:
                lat.append(s["latency_min"])
        per[fault] = {"correct": sum(r["ok"] for r in rows), "harmful": sum(r["harmful"] for r in rows),
                      "safe_mode": sum(r["safe_mode"] for r in rows), "damage": sum(r["damage"] for r in rows), "n": len(rows)}
    tot = {k: sum(v[k] for v in per.values()) for k in ("correct", "harmful", "safe_mode", "damage", "n")}
    tot["tick_ms_mean"] = round(sum(tick_ms) / len(tick_ms), 3)
    tot["latency_min_median"] = sorted(lat)[len(lat) // 2] if lat else None
    tot["within_20_min"] = sum(1 for x in lat if x <= 20)
    tot["twin_runs"] = twin
    OUT[label] = {"total": tot, "per_fault": per}
    print("%-26s correct %3d/%d  harmful %2d  safe-mode %3d  damage %2d  median latency %s min  tick %.3f ms  twin runs %d"
          % (label, tot["correct"], tot["n"], tot["harmful"], tot["safe_mode"], tot["damage"], tot["latency_min_median"],
             tot["tick_ms_mean"], twin))
    return per


def main():
    print("== 1. FAULT RESPONSE, HELD-OUT seeds %d-%d (%d scenarios)" % (HELD[0], HELD[-1], len(HELD) * len(FAULTS)))
    ds = evaluate(lambda s: FDIR(), "oracle7-deepspace")
    evaluate(lambda s: Threshold(), "baseline: limit->safe mode")
    evaluate(lambda s: type("N", (), {"decide": lambda self, tm: None})(), "no action")
    evaluate(lambda s: Random(s), "random admissible response")
    print("   per fault (deepspace):", {f: "%d/%d%s" % (v["correct"], v["n"], " H%d" % v["harmful"] if v["harmful"] else "")
                                         for f, v in ds.items()})

    print("\n== 2. SELF-MAINTENANCE: learn from DEV missions, admit only through the gate (suite seeds 200-203)")
    traces = []
    for fault in FAULTS:
        for seed in DEV:
            p = scenario(fault, seed)
            traces.append(run(p, FDIR().decide))
    props = propose(traces)
    rb = Rulebook(veto_path=os.path.join(HERE, "..", "results", "veto.json"))
    bad = [{"id": "MUTANT-bias->heater", "signature": ("sensor_disagree",), "action": "switch_heater_backup",
            "bind": {"zone": ("sensor_disagree", 1)}, "support": 99},
           {"id": "MUTANT-wheel->safe", "signature": ("wheel_overcurrent",), "action": "safe_mode", "bind": {}, "support": 99},
           {"id": "MUTANT-solar->safe", "signature": ("solar_low",), "action": "safe_mode", "bind": {}, "support": 99}]
    for rule in bad + props:
        v = rb.consider(rule)
        print("   %-6s %-52s %s" % (v["verdict"], rule["id"][:52], "; ".join(v["reasons"]) or
                                   "suite correct %d->%d harmful %d->%d twin runs %d->%d" % (
                                       v["baseline"]["correct"], v["candidate"]["correct"], v["baseline"]["harmful"],
                                       v["candidate"]["harmful"], v["baseline"]["twin_runs"], v["candidate"]["twin_runs"])))
    OUT["gate_log"] = rb.log
    learned = rb.current
    print("   admitted rules:", len(learned), "| mutants rejected:", sum(1 for v in rb.log if v["rule"].startswith("MUTANT") and v["verdict"] == "REJECT"), "of 3")
    evaluate(lambda s: FDIR(rulebook=copy.deepcopy(learned)), "deepspace + learned rules")
    if learned:
        vid = next(iter(learned.values()))["id"]
        os.makedirs(os.path.join(HERE, "..", "results"), exist_ok=True)
        json.dump([vid], open(os.path.join(HERE, "..", "results", "veto.json"), "w"))
        after = len(rb.current)
        json.dump([], open(os.path.join(HERE, "..", "results", "veto.json"), "w"))
        print("   ground veto of %s: rules %d -> %d (restored after test)" % (vid, len(learned), after))

    print("\n== 3. SELF-REPORT after real missions (answers cite telemetry frames and decision receipts)")
    panel = []
    for fault, seed, qs in [
            ("heater_stuck_on", 101, [("status?", ["Mode:", "Battery", "[tm t="]), ("what is wrong?", ["heater", "backup circuit", "[receipt #"]),
                                      ("why did you switch the heater?", ["because", "kept heating", "worst-case", "[receipt #"]),
                                      ("what is the battery temperature?", ["battery is at", "sensor A", "[tm t="]),
                                      ("why did you enter safe mode?", ["no onboard record"]), ("who won the 1998 world cup?", ["no onboard record"])]),
            ("sensor_bias", 102, [("any faults?", ["sensor", "invalid", "[receipt #"]), ("why did you mark the sensor invalid?", ["disagreed", "[receipt #"]),
                                  ("how confident are you?", ["beat the next option", "[receipt #"]), ("what is the temperature on europa?", ["no onboard record"])]),
            ("seu_spike", 103, [("what is wrong?", ["No confirmed fault", "did not persist"]), ("why did you shed the science load?", ["no onboard record"])]),
            ("wheel_friction", 104, [("why did you switch the wheel?", ["reaction wheel", "more than 1 A", "[receipt #"]),
                                     ("what is the pointing?", ["Pointing error", "[tm t="]), ("what did you learn?", ["admitted", "rule"])])]:
        p = scenario(fault, seed)
        f = FDIR()
        run(p, f.decide)
        rep = SelfReport(f, rulebook=rb)
        for q, must in qs:
            a, cites = rep.answer(q)
            panel.append({"q": q, "a": a, "must": must, "ok": all(m.lower() in a.lower() for m in must)})
    real = sum(x["ok"] for x in panel)
    fakes = {"echo": lambda q: q, "canned 'all nominal'": lambda q: "All systems nominal.",
             "always decline": lambda q: "I have no onboard record that answers that."}
    fake_scores = {k: sum(all(m.lower() in fn(x["q"]).lower() for m in x["must"]) for x in panel) for k, fn in fakes.items()}
    print("   real %d/%d | fakes: %s" % (real, len(panel), fake_scores))
    for x in panel[:4] + [x for x in panel if not x["ok"]][:3]:
        print("   Q: %s\n   A: %s%s" % (x["q"], x["a"][:260], "" if x["ok"] else "   <-- MISSED"))
    OUT["self_report"] = {"real": real, "n": len(panel), "fakes": fake_scores, "items": panel}

    print("\n== 4. COMPUTE (one held-out mission, full onboard stack)")
    tracemalloc.start()
    p = scenario("battery_short", 105)
    f = FDIR()
    t0 = time.perf_counter()
    tr = run(p, f.decide)
    dt = time.perf_counter() - t0
    cur, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    dec = [r["ms"] for r in f.receipts]
    OUT["compute"] = {"mission_ticks": 320, "mission_seconds": round(dt, 3), "ms_per_tick": round(dt * 1000 / 320, 3),
                      "max_decision_ms": max(dec) if dec else 0, "twin_runs": f.twin_runs, "peak_kib": peak // 1024}
    print("  ", OUT["compute"])
    os.makedirs(os.path.join(HERE, "..", "results"), exist_ok=True)
    json.dump(OUT, open(os.path.join(HERE, "..", "results", "eval.json"), "w"), indent=1, default=str)


if __name__ == "__main__":
    main()