"""Collect, classify, register, render, and send the personal inbox brief.""" from __future__ import annotations import argparse import hashlib import subprocess import json from datetime import date, datetime, timedelta, timezone from email.utils import parseaddr from typing import Any import streak from collect import Collection, collect from judgement import judgement_for from error_reporting import record_failure from notify import ROOT, send from register import REGISTER_PATH, append_deadlines from rules import write_findings LAST_BRIEF_PATH = ROOT / "state" / "last_brief.json" VOICE_DIR = ROOT / "inbox-voice" # 60, not 30. The visa is 56 days out: at 30 it was registered and invisible, which is # worse than not having it, because the system held the fact and showed a Vodafone bill # instead. Measured 2026-07-29: widening alone would not have fixed it either — the visa # sat at position 23 of 23 in the old register and the brief shows three. The noise fix # had to come first. 60 is also what the design document specified. HORIZON_DAYS = 60 # The timer unit whose firing marks a run as unattended. --scheduled is set in the unit # file, so `systemctl start` typed by a human carries it too; the flag alone proves # nothing. See _timer_initiated(). TIMER_UNIT = "personal-cos-brief.timer" def _sender_name(value: str) -> str: name, address = parseaddr(value) return name.strip('"') or address or "Unknown sender" def _local_time(value: str | None) -> str: if not value: return "time unknown" try: return datetime.fromisoformat(value).astimezone().strftime("%H:%M") except ValueError: return "time unknown" def _upcoming_deadlines() -> list[dict[str, Any]]: if not REGISTER_PATH.exists(): return [] today = datetime.now().astimezone().date() end = today + timedelta(days=HORIZON_DAYS) latest: dict[tuple[str, str, str, str], dict[str, Any]] = {} for line in REGISTER_PATH.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue record = json.loads(line) try: due = date.fromisoformat(record["date"]) except (KeyError, ValueError): continue if today <= due <= end: # Not keyed on account. The same obligation lands in two mailboxes — the # migration agent's email is in both rocky and signup — and a four-line # brief cannot afford to say the same thing twice. key = ( record["date"], record["source_subject"], record["source_sender"], ) previous = latest.get(key) if not previous or record["last_seen"] > previous["last_seen"]: latest[key] = record return sorted( latest.values(), key=lambda item: (item["date"], item.get("account", "signup")), ) def _voice_count() -> int: VOICE_DIR.mkdir(parents=True, exist_ok=True) return sum( 1 for path in VOICE_DIR.iterdir() if path.is_file() and path.name != ".gitkeep" ) def _coverage_line(collection: Collection) -> str: covered = datetime.fromisoformat(collection.covered_to_utc).astimezone() line = f"Covered to {covered.strftime('%H:%M')}." if collection.excluded_work_count: noun = "email" if collection.excluded_work_count == 1 else "emails" line += f" {collection.excluded_work_count} work {noun} excluded." if collection.coverage_rebaselined: line += " Coverage re-baselined over 7 days." return line def render( collection: Collection, findings: list[dict[str, Any]], voice_count: int ) -> str: act = [item for item in findings if item["bucket"] == "ACT"] deadlines = _upcoming_deadlines() coverage = _coverage_line(collection) if not act and not deadlines: lines = ["Nothing needing you."] lines.extend((collection.notes or [])[:3]) if voice_count: lines.append(f"Voice: {voice_count} captures waiting.") lines.append(coverage) return "\n".join(lines) now = datetime.now().astimezone() lines = [f"PERSONAL — {now.strftime('%a %d %b')}"] accounts = sorted( {item.get("account", "signup") for item in findings} | {item.get("account", "signup") for item in deadlines} ) for account in accounts: account_findings = [ item for item in findings if item.get("account", "signup") == account ] if not account_findings: continue account_act = [item for item in account_findings if item["bucket"] == "ACT"] review_count = sum(item["bucket"] == "REVIEW" for item in account_findings) note_count = sum(item["bucket"] == "NOTE" for item in account_findings) lines.extend(["", account.upper()]) for item in account_act[:2]: date_text = ( f", due {date.fromisoformat(item['normalized_date']).strftime('%d %b')}" if item.get("normalized_date") else "" ) lines.append(f"- {item['subject'] or '(no subject)'}{date_text}") lines.append( f" {_sender_name(item['sender'])}, " f"{_local_time(item['message_date'])}" ) if review_count: lines.append(f"REVIEW {review_count} items") if note_count: lines.append(f"NEW {note_count} others, nothing found") if len(lines) >= 13: break if deadlines: # Deadlines are the reason this system exists. They are never crowded out by # today's inbox chatter — the ACT block is trimmed to make room instead. lines = lines[:13] lines.extend(["", "COMING UP"]) for item in deadlines[:3]: due = date.fromisoformat(item["date"]).strftime("%d %b") # No account tag. These are de-duplicated across mailboxes, so whichever # account survived the dedupe is arbitrary — printing it would imply the # obligation lives in one inbox when it usually lives in two. lines.append(f"- {due} {item['source_subject']}") summary_lines = list(collection.notes or []) if voice_count: summary_lines.append(f"Voice: {voice_count} captures waiting.") lines.extend(summary_lines[: max(0, 19 - len(lines))]) lines = lines[:19] # Judgement is appended UNDERNEATH everything the rules found, never instead of # it, and it cannot shorten what is already here. If it is missing, stale or # malformed, judgement_for returns None and this block does nothing. thought = judgement_for() if thought: lines.extend(["", "WORTH A THOUGHT"]) lines.extend(f"- {line}" for line in thought["lines"]) lines.append(coverage) return "\n".join(lines) def _timer_initiated(max_skew_seconds: int = 1800) -> bool: """Did the systemd timer start this run, or did a human? Asks systemd when the timer last fired and compares it to now. This replaces a clock-window check (is it near 18:30?) which was wrong in BOTH directions: - It did not catch what it was built for. The false day that motivated it came from `systemctl start` at 18:46 — sixteen minutes from the slot, inside the twenty-minute window. It would have counted anyway. - It broke the case that matters more. The timer is Persistent=true, so a box that was down at 18:30 fires the brief late on catch-up. The brief lands, the day is real, and a clock window would refuse to count it — silently resetting the fourteen-day clock while the counter still looked healthy. Asking systemd is the honest question: a timer firing sets LastTriggerUSec, a human typing `systemctl start` does not. Returns False when it cannot tell — no systemd, no timer, unparseable value. A run that cannot be PROVEN unattended does not count toward the fourteen. The window is 30 minutes, not 5. A brief takes ~110 seconds, and a slow one takes longer — tonight a single IMAP account blocked for 60s on its own. Checked at the END of a slow run against a 5-minute window, a genuine timer fire would have failed the test and silently not counted, which is the same silent-clock-reset this replaced. 30 minutes covers any plausible run while still separating a timer fire from a human at a different hour; nothing fires this timer more than once a day. """ try: result = subprocess.run( ["systemctl", "show", TIMER_UNIT, "-p", "LastTriggerUSec", "--value"], capture_output=True, text=True, timeout=15, ) except (OSError, subprocess.SubprocessError): return False stamp = result.stdout.strip() if result.returncode != 0 or not stamp or stamp in {"n/a", "0"}: return False # systemd renders e.g. "Thu 2026-07-30 18:30:00 AEST" try: fired = datetime.strptime(stamp[4:23], "%Y-%m-%d %H:%M:%S").astimezone() except ValueError: return False return abs((datetime.now().astimezone() - fired).total_seconds()) <= max_skew_seconds def _record_success(text: str, scheduled: bool = False, unattended: bool | None = None) -> None: LAST_BRIEF_PATH.parent.mkdir(parents=True, exist_ok=True) value = { "sent_at_utc": datetime.now(timezone.utc).isoformat(), "brief_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), "trigger": "schedule" if scheduled else "manual", } if unattended if unattended is not None else (scheduled and _timer_initiated()): # Only unattended runs count toward the fourteen. A hand-fired brief proves # the code works; it proves nothing about whether the rail runs without him. value["unattended_streak"] = streak.record()["streak"] elif scheduled: value["trigger"] = "manual" value["_note"] = "carried --scheduled but the timer did not fire it" temporary = LAST_BRIEF_PATH.with_suffix(".json.tmp") temporary.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") temporary.replace(LAST_BRIEF_PATH) def _report_failure(error: Exception) -> None: safe_message = record_failure("Personal brief failed", error) try: if not send(f"PERSONAL BRIEF FAILED\n{safe_message}"): notification_error = RuntimeError("Telegram failure notification returned False") notification_summary = record_failure( "Personal brief failure notification failed", notification_error ) print(f"Personal brief failure notification failed: {notification_summary}") except Exception as notification_error: notification_summary = record_failure( "Personal brief failure notification failed", notification_error ) print(f"Personal brief failure notification failed: {notification_summary}") print(f"Personal brief failed: {safe_message}") def main(scheduled: bool = False) -> int: # Ask BEFORE the slow work, not after: collection can take minutes, and the # question is whether the timer started this process, not how long it then ran. unattended = scheduled and _timer_initiated() collection = collect() findings = write_findings(collection.messages) append_deadlines(findings) text = render(collection, findings, _voice_count()) if not send(text): return 1 _record_success(text, scheduled, unattended) print("Brief sent successfully.") return 0 if __name__ == "__main__": parser = argparse.ArgumentParser(description="Send the personal inbox brief.") parser.add_argument( "--scheduled", action="store_true", help="mark this run as unattended; only these count toward the streak", ) arguments = parser.parse_args() try: raise SystemExit(main(arguments.scheduled)) except Exception as error: _report_failure(error) raise SystemExit(1)