sei sorveglianti registrano i giri e stanno in monitor_health (Fixes #16)

src/live/giri.registra_giro: una riga jsonl {ts ms, campi} dopo una misura completata,
mai un'eccezione. edge_watch, fee_watch, scale_watch, vrp_f_watch, venue_news (cron_daily)
e venue_watch (cron_book, orario) scrivono data/<dir>/runs.jsonl; niente riga senza misura.
monitor_health: sei MonitorSpec (30h i giornalieri, 3h venue_watch), open_labeled=False.
Stessa riparazione di cc01 (#10, #12), generalizzata. Primo giro notturno: NUOVO, non allerta.
Test: incrocio P1 percorso/RUNS, misura -> una riga, non-misura -> nessuna. Suite: 1098 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.gitignore

@@ -81,6 +81,10 @@ data/chain_collect/
 data/vrp_f_watch/
 # giri misurati del trigger di regime CC01 (letti da monitor_health, #12)
 data/cc01_regime/
+# giri misurati dei sorveglianti generalizzati da cc01 (#16), letti da monitor_health
+data/edge_watch/
+data/scale_watch/
+data/venue_news/
 
 # log esecuzioni del book live (stato runtime, contiene fill/fee del conto reale)
 data/live/

scripts/live/edge_watch.py

@@ -42,10 +42,13 @@ import pandas as pd
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
+from src.live.giri import registra_giro                  # noqa: E402
 from src.live.notifier import notify                     # noqa: E402
 from src.portfolio.portfolio import combine_outer        # noqa: E402
 from src.portfolio.sleeves import deribit_book_sleeves, tp01_sleeve  # noqa: E402
 
+RUNS = ROOT / "data" / "edge_watch" / "runs.jsonl"
+
 # --- soglie CONGELATE (cambiarle invalida la taratura di r0726_edge_death.py) ---
 WIN_MONTHS = 36
 SHARPE_KILL = -0.5
@@ -119,6 +122,7 @@ def run() -> dict:
 
 def main() -> int:
     r = run()
+    registra_giro(RUNS, sharpe=r["sharpe"], n_alert=len(r["alerts"]))
     if "--quiet" not in sys.argv:
         print("=" * 78)
         print("  EDGE WATCH — il book live sta ancora funzionando?")

scripts/live/fee_watch.py

@@ -49,9 +49,11 @@ ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
 from src.live.book import INSTRUMENT as BOOK_INSTRUMENT  # noqa: E402
+from src.live.giri import registra_giro  # noqa: E402
 from src.live.notifier import notify  # noqa: E402
 
 STATE = ROOT / "data" / "fee_watch" / "state.json"
+RUNS = ROOT / "data" / "fee_watch" / "runs.jsonl"
 # ⚠️ CORREZIONE 2026-08-21. Qui c'era la tupla CABLATA ("BTC-PERPETUAL", "ETH-PERPETUAL") — gli
 # INVERSE, regolati in BTC/ETH — mentre il book esegue sui LINEARI USDC (BTC_USDC-PERPETUAL).
 # Sono due linee di prodotto distinte e Deribit le cambia in modo indipendente: il 18/08 ha
@@ -249,6 +251,7 @@ def main() -> int:
             **{f"cambio {i+1}": c for i, c in enumerate(r["changes"])},
         })
     save_state(r["current"])
+    registra_giro(RUNS, level=r["level"])
     return 2 if r["level"] == "AZIONE" else 0
 
 

scripts/live/scale_watch.py

@@ -13,11 +13,15 @@ from __future__ import annotations
 import sys
 from pathlib import Path
 
-sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
 
+from src.live.giri import registra_giro       # noqa: E402
 from src.live.notifier import notify          # noqa: E402
 from src.live.scale_watch import ALLARME, OK, run_once  # noqa: E402
 
+RUNS = ROOT / "data" / "scale_watch" / "runs.jsonl"
+
 
 def _sender(testo: str):
     try:
@@ -31,6 +35,7 @@ def main() -> None:
     secco = "--secco" in sys.argv
     quiet = "--quiet" in sys.argv
     rep = run_once(sender=None if secco else _sender)
+    registra_giro(RUNS, stato=rep["stato"])
     icona = {OK: "✅", ALLARME: "🚨"}.get(rep["stato"], "⚠️")
     if quiet:
         fb = rep["fallback"]

scripts/live/venue_news.py

@@ -40,10 +40,12 @@ from pathlib import Path
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
+from src.live.giri import registra_giro   # noqa: E402
 from src.live.notifier import notify   # noqa: E402
 
 FEED = "https://insights.deribit.com/exchange-updates/feed/"
 STATE = ROOT / "data" / "live" / "venue_news.jsonl"
+RUNS = ROOT / "data" / "venue_news" / "runs.jsonl"
 
 # Parole che alzano un annuncio a 🚨. Le prime sono FISSE (categorie di rischio che ci toccano
 # per costruzione); le altre si DERIVANO dal codice, sotto.
@@ -140,6 +142,7 @@ def main() -> int:
             v["seed"] = primo_giro
             f.write(json.dumps(v) + "\n")
             nuove.append(v)
+    registra_giro(RUNS, voci=len(tutte), nuove=len(nuove))
 
     # Primo giro: si semina lo storico SENZA allertare (P9: l'allarme massimo non si spende
     # per un arretrato di 10 voci, o non verra' letto il giorno che e' vero).

scripts/live/venue_watch.py

@@ -16,10 +16,13 @@ ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT))
 
 from src.live.deribit import check_specs        # noqa: E402
+from src.live.giri import registra_giro         # noqa: E402
 from src.live.notifier import notify, ultimo_errore   # noqa: E402
 from src.live.venue_watch import (PERSIST_HOURS, THRESHOLD_BPS,  # noqa: E402
                                   run_once)
 
+RUNS = ROOT / "data" / "venue_watch" / "runs.jsonl"
+
 
 def _titolo(rep: dict) -> str:
     """⚠️ Il titolo dice cosa FARE, non solo cosa e' successo: chi lo legge sul telefono deve
@@ -70,6 +73,7 @@ def main() -> int:
     # che manda ordini. Il 18/08 Deribit ha cambiato tick e size dei perpetual lineari USDC e
     # niente se n'era accorto: e' andata bene perche' i cambi erano riduzioni.
     spec = check_specs()
+    registra_giro(RUNS, livelli=rep["levels"])
     if spec["divergenze"]:
         rifiuto = [d for d in spec["divergenze"] if d["rischio"] == "rifiuto"]
         righe = {f"{d['strumento']} {d['campo']}": f"dichiarato {d['dichiarato']} / "

scripts/live/vrp_f_watch.py

@@ -51,9 +51,11 @@ sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research"))
 from cblib import (                                                        # noqa: E402
     dvol_series, f_factors, load_chain, pick_legs, puts, spot_series,
 )
+from src.live.giri import registra_giro                                    # noqa: E402
 from src.live.notifier import notify                                       # noqa: E402
 
 STATE = PROJECT_ROOT / "data" / "vrp_f_watch" / "state.json"
+RUNS = PROJECT_ROOT / "data" / "vrp_f_watch" / "runs.jsonl"
 
 # --- config CONGELATA il 2026-07-30 ---------------------------------------
 STRUCTS = {
@@ -190,6 +192,8 @@ def main() -> int:
         a["notified"] = bool(prev.get("notified", False))
 
     _save_state(a)
+    if not R.empty:
+        registra_giro(RUNS, pairs=a["pairs"], ready=a["ready"])
     return 0
 
 

src/live/giri.py

@@ -0,0 +1,35 @@
+"""registra_giro — il giro che un sorvegliante ha misurato, come riga jsonl (2026-09-21, #16).
+
+PERCHE' ESISTE. `cc01_regime_watch` e' morto 14 notti di fila (#10) senza che nessuno lo vedesse,
+perche' non scriveva niente da sorvegliare: `monitor_health` guarda solo le serie che i monitor
+scrivono, e un monitor silenzioso non e' distinguibile da uno che dice "tutto ok". La riparazione
+di cc01 (`data/cc01_regime/runs.jsonl`, commit 31f5915) e' il modello: qui si generalizza a ogni
+sorvegliante, cosi' un monitor fermo diventa FERMO in `monitor_health`, non invisibile.
+
+Un registro che fallisce non deve cambiare l'esito del sorvegliante che lo scrive: `registra_giro`
+non solleva mai, e su errore lo dice su stderr e ritorna False.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import time
+from pathlib import Path
+
+
+def registra_giro(path: Path, **campi) -> bool:
+    """Appende `{"ts": <ms epoch>, **campi}` a `path` (jsonl, una riga). Crea la cartella se manca.
+
+    Ritorna True se la riga e' stata scritta, False su OSError (disco pieno, permessi, percorso
+    invalido...) dopo aver scritto il motivo su stderr. Non solleva mai: la misura del
+    sorvegliante e' gia' fatta, e un registro che fallisce non deve buttarla via.
+    """
+    riga = {"ts": int(time.time() * 1000), **campi}
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps(riga, default=str) + "\n")
+        return True
+    except OSError as e:
+        print(f"registra_giro: non scrivo {path}: {type(e).__name__}: {e}", file=sys.stderr)
+        return False

src/live/monitor_health.py

@@ -105,6 +105,29 @@ MONITORS: tuple[MonitorSpec, ...] = (
     MonitorSpec("cc01_regime_watch", "cc01_regime", "runs.jsonl", 24.0, max_age_h=30.0,
                 gate="trigger di regime CC01: WARN>=10% / ALERT>=15% funding 30g (riapre il dossier)",
                 open_labeled=False),
+    # I sei sorveglianti che generalizzano la riparazione di cc01 (#16): ognuno registra il
+    # proprio giro in `data/<dir>/runs.jsonl` (src.live.giri.registra_giro), timbri di
+    # ESECUZIONE come cc01 -> open_labeled=False. cron_daily.sh (GIORNALIERO) per i primi
+    # cinque: max_age 30h = un giro mancato + margine.
+    MonitorSpec("edge_watch", "edge_watch", "runs.jsonl", 24.0, max_age_h=30.0,
+                gate="kill del libro live (Sharpe 36m < -0,5 / protezione TP01)",
+                open_labeled=False),
+    MonitorSpec("fee_watch", "fee_watch", "runs.jsonl", 24.0, max_age_h=30.0,
+                gate="regola fee §3 decisa in anticipo",
+                open_labeled=False),
+    MonitorSpec("scale_watch", "scale_watch", "runs.jsonl", 24.0, max_age_h=30.0,
+                gate="GATE SCALA-01 criterio A2",
+                open_labeled=False),
+    MonitorSpec("vrp_f_watch", "vrp_f_watch", "runs.jsonl", 24.0, max_age_h=30.0,
+                gate="criteri (a)(b) §5.18",
+                open_labeled=False),
+    MonitorSpec("venue_news", "venue_news", "runs.jsonl", 24.0, max_age_h=30.0,
+                gate="annunci Deribit, debito 8",
+                open_labeled=False),
+    # cron_book.sh: gira OGNI ORA accanto al libro vivo. max_age 3h = due giri persi.
+    MonitorSpec("venue_watch", "venue_watch", "runs.jsonl", 1.0, max_age_h=3.0,
+                gate="tripwire di venue orario",
+                open_labeled=False),
 )
 
 

tests/test_giri.py

@@ -0,0 +1,69 @@
+"""registra_giro (src/live/giri.py, #16): il registro che i sorveglianti scrivono a ogni giro.
+
+Il contratto e' quello di cc01_regime_watch, generalizzato: appende `{"ts": <ms>, **campi}` come
+riga jsonl e non solleva MAI — un registro che fallisce non deve buttare via una misura gia' fatta.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from src.live.giri import registra_giro  # noqa: E402
+
+
+def test_scrive_una_riga_con_ts_in_millisecondi(tmp_path):
+    p = tmp_path / "sub" / "runs.jsonl"
+    prima = time.time() * 1000
+    assert registra_giro(p, stato="OK", n=3) is True
+    riga = json.loads(p.read_text().strip())
+    assert abs(riga["ts"] - prima) < 5_000
+    assert isinstance(riga["ts"], int)
+    assert riga["stato"] == "OK" and riga["n"] == 3
+
+
+def test_crea_la_cartella_se_manca(tmp_path):
+    p = tmp_path / "non-esiste-ancora" / "runs.jsonl"
+    assert not p.parent.exists()
+    registra_giro(p, x=1)
+    assert p.exists()
+
+
+def test_appende_senza_troncare(tmp_path):
+    p = tmp_path / "runs.jsonl"
+    registra_giro(p, giro=1)
+    registra_giro(p, giro=2)
+    righe = [json.loads(r) for r in p.read_text().splitlines()]
+    assert [r["giro"] for r in righe] == [1, 2]
+
+
+def test_default_str_copre_i_valori_non_serializzabili(tmp_path):
+    """json.dumps(default=str): un valore che non e' nativamente JSON (es. un Path, o un
+    dataclass senza __dict__ diretto) non deve far fallire l'append — diventa la sua str()."""
+    p = tmp_path / "runs.jsonl"
+    assert registra_giro(p, percorso=Path("/tmp/x")) is True
+    riga = json.loads(p.read_text().strip())
+    assert riga["percorso"] == str(Path("/tmp/x"))
+
+
+def test_oserror_ritorna_false_senza_sollevare(tmp_path, capsys):
+    """Un percorso impossibile (qui: dentro un FILE, non una cartella) e' un OSError alla
+    mkdir/scrittura. Il chiamante non deve vedere un'eccezione: la misura e' gia' fatta."""
+    ostacolo = tmp_path / "non-e-una-cartella"
+    ostacolo.write_text("sono un file")
+    p = ostacolo / "runs.jsonl"
+    assert registra_giro(p, stato="OK") is False
+    err = capsys.readouterr().err
+    assert "registra_giro" in err and str(p) in err
+
+
+def test_un_valore_non_serializzabile_diventa_stringa_non_solleva(tmp_path):
+    p = tmp_path / "runs.jsonl"
+    oggetto = object()
+    assert registra_giro(p, oggetto=oggetto) is True
+    riga = json.loads(p.read_text().strip())
+    assert riga["oggetto"] == str(oggetto)

tests/test_watch_runs.py

@@ -0,0 +1,270 @@
+"""registra_giro nei sei sorveglianti generalizzati da cc01_regime_watch (2026-09-21, #16).
+
+PERCHE' ESISTE. `cc01_regime_watch` e' morto 14 notti di fila (#10) senza che nessuno lo vedesse,
+perche' non scriveva niente da sorvegliare (`monitor_health` guarda solo le serie che i monitor
+scrivono). La riparazione di cc01 (`data/cc01_regime/runs.jsonl`, commit 31f5915,
+tests/test_cc01_regime_watch.py) e' il modello: qui si verifica che sia stata applicata agli
+altri sei sorveglianti giornalieri/orari, con lo stesso incrocio P1 con `monitor_health.MONITORS`
+e lo stesso doppio ramo (misura -> una riga, non-misura -> nessuna riga).
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pandas as pd
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "scripts" / "live"))
+
+from src.live.monitor_health import MONITORS  # noqa: E402
+
+import edge_watch as EW    # noqa: E402
+import fee_watch as FW     # noqa: E402
+import scale_watch as SW   # noqa: E402
+
+
+def _carica(nome: str):
+    sp = importlib.util.spec_from_file_location(nome, ROOT / "scripts" / "live" / f"{nome}.py")
+    m = importlib.util.module_from_spec(sp)
+    sp.loader.exec_module(m)
+    return m
+
+
+VF = _carica("vrp_f_watch")
+VN = _carica("venue_news")
+VW = _carica("venue_watch")
+
+
+def _righe(path: Path) -> list[dict]:
+    import json
+    if not path.exists():
+        return []
+    return [json.loads(r) for r in path.read_text().splitlines() if r.strip()]
+
+
+def _spec(nome: str):
+    s = next((m for m in MONITORS if m.name == nome), None)
+    assert s is not None, f"{nome} non e' in monitor_health.MONITORS"
+    return s
+
+
+# ===========================================================================
+# P1 — il bersaglio della sorveglianza si deriva dal codice sorvegliato
+# ===========================================================================
+@pytest.mark.parametrize("nome,modulo,cadence_h,max_age_h", [
+    ("edge_watch", EW, 24.0, 30.0),
+    ("fee_watch", FW, 24.0, 30.0),
+    ("scale_watch", SW, 24.0, 30.0),
+    ("vrp_f_watch", VF, 24.0, 30.0),
+    ("venue_news", VN, 24.0, 30.0),
+    ("venue_watch", VW, 1.0, 3.0),
+])
+def test_registrato_in_monitor_health_sul_file_che_scrive(nome, modulo, cadence_h, max_age_h):
+    s = _spec(nome)
+    assert ROOT / "data" / s.directory / s.series == modulo.RUNS
+    assert s.open_labeled is False
+    assert s.cadence_h == cadence_h
+    assert s.max_age_h <= max_age_h
+
+
+def test_venue_watch_ha_la_cadenza_oraria_del_cron_book():
+    """⚠️ venue_watch gira OGNI ORA in cron_book.sh: max_age stretto, non i 30h dei giornalieri."""
+    assert _spec("venue_watch").cadence_h == 1.0 and _spec("venue_watch").max_age_h == 3.0
+
+
+# ===========================================================================
+# edge_watch — dopo run(): sharpe, n_alert
+# ===========================================================================
+def test_edge_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(EW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["edge_watch.py", "--quiet"])
+    monkeypatch.setattr(EW, "run", lambda: dict(sharpe=0.7, protection=[], consec_fail=0, alerts=[]))
+    assert EW.main() == 0
+    righe = _righe(runs)
+    assert len(righe) == 1
+    assert righe[0]["sharpe"] == 0.7 and righe[0]["n_alert"] == 0
+    assert isinstance(righe[0]["ts"], int)
+
+
+def test_edge_watch_run_che_fallisce_non_scrive(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(EW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["edge_watch.py", "--quiet"])
+
+    def boom():
+        raise RuntimeError("dati di mercato non leggibili")
+    monkeypatch.setattr(EW, "run", boom)
+    with pytest.raises(RuntimeError):
+        EW.main()
+    assert not runs.exists()
+
+
+# ===========================================================================
+# fee_watch — dopo save_state(): level
+# ===========================================================================
+def test_fee_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(FW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["fee_watch.py", "--quiet"])
+    monkeypatch.setattr(FW, "run", lambda: dict(current={}, changes=[], worst_taker_bps=0.0,
+                                                level="OK", reason="tutto in regola", first_read=True))
+    monkeypatch.setattr(FW, "save_state", lambda cur: None)
+    assert FW.main() == 0
+    righe = _righe(runs)
+    assert len(righe) == 1 and righe[0]["level"] == "OK"
+
+
+def test_fee_watch_run_che_fallisce_non_scrive(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(FW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["fee_watch.py", "--quiet"])
+
+    def boom():
+        raise RuntimeError("API Deribit irraggiungibile")
+    monkeypatch.setattr(FW, "run", boom)
+    with pytest.raises(RuntimeError):
+        FW.main()
+    assert not runs.exists()
+
+
+# ===========================================================================
+# scale_watch — dopo run_once(): stato
+# ===========================================================================
+_REP_SCALA = dict(stato="OK", scala_dichiarata=1.0, ladder="0.5/1/2", tetto=2.0,
+                  domande={}, fallback={"stato": "OK", "quota": 0.0, "fallback_veri": 0,
+                                        "giri": 10, "soglia": 0.1, "paper": 0},
+                  lettore={"stato": "OK", "motivo": "ok"}, invio="non tentato")
+
+
+def test_scale_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(SW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["scale_watch.py", "--quiet", "--secco"])
+    monkeypatch.setattr(SW, "run_once", lambda sender=None: dict(_REP_SCALA))
+    SW.main()
+    righe = _righe(runs)
+    assert len(righe) == 1 and righe[0]["stato"] == "OK"
+
+
+def test_scale_watch_run_once_che_fallisce_non_scrive(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(SW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["scale_watch.py", "--quiet", "--secco"])
+
+    def boom(sender=None):
+        raise RuntimeError("lettore di produzione irraggiungibile")
+    monkeypatch.setattr(SW, "run_once", boom)
+    with pytest.raises(RuntimeError):
+        SW.main()
+    assert not runs.exists()
+
+
+# ===========================================================================
+# vrp_f_watch — dopo _save_state(), SOLO se R non e' vuoto: pairs, ready
+# ===========================================================================
+def _frame_coppie(n: int = 45) -> pd.DataFrame:
+    rows = []
+    for i in range(n):
+        exp = pd.Timestamp("2026-01-01") + pd.Timedelta(days=i)
+        for name, v in (("canonico", 0.70), ("candidato", 0.80)):
+            rows.append(dict(asset="BTC", struct=name, exp=exp, ts=pd.Timestamp("2026-01-01"),
+                             dte=7.0, f_short=1.0, f_long=2.0, f_net=v, f_net_mid=v,
+                             quota_lunga=0.18))
+    return pd.DataFrame(rows)
+
+
+def test_vrp_f_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(VF, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["vrp_f_watch.py", "--quiet"])
+    frame = _frame_coppie()
+    monkeypatch.setattr(VF, "measure", lambda chain=None: frame)
+    monkeypatch.setattr(VF, "_save_state", lambda d: None)
+    monkeypatch.setattr(VF, "_load_state", lambda: {})
+    monkeypatch.setattr(VF, "notify", lambda *a, **k: True)
+    assert VF.main() == 0
+    righe = _righe(runs)
+    atteso = VF.assess(frame)
+    assert len(righe) == 1
+    assert righe[0]["pairs"] == atteso["pairs"] and righe[0]["ready"] == atteso["ready"]
+
+
+def test_vrp_f_watch_senza_osservazioni_non_scrive(tmp_path, monkeypatch):
+    """P5: R vuoto non e' un giro misurato — nessuna riga (a differenza di cc01, qui non
+    c'e' nemmeno un `return` anticipato: la guardia sta nella condizione prima di registrare)."""
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(VF, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["vrp_f_watch.py", "--quiet"])
+    monkeypatch.setattr(VF, "measure", lambda chain=None: pd.DataFrame())
+    monkeypatch.setattr(VF, "_save_state", lambda d: None)
+    monkeypatch.setattr(VF, "_load_state", lambda: {})
+    assert VF.main() == 0
+    assert not runs.exists()
+
+
+# ===========================================================================
+# venue_news — dopo la scrittura delle voci: voci, nuove. Feed illeggibile -> nessuna riga
+# ===========================================================================
+_FEED = """<rss><channel>
+<item><title>Deribit Welcomes Orbit Markets</title>
+<link>https://x/a</link><guid>https://x/?p=1</guid><pubDate>Fri, 10 Jul 2026 00:00:00 +0000</pubDate></item>
+<item><title>Another Unrelated Announcement</title>
+<link>https://x/b</link><guid>https://x/?p=2</guid><pubDate>Fri, 11 Jul 2026 00:00:00 +0000</pubDate></item>
+</channel></rss>"""
+
+
+def test_venue_news_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    state = tmp_path / "venue_news.jsonl"
+    monkeypatch.setattr(VN, "RUNS", runs)
+    monkeypatch.setattr(VN, "STATE", state)
+    monkeypatch.setattr(VN, "scarica", lambda *a, **k: _FEED)
+    monkeypatch.setattr(VN, "notify", lambda *a, **k: True)
+    monkeypatch.setattr(sys, "argv", ["venue_news.py", "--quiet"])
+    assert VN.main() == 0
+    righe = _righe(runs)
+    assert len(righe) == 1 and righe[0]["voci"] == 2 and righe[0]["nuove"] == 2
+
+
+def test_venue_news_feed_illeggibile_non_scrive(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(VN, "RUNS", runs)
+    monkeypatch.setattr(VN, "STATE", tmp_path / "venue_news.jsonl")
+    monkeypatch.setattr(VN, "scarica", lambda *a, **k: None)
+    monkeypatch.setattr(VN, "notify", lambda *a, **k: True)
+    monkeypatch.setattr(sys, "argv", ["venue_news.py", "--quiet"])
+    assert VN.main() == 2
+    assert not runs.exists()
+
+
+# ===========================================================================
+# venue_watch — dopo run_once() e check_specs(): livelli (rep["levels"])
+# ===========================================================================
+def test_venue_watch_un_giro_misurato_scrive_una_riga(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(VW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["venue_watch.py", "--quiet"])
+    monkeypatch.setattr(VW, "run_once", lambda sender=None: dict(
+        levels={"BTC": "OK", "ETH": "OK"}, alerts=[], platform_locked=False))
+    monkeypatch.setattr(VW, "check_specs", lambda: dict(divergenze=[]))
+    assert VW.main() == 0
+    righe = _righe(runs)
+    assert len(righe) == 1 and righe[0]["livelli"] == {"BTC": "OK", "ETH": "OK"}
+
+
+def test_venue_watch_run_once_che_fallisce_non_scrive(tmp_path, monkeypatch):
+    runs = tmp_path / "runs.jsonl"
+    monkeypatch.setattr(VW, "RUNS", runs)
+    monkeypatch.setattr(sys, "argv", ["venue_watch.py", "--quiet"])
+
+    def boom(sender=None):
+        raise RuntimeError("osservazione del consenso fallita")
+    monkeypatch.setattr(VW, "run_once", boom)
+    with pytest.raises(RuntimeError):
+        VW.main()
+    assert not runs.exists()