Retain audit diagnostics and post bundle results to Discord

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:

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]:

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()

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

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,

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
 

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")]}

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