harness: coverage reporting + --strict-coverage in tracecmp

This commit is contained in:
alex 2026-09-08 01:28:36 -04:00
parent 84a0b7ceae
commit cdada28d78
4 changed files with 426 additions and 15 deletions

View file

@ -43,6 +43,7 @@ online `compare` (shim runs original + ours) and offline `replay` (harness diffs
| `ours` | object | compare only | `{"ret": tv|null, "side": {name: {"after": tv}}}` — the reimplementation's outputs on the same snapshot. | | `ours` | object | compare only | `{"ret": tv|null, "side": {name: {"after": tv}}}` — the reimplementation's outputs on the same snapshot. |
| `diverged` | bool | compare only | the shim's own verdict. The harness recomputes and reports disagreements as warnings. | | `diverged` | bool | compare only | the shim's own verdict. The harness recomputes and reports disagreements as warnings. |
| `diff` | array of diff entries | compare only | the shim's own diff (may be `[]` or truncated). Advisory. | | `diff` | array of diff entries | compare only | the shim's own diff (may be `[]` or truncated). Advisory. |
| `coverage` | object | no | guard findings for this call (section 8). Present **iff** the hook declared at least one guard region, so `"undeclared":[]` means "watched, nothing moved" — which is not the same as the key being absent. |
| `err` | string | no | the hook could not capture (exception in ours, snapshot failure…). Counts as a divergence in `compare`. | | `err` | string | no | the hook could not capture (exception in ours, snapshot failure…). Counts as a divergence in `compare`. |
| `note` | string | no | free text. Ignored. | | `note` | string | no | free text. Ignored. |
@ -50,7 +51,9 @@ Unknown top-level keys are a validation **warning** (not an error) so the shim c
fields without breaking older harnesses. fields without breaking older harnesses.
`replace` records have no `ours`; `ret`/`side` are the reimplementation's. They are `replace` records have no `ours`; `ret`/`side` are the reimplementation's. They are
counted and validated but never diffed (there is nothing to diff against). counted and validated but never diffed (there is nothing to diff against). They are
emitted all the same: a replace run that writes no records can only be judged by the
save-file oracle, which is how B3's missing event went unnoticed for a milestone.
## 3. Typed values (`tv`) ## 3. Typed values (`tv`)
@ -101,8 +104,8 @@ is a divergence unless both are numbers and the tolerance policy is `"numeric"`.
The emitter is a set of `fprintf`s. What it must guarantee: The emitter is a set of `fprintf`s. What it must guarantee:
1. **Key order is fixed**: `ts, hook, mode, call_id, thread, depth, args, ret, side, ours, 1. **Key order is fixed**: `ts, hook, mode, call_id, thread, depth, args, ret, side, ours,
diverged, diff, err, note`. (The harness does not care, but fixed order makes logs diverged, diff, coverage, err, note`. (The harness does not care, but fixed order makes
`diff`-able by eye and grep-able.) logs `diff`-able by eye and grep-able.)
2. **Strings** (`hook`, `str`/`wstr` `v`, `n`, `err`, `note`, struct field names, side 2. **Strings** (`hook`, `str`/`wstr` `v`, `n`, `err`, `note`, struct field names, side
names): write `"`, then for each unit: names): write `"`, then for each unit:
- `"` → `\"`, `\` → `\\` - `"` → `\"`, `\` → `\\`
@ -130,13 +133,24 @@ The emitter is a set of `fprintf`s. What it must guarantee:
```json ```json
{"meta":{"format":1,"build":"sots-engine 0.0.3 g1a2b3c4","exe_sha256":"…", {"meta":{"format":1,"build":"sots-engine 0.0.3 g1a2b3c4","exe_sha256":"…",
"started":"2026-09-07T18:00:00Z","inline_max":256, "started":"2026-09-07T18:00:00Z","inline_max":256,
"hooks":{"CfgVar_RegisterKey":{"ftol":0,"ptr":"ignore"}, "hooks":{"CfgVar_RegisterKey":{"ftol":0,"ptr":"ignore",
"Mars::ParseBlock":{"ftol":1e-6,"ftol_kind":"rel","unordered":["ret.v.items"]}}}} "coverage":{"state":"complete","why":"the declared regions are every word "
"the original writes","unmodelled":[]}},
"Mars::ParseBlock":{"ftol":1e-6,"ftol_kind":"rel","unordered":["ret.v.items"],
"coverage":{"state":"partial","why":"","unmodelled":[
{"what":"appends EVENT_RESEARCH_OVERBUDGET to the owner's event list",
"risk":"high","why":"the message text is composed from the tech name",
"mitigation":"guard:player"}]}}}}}
``` ```
- `format`: this spec's version (1). The harness refuses other versions. - `format`: this spec's version (1). The harness refuses other versions.
- `hooks`: per-hook **policy**, the same keys `tracecmp.py --tolerance` accepts. CLI flags - `hooks`: per-hook **policy**, the same keys `tracecmp.py --tolerance` accepts. CLI flags
override the log's policy; the log's policy overrides harness defaults. override the log's policy; the log's policy overrides harness defaults.
- `hooks.<H>.coverage`: the descriptor's own statement of what its declared regions do **not**
cover (section 8). `state` ∈ `complete | partial | unstated`; `unmodelled` is a list of
`{what, risk, why, mitigation}` with `risk` ∈ `low | medium | high`. A hook with no
`coverage` key reads as `unstated`, which the harness warns about: a clean compare for such
a hook does not say what it did not check.
- Everything else is informational and copied into the report. - Everything else is informational and copied into the report.
## 6. Divergence rules ## 6. Divergence rules
@ -189,7 +203,40 @@ normalization by `oracle_parsers.canonical()`:
objects of their fields; objects of their fields;
- the file ends with a single `\n`. - the file ends with a single `\n`.
## 8. Replay (offline) input ## 8. Coverage: guard regions and the `coverage` block
A declared region is a **check**. Everything else the hooked function writes is invisible to
the diff, and a clean compare says nothing about it — B3 shipped a "13/15 zero-divergence"
result while replace mode wrote a save that was missing an event, because the event landed
outside every declared region. Two fields exist so that can never again be silent.
**Static** — `meta.hooks.<H>.coverage` (section 5): what the descriptor admits it does not
model. The shim requires it at compile time; the harness prints it under every report,
clean or not.
**Per call** — the record's `coverage`:
```json
"coverage":{"guards":["player","tree_header"],
"undeclared":[{"region":"player","off":688,"len":4}],"n":3}
```
- `guards`: the names of the coarse spans watched around this call. A *guard region* is
snapshotted before and after the original (or, in `replace`, around `ours`), is never handed
to the reimplementation and is never diffed.
- `undeclared`: byte runs inside a guard that moved and that **no declared region covers** —
writes the descriptor forgot. Capped (the shim emits at most 16); `n` is the true count.
- The key is present iff at least one guard was declared, so `"undeclared":[]` is a positive
statement ("watched, nothing moved outside the checks"), not an absence of information.
`tracecmp.py` folds these into a `coverage` section per hook: what was compared, what was
guarded, what the guards caught, and what the descriptor admits. Exit codes are unchanged, and
a guard finding on a hook that admits it is *not* a failure. One case is: a hook whose
`coverage.state` is `complete` while a guard caught an undeclared write has been proved wrong
about its own model, and that counts as a divergence (exit 1). `--strict-coverage` widens
that to any undeclared write or any `unstated` hook.
## 9. Replay (offline) input
`tracecmp.py --replay GOLDEN.jsonl IMPL.jsonl`: `GOLDEN` is a normal log (any mode; usually `tracecmp.py --replay GOLDEN.jsonl IMPL.jsonl`: `GOLDEN` is a normal log (any mode; usually
`trace`). `IMPL` is JSONL whose records need only `call_id`, `ret`, `side` (`side` may be `trace`). `IMPL` is JSONL whose records need only `call_id`, `ret`, `side` (`side` may be

View file

@ -30,7 +30,7 @@ import sys
from typing import Any from typing import Any
KEY_ORDER = ["ts", "hook", "mode", "call_id", "thread", "depth", "args", "ret", "side", KEY_ORDER = ["ts", "hook", "mode", "call_id", "thread", "depth", "args", "ret", "side",
"ours", "diverged", "diff", "err", "note"] "ours", "diverged", "diff", "coverage", "err", "note"]
INLINE_MAX = 256 INLINE_MAX = 256
HOOKS = ["CfgVar_RegisterKey", "Manifest_Load", "Mars::ParseBlock"] HOOKS = ["CfgVar_RegisterKey", "Manifest_Load", "Mars::ParseBlock"]
@ -242,8 +242,8 @@ def emit_record(rec: dict) -> str:
',"side":' + emit_side(v["side"]) + "}") ',"side":' + emit_side(v["side"]) + "}")
elif k == "diverged": elif k == "diverged":
parts.append('"diverged":' + ("true" if v else "false")) parts.append('"diverged":' + ("true" if v else "false"))
elif k == "diff": elif k in ("diff", "coverage"):
parts.append('"diff":' + emit_value(v)) parts.append(esc(k) + ":" + emit_value(v))
else: else:
raise AssertionError(k) raise AssertionError(k)
return "{" + ",".join(parts) + "}\n" return "{" + ",".join(parts) + "}\n"
@ -265,11 +265,35 @@ def write_log(path: str, meta: dict | None, records: list[dict], raw_lines: list
# --- scenario generation ---------------------------------------------------------------- # --- scenario generation ----------------------------------------------------------------
def coverage(state: str = "partial", why: str = "", unmodelled=()) -> dict:
"""meta.hooks[H].coverage -- the descriptor's admission of what it does not check."""
return {"state": state, "why": why, "unmodelled": [dict(u) for u in unmodelled]}
PARTIAL = coverage("partial", "", [
{"what": "appends to the owner's event list", "risk": "high",
"why": "the message text is composed by the game", "mitigation": "guard:player"}])
COMPLETE = coverage("complete", "the declared regions are every word the original writes")
def meta(**hooks) -> dict: def meta(**hooks) -> dict:
"""Per-hook policy. Any hook without an explicit `coverage` gets the partial default, so a
fixture log looks like a shim log that has been through the coverage audit."""
for h in HOOKS:
hooks.setdefault(h, {})
for d in hooks.values():
d.setdefault("coverage", copy.deepcopy(PARTIAL))
return {"format": 1, "build": "mkfixture 1", "exe_sha256": "0" * 64, return {"format": 1, "build": "mkfixture 1", "exe_sha256": "0" * 64,
"started": "2026-09-07T00:00:00Z", "inline_max": INLINE_MAX, "hooks": hooks} "started": "2026-09-07T00:00:00Z", "inline_max": INLINE_MAX, "hooks": hooks}
def guard_block(guards, spans=(), total=None) -> dict:
"""A record-level `coverage` block: what the guards watched and what they caught."""
spans = [dict(s) for s in spans]
return {"guards": list(guards), "undeclared": spans,
"n": len(spans) if total is None else total}
def gen_calls(n: int, seed: int = 1) -> list[dict]: def gen_calls(n: int, seed: int = 1) -> list[dict]:
"""n trace calls cycling over HOOKS, with every tv type represented.""" """n trace calls cycling over HOOKS, with every tv type represented."""
rng = random.Random(seed) rng = random.Random(seed)
@ -452,8 +476,37 @@ def build_all(outdir: str, n: int = 12, seed: int = 1) -> dict:
impl_bad.append({"call_id": 9999, "hook": "Extra", "ret": None, "side": {}}) # extra -> warning impl_bad.append({"call_id": 9999, "hook": "Extra", "ret": None, "side": {}}) # extra -> warning
emit_impl(paths["replay_impl_bad"], impl_bad) emit_impl(paths["replay_impl_bad"], impl_bad)
write_log(paths["invalid"], m, [], invalid_lines()) write_log(paths["invalid"], m, [], invalid_lines())
# --- coverage scenarios (docs/harness-audit.md) --------------------------------------
#
# coverage_guarded : the B3 shape. Every call diffs clean, and a guard region reports the
# write the descriptor admits it does not model -> exit 0, but the
# report says so.
# coverage_lying : same log, but the descriptor claims COMPLETE coverage. The guard
# contradicts the claim -> exit 1.
# coverage_unstated : a log whose meta carries no coverage at all -> warned, verdict
# "unstated", and --strict-coverage fails it.
span = {"region": "player", "off": 0x2b0, "len": 4}
guarded = _deep(clean)
for i, r in enumerate(guarded):
r["coverage"] = guard_block(["player"], [span] if i % 3 == 0 else [])
paths["coverage_guarded"] = os.path.join(outdir, "coverage_guarded.jsonl")
write_log(paths["coverage_guarded"], m, guarded)
lying = meta(**{h: {"coverage": copy.deepcopy(COMPLETE)} for h in HOOKS})
paths["coverage_lying"] = os.path.join(outdir, "coverage_lying.jsonl")
write_log(paths["coverage_lying"], lying, guarded)
silent = {"format": 1, "build": "mkfixture 1", "exe_sha256": "0" * 64,
"started": "2026-09-07T00:00:00Z", "inline_max": INLINE_MAX,
"hooks": {h: {"ftol": 0} for h in HOOKS}}
paths["coverage_unstated"] = os.path.join(outdir, "coverage_unstated.jsonl")
write_log(paths["coverage_unstated"], silent, clean)
return {"paths": paths, "trace": trace, "compare_bad": bad, "expect": exp, return {"paths": paths, "trace": trace, "compare_bad": bad, "expect": exp,
"replay_bad": {"wrong": 0, "missing": dropped, "extra": 9999}} "replay_bad": {"wrong": 0, "missing": dropped, "extra": 9999},
"guard_span": span,
"guard_calls": sum(1 for i in range(len(clean)) if i % 3 == 0)}
def main(argv=None) -> int: def main(argv=None) -> int:

View file

@ -166,8 +166,14 @@ class CompareTest(unittest.TestCase):
def test_clean_log_is_clean(self): def test_clean_log_is_clean(self):
rc, rep, out = run(P["compare_clean"]) rc, rep, out = run(P["compare_clean"])
self.assertEqual(rc, 0) self.assertEqual(rc, 0)
self.assertEqual(rep["totals"], {"calls": 12, "compared": 12, "diverged": 0, "invalid_records": 0}) self.assertEqual(rep["totals"], {"calls": 12, "compared": 12, "diverged": 0,
"invalid_records": 0, "guarded_calls": 0,
"undeclared_calls": 0, "undeclared_writes": 0,
"coverage_unstated": 0, "coverage_contradicted": 0})
self.assertIn("| CfgVar_RegisterKey | 4 |", out) self.assertIn("| CfgVar_RegisterKey | 4 |", out)
# a clean run still states what it did not check
self.assertIn("### coverage", out)
self.assertIn("appends to the owner's event list", out)
def test_injected_divergences_exact(self): def test_injected_divergences_exact(self):
rc, rep, out = run(P["compare_bad"], "--first", "20") rc, rep, out = run(P["compare_bad"], "--first", "20")
@ -333,6 +339,100 @@ class CompareTest(unittest.TestCase):
self.assertEqual(rc, 1) self.assertEqual(rc, 1)
class CoverageTest(unittest.TestCase):
"""The audit mechanism: a clean compare must state what it did not check, and a descriptor
that claims complete coverage must be contradicted when a guard region says otherwise.
See sots-engine/docs/harness-audit.md."""
def test_meta_coverage_reaches_the_report(self):
_, rep, out = run(P["compare_clean"])
cov = rep["hooks"]["CfgVar_RegisterKey"]["coverage"]
self.assertEqual(cov["verdict"], "partial")
self.assertEqual(cov["checked_regions"], ["cfg_table"])
self.assertEqual(len(cov["unmodelled"]), 1)
self.assertEqual(cov["unmodelled"][0]["risk"], "high")
self.assertIn("| CfgVar_RegisterKey | partial | cfg_table |", out)
def test_guard_findings_are_reported_but_do_not_fail_an_honest_hook(self):
rc, rep, out = run(P["coverage_guarded"])
self.assertEqual(rc, 0) # ours matched every declared region: no divergence
cov = rep["hooks"]["CfgVar_RegisterKey"]["coverage"]
self.assertEqual(cov["verdict"], "partial")
self.assertEqual(cov["guards"], ["player"])
self.assertEqual(cov["guarded_calls"], 4)
self.assertEqual(cov["undeclared_calls"], FX["guard_calls"])
self.assertEqual(cov["undeclared_writes"], FX["guard_calls"])
self.assertEqual(cov["spans"]["compare"], ["player+0x2b0:4"])
self.assertEqual(rep["totals"]["undeclared_writes"], FX["guard_calls"])
self.assertIn("guard hits in compare mode: player+0x2b0:4", out)
def test_strict_coverage_fails_on_any_undeclared_write(self):
self.assertEqual(run(P["coverage_guarded"], "--strict-coverage")[0], 1)
self.assertEqual(run(P["compare_clean"], "--strict-coverage")[0], 0)
def test_a_false_completeness_claim_is_a_divergence(self):
rc, rep, out = run(P["coverage_lying"])
self.assertEqual(rc, 1)
self.assertEqual(rep["coverage_contradicted"], ["CfgVar_RegisterKey"])
self.assertEqual(rep["totals"]["coverage_contradicted"], 1)
self.assertEqual(rep["totals"]["diverged"], 0) # not a per-call diff: a false claim
self.assertEqual(rep["hooks"]["CfgVar_RegisterKey"]["coverage"]["verdict"], "contradicted")
self.assertEqual(rep["hooks"]["Manifest_Load"]["coverage"]["verdict"], "complete")
self.assertTrue(any("claims complete coverage but a guard saw" in w for w in rep["warnings"]))
def test_a_log_with_no_coverage_statement_is_called_out(self):
rc, rep, out = run(P["coverage_unstated"])
self.assertEqual(rc, 0) # legacy logs still pass...
self.assertEqual(rep["coverage_unstated"],
["CfgVar_RegisterKey", "Manifest_Load", "Mars::ParseBlock"])
self.assertTrue(all(s["coverage"]["verdict"] == "unstated" for s in rep["hooks"].values()))
self.assertTrue(any("no coverage statement in meta" in w for w in rep["warnings"]))
self.assertIn("| CfgVar_RegisterKey | unstated |", out)
self.assertEqual(run(P["coverage_unstated"], "--strict-coverage")[0], 1) # ...but not here
def test_record_coverage_block_is_validated(self):
good = {"guards": ["p"], "undeclared": [{"region": "p", "off": 4, "len": 4}], "n": 1}
for bad in ({"guards": "p", "undeclared": []},
{"guards": [], "undeclared": [{"region": "p", "off": "x", "len": 4}], "n": 1},
{"guards": [], "undeclared": [{"region": "p", "off": 4, "len": 4}], "n": 0},
{"guards": [], "undeclared": [], "n": -1},
{"guards": []},
"not-an-object"):
errs: list = []
tc.validate_coverage(bad, errs)
self.assertTrue(errs, bad)
errs = []
tc.validate_coverage(good, errs)
self.assertEqual(errs, [])
def test_a_bad_coverage_block_invalidates_the_record(self):
path = os.path.join(TMP, "cov_invalid.jsonl")
rec = dict(mk.to_compare(FX["trace"])[0])
rec["coverage"] = {"guards": [1], "undeclared": []}
mk.write_log(path, mk.meta(), [rec])
rc, rep, _ = run(path)
self.assertEqual(rc, 2)
self.assertIn("coverage.guards", rep["invalid"][0]["error"])
def test_compare_vs_replace_asymmetry_is_named(self):
"""The B3 shape: the original writes something in compare mode that ours never writes in
replace mode. Both logs are individually clean; only the two together show the gap."""
path = os.path.join(TMP, "cov_asym.jsonl")
recs = mk.to_compare(FX["trace"])[:2]
a, b = dict(recs[0]), dict(recs[1])
a["coverage"] = mk.guard_block(["player"], [{"region": "player", "off": 0x2b0, "len": 4}])
b["hook"] = a["hook"] # the same hook in both modes: that is what makes them comparable
b["mode"] = "replace"
b.pop("ours", None)
b.pop("diverged", None)
b.pop("diff", None)
b["coverage"] = mk.guard_block(["player"])
mk.write_log(path, mk.meta(), [a, b])
rc, _, out = run(path)
self.assertEqual(rc, 0)
self.assertIn("ONLY the original writes these: player+0x2b0:4", out)
class ReplayTest(unittest.TestCase): class ReplayTest(unittest.TestCase):
def test_replay_ok(self): def test_replay_ok(self):
rc, rep, _ = run(P["trace_clean"], "--replay", P["replay_impl_ok"]) rc, rep, _ = run(P["trace_clean"], "--replay", P["replay_impl_ok"])

View file

@ -18,10 +18,18 @@ flags:
--first N diffs shown per hook (default 5) --first N diffs shown per hook (default 5)
--json-out PATH write the full report as JSON (convention: --json-out PATH write the full report as JSON (convention:
verify/results/compare/<run>.json) verify/results/compare/<run>.json)
--strict-coverage also fail on any undeclared write or unstated hook
--skip-invalid drop unparsable/invalid records with a warning instead of --skip-invalid drop unparsable/invalid records with a warning instead of
failing the run failing the run
Every report ends with a `coverage` section stating, per hook, which regions were
actually compared, which coarse spans were guarded, what the guards saw the original
write outside those regions, and what the hook's descriptor admits it does not model.
A clean compare therefore always says what it did not check.
exit: 0 = no divergences; 1 = divergences; 2 = invalid input (or usage). exit: 0 = no divergences; 1 = divergences; 2 = invalid input (or usage).
A hook whose descriptor claims complete coverage while a guard region caught an
undeclared write is counted as a divergence (it is one: the model is wrong).
Stdlib only. Stdlib only.
""" """
@ -44,7 +52,8 @@ FLOAT_TYPES = {"f32", "f64"}
TV_TYPES = INT_TYPES | FLOAT_TYPES | {"bool", "str", "wstr", "ptr", "enum", "null", TV_TYPES = INT_TYPES | FLOAT_TYPES | {"bool", "str", "wstr", "ptr", "enum", "null",
"bytes", "list", "set", "struct", "json"} "bytes", "list", "set", "struct", "json"}
RECORD_KEYS = {"ts", "hook", "mode", "call_id", "thread", "depth", "args", "ret", "side", RECORD_KEYS = {"ts", "hook", "mode", "call_id", "thread", "depth", "args", "ret", "side",
"ours", "diverged", "diff", "err", "note"} "ours", "diverged", "diff", "coverage", "err", "note"}
COVERAGE_STATES = ("complete", "partial", "unstated")
INT_RANGE = {"i8": (-2**7, 2**7 - 1), "i16": (-2**15, 2**15 - 1), "i32": (-2**31, 2**31 - 1), INT_RANGE = {"i8": (-2**7, 2**7 - 1), "i16": (-2**15, 2**15 - 1), "i32": (-2**31, 2**31 - 1),
"i64": (-2**63, 2**63 - 1), "u8": (0, 2**8 - 1), "u16": (0, 2**16 - 1), "i64": (-2**63, 2**63 - 1), "u8": (0, 2**8 - 1), "u16": (0, 2**16 - 1),
"u32": (0, 2**32 - 1), "u64": (0, 2**64 - 1)} "u32": (0, 2**32 - 1), "u64": (0, 2**64 - 1)}
@ -220,6 +229,44 @@ def validate_side(side: Any, path: str, errs: list[str], need_before: bool = Fal
validate_tv(ent["after"], f"{path}.{name}.after", errs) validate_tv(ent["after"], f"{path}.{name}.after", errs)
def validate_coverage(cov: Any, errs: list[str]) -> None:
"""Record-level `coverage`: {guards: [name], undeclared: [{region, off, len}], n: int}.
Present iff the call declared at least one Guard region, so an empty `undeclared` means
"watched and nothing moved" -- which is different from, and much stronger than, absent.
"""
if not isinstance(cov, dict):
errs.append("coverage must be an object")
return
guards = cov.get("guards")
if not isinstance(guards, list) or not all(isinstance(g, str) for g in guards):
errs.append("coverage.guards must be an array of strings")
und = cov.get("undeclared")
if not isinstance(und, list):
errs.append("coverage.undeclared must be an array")
else:
for i, w in enumerate(und):
if (not isinstance(w, dict) or not isinstance(w.get("region"), str)
or not _is_int(w.get("off")) or not _is_int(w.get("len"))):
errs.append(f"coverage.undeclared[{i}]: must be {{region, off, len}}")
if "n" in cov and (not _is_int(cov["n"]) or cov["n"] < 0):
errs.append("coverage.n must be int >= 0")
if isinstance(und, list) and _is_int(cov.get("n")) and cov["n"] < len(und):
errs.append("coverage.n is smaller than the spans it reports")
def validate_meta_coverage(hook: str, cov: Any, warns: list[str]) -> None:
"""meta.hooks[H].coverage: the descriptor's own admission of what it does not check."""
if not isinstance(cov, dict):
warns.append(f"meta.hooks.{hook}.coverage must be an object")
return
if cov.get("state") not in COVERAGE_STATES:
warns.append(f"meta.hooks.{hook}.coverage.state must be one of {COVERAGE_STATES}")
for i, n in enumerate(cov.get("unmodelled") or []):
if not isinstance(n, dict) or not isinstance(n.get("what"), str) or not n.get("what"):
warns.append(f"meta.hooks.{hook}.coverage.unmodelled[{i}]: needs a non-empty 'what'")
def validate_record(rec: Any) -> tuple[list[str], list[str]]: def validate_record(rec: Any) -> tuple[list[str], list[str]]:
"""-> (errors, warnings). Errors make the record invalid.""" """-> (errors, warnings). Errors make the record invalid."""
errs: list[str] = [] errs: list[str] = []
@ -251,6 +298,8 @@ def validate_record(rec: Any) -> tuple[list[str], list[str]]:
if rec["ret"] is not None: if rec["ret"] is not None:
validate_tv(rec["ret"], "ret", errs) validate_tv(rec["ret"], "ret", errs)
validate_side(rec["side"], "side", errs) validate_side(rec["side"], "side", errs)
if "coverage" in rec:
validate_coverage(rec["coverage"], errs)
if "err" in rec and not isinstance(rec["err"], str): if "err" in rec and not isinstance(rec["err"], str):
errs.append("err must be a string") errs.append("err must be a string")
if "note" in rec and not isinstance(rec["note"], str): if "note" in rec and not isinstance(rec["note"], str):
@ -583,9 +632,59 @@ def load_impl(path: str) -> tuple[dict[int, dict], list[tuple[int, str]]]:
# --- report ------------------------------------------------------------------------ # --- report ------------------------------------------------------------------------
MAX_SPANS_REPORTED = 12
def new_hook_stats() -> dict: def new_hook_stats() -> dict:
return {"calls": 0, "modes": {}, "compared": 0, "diverged": 0, "errors": 0, return {"calls": 0, "modes": {}, "compared": 0, "diverged": 0, "errors": 0,
"diverged_call_ids": [], "diffs": []} "diverged_call_ids": [], "diffs": [],
# --- coverage: what this run actually checked, and what it admits it did not ---
"coverage": {
"state": "unstated", # from meta.hooks[H].coverage (absent -> unstated)
"why": "",
"unmodelled": [], # [{what, risk, why, mitigation}]
"checked_regions": [], # side names actually diffed (union over records)
"guards": [], # guard names the shim watched
"guarded_calls": 0, # calls that carried a coverage block
"undeclared_calls": 0, # calls where a guard saw an undeclared write
"undeclared_writes": 0, # total spans, honest even when the list truncates
"spans": {}, # mode -> ["region+off:len", ...] (capped)
}}
def _hook_coverage_from_meta(logs: list) -> dict:
"""meta.hooks[H].coverage, merged across the input logs (last log wins on conflict)."""
out: dict[str, dict] = {}
for log in logs:
for h, d in (log.meta.get("hooks") or {}).items():
if isinstance(d, dict) and isinstance(d.get("coverage"), dict):
out[h] = d["coverage"]
return out
def _account_coverage(st: dict, rec: dict) -> None:
"""Fold one record's declared regions and guard findings into the hook's coverage stats."""
cov = st["coverage"]
for name in rec.get("side") or {}:
if name not in cov["checked_regions"]:
cov["checked_regions"].append(name)
block = rec.get("coverage")
if not isinstance(block, dict):
return
cov["guarded_calls"] += 1
for g in block.get("guards") or []:
if g not in cov["guards"]:
cov["guards"].append(g)
spans = block.get("undeclared") or []
total = block.get("n", len(spans))
if total:
cov["undeclared_calls"] += 1
cov["undeclared_writes"] += total
bucket = cov["spans"].setdefault(rec["mode"], [])
for w in spans:
key = "%s+0x%x:%d" % (w["region"], w["off"], w["len"])
if key not in bucket and len(bucket) < MAX_SPANS_REPORTED:
bucket.append(key)
def run_report(logs: list[Log], policies: PolicyTable, hooks: set[str] | None, first: int) -> dict: def run_report(logs: list[Log], policies: PolicyTable, hooks: set[str] | None, first: int) -> dict:
@ -601,6 +700,7 @@ def run_report(logs: list[Log], policies: PolicyTable, hooks: set[str] | None, f
st = per_hook.setdefault(h, new_hook_stats()) st = per_hook.setdefault(h, new_hook_stats())
st["calls"] += 1 st["calls"] += 1
st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1 st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1
_account_coverage(st, rec)
if rec["mode"] != "compare": if rec["mode"] != "compare":
if "err" in rec: if "err" in rec:
st["errors"] += 1 st["errors"] += 1
@ -634,6 +734,7 @@ def run_replay(golden: list[Log], impl: dict[int, dict], policies: PolicyTable,
st = per_hook.setdefault(h, new_hook_stats()) st = per_hook.setdefault(h, new_hook_stats())
st["calls"] += 1 st["calls"] += 1
st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1 st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1
_account_coverage(st, rec)
st["compared"] += 1 st["compared"] += 1
cid = rec["call_id"] cid = rec["call_id"]
seen.add(cid) seen.add(cid)
@ -659,6 +760,42 @@ def run_replay(golden: list[Log], impl: dict[int, dict], policies: PolicyTable,
def _finish(per_hook: dict, logs: list[Log], warnings: list[str], kind: str) -> dict: def _finish(per_hook: dict, logs: list[Log], warnings: list[str], kind: str) -> dict:
total_div = sum(s["diverged"] for s in per_hook.values()) total_div = sum(s["diverged"] for s in per_hook.values())
invalid = [(os.path.basename(l.path), ln, m) for l in logs for ln, m in l.invalid] invalid = [(os.path.basename(l.path), ln, m) for l in logs for ln, m in l.invalid]
# Fold the shim's own coverage declaration in, and decide the coverage verdict per hook.
#
# unstated the descriptor never said what it does not check -> the compare's
# completeness is unknown, and a clean run means nothing on its own.
# contradicted the descriptor claims complete coverage, but a Guard region caught the
# original writing outside every declared region. The claim is false, and
# that counts as a divergence (exit 1) -- it is the exact failure B3 hit.
# partial the honest normal case: admissions on record, guards agreeing with them.
# complete claims completeness, and no guard contradicted it.
meta_cov = _hook_coverage_from_meta(logs)
unstated, contradicted = [], []
for h, s in per_hook.items():
cov = s["coverage"]
m = meta_cov.get(h)
if isinstance(m, dict):
cov["state"] = m.get("state", "unstated")
cov["why"] = m.get("why", "")
cov["unmodelled"] = list(m.get("unmodelled") or [])
for k in ("checked_regions", "guards"):
cov[k] = sorted(cov[k])
if cov["state"] == "unstated":
cov["verdict"] = "unstated"
unstated.append(h)
elif cov["state"] == "complete" and cov["undeclared_writes"]:
cov["verdict"] = "contradicted"
contradicted.append(h)
else:
cov["verdict"] = cov["state"]
if cov["verdict"] == "unstated":
warnings.append(f"{h}: no coverage statement in meta -- a clean compare for this hook "
f"does not say what it did not check")
if cov["verdict"] == "contradicted":
warnings.append(f"{h}: claims complete coverage but a guard saw "
f"{cov['undeclared_writes']} undeclared write(s)")
return { return {
"kind": kind, "kind": kind,
"format": FORMAT_VERSION, "format": FORMAT_VERSION,
@ -668,7 +805,14 @@ def _finish(per_hook: dict, logs: list[Log], warnings: list[str], kind: str) ->
"totals": {"calls": sum(s["calls"] for s in per_hook.values()), "totals": {"calls": sum(s["calls"] for s in per_hook.values()),
"compared": sum(s["compared"] for s in per_hook.values()), "compared": sum(s["compared"] for s in per_hook.values()),
"diverged": total_div, "diverged": total_div,
"invalid_records": len(invalid)}, "invalid_records": len(invalid),
"guarded_calls": sum(s["coverage"]["guarded_calls"] for s in per_hook.values()),
"undeclared_calls": sum(s["coverage"]["undeclared_calls"] for s in per_hook.values()),
"undeclared_writes": sum(s["coverage"]["undeclared_writes"] for s in per_hook.values()),
"coverage_unstated": len(unstated),
"coverage_contradicted": len(contradicted)},
"coverage_unstated": sorted(unstated),
"coverage_contradicted": sorted(contradicted),
"invalid": [{"file": f, "line": ln, "error": m} for f, ln, m in invalid], "invalid": [{"file": f, "line": ln, "error": m} for f, ln, m in invalid],
"warnings": warnings, "warnings": warnings,
} }
@ -679,6 +823,59 @@ def _short(x: Any, n: int = 60) -> str:
return s if len(s) <= n else s[:n - 1] + "…" return s if len(s) <= n else s[:n - 1] + "…"
def print_coverage(rep: dict, out=None) -> None:
"""What this run checked, and what it says it did not.
Printed on every run, clean or not. A compare that reports zero divergences bounds only the
regions it declared; without this section that fact is invisible, which is exactly how B3's
unposted event survived a clean pass.
"""
out = out or sys.stdout
P = lambda s="": print(s, file=out) # noqa: E731
if not rep["hooks"]:
return
P()
P("### coverage")
P()
P("| hook | verdict | compared regions | guards | undeclared writes | unmodelled |")
P("|---|---|---|---|---|---|")
for h, s in rep["hooks"].items():
c = s["coverage"]
regions = ", ".join(c["checked_regions"][:6]) or "-"
if len(c["checked_regions"]) > 6:
regions += f", +{len(c['checked_regions']) - 6}"
guards = ", ".join(c["guards"]) or "-"
und = (f"{c['undeclared_writes']} in {c['undeclared_calls']} call(s)"
if c["undeclared_writes"] else ("0" if c["guarded_calls"] else "not watched"))
P(f"| {h} | {c.get('verdict', c['state'])} | {regions} | {guards} | {und} | "
f"{len(c['unmodelled'])} |")
for h, s in rep["hooks"].items():
c = s["coverage"]
if not c["unmodelled"] and not c["spans"]:
continue
P()
P(f"#### {h} — not checked by this run")
if c.get("why"):
P(f"- claims complete coverage: {c['why']}")
for n in c["unmodelled"]:
mit = f" [{n['mitigation']}]" if n.get("mitigation") else ""
P(f"- ({n.get('risk', '?')}) {n.get('what', '?')} — {n.get('why', '')}{mit}")
for mode, spans in sorted(c["spans"].items()):
if spans:
P(f"- guard hits in {mode} mode: {', '.join(spans)}")
# A write the original makes in compare mode but ours never makes in replace mode (or
# the reverse) is the asymmetry that a save-hash oracle would eventually catch the hard
# way. Say it here instead.
if "compare" in c["spans"] and "replace" in c["spans"]:
cmp_s, rep_s = set(c["spans"]["compare"]), set(c["spans"]["replace"])
only_c = sorted(cmp_s - rep_s)
only_r = sorted(rep_s - cmp_s)
if only_c:
P(f"- ONLY the original writes these: {', '.join(only_c)}")
if only_r:
P(f"- ONLY ours writes these: {', '.join(only_r)}")
def print_report(rep: dict, out=None) -> None: def print_report(rep: dict, out=None) -> None:
out = out or sys.stdout out = out or sys.stdout
P = lambda s="": print(s, file=out) # noqa: E731 P = lambda s="": print(s, file=out) # noqa: E731
@ -689,12 +886,17 @@ def print_report(rep: dict, out=None) -> None:
t = rep["totals"] t = rep["totals"]
P(f"- calls: {t['calls']} compared: {t['compared']} diverged: {t['diverged']} " P(f"- calls: {t['calls']} compared: {t['compared']} diverged: {t['diverged']} "
f"invalid records: {t['invalid_records']} warnings: {len(rep['warnings'])}") f"invalid records: {t['invalid_records']} warnings: {len(rep['warnings'])}")
P(f"- coverage: {t.get('guarded_calls', 0)} guarded call(s), "
f"{t.get('undeclared_writes', 0)} undeclared write(s) in {t.get('undeclared_calls', 0)} call(s); "
f"{t.get('coverage_unstated', 0)} hook(s) unstated, "
f"{t.get('coverage_contradicted', 0)} contradicted")
P() P()
P("| hook | calls | modes | compared | diverged | errors |") P("| hook | calls | modes | compared | diverged | errors |")
P("|---|---|---|---|---|---|") P("|---|---|---|---|---|---|")
for h, s in rep["hooks"].items(): for h, s in rep["hooks"].items():
modes = " ".join(f"{k}:{v}" for k, v in sorted(s["modes"].items())) modes = " ".join(f"{k}:{v}" for k, v in sorted(s["modes"].items()))
P(f"| {h} | {s['calls']} | {modes} | {s['compared']} | {s['diverged']} | {s['errors']} |") P(f"| {h} | {s['calls']} | {modes} | {s['compared']} | {s['diverged']} | {s['errors']} |")
print_coverage(rep, out)
for h, s in rep["hooks"].items(): for h, s in rep["hooks"].items():
if not s["diffs"]: if not s["diffs"]:
continue continue
@ -735,6 +937,9 @@ def build_parser() -> argparse.ArgumentParser:
ap.add_argument("--numeric", action="store_true") ap.add_argument("--numeric", action="store_true")
ap.add_argument("--first", type=int, default=5, metavar="N") ap.add_argument("--first", type=int, default=5, metavar="N")
ap.add_argument("--json-out", metavar="PATH") ap.add_argument("--json-out", metavar="PATH")
ap.add_argument("--strict-coverage", action="store_true",
help="also fail (exit 1) on any undeclared write or unstated hook, not only "
"on a hook whose 'complete' claim a guard contradicted")
ap.add_argument("--skip-invalid", action="store_true") ap.add_argument("--skip-invalid", action="store_true")
ap.add_argument("--quiet", action="store_true", help="no markdown report on stdout") ap.add_argument("--quiet", action="store_true", help="no markdown report on stdout")
return ap return ap
@ -785,7 +990,13 @@ def main(argv: list[str] | None = None) -> int:
print_report(rep) print_report(rep)
if rep["totals"]["invalid_records"] and not a.skip_invalid: if rep["totals"]["invalid_records"] and not a.skip_invalid:
return 2 return 2
return 1 if rep["totals"]["diverged"] else 0 # A descriptor that claims complete coverage while a guard watched the original write
# outside every declared region has been proved wrong: that is a divergence between the
# harness's model and the game, and it fails the run like any other.
failed = rep["totals"]["diverged"] or rep["totals"]["coverage_contradicted"]
if a.strict_coverage:
failed = failed or rep["totals"]["undeclared_writes"] or rep["totals"]["coverage_unstated"]
return 1 if failed else 0
if __name__ == "__main__": if __name__ == "__main__":