freedom-intelligence/scripts/verify_brief_durability.py
tegwick a78187c33e Make the sensing loop durable and pin the reserve to Scaleway
The weekday brief only counts when the file is on origin/main. Hub
events without git objects are failures. The open-weight SoT moves
from the VAULT 850 GiB quota to a dedicated Scaleway bucket (One Zone
IA for R, Glacier for S) so V4-Flash and K3 are economically in-scope.

Catalogs the real DeepSeek-V4-Flash-0731 (MIT ~167 GiB MoE, not 12B)
and nominates Qwen3.8-27B. Bucket create remains FI-WP-0004-T08.

Assistant: grok
Assistant-Session: 01a09c6a-1cfc-75b1-a78d-c13eaf22241d
2026-09-13 22:45:43 +02:00

114 lines
3.6 KiB
Python
Executable file

#!/usr/bin/env python3
"""Fail if State Hub fi_daily_brief events are not on origin/main.
A brief day is complete only when briefs/YYYY/MM/YYYY-MM-DD.md exists in
this git repository (the clone should be origin/main). Hub events that
claim wrote/committed without a matching file are the 2026-08/09 failure
mode: local commit on railiance, due-bit cleared, durable memory empty.
Usage:
python3 scripts/verify_brief_durability.py
STATE_HUB_URL=http://127.0.0.1:8000 python3 scripts/verify_brief_durability.py
Exit 0 = every hub success event has a brief file.
Exit 1 = mismatch (or hub unreachable when --require-hub is set).
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def brief_path(date: str) -> Path:
year, month, _ = date.split("-")
return ROOT / "briefs" / year / month / f"{date}.md"
def load_events(url: str, timeout: float) -> list[dict]:
req = urllib.request.Request(
f"{url.rstrip('/')}/progress/?event_type=fi_daily_brief",
headers={"Accept": "application/json"},
)
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode())
if isinstance(data, list):
return data
if isinstance(data, dict):
for key in ("items", "events", "results"):
if isinstance(data.get(key), list):
return data[key]
raise RuntimeError(f"unexpected hub payload type {type(data)}")
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument(
"--hub",
default=os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000"),
)
ap.add_argument("--timeout", type=float, default=8.0)
ap.add_argument(
"--require-hub",
action="store_true",
help="exit 1 if the hub cannot be reached (default: warn and check git only)",
)
args = ap.parse_args()
git_dates = sorted(
p.stem
for p in (ROOT / "briefs").rglob("20*.md")
if p.name[0].isdigit()
)
print(f"git briefs: {len(git_dates)} last={git_dates[-1] if git_dates else '-'}")
try:
events = load_events(args.hub, args.timeout)
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, RuntimeError) as exc:
msg = f"hub unreachable or unreadable at {args.hub}: {exc}"
if args.require_hub:
print(f"FAIL {msg}", file=sys.stderr)
return 1
print(f"WARN {msg}")
return 0
missing: list[str] = []
present: list[str] = []
for ev in events:
detail = ev.get("detail") or {}
date = str(detail.get("date") or "")
if not date:
continue
repo = detail.get("repo")
if repo and repo != "freedom-intelligence":
continue
path = brief_path(date)
if path.is_file():
present.append(date)
else:
missing.append(date)
missing = sorted(set(missing))
present = sorted(set(present))
print(f"hub fi_daily_brief dates on git: {len(present)}")
if missing:
print("FAIL hub events with no brief file:")
for date in missing:
print(f" {date} expected {brief_path(date).relative_to(ROOT)}")
print(
"A local-only commit must not clear due. "
"Push to origin, or treat the hub event as a false completion."
)
return 1
print("ok: every hub fi_daily_brief date has a brief file")
return 0
if __name__ == "__main__":
sys.exit(main())