"""Read-only #302 STEP1 joiner. Exit 2 means evidence is insufficient, not healthy.""" import argparse from collections import Counter, defaultdict import hashlib import json from pathlib import Path import re import sys from step1_rounds import parse_rounds, align_rounds BASES = ("NET_IF_ENUM", "NET_GATEWAY_LOOKUP", "NET_CANCEL_MIB", "NET_MEET_BIND", "NET_MEET_RETIRE", "NET_NTP_REFRESH") THRESHOLDS = {"stale_ms": 1000, "severe_ms": 2000, "query_ms": 250, "cadence_gap_ms": 1500, "long_sync_ms": 250, "clock_offset_spread_ms": 100, "capture_span_ms": 100, "capture_separation_ms": 250, "pty_rtt_ms": 2000, "usable_fraction": .95, "default_min_duration_ms": 170000} SYMBOLS = { "NET_IF_ENUM": re.compile(r"GetAdaptersAddresses|netdev.*(?:get_interfaces|get_adapters)", re.I), "NET_GATEWAY_LOOKUP": re.compile(r"(?:portmapper|netwatch).*(?:ip_and_gateway|HomeRouter)|GetAdaptersAddresses", re.I), "NET_CANCEL_MIB": re.compile(r"CancelMibChangeNotify2", re.I), "NET_NTP_REFRESH": re.compile(r"(?:pairing.*ntp|ntp::).*(?:refresh|query|offset)|(?:GetAddrInfo|getaddrinfo|recvfrom)", re.I), } IDLE = re.compile(r"tokio::runtime::.*(?:\bpark\b|::park::|::Park)", re.I) def number(value): return type(value) in (int, float) and value >= 0 and value < float("inf") def load_jsonl(path): rows = [] for n, line in enumerate(path.read_text(encoding="utf-8-sig").splitlines(), 1): if line.strip(): row = json.loads(line) if not isinstance(row, dict): raise ValueError(f"{path}:{n}: object required") rows.append(row) return rows def normalize_samples(rows): normalized = [] previous_time = None for index, row in enumerate(rows): probe = row.get("probe") or {} reasons = [] if not isinstance(probe, dict): probe = {} if row.get("exit") != 0 or row.get("timed_out") or row.get("failure") or probe.get("probe_error"): reasons.append("query_failed") if probe.get("net_enabled") is not True: reasons.append("net_disabled_or_unknown") for key in ("sampled_at_ms", "net_canary_age_ms", "active_dial_tasks", "query_elapsed_ms", "broker_pid"): if not number(probe.get(key)): reasons.append("missing_or_invalid_" + key) elapsed = row.get("outer_elapsed_ms") query_elapsed = probe.get("query_elapsed_ms") if (not number(elapsed) or elapsed > THRESHOLDS["query_ms"] or not number(query_elapsed) or query_elapsed > THRESHOLDS["query_ms"]): reasons.append("query_timing_uncertain") stamp = probe.get("sampled_at_ms") if number(stamp): if previous_time is not None and stamp <= previous_time: reasons.append("nonincreasing_sample_clock") previous_time = stamp normalized.append({"index": index, "sequence": row.get("sequence"), "wall_ms": stamp, "age_ms": probe.get("net_canary_age_ms"), "dials": probe.get("active_dial_tasks"), "broker_pid": probe.get("broker_pid"), "broker_version": probe.get("broker_version"), "uncertainty_ms": max((v for v in (elapsed, query_elapsed) if number(v)), default=0), "usable": not reasons, "reasons": reasons}) return normalized def stale_intervals(samples): intervals, current, previous = [], None, None def finish(recovery=None, reason=None): nonlocal current if current: current["recovery_bound_ms"] = recovery current["right_censored"] = recovery is None current["end_reason"] = reason current["id"] = len(intervals) + 1 intervals.append(current) current = None for sample in samples: if not sample["usable"]: finish(reason="invalid_sample") previous = sample continue t, age = sample["wall_ms"], sample["age_ms"] continuous = (previous is not None and previous["usable"] and 0 < t - previous["wall_ms"] <= THRESHOLDS["cadence_gap_ms"] and sample["broker_pid"] == previous["broker_pid"]) if current and not continuous: finish(reason="cadence_or_broker_gap") if age < THRESHOLDS["stale_ms"]: finish(recovery=t, reason="healthy_sample") else: support = t - age + THRESHOLDS["stale_ms"] if current is None: current = {"broker_pid": sample["broker_pid"], "start_ms": support, "end_ms": t, "first_observed_ms": t, "last_observed_ms": t, "peak_ms": t, "peak_age_ms": age, "left_censored": not continuous, "samples": [], "dials": [], "uncertainty_ms": sample["uncertainty_ms"]} current["start_ms"] = min(current["start_ms"], support) current["end_ms"] = t current["last_observed_ms"] = t current["samples"].append(sample["index"]) current["dials"].append(sample["dials"]) current["uncertainty_ms"] = max(current["uncertainty_ms"], sample["uncertainty_ms"]) if age > current["peak_age_ms"]: current["peak_ms"], current["peak_age_ms"] = t, age previous = sample finish(reason="end_of_samples") return intervals def parse_stamps(lines): epochs = {} for line_number, line in enumerate(lines, 1): marker = "NET_DIAG_V1:" if marker not in line: continue row = json.loads(line.split(marker, 1)[1].strip()) required = ("seq", "broker_pid", "wall_ms", "mono_ms") if not isinstance(row, dict) or row.get("v") != 1 or not all(number(row.get(k)) for k in required): raise ValueError(f"line {line_number}: invalid diagnostic common fields") if not isinstance(row.get("run_id"), str) or not row["run_id"] or not isinstance(row.get("event"), str): raise ValueError(f"line {line_number}: missing diagnostic epoch/event") row = {**row, "line_number": line_number} epoch = epochs.setdefault(row["run_id"], {"run_id": row["run_id"], "records": [], "issues": [], "spans": [], "stacks": []}) epoch["records"].append(row) for epoch in epochs.values(): records = sorted(epoch["records"], key=lambda r: r["seq"]) epoch["records"] = records issues = epoch["issues"] sequences = [r["seq"] for r in records] if any(type(n) is not int for n in sequences) or sequences != list(range(len(records))): issues.append("sequence_gap_duplicate_or_missing_start") starts = [r for r in records if r["event"] == "NET_DIAG_START"] epoch["start"] = starts[0] if len(starts) == 1 else None start = epoch["start"] if not start or start["seq"] != 0: issues.append("missing_or_duplicate_start") if len({r["broker_pid"] for r in records}) != 1: issues.append("broker_changed_inside_epoch") epoch["broker_pid"] = records[0]["broker_pid"] epoch["wall_start_ms"] = min(r["wall_ms"] for r in records) epoch["wall_end_ms"] = max(r["wall_ms"] for r in records) offsets = [r["wall_ms"] - r["mono_ms"] for r in records] epoch["clock_offset_spread_ms"] = max(offsets) - min(offsets) if epoch["clock_offset_spread_ms"] > THRESHOLDS["clock_offset_spread_ms"]: issues.append("clock_alignment_unstable") tids = start.get("worker_tids", []) if start else [] if len(tids) != 2 or any(type(t) is not int or t <= 0 for t in tids) or len(set(tids)) != 2: issues.append("both_worker_identity_missing") epoch["worker_tids"] = tids if not start or set(start.get("capabilities", [])) != set(BASES) or start.get("canary_period_ms") != 25: issues.append("boundary_capabilities_or_original_canary_missing") epoch["heartbeats"] = [r for r in records if r["event"] in ("NET_DIAG_START", "NET_DIAG_HEARTBEAT", "NET_DIAG_END")] if any(r.get("dropped_records") != 0 for r in epoch["heartbeats"]): issues.append("dropped_or_unknown_record_count") pending, seen_ops = {}, set() for row in records: event = row["event"] if event == "NET_WORKERS_SNAPSHOT": epoch["stacks"].append(row) continue base_name, phase = event.rsplit("_", 1) if "_" in event else (None, None) if base_name not in BASES or phase not in ("BEGIN", "END"): if event not in ("NET_DIAG_START", "NET_DIAG_HEARTBEAT", "NET_DIAG_END", "NET_MEET_ROUND", "NET_FAMILY_GATE", "PAIR_MEET_UP"): issues.append("unknown_stamp_event_" + event) continue op = row.get("op_id") if (not isinstance(op, str) or not op or type(row.get("tid")) is not int or row["tid"] <= 0 or "parent_op_id" not in row or row.get("execution") not in ("sync", "async") or row.get("runtime") not in ("spt-broker-net", "blocking-pool", "other")): issues.append(f"invalid_boundary_fields_line_{row['line_number']}") continue if phase == "BEGIN": if op in seen_ops: issues.append("reused_operation_id_" + op) seen_ops.add(op) pending[op] = row else: begin = pending.pop(op, None) if not begin or begin["event"] != base_name + "_BEGIN": issues.append("unmatched_boundary_end_" + op) continue if any(begin.get(key) != row.get(key) for key in ("runtime", "execution", "parent_op_id", "subnet", "step", "stage")) or (begin["execution"] == "sync" and begin["tid"] != row["tid"]): issues.append("boundary_identity_changed_" + op) duration = row["mono_ms"] - begin["mono_ms"] if duration < 0 or row["wall_ms"] < begin["wall_ms"] or row.get("outcome") not in ("ok", "error", "cancelled"): issues.append("invalid_boundary_end_" + op) epoch["spans"].append({"run_id": epoch["run_id"], "base": base_name, "op_id": op, "begin": begin, "end": row, "duration_ms": duration}) for op, begin in pending.items(): epoch["spans"].append({"run_id": epoch["run_id"], "base": begin["event"].removesuffix("_BEGIN"), "op_id": op, "begin": begin, "end": None, "duration_ms": None}) issues.append("open_boundary_" + op) previous_processing_end = None for capture in sorted(epoch["stacks"], key=lambda r: r.get("capture_begin_mono_ms") if number(r.get("capture_begin_mono_ms")) else float("inf")): begin = capture.get("capture_begin_mono_ms") if (number(begin) and previous_processing_end is not None and begin < previous_processing_end): issues.append("capture_overlaps_previous_processing") if capture.get("capture_timing") == "raw-copy-resume-v1": end = capture.get("processing_end_mono_ms") if number(end): previous_processing_end = max(previous_processing_end or 0, end) epoch["issues"] = sorted(set(issues)) return epochs def interval_coverage(epoch, interval): issues = list(epoch["issues"]) lo = interval["start_ms"] - interval["uncertainty_ms"] hi = interval["end_ms"] + interval["uncertainty_ms"] beats = sorted(r["wall_ms"] for r in epoch["heartbeats"]) before = [t for t in beats if t <= lo] after = [t for t in beats if t >= hi] if not before or not after: issues.append("heartbeat_does_not_bracket_peak") else: window = [max(before)] + [t for t in beats if lo < t < hi] + [min(after)] if any(b-a > THRESHOLDS["cadence_gap_ms"] for a, b in zip(window, window[1:])): issues.append("heartbeat_gap_inside_peak") return sorted(set(issues)) def stack_evidence(epoch, interval): inner_lo = interval["start_ms"] + interval["uncertainty_ms"] inner_hi = interval["end_ms"] - interval["uncertainty_ms"] accepted, rejected = [], [] spans = {s["op_id"]: s for s in epoch["spans"]} for capture in epoch["stacks"]: begin, end = capture.get("capture_begin_mono_ms"), capture.get("capture_end_mono_ms") if not number(begin) or not number(end): rejected.append({"line": capture["line_number"], "reason": "missing_capture_times"}) continue offset = capture["wall_ms"] - capture["mono_ms"] wall_begin, wall_end = begin + offset, end + offset if wall_end < interval["start_ms"] or wall_begin > interval["end_ms"]: continue workers = capture.get("workers", []) reasons = [] timing = {"capture_timing": capture.get("capture_timing", "legacy-total-span"), "capture_span_ms": end-begin, "total_span_ms": end-begin, **{key: capture.get(key) for key in ("processing_begin_mono_ms", "processing_end_mono_ms", "processing_duration_ms", "total_duration_ms")}} if "capture_timing" in capture: if capture["capture_timing"] != "raw-copy-resume-v1": reasons.append("unknown_capture_timing") else: processing_begin = capture.get("processing_begin_mono_ms") processing_end = capture.get("processing_end_mono_ms") processing_duration = capture.get("processing_duration_ms") total_duration = capture.get("total_duration_ms") if not all(number(n) for n in (processing_begin, processing_end, processing_duration, total_duration)): reasons.append("missing_or_invalid_processing_times") else: timing["processing_span_ms"] = processing_end-processing_begin timing["total_span_ms"] = processing_end-begin if (processing_begin != end or processing_end < processing_begin or processing_duration != processing_end-processing_begin or total_duration != processing_end-begin): reasons.append("inconsistent_processing_times") if begin > end or end-begin > THRESHOLDS["capture_span_ms"] or wall_begin < inner_lo or wall_end > inner_hi: reasons.append("capture_outside_conservative_peak_or_too_slow") if not number(capture.get("trigger_canary_age_ms")) or capture["trigger_canary_age_ms"] < THRESHOLDS["stale_ms"]: reasons.append("capture_not_stale_triggered") if not isinstance(workers, list) or len(workers) != 2 or any(not isinstance(w, dict) for w in workers) or sorted(w.get("tid", -1) for w in workers) != sorted(epoch["worker_tids"]): reasons.append("both_workers_not_captured") workers = [] states, details = [], [] for worker in workers: frames = worker.get("frames") if worker.get("error") or not isinstance(frames, list) or not frames or any(not isinstance(f, str) or not f for f in frames): reasons.append("worker_unwind_failed") continue activity = worker.get("activity") classified = "unknown" if activity == "idle" and any(IDLE.search(frame) for frame in frames): classified = "idle" elif activity == "unrelated" and number(worker.get("progress_counter")): classified = "unrelated" elif activity == "boundary": span = spans.get(worker.get("op_id")) if span and span["end"] and span["base"] in SYMBOLS: b, e = span["begin"], span["end"] if (b["execution"] == "sync" and b["runtime"] == "spt-broker-net" and b["tid"] == worker["tid"] and b["mono_ms"] <= begin <= end <= e["mono_ms"] and span["duration_ms"] >= THRESHOLDS["long_sync_ms"] and any(SYMBOLS[span["base"]].search(frame) for frame in frames)): classified = "boundary" states.append(classified) details.append({"tid": worker["tid"], "classified": classified, "op_id": worker.get("op_id"), "progress_counter": worker.get("progress_counter"), "frames": frames}) if reasons: rejected.append({"line": capture["line_number"], "reason": reasons, "timing": timing}) else: accepted.append({"capture_id": capture.get("capture_id"), "line": capture["line_number"], "mono_ms": begin, "wall_ms": wall_begin, "states": states, "workers": details, "timing": timing}) accepted.sort(key=lambda r: r["mono_ms"]) outcomes = {"both_boundary": False, "both_idle": False, "both_unrelated_progress": False} for i, first in enumerate(accepted): for second in accepted[i+1:]: if second["mono_ms"] - first["mono_ms"] < THRESHOLDS["capture_separation_ms"]: continue if first["states"] == second["states"] == ["boundary", "boundary"]: outcomes["both_boundary"] = True if first["states"] == second["states"] == ["idle", "idle"]: outcomes["both_idle"] = True if first["states"] == second["states"] == ["unrelated", "unrelated"]: old = {w["tid"]: w["progress_counter"] for w in first["workers"]} if all(w["progress_counter"] > old[w["tid"]] for w in second["workers"]): outcomes["both_unrelated_progress"] = True return {**outcomes, "accepted": accepted, "rejected": rejected} def judge_interval(interval, epochs, rounds, pty): row = {**interval, "round_alignment": align_rounds(rounds, interval["start_ms"], interval["peak_ms"]), "verdict": "insufficient evidence", "mechanism_verdict": "INSUFFICIENT_EVIDENCE", "reasons": [], "boundaries": [], "stacks": None} row["family_gate_alignment"] = align_rounds( {"rounds": rounds["family_gates"]}, interval["start_ms"], interval["peak_ms"]) candidates = [e for e in epochs.values() if e["broker_pid"] == interval["broker_pid"] and e["wall_start_ms"] <= interval["start_ms"] and e["wall_end_ms"] >= interval["end_ms"]] if len(candidates) != 1: row["reasons"].append("missing_or_ambiguous_instrumented_epoch") else: epoch = candidates[0] row["run_id"] = epoch["run_id"] for key in ("round_alignment", "family_gate_alignment"): row[key] = [r for r in row[key] if not r["structured"] or r["run_id"] == epoch["run_id"]] issues = interval_coverage(epoch, interval) row["reasons"].extend(issues) lo, hi = interval["start_ms"] - interval["uncertainty_ms"], interval["end_ms"] + interval["uncertainty_ms"] for span in epoch["spans"]: start = span["begin"]["wall_ms"] end = span["end"]["wall_ms"] if span["end"] else epoch["wall_end_ms"] overlap = max(0, min(end, hi) - max(start, lo)) row["boundaries"].append({"base": span["base"], "op_id": span["op_id"], "begin_ms": start, "end_ms": span["end"]["wall_ms"] if span["end"] else None, "duration_ms": span["duration_ms"], "overlap_ms": overlap, "relation": "inside_peak" if overlap else "outside_peak", "runtime": span["begin"]["runtime"], "execution": span["begin"]["execution"], "tid": span["begin"]["tid"]}) stacks = stack_evidence(epoch, interval) row["stacks"] = stacks overlapping = [b for b in row["boundaries"] if b["overlap_ms"] > 0] if overlapping: row["verdict"] = "boundary inside peak" if not issues: positive = stacks["both_boundary"] negative = stacks["both_idle"] or stacks["both_unrelated_progress"] if positive and negative: row["reasons"].append("mixed_worker_states_inside_interval") elif positive: row.update(verdict="boundary inside peak", mechanism_verdict="CONFIRMED") elif stacks["both_idle"]: row.update(verdict="workers idle", mechanism_verdict="FALSIFIED") elif stacks["both_unrelated_progress"]: row.update(verdict="workers servicing unrelated work", mechanism_verdict="FALSIFIED") elif not overlapping: row.update(verdict="boundary outside peak", mechanism_verdict="FALSIFIED") else: row["reasons"].append("overlap_without_both_worker_synchronous_evidence") if not stacks["accepted"]: row["reasons"].append("no_complete_in_peak_both_worker_capture") observations = [] for p in pty: end = p["ack_ms"] if p["ack_ms"] is not None else p["wall_ms"] if p["sent_ms"] <= interval["end_ms"] and end >= interval["start_ms"]: observations.append({**p, "rtt_ms": end-p["sent_ms"] if p["ack_ms"] is not None else None, "timely": p["ack_ms"] is not None and p["ack_ms"]-p["sent_ms"] <= THRESHOLDS["pty_rtt_ms"] and p["state_confirmed"]}) row["pty"] = {"status": "observed" if observations else "not_observed", "observations": observations} row["dials_interpretation"] = "Counts started submit_dial tasks only; zero is not proof of no queued or uncounted work." return row def write_json(path, value): path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") def write_jsonl(path, rows): with path.open("w", encoding="utf-8") as stream: for row in rows: stream.write(json.dumps(row) + "\n") def analyze(samples_path, log_path, out, pty_path=None): raw_samples = load_jsonl(samples_path) if not raw_samples: raise ValueError("empty sample input") lines = log_path.read_text(encoding="utf-8-sig", errors="replace").splitlines() pty = load_jsonl(pty_path) if pty_path else [] for p in pty: if (not all(number(p.get(k)) for k in ("wall_ms", "sent_ms")) or p.get("route") not in ("local", "remote") or not isinstance(p.get("tag"), str) or type(p.get("state_confirmed")) is not bool or "ack_ms" not in p or (p["ack_ms"] is not None and (not number(p["ack_ms"]) or p["ack_ms"] < p["sent_ms"]))): raise ValueError("invalid PTY observation contract") samples = normalize_samples(raw_samples) intervals = stale_intervals(samples) rounds = parse_rounds(lines) epochs = parse_stamps(lines) verdicts = [judge_interval(i, epochs, rounds, pty) for i in intervals] usable = [s for s in samples if s["usable"]] gaps = [b["wall_ms"]-a["wall_ms"] for a, b in zip(usable, usable[1:])] duration = usable[-1]["wall_ms"]-usable[0]["wall_ms"] if usable else 0 coverage_reasons = [] if len(usable)/len(samples) < THRESHOLDS["usable_fraction"]: coverage_reasons.append("usable_samples_below_95_percent") if duration < THRESHOLDS["default_min_duration_ms"]: coverage_reasons.append("window_shorter_than_170_seconds") if not gaps or max(gaps) > THRESHOLDS["cadence_gap_ms"]: coverage_reasons.append("sampling_cadence_gap") if len({s["broker_pid"] for s in usable}) != 1: coverage_reasons.append("multiple_or_unknown_sample_brokers") if not verdicts: coverage_reasons.append("no_stale_episode_observed") if any(v["mechanism_verdict"] == "INSUFFICIENT_EVIDENCE" for v in verdicts): coverage_reasons.append("one_or_more_intervals_undiscriminated") if any(not v.get("stacks") or not v["stacks"]["accepted"] for v in verdicts): coverage_reasons.append("both_worker_capture_missing") counts = Counter(v["mechanism_verdict"] for v in verdicts) mechanism = ("MIXED" if counts["CONFIRMED"] and counts["FALSIFIED"] else "CONFIRMED" if counts["CONFIRMED"] else "FALSIFIED" if counts["FALSIFIED"] and not counts["INSUFFICIENT_EVIDENCE"] else "INSUFFICIENT_EVIDENCE") marked_captures = [c for e in epochs.values() for c in e["stacks"] if c.get("capture_timing") == "raw-copy-resume-v1"] processing_durations = [c["processing_duration_ms"] for c in marked_captures if number(c.get("processing_duration_ms"))] summary = {"schema": "spt-302-step1-v1", "coverage": "PASS" if not coverage_reasons else "INSUFFICIENT_EVIDENCE", "coverage_reasons": coverage_reasons, "mechanism": mechanism, "thresholds": THRESHOLDS, "sample_count": len(samples), "usable_samples": len(usable), "duration_ms": duration, "max_sample_gap_ms": max(gaps) if gaps else None, "stale_samples": sum(s["age_ms"] >= 1000 for s in usable), "severe_samples": sum(s["age_ms"] >= 2000 for s in usable), "max_canary_age_ms": max((s["age_ms"] for s in usable), default=None), "all_observed_dials_zero": bool(usable) and all(s["dials"] == 0 for s in usable), "stale_intervals": len(verdicts), "verdict_counts": dict(counts), "instrumented_epochs": len(epochs), "round_count": len(rounds["rounds"]), "family_gate_count": len(rounds["family_gates"]), "pty_input": str(pty_path) if pty_path else None, "processing_duration_ms": { "reported_count": len(processing_durations), "missing_or_invalid_count": len(marked_captures)-len(processing_durations), "min": min(processing_durations, default=None), "max": max(processing_durations, default=None), "sum": sum(processing_durations)}, "inputs": {str(p): hashlib.sha256(p.read_bytes()).hexdigest() for p in (samples_path, log_path, *([pty_path] if pty_path else []))}} out.mkdir(parents=True, exist_ok=False) write_jsonl(out/"samples.jsonl", samples) write_jsonl(out/"intervals.jsonl", verdicts) write_json(out/"rounds.json", rounds) write_json(out/"epochs.json", [{k: v for k, v in e.items() if k not in ("records", "spans", "stacks")} for e in epochs.values()]) write_jsonl(out/"boundaries.jsonl", [s for e in epochs.values() for s in e["spans"]]) write_jsonl(out/"stacks.jsonl", [s for e in epochs.values() for s in e["stacks"]]) write_json(out/"summary.json", summary) return summary, 0 if not coverage_reasons else 2 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--samples", type=Path, required=True) parser.add_argument("--log", type=Path, required=True) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--pty", type=Path) args = parser.parse_args() try: summary, status = analyze(args.samples, args.log, args.out, args.pty) print(json.dumps(summary)) return status except (OSError, ValueError, TypeError, KeyError) as error: print(json.dumps({"error": str(error)}), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())