diff --git a/cacheon/audit.py b/cacheon/audit.py index 521d8f7e..775d006b 100644 --- a/cacheon/audit.py +++ b/cacheon/audit.py @@ -37,9 +37,13 @@ from __future__ import annotations +import json import logging +import math import os import random +import sys +import time from typing import Callable, Optional import torch @@ -51,6 +55,9 @@ _state: dict = {"rate": None, "rng": None} _stats: dict[str, dict] = {} +_diagnostic_bytes = 0 +_diagnostic_limit_reported = False +_DIAGNOSTIC_LIMIT = 1 << 20 def _rate() -> float: @@ -120,6 +127,41 @@ def compare_error(slot: str) -> None: _receipt(slot) +def window_diagnostic(slot: str, node: str, position: int, window: int, + calls: list[dict], fraction: float, bar: float) -> None: + """Keep failed-window numerics in the existing request log, within 1 MiB per worker. + + The caller retains at most 4,096 rows per window. Counts always cover the + full comparison; omitted rows and an exhausted byte budget are explicit. + Diagnostics cannot change the audit verdict or mask its original failure. + """ + global _diagnostic_bytes, _diagnostic_limit_reported + if fraction >= bar or _diagnostic_limit_reported: + return + try: + def finite(value): + return str(value) if isinstance(value, float) and not math.isfinite(value) else value + + rows = [{key: [finite(x) for x in value] if isinstance(value, list) else finite(value) + for key, value in call.items()} for call in calls] + packet = {"time_unix_ns": time.time_ns(), **receipts.identity(), + "slot": slot, "node": node, "tensor_position": position, + "window": window, "passing_fraction": fraction, "minimum_fraction": bar, + "calls": rows} + packet["rows_omitted"] = sum(row["rows"] - len(row["row_errors"]) for row in rows) + prefix = "CACHEON-AUDIT-WINDOW: " + line = prefix + json.dumps(packet, separators=(",", ":"), allow_nan=False) + if _diagnostic_bytes + len(line.encode()) + 1 > _DIAGNOSTIC_LIMIT: + print("CACHEON-AUDIT-DIAGNOSTICS-LIMIT: 1 MiB reached; later row details omitted; " + "audit counts and verdict remain complete", file=sys.stderr, flush=True) + _diagnostic_limit_reported = True + return + _diagnostic_bytes += len(line.encode()) + 1 + print(line, file=sys.stderr, flush=True) + except Exception: # diagnostics must not turn a numerical result into infrastructure failure + logger.exception("cacheon: failed-window diagnostic could not be emitted") + + def capture_reference( slot: str, baseline_thunk: Callable, ) -> tuple[Optional[torch.Tensor], ...] | None: diff --git a/cacheon/chain/baseline_band.py b/cacheon/chain/baseline_band.py index 05426003..7d917541 100644 --- a/cacheon/chain/baseline_band.py +++ b/cacheon/chain/baseline_band.py @@ -91,7 +91,10 @@ def qualification_speed_from_payload( return None if not isinstance(witness, dict) or witness.get("goodput") is None: return None - return replay_measurements(witness) + speed = replay_measurements(witness) + if audit := result.get("audit_witness"): + speed["audit"] = {key: audit[key] for key in ("decision", "detail")} + return speed def replay_measurements(witness: dict[str, Any]) -> dict[str, Any]: diff --git a/cacheon/chain/bundle_result_monitor.py b/cacheon/chain/bundle_result_monitor.py new file mode 100644 index 00000000..befd981c --- /dev/null +++ b/cacheon/chain/bundle_result_monitor.py @@ -0,0 +1,86 @@ +"""Concise Discord results from the same read-only API the dashboard displays.""" + +from __future__ import annotations + +import json +from urllib.parse import quote, urlencode +from urllib.request import Request, urlopen + + +def _get(url: str) -> dict: + request = Request(url, headers={"User-Agent": "Cacheon/1.0"}) + with urlopen(request, timeout=15) as response: + return json.load(response) + + +def result_message(base: str, arena: dict, row: dict) -> str: + """Describe qualification separately from speed and reward eligibility.""" + verdict = row.get("decision") or row["status"].upper() + rid, hotkey = row["reservation_id"], row["hotkey"] + lines = [f"**Cacheon · {arena['label'][:80]} · {verdict}**", + f"Bundle `{rid[:12]}` · miner `{hotkey[:8]}…{hotkey[-8:]}`", + f"Target: {row['target_id'][:80]}"] + reason = row.get("reason") or row.get("invalid_reason") + if reason and reason != "qualified": + lines.append("Reason: " + reason.replace("_", " ").replace("\n", " ")[:180]) + result = row.get("result") + if result: + measured, required = result["speedup"], result["required_speedup"] + lines.append(f"Speed: {(measured - 1) * 100:+.2f}% · needed {(required - 1) * 100:+.2f}%" + f" · {result['passes']}/{result['pass_limit']} replay passes") + for key, label, unit, places in (("decode_tps", "Decode", "tok/s", 1), + ("ttft_s", "First token", "s", 2)): + values = result.get(key) or [] + if len(values) == 2 and all(v is not None for v in values): + lines.append(f"{label}: {values[0]:.{places}f} → {values[1]:.{places}f} {unit}") + audit = result.get("audit") + if audit: + lines.append(f"Audit {audit['decision']}: {audit['detail'][:280]}") + if verdict == "PASS": + reward = row.get("reward") or {} + eligible = reward.get("reward_eligible") + lines.append("Reward: " + ("eligible; on-chain activation is reported separately" + if eligible is True else "not currently reward-eligible" + if eligible is False else "pending settlement")) + url = base.rstrip("/") + "/" + quote(arena.get("slug") or arena["key"], safe="") + lines.append("[Details and logs](" + url + "?" + urlencode({"submission": rid}) + ")") + return "\n".join(lines) + + +def check_results(base: str, state: dict, *, notify, persist, get=_get) -> None: + """Announce new outcomes, checkpointing each delivery across ordinary restarts. + + First observation of an arena seeds its existing results without flooding + Discord. Unfinished submissions are not seeded: their later result is new. + A failed send leaves that result unacknowledged for the next poll. + """ + base = base.rstrip("/") + known = state.setdefault("bundle_results", {}) + for arena in get(base + "/api/arenas")["items"]: + items, offset = [], 0 + while True: + page = get(base + "/api/submissions?" + urlencode({ + "arena": arena["key"], "order": "asc", "limit": 1000, "offset": offset, + })) + items.extend(page["items"]) + offset += len(page["items"]) + if not page["items"] or offset >= page["total"]: + break + first = arena["key"] not in known + seen = known.setdefault(arena["key"], {}) + for row in items: + if (row.get("decision") not in {"PASS", "FAIL", "HOLD", "NO_DECISION"} + and row["status"] not in {"qualified", "failed", "expired", "held", "no_decision"}): + continue + reward = row.get("reward") or {} + outcome = [row["status"], row.get("decision"), row.get("reason"), + reward.get("reward_eligible")] + rid = row["reservation_id"] + if seen.get(rid) == outcome: + continue + if not first: + notify(result_message(base, arena, row)) + seen[rid] = outcome + if not first: + persist() + persist() diff --git a/cacheon/chain/weight_reveal_monitor.py b/cacheon/chain/weight_reveal_monitor.py index 3621f248..53a25532 100644 --- a/cacheon/chain/weight_reveal_monitor.py +++ b/cacheon/chain/weight_reveal_monitor.py @@ -1,4 +1,4 @@ -"""Read-only reveal monitoring in a separate process; Discord alerts never gate signing.""" +"""Read-only result and reveal notifications; Discord delivery never gates evaluation or signing.""" from __future__ import annotations @@ -12,6 +12,7 @@ from cacheon import chain from cacheon.chain.weight_share import CurrentWeightOffer +from cacheon.chain.bundle_result_monitor import check_results def send_discord(webhook: str, message: str) -> None: @@ -117,11 +118,20 @@ def check_reveals(subtensor, journal: Path, state: dict, *, deadline_blocks: int return _send_info(state, notify) +def _save_state(path: Path, state: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(state, sort_keys=True) + "\n") + temporary.replace(path) + + def main() -> None: + """Poll the existing dashboard and publication journal without controlling either.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--journal-db", type=Path, required=True) parser.add_argument("--state", type=Path, required=True) parser.add_argument("--webhook-file", type=Path, required=True) + parser.add_argument("--dashboard-url", help="Dashboard origin to watch for bundle results") parser.add_argument("--network", default="finney") parser.add_argument("--deadline-blocks", type=int, default=600) parser.add_argument("--interval", type=int, default=60) @@ -137,16 +147,20 @@ def main() -> None: client = None failures = 0 while True: + if args.dashboard_url: + try: + check_results(args.dashboard_url, state, + notify=lambda message: send_discord(webhook, message), + persist=lambda: _save_state(args.state, state)) + except Exception as exc: + print(f"bundle-result-monitor check failed: {type(exc).__name__}", flush=True) try: if client is None: client = chain.connect(args.network) state = check_reveals(client, args.journal_db, state, deadline_blocks=args.deadline_blocks, notify=lambda message: send_discord(webhook, message)) - args.state.parent.mkdir(parents=True, exist_ok=True) - temporary = args.state.with_suffix(".tmp") - temporary.write_text(json.dumps(state, sort_keys=True) + "\n") - temporary.replace(args.state) + _save_state(args.state, state) print(f"weight-reveal-monitor block={state.get('checked_block')} " f"pending={len(state.get('pending', []))} alerted={bool(state.get('alerted'))}", flush=True) failures = 0 diff --git a/cacheon/eval/oci_process.py b/cacheon/eval/oci_process.py index 229539ba..3dfe13a4 100644 --- a/cacheon/eval/oci_process.py +++ b/cacheon/eval/oci_process.py @@ -517,23 +517,6 @@ def snapshot(self) -> OCIAttachedDiagnostic: self._artifact, ) - def discard_artifact(self) -> None: - with self._lock: - artifact = self._artifact - writer = self._artifact_writer - if artifact is None: - return - if writer is None: - raise OCIProcessError("stderr artifact lacks its manager authority") - writer.manager.reopen_stderr_artifact(artifact) - artifact.receipt_path.unlink(missing_ok=True) - OCIProcessManager._fsync_directory(artifact.artifact_path.parent) - artifact.artifact_path.unlink(missing_ok=True) - OCIProcessManager._fsync_directory(artifact.artifact_path.parent) - with self._lock: - if self._artifact == artifact: - self._artifact = None - @dataclass(frozen=True) class OCIQuiescenceReceipt: @@ -1507,7 +1490,7 @@ def run( process = subprocess.Popen( list(argv), stdin=subprocess.PIPE, - stdout=subprocess.DEVNULL, + stdout=stderr_write_fd, # Keep Docker stderr off ``communicate`` so the dedicated drain # can hash every byte while retaining only a bounded memory tail # and bounded private disk prefix. @@ -1565,8 +1548,6 @@ def run( diagnostic, ) from None self.force_remove_container(lease) - if int(process.returncode) == 0: - stderr_capture.discard_artifact() diagnostic = stderr_capture.snapshot() diagnostic = OCIAttachedDiagnostic( diagnostic.stderr_tail, diff --git a/cacheon/integrations/sglang_nodes.py b/cacheon/integrations/sglang_nodes.py index e68af964..9053e285 100644 --- a/cacheon/integrations/sglang_nodes.py +++ b/cacheon/integrations/sglang_nodes.py @@ -85,10 +85,11 @@ # The wire vocabulary (oci_session_protocol.AuditReceiptFacts) is closed: a share of # rows within tolerance against a bar is its matched_ratio. _MODE = "matched_ratio" -# Per bound node and graded-tensor position: recent twin noise, and [passed, seen] -# rows. Per node, not per address: layers under one `*` differ severalfold in noise. +# Per node, not per address: layers under one `*` differ severalfold in noise. +# Diagnostic rows travel with pooled counts, including earlier calls in a window. _noise: dict[tuple[int, int], deque] = {} -_pooled: dict[tuple[int, int], list[int]] = {} +_pooled: dict[tuple[int, int], list] = {} +_DIAGNOSTIC_ROWS = 4096 def node_pattern(slot: str) -> re.Pattern[str]: @@ -429,7 +430,8 @@ def _row_errors(actual: torch.Tensor, expected: torch.Tensor, dim: int, return errors[torch.isfinite(e).all(dim=1)] -def _grade(slot: str, node: int, actual: list, twin: list) -> tuple[float, int] | None: +def _grade(slot: str, node: int, actual: list, twin: list, + *, node_name: str = "") -> tuple[float, int] | None: """Pool each graded tensor's passing rows; record a unit when a window fills.""" filled = [] @@ -441,14 +443,25 @@ def _grade(slot: str, node: int, actual: list, twin: list) -> tuple[float, int] noise.append(torch.quantile(honest, 0.9).item()) # History stabilizes one-row calls, but cannot discard this call's reference # noise: quiet decode history falsely rejected GLM's 4096-row prefill calls. - measured = _TWIN_FACTOR * max(noise[-1], sorted(noise)[int(0.9 * (len(noise) - 1))]) + history_p90 = sorted(noise)[int(0.9 * (len(noise) - 1))] + measured = _TWIN_FACTOR * max(noise[-1], history_p90) tolerance = min(_CEILING, max(_FLOOR, measured)) - pooled = _pooled.setdefault(key, [0, 0]) - pooled[0] += int((errors <= tolerance).sum().item()) + pooled = _pooled.setdefault(key, [0, 0, [], 1]) + kept = max(0, _DIAGNOSTIC_ROWS - pooled[1]) + passed = int((errors <= tolerance).sum().item()) + pooled[2].append({ + "rows": errors.numel(), "passed": passed, "tolerance": tolerance, + "reference_p90": noise[-1], "history_p90": history_p90, + "row_errors": errors[:kept].tolist(), "reference_row_errors": honest[:kept].tolist(), + }) + pooled[0] += passed pooled[1] += errors.numel() if pooled[1] >= _WINDOW: - filled.append((pooled[0] / pooled[1], position)) - pooled[:] = [0, 0] + fraction = pooled[0] / pooled[1] + _audit.window_diagnostic(slot, node_name or slot, position, pooled[3], + pooled[2], fraction, _ROW_BAR) + filled.append((fraction, position)) + pooled[:] = [0, 0, [], pooled[3] + 1] if filled: worst = min(filled) _audit.record_fraction(slot, worst[0], _ROW_BAR, _MODE) @@ -554,7 +567,7 @@ def dispatched(*args, **kwargs): [(t, 0) for t in _tensors(result, [], fields=True)], _state_rows(runner, (*args, *kwargs.values()), layer), expected, module=module, ) - grade = _grade(slot, id(module), actual, twin) + grade = _grade(slot, id(module), actual, twin, node_name=node_name or slot) if grade is not None and grade[0] < _ROW_BAR and not reported: logging.getLogger(__name__).error( "node audit mismatch: node=%s tensor_position=%d passing_fraction=%.6f", @@ -563,6 +576,8 @@ def dispatched(*args, **kwargs): reported = True except ValueError: _audit.compare_error(slot) + logging.getLogger(__name__).exception( + "node audit comparison failed: node=%s", node_name or slot) _receipts.completed(slot) return result diff --git a/dashboard/winners.py b/dashboard/winners.py index c0dcf91e..f82c5f78 100644 --- a/dashboard/winners.py +++ b/dashboard/winners.py @@ -93,6 +93,7 @@ def mean(role: str, field: str) -> float | None: values = [lane[field] for lane in speed["lanes"] if lane["role"] == role and lane[field] is not None] return sum(values) / len(values) if values else None return {"speedup": speed["speedup"], "required_speedup": grade["required_speedup"], + "audit": speed.get("audit"), "passes": speed["windows"], "pass_limit": speed["window_limit"], "detail": grade["detail"], "decode_tps": [mean("B", "decode_tps"), mean("C", "decode_tps")], "ttft_s": [mean("B", "mean_ttft_s"), mean("C", "mean_ttft_s")]} diff --git a/docs/security/evidence.md b/docs/security/evidence.md index eaeb6229..3ca20f83 100644 --- a/docs/security/evidence.md +++ b/docs/security/evidence.md @@ -86,6 +86,28 @@ Not every artifact is public. Selection secrets and hidden tasks remain in priva validator storage; serializable authority records contain references and commitments, not the secret bytes. +Failed node-audit windows also emit `CACHEON-AUDIT-WINDOW` diagnostics into the +existing retained worker log. They identify the node, tensor position, rank, +window and timestamp, and retain each contributing call's counts, numerical +tolerance, current/reference-history noise measurements, and candidate/reference +row errors. Earlier calls pooled into the same failed window are included. +The unchanged audit receipt remains the verdict authority. + +Diagnostic row arrays are limited to 4,096 rows per window and their combined +output to 1 MiB per worker. `rows_omitted` and +`CACHEON-AUDIT-DIAGNOSTICS-LIMIT` identify incomplete numerical detail explicitly; +they do not change the comparison counts or verdict. These records explain the +comparison without retaining full model tensors or a GPU-memory snapshot. +They cannot recover measurements from older runs that never emitted them. + +One-shot build stdout and stderr use the same retained diagnostic stream, even +when the build succeeds. Runtime streams remain attached to the request's +downloadable worker log after container teardown. The stream receipt states +whether its bounded artifact is truncated; operators must retain VM/service +logs separately on durable storage and inspect that flag before claiming a +complete log. Deployment paths and log-rotation configuration belong to the +operator's VM configuration. + Pristine-reference requests can carry text or canonical input token IDs. The `ORQ3` frame carries the IDs directly, so the reference engine scores the same chat-template input used by a replay without decoding and tokenizing it again. Its response must diff --git a/docs/validator-guide/settlement-and-weights.md b/docs/validator-guide/settlement-and-weights.md index 865a59db..612c28af 100644 --- a/docs/validator-guide/settlement-and-weights.md +++ b/docs/validator-guide/settlement-and-weights.md @@ -585,6 +585,7 @@ python -m cacheon.chain.weight_reveal_monitor \ --journal-db /absolute/path/follow.sqlite3 \ --state /absolute/path/reveal-monitor.json \ --webhook-file /absolute/path/discord-webhook.txt \ + --dashboard-url https://dash.cacheon.ai \ --network finney --deadline-blocks 600 --interval 60 ``` @@ -603,6 +604,21 @@ seed its initial inventory; subsequent share changes and ordinary restarts do not repeat the new-hotkey announcement. Undelivered info messages remain queued in the monitor's own state. Activation describes this validator's weight row, not network-wide emissions. +With `--dashboard-url`, the same monitor also watches every arena advertised by +that dashboard. Each new bundle result posts its verdict and reason, measured +and required speed gain, replay count, decode/first-token measurements when +available, audit details, reward eligibility, and an arena-qualified link to +the submission and its downloadable logs. It does not post raw stdout. +Qualification PASS, reward eligibility, and observed on-chain activation remain +separate statements. + +The first successful observation seeds each arena's existing results without +posting historical messages; unfinished submissions are announced when they +finish. A changed result or reward-eligibility decision is announced again. +Every successful delivery checkpoints the monitor's state. Failed sends retry +on the next poll, and a chain outage does not prevent result checks. A crash +after Discord accepts a message but before its checkpoint can repeat that +message; deleting the state seeds a new starting inventory. The legacy combined `set-weights` reconciler described above remains separate. When `serve-weights` is deliberately run without push credentials, raw local diff --git a/scripts/file_ceilings.txt b/scripts/file_ceilings.txt index 07f0baa0..f966e4ee 100644 --- a/scripts/file_ceilings.txt +++ b/scripts/file_ceilings.txt @@ -19,7 +19,7 @@ cacheon/eval/device_state.py 1461 cacheon/eval/native_artifact.py 1220 cacheon/eval/oci_backend.py 1716 cacheon/eval/oci_outer_session.py 907 -cacheon/eval/oci_process.py 1783 +cacheon/eval/oci_process.py 1764 cacheon/eval/oci_session_protocol.py 1239 cacheon/eval/oci_session_worker.py 939 cacheon/eval/qualification.py 977 diff --git a/scripts/loc_baseline.txt b/scripts/loc_baseline.txt index 1cb851c4..c236d450 100644 --- a/scripts/loc_baseline.txt +++ b/scripts/loc_baseline.txt @@ -4,10 +4,10 @@ (root) 1158 .github 345 bundles 2398 -cacheon 90047 +cacheon 90188 cacheon_kernels 936 -dashboard 4269 -docs 12414 +dashboard 4270 +docs 12452 examples 378 scripts 1893 -tests 65765 +tests 65931 diff --git a/scripts/surface_baseline/modules.json b/scripts/surface_baseline/modules.json index caccb151..6975c2e9 100644 --- a/scripts/surface_baseline/modules.json +++ b/scripts/surface_baseline/modules.json @@ -3,7 +3,7 @@ "cacheon._strict": {"all":null,"exceptions":[],"public":["NODE_ADDRESS","duplicate_key_pairs","members_overlap","require_digest","require_driver_integer","require_exact_fields","require_identifier","require_int","require_node_members","truthy_flag"]}, "cacheon.arena_allocation": {"all":null,"exceptions":[],"public":["ArenaAllocation","PPM","allocate_submission_weights","arena_base_credits","normalize_weights"]}, "cacheon.arena_service": {"all":["ArenaCandidateBinding","ArenaCapacityPolicy","ArenaQualificationRequest","ArenaQualificationWork","ArenaRuntimeIdentity","ArenaService","ArenaServiceError","ArenaServiceManifest","ArenaServiceProvider","Workload","WorkloadCell"],"exceptions":["ArenaServiceError"]}, - "cacheon.audit": {"all":null,"exceptions":[],"public":["baseline_refused","capture_reference","compare_error","record_fraction","sampled"]}, + "cacheon.audit": {"all":null,"exceptions":[],"public":["baseline_refused","capture_reference","compare_error","record_fraction","sampled","window_diagnostic"]}, "cacheon.audit_gate": {"all":null,"exceptions":[],"public":["gate","grade","infrastructure_failure"]}, "cacheon.bootstrap": {"all":null,"exceptions":[],"public":["install"]}, "cacheon.bundle_hash": {"all":null,"exceptions":[],"public":["CARRIER_RECEIPT_NAME","content_hash"]}, @@ -17,6 +17,7 @@ "cacheon.chain.baseline_band": {"all":["qualification_evidence_roots","qualification_speed","replay_measurements"],"exceptions":[]}, "cacheon.chain.baseline_segments": {"all":["bind_reservation_baseline_segment","bind_unbound_queue_to_stack","commission_boundary","qualification_queue_baseline","reservation_baseline_segment"],"exceptions":[]}, "cacheon.chain.bundle_privacy": {"all":null,"exceptions":[],"public":["KEY_ENV","MAGIC","decrypt_archive","encrypt_archive","recipient_key","validator_key"]}, + "cacheon.chain.bundle_result_monitor": {"all":null,"exceptions":[],"public":["check_results","result_message"]}, "cacheon.chain.duplicate_replay": {"all":null,"exceptions":[],"public":["PriorVerdict","REPLAYABLE_DECISIONS","ReplayDecision","decide_replay","replay_reason"]}, "cacheon.chain.eval_cost": {"all":null,"exceptions":["EvalCostCommitError","EvalCostError","EvalCostFetchError"],"public":["DEFAULT_EVAL_COST_PAYMENT_WINDOW_BLOCKS","DEFAULT_EVAL_COST_QUOTE_TTL_BLOCKS","EVAL_COST_ASSET","EVAL_COST_INSTRUMENT","EVAL_COST_PAYMENT_DOMAIN","EVAL_COST_QUOTE_VERSION","EvalCostCommitError","EvalCostError","EvalCostFetchError","EvalCostPaymentProof","EvalCostPolicy","EvalCostQuote","EvalCostRequest","PUBLISHED_EVAL_COST_TAO_RAO","REASON_INVALID","REASON_MISSING","REASON_QUOTE_EXPIRED","REASON_USED","REASON_WINDOW","decode_payment_remark","encode_payment_remark","proof_from_decoded_extrinsic","quote_eval_cost","unused_eval_cost_retry_flags","verify_eval_cost_payment"]}, "cacheon.chain.eval_cost_credit": {"all":["EVAL_COST_CREDITS_DDL","EvalCostCredit","grant_eval_cost_credit","list_eval_cost_credits"],"exceptions":[]}, diff --git a/tests/test_audit_diagnostics.py b/tests/test_audit_diagnostics.py new file mode 100644 index 00000000..0a14178f --- /dev/null +++ b/tests/test_audit_diagnostics.py @@ -0,0 +1,56 @@ +"""A failed pooled comparison remains explainable after its worker exits.""" + +import json + +import pytest +import torch + +from cacheon import audit +from cacheon.integrations import sglang_nodes as nodes + + +@pytest.fixture(autouse=True) +def clean_audit(monkeypatch): + for module, name, value in ((nodes, "_noise", {}), (nodes, "_pooled", {}), + (audit, "_stats", {}), (audit, "_diagnostic_bytes", 0), + (audit, "_diagnostic_limit_reported", False)): + monkeypatch.setattr(module, name, value) + monkeypatch.delenv("CACHEON_SEAM_RECEIPT_DIR", raising=False) + + +def test_failed_window_keeps_prior_call_errors_and_their_own_tolerances(capsys): + # The October 11 failure was 183/256, pooled across calls. The diagnostic + # must explain all 256 rows, not just the final call that completed it. + for passed in (93, 90): + errors = torch.cat((torch.zeros(passed), torch.ones(128 - passed))) + nodes._grade("model", 1, [errors], [torch.zeros(128)], node_name="model") + line = next(x for x in capsys.readouterr().err.splitlines() + if x.startswith("CACHEON-AUDIT-WINDOW: ")) + event = json.loads(line.partition(": ")[2]) + calls = event["calls"] + replayed = sum(sum(error <= call["tolerance"] for error in call["row_errors"]) + for call in calls) / sum(call["rows"] for call in calls) + assert replayed == event["passing_fraction"] == 183 / 256 + assert event["rows_omitted"] == 0 and event["node"] == "model" + assert len(calls) == 2 and all(len(call["reference_row_errors"]) == 128 for call in calls) + assert all(call["reference_p90"] == call["history_p90"] == 0 for call in calls) + assert audit._stats["model"]["violations"] == 1 + + +def test_diagnostics_bound_rows_and_bytes_without_changing_the_verdict(capsys, monkeypatch): + nodes._grade("model", 1, [torch.ones(8192)], [torch.zeros(8192)]) + event = json.loads(capsys.readouterr().err.partition(": ")[2]) + assert event["rows_omitted"] == 4096 and event["calls"][0]["rows"] == 8192 + assert event["passing_fraction"] == 0 + monkeypatch.setattr(audit, "_DIAGNOSTIC_LIMIT", 1) + for _ in range(2): + nodes._grade("model", 1, [torch.ones(256)], [torch.zeros(256)]) + assert capsys.readouterr().err.count("CACHEON-AUDIT-DIAGNOSTICS-LIMIT:") == 1 + assert (audit._stats["model"]["n"], audit._stats["model"]["violations"]) == (3, 3) + + +def test_nonfinite_rows_remain_explicit_in_the_diagnostic(capsys): + nodes._grade("model", 1, [torch.full((256,), float("nan"))], [torch.zeros(256)]) + event = json.loads(capsys.readouterr().err.partition(": ")[2]) + assert event["calls"][0]["row_errors"] == ["nan"] * 256 + assert event["passing_fraction"] == 0 and audit._stats["model"]["compare_errors"] == 0 diff --git a/tests/test_bundle_result_monitor.py b/tests/test_bundle_result_monitor.py new file mode 100644 index 00000000..a5fef4f4 --- /dev/null +++ b/tests/test_bundle_result_monitor.py @@ -0,0 +1,101 @@ +"""Result delivery survives restart/outage and uses arena-scoped dashboard links.""" + +import json +from urllib.parse import parse_qs, urlparse + +import pytest + +from cacheon.chain.bundle_result_monitor import check_results, result_message + + +ARENAS = [{"key": name, "slug": name + "-model", "label": name.title()} + for name in ("first", "second")] + + +def submission(rid, *, status="qualified", decision="PASS", reason="qualified"): + return {"reservation_id": rid * 64, "hotkey": "miner-" + rid * 32, + "target_id": "forward_pass", "status": status, "decision": decision, + "reason": reason, "reward": {"reward_eligible": decision == "PASS"}, + "result": {"speedup": 1.2799, "required_speedup": 1.0428, "passes": 3, + "pass_limit": 4, "decode_tps": [158.3, 197.3], "ttft_s": [3.46, 2.69]}} + + +def api(rows): + def get(url): + if url.endswith("/api/arenas"): + return {"items": ARENAS} + query = parse_qs(urlparse(url).query) + items = rows[query["arena"][0]] + offset = int(query["offset"][0]) + return {"total": len(items), "items": items[offset:offset + 2]} + return get + + +def test_all_arenas_paginate_seed_history_and_deliver_new_results_once(): + rows = {a["key"]: [submission("a"), submission("b", status="qualifying", decision="", reason="")] + for a in ARENAS} + state, messages, saved = {}, [], [] + persist = lambda: saved.append(json.loads(json.dumps(state))) + check_results("https://dash.example", state, notify=messages.append, persist=persist, get=api(rows)) + assert messages == [] + rows["first"][1] = submission("b", status="failed", decision="FAIL", reason="slot_audit_failed") + rows["first"].append(submission("c", status="held", decision="HOLD", reason="worker_unavailable")) + rows["second"].append(submission("d")) + check_results("https://dash.example", state, notify=messages.append, persist=persist, get=api(rows)) + assert len(messages) == 3 and "FAIL" in messages[0] and "HOLD" in messages[1] + assert "/first-model?submission=" in messages[0] and "/second-model?submission=" in messages[2] + state = saved[-1] + check_results("https://dash.example", state, notify=messages.append, persist=persist, get=api(rows)) + assert len(messages) == 3 + + +def test_delivery_failure_retries_only_the_unacknowledged_result(): + state, messages, saved = {"bundle_results": {a["key"]: {} for a in ARENAS}}, [], [] + rows = {"first": [submission("a"), submission("b")], "second": []} + + def flaky(message): + if messages: + raise OSError("Discord unavailable") + messages.append(message) + + with pytest.raises(OSError): + check_results("https://dash.example", state, notify=flaky, + persist=lambda: saved.append(json.loads(json.dumps(state))), get=api(rows)) + state = saved[-1] + check_results("https://dash.example", state, notify=messages.append, persist=lambda: None, get=api(rows)) + assert len(messages) == 2 and "`aaaaaaaaaaaa`" in messages[0] and "`bbbbbbbbbbbb`" in messages[1] + + +def test_message_explains_audit_failure_without_calling_speed_a_qualification(): + row = submission("f", status="failed", decision="FAIL", reason="slot_audit_failed") + row["result"]["audit"] = {"decision": "FAIL", "detail": "872 compared units, 2 violations; worst 0.7148"} + message = result_message("https://dash.example/", ARENAS[0], row) + assert "FAIL" in message and "+27.99%" in message and "needed +4.28%" in message + assert "3/4 replay passes" in message and "872 compared units" in message + assert "158.3 → 197.3 tok/s" in message and "3.46 → 2.69 s" in message + assert "Reward:" not in message and len(message) < 1200 + row["reason"] = "unexpected output\n" * 500 + message = result_message("https://dash.example", ARENAS[0], row) + assert len(message) < 1200 and message.endswith(row["reservation_id"] + ")") + + +def test_main_checks_results_and_saves_them_even_when_chain_is_unavailable(tmp_path, monkeypatch): + from cacheon.chain import weight_reveal_monitor as monitor + + state_file, webhook = tmp_path / "state.json", tmp_path / "webhook" + webhook.write_text("https://discord.com/api/webhooks/test") + monkeypatch.setattr("sys.argv", ["monitor", "--state", str(state_file), "--webhook-file", str(webhook), + "--journal-db", str(tmp_path / "journal"), + "--dashboard-url", "https://dash.example"]) + + def checked(base, state, *, notify, persist): + assert base == "https://dash.example" + state["bundle_results"] = {"first": {"delivered": ["FAIL"]}} + persist() + + monkeypatch.setattr(monitor, "check_results", checked) + monkeypatch.setattr(monitor.chain, "connect", lambda _: (_ for _ in ()).throw(OSError("offline"))) + monkeypatch.setattr(monitor.time, "sleep", lambda _: (_ for _ in ()).throw(KeyboardInterrupt())) + with pytest.raises(KeyboardInterrupt): + monitor.main() + assert json.loads(state_file.read_text())["bundle_results"]["first"]["delivered"] == ["FAIL"] diff --git a/tests/test_dashboard_replay.py b/tests/test_dashboard_replay.py index 1416c3fb..162c4705 100644 --- a/tests/test_dashboard_replay.py +++ b/tests/test_dashboard_replay.py @@ -60,6 +60,13 @@ def test_corrupt_replay_witness_reports_its_original_error(): assert "speedup" not in speed +def test_result_includes_the_retained_audit_reason_beside_passing_speed(): + audit = {"decision": "FAIL", "detail": "872 audited calls, 2 violations, worst_frac=0.7148"} + payload = {"speed_witness": _witness(statistical=True).to_dict(), "audit_witness": audit} + speed = qualification_speed_from_payload(json.dumps(payload).encode()) + assert result_summary(speed)["audit"] == audit + + def test_replay_measurements_reach_submission_and_winner_apis(tmp_path, client, monkeypatch): # noqa: F811 from cacheon.eval.evidence_store import prepare_evidence_root, publish_canonical_json_evidence diff --git a/tests/test_oci_process.py b/tests/test_oci_process.py index 254912bf..3d63170d 100644 --- a/tests/test_oci_process.py +++ b/tests/test_oci_process.py @@ -255,8 +255,9 @@ def popen(argv, **kwargs): assert lease.record_path.exists(), "caller retains staged state until publication" -def test_run_continuously_drains_and_retains_only_bounded_stderr_tail( - tmp_path: Path, monkeypatch +@pytest.mark.parametrize("exit_code", [0, 7]) +def test_run_continuously_drains_and_retains_both_streams_after_any_exit( + tmp_path: Path, monkeypatch, exit_code ) -> None: manager = _manager(tmp_path) lease = manager.register(lease_id="lease-1", container_name="container-1") @@ -266,7 +267,8 @@ def test_run_continuously_drains_and_retains_only_bounded_stderr_tail( "import sys; " f"sys.stderr.buffer.write(b'A' * {ATTACHED_STDERR_MAX_BYTES * 4}); " f"sys.stderr.buffer.write({marker!r}); sys.stderr.buffer.flush(); " - "raise SystemExit(7)" + "sys.stdout.buffer.write(b'BUILD-STDOUT'); sys.stdout.buffer.flush(); " + f"raise SystemExit({exit_code})" ) def spawn(_argv, **kwargs): @@ -279,14 +281,14 @@ def spawn(_argv, **kwargs): timeout_s=5.0, ) diagnostic = result.stderr_diagnostic - assert result.returncode == 7 and diagnostic is not None + assert result.returncode == exit_code and diagnostic is not None assert len(diagnostic.stderr_tail) == ATTACHED_STDERR_MAX_BYTES assert diagnostic.stderr_truncated and diagnostic.capture_complete - assert diagnostic.client_returncode == 7 - assert diagnostic.stderr_tail.endswith(marker) + assert diagnostic.client_returncode == exit_code + assert diagnostic.stderr_tail.endswith(marker + b"BUILD-STDOUT") artifact = diagnostic.artifact assert type(artifact) is OCIStderrArtifactReceipt - expected = b"A" * (ATTACHED_STDERR_MAX_BYTES * 4) + marker + expected = b"A" * (ATTACHED_STDERR_MAX_BYTES * 4) + marker + b"BUILD-STDOUT" assert diagnostic.stream_bytes == len(expected) assert diagnostic.stream_sha256 == hashlib.sha256(expected).hexdigest() assert artifact.stream_bytes == len(expected)