#!/usr/bin/env python3 """ vault-digest — daily 24h state snapshot. Reads last 24h of vault activity (commits, events.ndjson, HealthLog, CuratorQueue additions, daily files), formats as markdown, writes to HQ-Vault/01_Daily/YYYY-MM-DD-Digest.md, and pushes a 5-line Telegram summary to Rocky. Schedule: daily 21:00 UTC (~7am AEST) via /etc/cron.d/vault-digest. """ import json import os import re import subprocess import sys import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path ROOT = Path("HQ-Vault") DAILY_DIR = ROOT / "01_Daily" EVENTS = ROOT / "_System" / "AgentLog" / "events.ndjson" HEALTH = ROOT / "_System" / "HealthLog.md" CURATOR_QUEUE = ROOT / "_System" / "CuratorQueue.md" NOW_UTC = datetime.now(timezone.utc) TODAY = NOW_UTC.strftime("%Y-%m-%d") SINCE = (NOW_UTC - timedelta(hours=24)).isoformat() # Telegram from /root/.hermes/.env (read at runtime) def load_env(path: str) -> dict: out = {} if not os.path.exists(path): return out for line in open(path, "r", encoding="utf-8", errors="ignore"): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") out[k.strip()] = v.strip().strip('"').strip("'") return out def section_commits() -> str: try: out = subprocess.check_output( ["git", "log", f"--since={SINCE}", "--oneline", "--no-merges"], text=True, encoding="utf-8", ) except subprocess.CalledProcessError: return "_(git log error)_" if not out.strip(): return "_No commits in last 24h._" lines = out.strip().splitlines() return f"**{len(lines)} commits** in last 24h:\n\n" + "\n".join(f"- `{l}`" for l in lines[:25]) def section_events() -> str: if not EVENTS.exists(): return "_events.ndjson missing_" cutoff = (NOW_UTC - timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ") by_action: dict[str, int] = {} by_agent: dict[str, int] = {} failures: list[dict] = [] total = 0 for line in EVENTS.read_text(encoding="utf-8", errors="ignore").splitlines(): line = line.strip() if not line: continue try: ev = json.loads(line) except json.JSONDecodeError: continue if ev.get("ts", "") < cutoff: continue total += 1 by_action[ev.get("action", "unknown")] = by_action.get(ev.get("action", "unknown"), 0) + 1 by_agent[ev.get("agent", "unknown")] = by_agent.get(ev.get("agent", "unknown"), 0) + 1 if ev.get("result") in ("fail", "warn"): failures.append(ev) if total == 0: return "_No events.ndjson activity in 24h._" parts = [f"**{total} events** in last 24h."] parts.append("") parts.append("By agent: " + ", ".join(f"`{k}`={v}" for k, v in sorted(by_agent.items(), key=lambda x: -x[1]))) parts.append("") parts.append("By action: " + ", ".join(f"`{k}`={v}" for k, v in sorted(by_action.items(), key=lambda x: -x[1]))) if failures: parts.append("") parts.append(f"**{len(failures)} fail/warn events:**") for ev in failures[:10]: parts.append(f"- `{ev.get('agent')}` {ev.get('action')} `{ev.get('target')}` ({ev.get('result')}): {ev.get('notes', '')}") return "\n".join(parts) def section_curator_queue() -> str: if not CURATOR_QUEUE.exists(): return "" text = CURATOR_QUEUE.read_text(encoding="utf-8", errors="ignore") today_marker = f"--- vault-curator daily {TODAY}" if today_marker not in text: return "_No new curator items today._" after = text.split(today_marker, 1)[1] lines = [l for l in after.splitlines() if l.startswith("- ")] return f"**{len(lines)} new items** queued by vault-curator today.\n\n" + "\n".join(lines[:10]) def section_healthlog_latest() -> str: if not HEALTH.exists(): return "" text = HEALTH.read_text(encoding="utf-8", errors="ignore") headers = re.findall(r"^## .+$", text, re.MULTILINE) if not headers: return "" last_header = headers[-1] idx = text.rfind(last_header) snippet = text[idx:idx + 600] return f"Latest HealthLog entry:\n\n{snippet}" def write_digest(content: str) -> Path: out_path = DAILY_DIR / f"{TODAY}-Digest.md" DAILY_DIR.mkdir(parents=True, exist_ok=True) out_path.write_text(content, encoding="utf-8") return out_path def telegram_send(token: str, chat_id: str, text: str) -> bool: if not token or not chat_id: return False url = f"https://api.telegram.org/bot{token}/sendMessage" data = urllib.parse.urlencode({ "chat_id": chat_id, "text": text, "parse_mode": "Markdown", "disable_web_page_preview": "true", }).encode("utf-8") try: req = urllib.request.Request(url, data=data, method="POST") with urllib.request.urlopen(req, timeout=10) as resp: return resp.status == 200 except Exception as e: print(f"telegram error: {e}", file=sys.stderr) return False def emit_event(action: str, target: str, result: str, notes: str) -> None: try: subprocess.run( ["bash", "scripts/log-event.sh", "cron:vault-digest", action, target, result, notes], check=False, timeout=5, ) except Exception: pass def main() -> int: body = [ f"# Vault Digest — {TODAY}", "", f"_Generated {NOW_UTC.strftime('%Y-%m-%d %H:%M UTC')} by cron:vault-digest. Covers last 24h._", "", "## Commits", "", section_commits(), "", "## AgentLog events", "", section_events(), "", "## Curator findings", "", section_curator_queue(), "", "## Vault health", "", section_healthlog_latest(), "", ] digest_md = "\n".join(body) out_path = write_digest(digest_md) print(f"vault-digest: wrote {out_path}") # Telegram TLDR — first ~10 lines of body essentials env = load_env("/root/.hermes/.env") token = env.get("TELEGRAM_BOT_TOKEN", "") chat_id = env.get("TELEGRAM_HOME_CHANNEL", "") # Compact summary for Telegram (Markdown) commits_text = section_commits().splitlines()[0] if section_commits() else "no commits" events_text = section_events().splitlines()[0] if section_events() else "no events" queue_text = section_curator_queue().splitlines()[0] if section_curator_queue() else "no curator items" tg_msg = ( f"*Vault Digest — {TODAY}*\n\n" f"• {commits_text}\n" f"• {events_text}\n" f"• {queue_text}\n\n" f"Full report: `01_Daily/{TODAY}-Digest.md`" ) sent = telegram_send(token, chat_id, tg_msg) if sent: emit_event("cron_run", "vault-digest.py", "ok", f"digest written + telegram pushed") print("vault-digest: telegram pushed") else: emit_event("cron_run", "vault-digest.py", "warn", f"digest written but telegram push failed") print("vault-digest: telegram push failed (digest still written)") return 0 if __name__ == "__main__": sys.exit(main())