#!/usr/bin/env python3 """report.py — monitoraggio permanente del consumo token per istanza. Nato dal thread di diagnosi del 04/09/2026 (sentinella-inbox-documenti a 62MB di sessione mai ruotata) — quella diagnosi si è basata sulla dimensione grezza dei file .jsonl, una stima grossolana. Qui si legge il dato vero: ogni turno "assistant" nei transcript porta il campo `message.usage` con i token reali di quella chiamata (cache_creation_input_tokens, cache_read_input_tokens, output_tokens) — sommandoli per sessione si ottiene il consumo reale, non un proxy. Non è un contatore di sistema esterno: legge gli stessi .jsonl che Claude Code già scrive in ~/.claude/projects/, nessuna integrazione con l'account o con /usage (quello resta l'unica fonte per il consumo di piano/quota in senso stretto - questo script guarda "quale istanza consuma quanto", non "quanto resta del piano"). Bucket per istanza, per `cwd` (letto direttamente dalle righe del transcript, non dedotto dal nome della cartella .claude/projects — più robusto): - scripts/slave-sentinels// -> quel nome (sentinella o esperto persistente) - root del progetto (/mnt/ssd/data/Dropbox/adrian) -> smistato in due bucket via il marcatore fisso che avvia_slave.sh antepone sempre al prompt di una slave one-shot ("Sei una slave Adrian one-shot..." — indipendente da --with-memory, vedi avvia_slave.sh, capovolto il 05/09/2026 a "niente memory di default"): "slave one-shot" se il primo messaggio utente inizia così, "sessioni interattive Adrian" altrimenti. - qualunque altro cwd (worktree, sandbox, ecc.) -> bucket "altro" per cwd. Soglie di attenzione (di partenza, tarabili — non validate su uno storico lungo). La metrica di allarme è `cache_creation_input_tokens + output_tokens` ("costo reale"), non il totale grezzo: `cache_read_input_tokens` da solo può essere enorme in una sessione interattiva lunga e normale (riuso di contesto già in cache, economico) — usarlo nella soglia produceva solo falsi allarmi, scoperto testando questo stesso script sui dati reali il 04/09/2026. - una singola sessione con oltre SOGLIA_COSTO_REALE token di costo reale. - una sessione persistente (stesso session_id) attiva da oltre SOGLIA_ORE_SENZA_RESTART ore senza restart — il pattern strutturale che ha causato il problema del 04/09 (sessione che non ruota mai, indipendentemente dal contenuto). Girato via cron giornaliero (vedi crontab, stesso pattern di collect_logs.py/check_job_health.py) — non serve un servizio sempre acceso, il consumo si guarda a bocce ferme una volta al giorno. """ from __future__ import annotations import json from collections import defaultdict from datetime import datetime, timezone from pathlib import Path HERE = Path(__file__).resolve().parent PROJECTS_DIR = Path.home() / ".claude" / "projects" ROOT = Path("/mnt/ssd/data/Dropbox/adrian") SENTINELS_DIR = ROOT / "scripts" / "slave-sentinels" REPORT_PATH = HERE / "report_latest.txt" HISTORY_PATH = HERE / "history.jsonl" SLAVE_PREFIX = "Sei una slave Adrian one-shot lanciata da avvia_slave.sh." SOGLIA_COSTO_REALE = 2_000_000 SOGLIA_ORE_SENZA_RESTART = 72 # Senza questo filtro il report include ogni sessione mai esistita in ~/.claude/projects/ — una # sentinella smantellata (come le tre del 04/09/2026) resterebbe segnalata per sempre, il suo # .jsonl storico non sparisce. Guardiamo solo l'attività recente: una sessione senza attività da # più di RECENCY_DAYS esce da sola dal radar, comportamento "autopulente" invece di dover # ricordarsi di escludere a mano le istanze decommissionate. RECENCY_DAYS = 3 # Bucket esclusi dalla segnalazione ⚠️ (mostrati comunque in tabella, per confronto): lavoro # umano/deliberato, non automazione che può derapare da sola. Una sessione interattiva lunga con # Mauro che costa milioni di token e' normale, non un'anomalia da segnalare ogni giorno - il senso # di questo monitoraggio e' sulle istanze secondarie (sentinelle/esperti/servizi), non sul lavoro # principale. Scoperto testando questo script sui dati reali il 04/09/2026 (senza # quest'esclusione, la lista di anomalie era dominata da normali sessioni Adrian). BUCKET_ESCLUSI_DA_ALLARME = {"sessioni interattive Adrian", "slave one-shot"} def parse_ts(s: str | None) -> datetime | None: if not s: return None try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except ValueError: return None def classify(cwd: str | None, first_user_text: str | None) -> str: if not cwd: return "altro (cwd sconosciuto)" cwd_path = Path(cwd) try: rel = cwd_path.relative_to(SENTINELS_DIR) return str(rel) except ValueError: pass if cwd_path == ROOT: if first_user_text and first_user_text.startswith(SLAVE_PREFIX): return "slave one-shot" return "sessioni interattive Adrian" return f"altro ({cwd})" def session_totals(jsonl_path: Path) -> dict | None: cache_creation = cache_read = output = 0 n_turns = 0 cwd = None first_ts = last_ts = None first_user_text = None with open(jsonl_path, "r", encoding="utf-8", errors="replace") as fh: for line in fh: line = line.strip() if not line: continue try: d = json.loads(line) except json.JSONDecodeError: continue t = d.get("type") ts = parse_ts(d.get("timestamp")) if ts: if first_ts is None or ts < first_ts: first_ts = ts if last_ts is None or ts > last_ts: last_ts = ts if cwd is None and d.get("cwd"): cwd = d["cwd"] if t == "user" and first_user_text is None: msg = d.get("message", {}) content = msg.get("content") if isinstance(content, str): first_user_text = content elif isinstance(content, list): for c in content: if c.get("type") == "text": first_user_text = c.get("text") break if t == "assistant": usage = d.get("message", {}).get("usage") if usage: cache_creation += usage.get("cache_creation_input_tokens", 0) or 0 cache_read += usage.get("cache_read_input_tokens", 0) or 0 output += usage.get("output_tokens", 0) or 0 n_turns += 1 if n_turns == 0: return None return { "path": str(jsonl_path), "cwd": cwd, "cwd_exists": bool(cwd) and Path(cwd).is_dir(), "cache_creation": cache_creation, "cache_read": cache_read, "output": output, "total": cache_creation + cache_read + output, "costo_reale": cache_creation + output, "n_turns": n_turns, "first_ts": first_ts.isoformat() if first_ts else None, "last_ts": last_ts.isoformat() if last_ts else None, "span_ore": (last_ts - first_ts).total_seconds() / 3600 if first_ts and last_ts else 0, "bucket": classify(cwd, first_user_text), } def scan_all() -> list[dict]: sessions = [] if not PROJECTS_DIR.exists(): return sessions cutoff = datetime.now(timezone.utc).timestamp() - RECENCY_DAYS * 86400 for proj_dir in PROJECTS_DIR.iterdir(): if not proj_dir.is_dir(): continue for jsonl_path in proj_dir.glob("*.jsonl"): if jsonl_path.stat().st_mtime < cutoff: continue info = session_totals(jsonl_path) if info: sessions.append(info) return sessions def build_report(sessions: list[dict]) -> tuple[str, bool]: now = datetime.now(timezone.utc) by_bucket: dict[str, list[dict]] = defaultdict(list) for s in sessions: by_bucket[s["bucket"]].append(s) issues: list[str] = [] lines = [f"# Consumo token per istanza – {now.isoformat(timespec='seconds')}", ""] rows = [] for bucket, sess_list in sorted(by_bucket.items(), key=lambda kv: -sum(s["costo_reale"] for s in kv[1])): total = sum(s["total"] for s in sess_list) costo_reale = sum(s["costo_reale"] for s in sess_list) cache_read = sum(s["cache_read"] for s in sess_list) n = len(sess_list) max_span = max((s["span_ore"] for s in sess_list), default=0) rows.append((bucket, costo_reale, cache_read, total, n, max_span)) if bucket in BUCKET_ESCLUSI_DA_ALLARME: continue for s in sess_list: if not s.get("cwd_exists", True): # cwd non esiste più su disco: istanza smantellata (cartella rimossa/rinominata, # es. inbox-documenti/sentinella-madre/gate-ufficio il 04/09/2026) — il suo .jsonl # storico resta nella finestra RECENCY_DAYS ma non ha più senso segnalarlo come # anomalia viva, senza aspettare i 3 giorni di decadimento naturale. continue if s["costo_reale"] > SOGLIA_COSTO_REALE: issues.append( f"{bucket}: sessione con {s['costo_reale']:,} token di costo reale " f"(soglia {SOGLIA_COSTO_REALE:,}) — {Path(s['path']).name}" ) if s["span_ore"] > SOGLIA_ORE_SENZA_RESTART: issues.append( f"{bucket}: sessione attiva da {s['span_ore']/24:.1f} giorni senza restart " f"(soglia {SOGLIA_ORE_SENZA_RESTART}h) — {Path(s['path']).name}" ) header = ( "⚠️ Problemi individuati: " + "; ".join(issues) if issues else "✅ Nessuna istanza sopra soglia." ) lines.insert(1, header) lines.append("") lines.append(f"{'Istanza':<38} {'Costo reale':>14} {'cache_read (riuso)':>19} {'Totale grezzo':>15} {'Sessioni':>9} {'Span max (h)':>13}") lines.append("-" * 116) for bucket, costo_reale, cache_read, total, n, max_span in rows: lines.append(f"{bucket:<38} {costo_reale:>14,} {cache_read:>19,} {total:>15,} {n:>9} {max_span:>13.1f}") lines.append("") lines.append( "Nota: 'Costo reale' = cache_creation_input_tokens + output_tokens, i token pagati a " "prezzo pieno per quella sessione — è la colonna da guardare per capire chi consuma " "davvero. 'cache_read (riuso)' è mostrato solo per contesto: può essere enorme in una " "sessione interattiva lunga e normale, non è un segnale d'allarme di per sé. 'Span max' = " "ore tra primo e ultimo turno della sessione più longeva del bucket — un valore alto su " "un'istanza persistente è il segnale strutturale che ha causato il problema del 04/09 " "(sessione mai ruotata), non solo il contenuto." ) text = "\n".join(lines).strip() + "\n" return text, bool(issues) def append_history(sessions: list[dict]) -> None: now = datetime.now(timezone.utc) by_bucket: dict[str, int] = defaultdict(int) for s in sessions: by_bucket[s["bucket"]] += s["costo_reale"] entry = {"timestamp": now.isoformat(timespec="seconds"), "per_bucket_costo_reale": dict(by_bucket)} with open(HISTORY_PATH, "a", encoding="utf-8") as fh: fh.write(json.dumps(entry, ensure_ascii=False) + "\n") def main() -> None: sessions = scan_all() report_text, has_issues = build_report(sessions) REPORT_PATH.write_text(report_text, encoding="utf-8") append_history(sessions) print(report_text) if has_issues: print(f"⚠️ Anomalie rilevate. Vedi {REPORT_PATH}") else: print(f"Report salvato in {REPORT_PATH}") if __name__ == "__main__": main()