"""Local #302 STEP2 gate. Exit 0=PASS, 1=FAIL, 2=INSUFFICIENT; never collects.""" import argparse import hashlib import json from pathlib import Path, PureWindowsPath import subprocess import sys HERE = Path(__file__).resolve().parent FROZEN = HERE / "raw-copy-resume-v2" APPROVED_ANALYZER = "6314743db493b01ee01e550bcbed5558ee7e0d86334d019ff65d03e29cfad339" MIN_DURATION_MS = 180000 TIMEOUT_SECONDS = 30 # Execute the unchanged CLI, then use its parser and coverage function for the # healthy window (the frozen CLI only checks epoch coverage for stale intervals). # Compile the verified helper source explicitly: no cached bytecode is trusted. WORKER = r''' import json, pathlib, runpy, sys, types frozen, capture, out = map(pathlib.Path, sys.argv[1:]) helper = types.ModuleType("step1_rounds") helper.__file__ = str(frozen / "step1_rounds.py") sys.modules[helper.__name__] = helper exec(compile(pathlib.Path(helper.__file__).read_bytes(), helper.__file__, "exec"), helper.__dict__) a = runpy.run_path(str(frozen / "step1-analyze.py")) sys.argv = [str(frozen / "step1-analyze.py"), "--samples", str(capture / "samples.jsonl"), "--log", str(capture / "log-after.log"), "--out", str(out / "analysis")] status = a["main"]() with (out / "analysis-exit.json").open("x", encoding="utf-8", newline="\n") as stream: json.dump({"exit_code": status}, stream, sort_keys=True) stream.write("\n") sys.stdout.flush() sys.stderr.flush() if status in (0, 2): samples = a["load_jsonl"](out / "analysis" / "samples.jsonl") usable = [s for s in samples if s["usable"]] epochs = a["parse_stamps"]((capture / "log-after.log").read_text(encoding="utf-8-sig", errors="replace").splitlines()) window = None if usable: window = {"start_ms": usable[0]["wall_ms"], "end_ms": usable[-1]["wall_ms"], "uncertainty_ms": max(s["uncertainty_ms"] for s in usable)} evidence = {"window": window, "epochs": []} for e in epochs.values(): # Use the START's physical worker identities, not a runtime label alone. # Completed sync spans are poll/native-call durations, not async lifetimes. spans = [s for s in e["spans"] if window and s["end"] and s["begin"]["execution"] == "sync" and s["begin"]["tid"] in e["worker_tids"] and s["begin"]["wall_ms"] <= window["end_ms"] + window["uncertainty_ms"] and s["end"]["wall_ms"] >= window["start_ms"] - window["uncertainty_ms"]] evidence["epochs"].append({ "run_id": e["run_id"], "broker_pid": e["broker_pid"], "wall_start_ms": e["wall_start_ms"], "wall_end_ms": e["wall_end_ms"], "coverage_issues": a["interval_coverage"](e, window) if window else e["issues"], "pair_meet_up": [r for r in e["records"] if r["event"] == "PAIR_MEET_UP"], "broker_sync_spans": [{"base": s["base"], "op_id": s["op_id"], "duration_ms": s["duration_ms"], "tid": s["begin"]["tid"], "runtime": s["begin"]["runtime"], "start_ms": s["begin"]["wall_ms"], "end_ms": s["end"]["wall_ms"], "begin_seq": s["begin"]["seq"], "end_seq": s["end"]["seq"]} for s in spans]}) with (out / "window-evidence.json").open("x", encoding="utf-8", newline="\n") as stream: json.dump(evidence, stream, indent=2, sort_keys=True) stream.write("\n") raise SystemExit(status) ''' def sha(data): return hashlib.sha256(data).hexdigest() def save(path, data): with path.open("xb") as stream: stream.write(data) def load(path, hashes): data = path.read_bytes() hashes[str(path)] = sha(data) return data def json_load(path, hashes): return json.loads(load(path, hashes).decode("utf-8-sig")) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--capture", required=True, type=Path) parser.add_argument("--out", required=True, type=Path, help="NEW local directory; never overwritten") parser.add_argument("--subnet", required=True, action="append", help="repeat exactly three times with distinct names") parser.add_argument("--broker-pid", required=True, type=int) parser.add_argument("--run-id", required=True) args = parser.parse_args() capture, out = args.capture.resolve(), args.out.resolve() receipt = {"schema": "spt-302-step2-gate-v1", "capture": str(capture), "expected": {"broker_pid": args.broker_pid, "run_id": args.run_id, "subnets": sorted(args.subnet)}, "minimum_duration_ms": MIN_DURATION_MS, "hashes": {}, "reasons": [], "analyzer": {"sha256": APPROVED_ANALYZER, "timeout_seconds": TIMEOUT_SECONDS, "exit_code": None, "timed_out": False}} failures, insufficient = [], [] made_out = False hashes = receipt["hashes"] try: if len(args.subnet) != 3 or len(set(args.subnet)) != 3 or any(not s.strip() or s != s.strip() for s in args.subnet): raise ValueError("exactly_three_distinct_nonempty_subnets_required") if args.broker_pid <= 0 or not args.run_id.strip() or args.run_id != args.run_id.strip(): raise ValueError("positive_broker_pid_and_nonempty_run_id_required") # Do not create output inside any immutable predecessor/capture directory. protected = (capture, FROZEN, HERE / "raw-copy-resume-v1") if any(out == path or path in out.parents for path in protected): raise ValueError("output_must_be_outside_capture_and_frozen_analyzer") out.mkdir(parents=True, exist_ok=False) made_out = True hashes[str(Path(__file__).resolve())] = sha(Path(__file__).read_bytes()) delivery = json_load(FROZEN / "DELIVERY.json", hashes) if delivery["sha256"]["step1-analyze.py"] != APPROVED_ANALYZER: raise ValueError("delivery_analyzer_hash_not_approved") for name in ("step1-analyze.py", "step1_rounds.py"): data = load(FROZEN / name, hashes) if sha(data) != delivery["sha256"][name]: raise ValueError("frozen_hash_mismatch:" + name) field = json_load(capture / "field-receipt.json", hashes) samples = load(capture / "samples.jsonl", hashes) log = load(capture / "log-after.log", hashes) # The unchanged collector declares samples SHA in its v1 summary, not # in field-receipt.json. Keep that mandatory predecessor intact. original = json_load(capture / "analysis" / "summary.json", hashes) if type(field.get("collector_exit")) is not int or field["collector_exit"] != 0: raise ValueError("collector_did_not_complete_zero") if (field.get("status") not in ("DISCRIMINATED", "INSUFFICIENT_EVIDENCE") or type(field.get("exit_code")) is not int or field["exit_code"] not in (0, 2) or field.get("analysis_exit") != field["exit_code"] or not field.get("ended_utc")): raise ValueError("collector_receipt_not_complete") if field.get("log_append_only") is not True: raise ValueError("log_append_only_not_declared") before, after = field["log_before"], field["log_after"] if (type(before["bytes"]) is not int or not 0 <= before["bytes"] <= len(log) or after["bytes"] != len(log) or before["offset"] != 0 or after["offset"] != 0 or before["path"] != after["path"] or field["log_bytes_added"] != len(log) - before["bytes"]): raise ValueError("log_snapshot_metadata_mismatch") if sha(log) != after["sha256"] or sha(log[:before["bytes"]]) != before["sha256"]: raise ValueError("log_hash_or_append_only_prefix_mismatch") for name, data in (("samples.jsonl", samples), ("log-after.log", log)): declared = [value for path, value in original["inputs"].items() if PureWindowsPath(path).name == name] if len(declared) != 1 or declared[0] != sha(data): raise ValueError("declared_input_hash_missing_or_mismatched:" + name) receipt["collector"] = {"exit_code": field["collector_exit"], "log_append_only_verified": True, "original_analysis_exit": field["analysis_exit"]} command = [sys.executable, "-I", "-B", "-c", WORKER, str(FROZEN), str(capture), str(out)] try: result = subprocess.run(command, stdin=subprocess.DEVNULL, capture_output=True, timeout=TIMEOUT_SECONDS) stdout, stderr = result.stdout, result.stderr receipt["analyzer"]["worker_exit_code"] = result.returncode except subprocess.TimeoutExpired as error: stdout, stderr = error.stdout or b"", error.stderr or b"" receipt["analyzer"]["timed_out"] = True insufficient.append("frozen_analyzer_timeout") save(out / "analysis.stdout.txt", stdout) save(out / "analysis.stderr.txt", stderr) exit_path = out / "analysis-exit.json" if exit_path.is_file(): receipt["analyzer"]["exit_code"] = json.loads(exit_path.read_text(encoding="utf-8"))["exit_code"] if receipt["analyzer"].get("worker_exit_code") not in (0, 2): insufficient.append("analyzer_evidence_worker_incomplete") # Also detect inputs or verified code changing while the child ran. for path, digest in list(hashes.items()): if sha(Path(path).read_bytes()) != digest: failures.append("input_changed_during_gate:" + path) code = receipt["analyzer"]["exit_code"] if code not in (0, 2): insufficient.append("frozen_analyzer_did_not_complete:exit=" + str(code)) else: summary = json.loads((out / "analysis" / "summary.json").read_text(encoding="utf-8")) receipt["analyzer"].update(summary=summary) normalized = [json.loads(line) for line in (out / "analysis" / "samples.jsonl").read_text(encoding="utf-8").splitlines() if line.strip()] evidence = json.loads((out / "window-evidence.json").read_text(encoding="utf-8")) for name in ("samples.jsonl", "log-after.log"): if summary["inputs"].get(str(capture / name)) != hashes[str(capture / name)]: failures.append("analyzer_input_hash_mismatch:" + name) if summary["stale_intervals"] != 0: failures.append("stale_intervals_observed:" + str(summary["stale_intervals"])) if summary["duration_ms"] < MIN_DURATION_MS: insufficient.append("valid_sample_span_below_180000_ms:" + str(summary["duration_ms"])) # A healthy window necessarily has no positive stale episode to # discriminate. Preserve that raw v2 result, waive only this reason. waived = [r for r in summary["coverage_reasons"] if r == "no_stale_episode_observed" and summary["stale_intervals"] == 0] receipt["analyzer"]["healthy_window_waived_reasons"] = waived insufficient.extend("v2_coverage:" + r for r in summary["coverage_reasons"] if r not in waived) if any(s["broker_pid"] is not None and s["broker_pid"] != args.broker_pid for s in normalized): failures.append("sample_broker_pid_mismatch") if any("nonincreasing_sample_clock" in s["reasons"] for s in normalized): insufficient.append("nonincreasing_sample_clock") window = evidence["window"] receipt["sample_window"] = window meets = {name: [] for name in sorted(args.subnet)} receipt["pair_meet_up"] = meets if window is None: insufficient.append("no_usable_sample_window") else: lo, hi = window["start_ms"], window["end_ms"] candidates = [e for e in evidence["epochs"] if e["broker_pid"] == args.broker_pid and e["wall_start_ms"] <= lo and e["wall_end_ms"] >= hi] if len(candidates) != 1: insufficient.append("missing_or_ambiguous_instrumented_epoch") elif candidates[0]["run_id"] != args.run_id: failures.append("instrumented_run_id_mismatch") else: epoch = candidates[0] receipt["epoch"] = {k: v for k, v in epoch.items() if k not in ("pair_meet_up", "broker_sync_spans")} spans = epoch["broker_sync_spans"] limit = summary["thresholds"]["long_sync_ms"] long_spans = [s for s in spans if s["duration_ms"] >= limit] receipt["broker_sync_span_gate"] = { "threshold_ms": limit, "observed_count": len(spans), "max_duration_ms": max((s["duration_ms"] for s in spans), default=None), "long_spans": long_spans} if long_spans: failures.append("broker_sync_spans_at_or_above_" + str(limit) + "_ms:" + str(len(long_spans))) insufficient.extend("v2_epoch:" + r for r in epoch["coverage_issues"]) if any(e["run_id"] != args.run_id and e["wall_start_ms"] <= hi and e["wall_end_ms"] >= lo for e in evidence["epochs"]): insufficient.append("other_epoch_overlaps_sample_window") for row in epoch["pair_meet_up"]: if (row.get("subnet") in meets and row["broker_pid"] == args.broker_pid and row["run_id"] == args.run_id and lo <= row["wall_ms"] <= hi): meets[row["subnet"]].append({ **{k: row[k] for k in ("wall_ms", "run_id", "broker_pid", "subnet", "seq", "line_number")}, "step": row.get("step")}) insufficient.extend("missing_structured_pair_meet_up_in_sample_window:" + name for name, rows in meets.items() if not rows) steps = {name: sorted({r["step"] for r in rows if type(r["step"]) is int and r["step"] >= 0}) for name, rows in meets.items()} receipt["pair_meet_steps"] = steps insufficient.extend("fewer_than_two_distinct_meet_steps:" + name + ":" + str(len(values)) for name, values in steps.items() if len(values) < 2) except FileNotFoundError as error: insufficient.append("missing_input:" + str(error.filename)) except FileExistsError: failures.append("output_already_exists_refusing_overwrite") except (OSError, ValueError, TypeError, KeyError, AttributeError) as error: failures.append("invalid_evidence:" + str(error)) if made_out: receipt["output_sha256"] = {str(p.relative_to(out)).replace("\\", "/"): sha(p.read_bytes()) for p in sorted(out.rglob("*")) if p.is_file()} receipt["status"] = "FAIL" if failures else "INSUFFICIENT" if insufficient else "PASS" receipt["reasons"] = [{"kind": "FAIL", "reason": r} for r in sorted(set(failures))] receipt["reasons"] += [{"kind": "INSUFFICIENT", "reason": r} for r in sorted(set(insufficient))] receipt["exit_code"] = {"PASS": 0, "FAIL": 1, "INSUFFICIENT": 2}[receipt["status"]] rendered = json.dumps(receipt, indent=2, sort_keys=True) + "\n" if made_out: save(out / "gate-receipt.json", rendered.encode("utf-8")) print(rendered, end="") return receipt["exit_code"] if __name__ == "__main__": raise SystemExit(main())