#!/usr/bin/env python3
"""
Local Screen Time reader (read-only)
====================================
Reads the local macOS Screen Time app-usage store
(``~/Library/Application Support/Knowledge/knowledgeC.db``) so the vault agent
can answer "where did my screen time go?" — the owner asks in plain language
and never runs a command. The `screentime` skill drives this.

Read-only by construction: the live DB is never even opened — it is
snapshot-copied (db + -wal + -shm) to a temp dir and the *copy* is queried, so
knowledged's locks are never contended and the original is never touched. The
only files this script ever writes are its own two history series (below).

Runs on the local machine, like the iMessage reader and unlike the digest/sweep
engine: knowledgeC.db records *this machine's* usage — an automation host's
copy would show the host's own usage, not the owner's. Mac-only (nothing syncs
into this store, so there is no iPhone/iPad data). Requires the app running the
assistant (Claude, or the terminal it runs in) to have macOS Full Disk Access.

Days end at a boundary hour — default 04:00 local, so a 1am session counts
toward the evening's day, not the next morning's — tunable via config (below).
Events carry their own GMT offset (ZSECONDSFROMGMT), so days bucket correctly
across travel.

knowledgeC only retains ~4 weeks, so each run upserts a per-day rollup into
``<vault>/Agent/scripts/screentime/history.json`` (created on first save; meant
to be committed) — long-run trends survive the rolling window. One entry per
logical day; a final day is never overwritten by a partial one. Zero-usage days
are never upserted — a gap is honest, a zero is a claim about the owner's
behaviour the data can't support.

FALLBACK — /app/usage can die. The stream can silently stop being donated (an
OS-level outage seen in the wild), and once the old rows age out of the ~4-week
window (0 rows) a naive reader becomes indistinguishable from "laptop never
opened". This reader detects a dead/stale stream (judged against
/display/isBacklit, so a genuinely-shut laptop is not misread as death; see
``app_usage_status`` in --json) and falls back to SCREEN-ON time — the
/display/isBacklit lit intervals, split at the same day boundary. No per-app
breakdown; slightly generous (idle-with-display-lit counts, but it tracks
app-usage within a few percent where both streams overlap). The fallback series
persists to ``history-screen-on.json`` — its own file, same upsert rules; it
never contaminates the app-usage series in ``history.json``.

Every run prints a one-line coverage summary to stderr (window · days with
data · headline metric · /app/usage status), so `--json` stdout stays clean and
a dead stream can never pass silently for a quiet week.

Optional config — read from Agent/config.json under the `screentime` key, all
optional (built-in defaults apply when absent):
    screentime.day_boundary_hour   local hour a logical day rolls over, 0-23 (default 4)
    screentime.default_since       window when none is asked for            (default "7d")

Usage:
    python -m alexandria_vault.engine.screentime_read --vault .                 # last 7 logical days + save history
    python -m alexandria_vault.engine.screentime_read --vault . --since today   # also: yesterday, 2w, 30d, YYYY-MM-DD, all
    python -m alexandria_vault.engine.screentime_read --vault . --json          # machine-readable (prefer for synthesis)
    python -m alexandria_vault.engine.screentime_read --vault . --history       # both saved series (app-usage + screen-on)
    python -m alexandria_vault.engine.screentime_read --vault . --no-save       # ad-hoc run, skip the history writes
    python -m alexandria_vault.engine.screentime_read --vault . --top 15        # more apps in the human summary

Read-only against the source. Python 3.9+, standard library only; no secrets,
no network.
"""
import argparse
import datetime as dt
import json
import os
import shutil
import socket
import sqlite3
import sys
import tempfile
from pathlib import Path

from alexandria_vault.engine import consume_vault_arg, resolve_vault
from alexandria_vault.vault_core import load_config as _vc_load_config

# `--vault` must land in VAULT_DIR before the module-level config load below.
consume_vault_arg()

VAULT = resolve_vault()
DB_PATH = Path.home() / "Library" / "Application Support" / "Knowledge" / "knowledgeC.db"
HIST_DIR = VAULT / "Agent" / "scripts" / "screentime"
HISTORY = HIST_DIR / "history.json"                      # /app/usage series (committed)
HISTORY_SCREEN_ON = HIST_DIR / "history-screen-on.json"  # /display/isBacklit fallback series
COCOA_EPOCH = 978307200          # seconds between 1970-01-01 and 2001-01-01 UTC
GAP_MIN = 5                      # merge usage segments ≤ this many minutes apart into one "block"


# ── config (Core-reads-Config seam; personalization never hardcoded here) ─────
def _load_config():
    """Read optional `screentime` tuning from the vault's Config (vault_core: the
    one locator + loader). A missing file or key is fine — built-in defaults apply.
    No identity is read here; this only tunes defaults (day boundary, window)."""
    cfg = _vc_load_config(VAULT)
    return cfg.get("screentime", {}) if isinstance(cfg, dict) else {}


ST = _load_config()
DEFAULT_SINCE = ST.get("default_since", "7d")
try:
    BOUNDARY_HOUR = int(ST.get("day_boundary_hour", 4))
except (TypeError, ValueError):
    BOUNDARY_HOUR = 4
if not 0 <= BOUNDARY_HOUR <= 23:
    BOUNDARY_HOUR = 4


# ── bundle id → friendly name (fallback: prettified last component) ──────────
APP_NAMES = {
    "com.anthropic.claudefordesktop": "Claude",
    "com.openai.chat": "ChatGPT",
    "md.obsidian": "Obsidian",
    "com.brave.Browser": "Brave",
    "com.apple.Safari": "Safari",
    "com.google.Chrome": "Chrome",
    "company.thebrowser.Browser": "Arc",
    "org.mozilla.firefox": "Firefox",
    "com.apple.MobileSMS": "Messages",
    "com.apple.FaceTime": "FaceTime",
    "net.whatsapp.WhatsApp": "WhatsApp",
    "com.tinyspeck.slackmacgap": "Slack",
    "com.apple.mail": "Mail",
    "us.zoom.xos": "Zoom",
    "ru.keepcoder.Telegram": "Telegram",
    "org.whispersystems.signal-desktop": "Signal",
    "com.hnc.Discord": "Discord",
    "com.apple.Terminal": "Terminal",
    "com.googlecode.iterm2": "iTerm",
    "dev.warp.Warp-Stable": "Warp",
    "com.mitchellh.ghostty": "Ghostty",
    "com.microsoft.VSCode": "VS Code",
    "com.todesktop.230313mzl4w4u92": "Cursor",
    "com.apple.dt.Xcode": "Xcode",
    "com.spotify.client": "Spotify",
    "com.apple.Music": "Music",
    "com.apple.TV": "TV",
    "com.apple.Photos": "Photos",
    "com.apple.podcasts": "Podcasts",
    "com.apple.iBooksX": "Books",
    "com.apple.news": "News",
    "com.apple.QuickTimePlayerX": "QuickTime",
    "com.apple.finder": "Finder",
    "com.apple.Preview": "Preview",
    "com.apple.systempreferences": "System Settings",
    "com.apple.ActivityMonitor": "Activity Monitor",
    "com.apple.AppStore": "App Store",
    "com.apple.iCal": "Calendar",
    "com.apple.Notes": "Notes",
    "com.apple.reminders": "Reminders",
    "com.apple.freeform": "Freeform",
    "com.apple.Passwords": "Passwords",
    "com.apple.findmy": "Find My",
    "com.apple.weather": "Weather",
    "com.apple.ScreenContinuity": "iPhone Mirroring",
}

# category → bundle ids (unlisted apps land in "Other")
CATEGORIES = {
    "Building": {
        "com.anthropic.claudefordesktop", "com.openai.chat", "md.obsidian",
        "com.microsoft.VSCode", "com.todesktop.230313mzl4w4u92", "com.apple.dt.Xcode",
        "com.apple.Terminal", "com.googlecode.iterm2", "dev.warp.Warp-Stable",
        "com.mitchellh.ghostty", "com.apple.Notes", "com.apple.freeform",
    },
    "Comms": {
        "com.apple.MobileSMS", "com.apple.FaceTime", "net.whatsapp.WhatsApp",
        "com.tinyspeck.slackmacgap", "com.apple.mail", "us.zoom.xos",
        "ru.keepcoder.Telegram", "org.whispersystems.signal-desktop", "com.hnc.Discord",
    },
    "Browsing": {
        "com.brave.Browser", "com.apple.Safari", "com.google.Chrome",
        "company.thebrowser.Browser", "org.mozilla.firefox",
    },
    "Media": {
        "com.spotify.client", "com.apple.Music", "com.apple.TV", "com.apple.Photos",
        "com.apple.podcasts", "com.apple.iBooksX", "com.apple.news",
        "com.apple.QuickTimePlayerX",
    },
    "System": {
        "com.apple.finder", "com.apple.Preview", "com.apple.systempreferences",
        "com.apple.ActivityMonitor", "com.apple.AppStore", "com.apple.Passwords",
        "com.apple.findmy", "com.apple.weather",
    },
}
BUNDLE_CATEGORY = {b: cat for cat, ids in CATEGORIES.items() for b in ids}


def app_name(bundle):
    if bundle in APP_NAMES:
        return APP_NAMES[bundle]
    tail = (bundle or "?").rsplit(".", 1)[-1].replace("-", " ").replace("_", " ")
    return tail.title() if tail.islower() else tail


def fmt_h(minutes):
    return f"{minutes / 60:.1f}h" if minutes >= 60 else f"{minutes:.0f}m"


# ── time helpers ──────────────────────────────────────────────────────────────
def local_offset(ts):
    """This machine's UTC offset (seconds) at a given unix timestamp."""
    return dt.datetime.fromtimestamp(ts).astimezone().utcoffset().total_seconds()


def to_local(ts, off):
    """Unix ts + event GMT offset → naive local datetime (the event's own tz)."""
    return dt.datetime.fromtimestamp(ts, tz=dt.timezone.utc).replace(tzinfo=None) + dt.timedelta(seconds=off)


def logical_date(local_dt):
    return (local_dt - dt.timedelta(hours=BOUNDARY_HOUR)).date()


def split_by_day(start_dt, end_dt):
    """Split a local-time interval at day boundaries → [(logical_date, seg_start, seg_end)]."""
    out = []
    cur = start_dt
    while cur < end_dt:
        day = logical_date(cur)
        boundary = dt.datetime.combine(day + dt.timedelta(days=1), dt.time(BOUNDARY_HOUR))
        seg_end = min(end_dt, boundary)
        out.append((day, cur, seg_end))
        cur = seg_end
    return out


def parse_since(s, today):
    """--since grammar → (start_date, end_date) of logical days, inclusive."""
    s = (s or "7d").strip().lower()
    if s == "all":
        return None, today
    if s == "today":
        return today, today
    if s == "yesterday":
        y = today - dt.timedelta(days=1)
        return y, y
    if s.endswith("d") and s[:-1].isdigit():
        return today - dt.timedelta(days=int(s[:-1]) - 1), today
    if s.endswith("w") and s[:-1].isdigit():
        return today - dt.timedelta(days=int(s[:-1]) * 7 - 1), today
    try:
        return dt.date.fromisoformat(s), today
    except ValueError:
        sys.exit(f"--since: can't parse {s!r} (want 7d / 2w / 30d / today / yesterday / YYYY-MM-DD / all)")


# ── database ──────────────────────────────────────────────────────────────────
FDA_MESSAGE = (
    "knowledgeC.db: access not permitted (macOS Full Disk Access).\n"
    "Grant Full Disk Access to the app running your assistant — Claude, or the terminal it runs in:\n"
    "  System Settings -> Privacy & Security -> Full Disk Access -> enable it -> re-open the app.\n"
    "Until then every read of ~/Library/Application Support/Knowledge/knowledgeC.db is blocked by macOS TCC."
)


def snapshot_and_open(db_path):
    """Copy db (+wal/shm) to a temp dir and open the copy; the live DB is never opened.
    Any permission failure on the copy is the TCC / Full Disk Access gate, reported as
    one clear message."""
    if not db_path.exists():
        sys.exit(f"No Screen Time database at {db_path} — is this a Mac with usage recorded?")
    tmp = Path(tempfile.mkdtemp(prefix="screentime-"))
    try:
        for suffix in ("", "-wal", "-shm"):
            src = Path(str(db_path) + suffix)
            if src.exists():
                shutil.copy2(src, tmp / src.name)
    except (PermissionError, OSError):
        shutil.rmtree(tmp, ignore_errors=True)
        sys.exit(FDA_MESSAGE)
    copy = tmp / db_path.name
    try:
        con = sqlite3.connect(f"file:{copy}?mode=ro", uri=True)
        con.execute("SELECT 1 FROM ZOBJECT LIMIT 1")
    except sqlite3.OperationalError:
        con = sqlite3.connect(str(copy))  # ro open can fail on WAL recovery; the copy is ours
    return con, tmp


def meta_column(con, *needles):
    """Find a ZSTRUCTUREDMETADATA column whose name contains all needles (schema drifts by macOS)."""
    try:
        cols = [r[1] for r in con.execute("PRAGMA table_info(ZSTRUCTUREDMETADATA)")]
    except sqlite3.OperationalError:
        return None
    for c in cols:
        if all(n in c.upper() for n in needles):
            return c
    return None


def load_events(con):
    """Pull the streams we use; convert Cocoa timestamps → unix. Returns dict of row lists."""
    def rows(sql):
        try:
            return con.execute(sql).fetchall()
        except sqlite3.OperationalError:
            return []

    usage = rows(
        "SELECT ZVALUESTRING, ZSTARTDATE+%d, ZENDDATE+%d, ZSECONDSFROMGMT FROM ZOBJECT "
        "WHERE ZSTREAMNAME='/app/usage' AND ZVALUESTRING IS NOT NULL "
        "AND ZENDDATE > ZSTARTDATE" % (COCOA_EPOCH, COCOA_EPOCH)
    )
    unlocks = rows(
        "SELECT ZSTARTDATE+%d, ZSECONDSFROMGMT FROM ZOBJECT "
        "WHERE ZSTREAMNAME='/device/isLocked' AND ZVALUEINTEGER=0" % COCOA_EPOCH
    )
    notif_col = meta_column(con, "NOTIFICATIONUSAGE", "BUNDLE")
    notifs = rows(
        "SELECT o.ZSTARTDATE+%d, o.ZSECONDSFROMGMT, %s FROM ZOBJECT o "
        "LEFT JOIN ZSTRUCTUREDMETADATA m ON o.ZSTRUCTUREDMETADATA = m.Z_PK "
        "WHERE o.ZSTREAMNAME='/notification/usage'"
        % (COCOA_EPOCH, ("m." + notif_col) if notif_col else "NULL")
    )
    web_col = meta_column(con, "WEBUSAGE", "DOMAIN") or meta_column(con, "WEBUSAGE", "URL")
    web = rows(
        "SELECT %s, SUM(o.ZENDDATE - o.ZSTARTDATE)/60.0 FROM ZOBJECT o "
        "LEFT JOIN ZSTRUCTUREDMETADATA m ON o.ZSTRUCTUREDMETADATA = m.Z_PK "
        "WHERE o.ZSTREAMNAME='/app/webUsage' GROUP BY 1 ORDER BY 2 DESC"
        % (("m." + web_col) if web_col else "o.ZVALUESTRING")
    ) if web_col else []
    backlit = rows(
        "SELECT ZSTARTDATE+%d, ZENDDATE+%d, ZSECONDSFROMGMT FROM ZOBJECT "
        "WHERE ZSTREAMNAME='/display/isBacklit' AND ZVALUEINTEGER=1 "
        "AND ZENDDATE > ZSTARTDATE" % (COCOA_EPOCH, COCOA_EPOCH)
    )
    return {"usage": usage, "unlocks": unlocks, "notifs": notifs, "web": web, "backlit": backlit}


# ── aggregation ───────────────────────────────────────────────────────────────
def aggregate(events, now_ts):
    """Bucket every event into logical days. Returns {date: day_record} over all coverage."""
    days = {}

    def day(d):
        return days.setdefault(d, {
            "total_min": 0.0, "apps": {}, "unlocks": 0, "notifications": 0,
            "notif_apps": {}, "first": None, "last": None, "intervals": [],
            "app_intervals": {}, "hourly": [0.0] * 24,
        })

    for bundle, start, end, off in events["usage"]:
        off = off if off is not None else local_offset(start)
        end = min(end, now_ts)
        if end <= start:
            continue
        s_dt, e_dt = to_local(start, off), to_local(end, off)
        for d, seg_s, seg_e in split_by_day(s_dt, e_dt):
            mins = (seg_e - seg_s).total_seconds() / 60
            rec = day(d)
            rec["total_min"] += mins
            rec["apps"][bundle] = rec["apps"].get(bundle, 0.0) + mins
            rec["first"] = min(rec["first"] or seg_s, seg_s)
            rec["last"] = max(rec["last"] or seg_e, seg_e)
            rec["intervals"].append((seg_s, seg_e))
            rec["app_intervals"].setdefault(bundle, []).append((seg_s, seg_e))
            cur = seg_s                                   # allocate minutes to hour buckets
            while cur < seg_e:
                nxt = min(seg_e, cur.replace(minute=0, second=0, microsecond=0) + dt.timedelta(hours=1))
                rec["hourly"][cur.hour] += (nxt - cur).total_seconds() / 60
                cur = nxt

    for ts, off in events["unlocks"]:
        off = off if off is not None else local_offset(ts)
        day(logical_date(to_local(ts, off)))["unlocks"] += 1

    for ts, off, bundle in events["notifs"]:
        off = off if off is not None else local_offset(ts)
        rec = day(logical_date(to_local(ts, off)))
        rec["notifications"] += 1
        if bundle:
            rec["notif_apps"][bundle] = rec["notif_apps"].get(bundle, 0) + 1

    def merge(intervals):
        """Group intervals ≤ GAP_MIN apart (raw events chunk on every focus switch).
        Returns [span_start, span_end, in_use_min] per group — in_use is the union
        of raw chunks, so an app's longest stretch never exceeds its total."""
        groups = []
        for s, e in sorted(intervals):
            if groups and (s - groups[-1][1]).total_seconds() / 60 <= GAP_MIN:
                g = groups[-1]
                g[2] += max(0.0, (e - max(s, g[1])).total_seconds() / 60)
                g[1] = max(g[1], e)
            else:
                groups.append([s, e, (e - s).total_seconds() / 60])
        return groups

    for rec in days.values():
        blocks = merge(rec["intervals"])                  # longest continuous screen block (wall-clock span)
        rec["longest_block_min"] = max(((e - s).total_seconds() / 60 for s, e, _ in blocks), default=0.0)
        rec["sessions"], rec["longest"] = {}, {}          # per-app: merged stretches, not focus flips
        for b, ivs in rec["app_intervals"].items():
            m = merge(ivs)
            rec["sessions"][b] = len(m)
            rec["longest"][b] = max(g[2] for g in m)      # in-app minutes within the longest stretch
        del rec["intervals"], rec["app_intervals"]
    return days


# ── screen-on fallback (/display/isBacklit) ──────────────────────────────────
def aggregate_screen_on(backlit, now_ts):
    """Bucket display-lit intervals into logical days → {date: rec}.

    The fallback metric for when /app/usage is dead: wall-clock minutes with
    the display lit. No per-app data; slightly generous vs app-usage (idle
    with the display lit counts). NOTE: the *running* lit block is only
    donated when the display next sleeps, so "today so far" lags by the
    current block.
    """
    days = {}
    for start, end, off in backlit:
        off = off if off is not None else local_offset(start)
        end = min(end, now_ts)
        if end <= start:
            continue
        for d, seg_s, seg_e in split_by_day(to_local(start, off), to_local(end, off)):
            days.setdefault(d, []).append((seg_s, seg_e))

    out = {}
    for d, ivs in sorted(days.items()):
        union = []                                # lit intervals shouldn't overlap; union to be safe
        for s, e in sorted(ivs):
            if union and s <= union[-1][1]:
                union[-1][1] = max(union[-1][1], e)
            else:
                union.append([s, e])
        blocks = []                               # bridge gaps ≤ GAP_MIN, like app-usage blocks
        for s, e in union:
            if blocks and (s - blocks[-1][1]).total_seconds() / 60 <= GAP_MIN:
                blocks[-1][1] = max(blocks[-1][1], e)
            else:
                blocks.append([s, e])
        hourly = [0.0] * 24
        for s, e in union:
            cur = s
            while cur < e:
                nxt = min(e, cur.replace(minute=0, second=0, microsecond=0) + dt.timedelta(hours=1))
                hourly[cur.hour] += (nxt - cur).total_seconds() / 60
                cur = nxt
        out[d] = {
            "weekday": d.strftime("%a"),
            "total_min": sum((e - s).total_seconds() / 60 for s, e in union),
            "blocks": blocks,
            "longest_block_min": max(((e - s).total_seconds() / 60 for s, e in blocks), default=0.0),
            "first": union[0][0],
            "last": union[-1][1],
            "after_midnight_min": sum(hourly[h] for h in range(BOUNDARY_HOUR)),
            "hourly": hourly,
        }
    return out


def app_usage_state(cov, screen_days):
    """Judge the /app/usage stream: 'ok', or dead — the false-zero guard.

    Dead is judged against the OTHER stream, not the calendar: display-lit
    days continuing more than a day past the last app-usage event mean the
    stream stopped, while both streams quiet together just means the laptop
    stayed shut. Returns (status, ok, usage_latest_logical_day_or_None).
    """
    latest_ts = (cov[1] + COCOA_EPOCH) if cov and cov[1] else None
    latest_dt = dt.datetime.fromtimestamp(latest_ts) if latest_ts else None
    latest_day = logical_date(latest_dt) if latest_dt else None
    lit_latest = max(screen_days) if screen_days else None

    if latest_day is None:
        if lit_latest is not None:
            hist_days = load_history().get("days", {})
            tail = f"history.json ends {max(hist_days)}" if hist_days else "no saved history"
            return (f"dead since unknown (0 /app/usage rows in knowledgeC — all aged out; {tail})",
                    False, None)
        return ("no data (0 rows in /app/usage and /display/isBacklit)", False, None)
    if lit_latest is not None and (lit_latest - latest_day).days > 1:
        return (f"dead since {latest_dt:%Y-%m-%d %H:%M} (display-lit days continue after it)",
                False, latest_day)
    return ("ok", True, latest_day)


# ── history (run-over-run retention beyond knowledgeC's ~4-week window) ──────
def load_history():
    if not HISTORY.exists():
        return {"boundary_hour": BOUNDARY_HOUR, "days": {}}
    try:
        return json.loads(HISTORY.read_text())
    except Exception:
        return {"boundary_hour": BOUNDARY_HOUR, "days": {}}


def save_history(days, today):
    hist = load_history()
    saved = 0
    for d, rec in days.items():
        if rec["total_min"] <= 0:
            continue                                     # zero-usage days are never upserted — with a dead
                                                         # /app/usage stream they're manufactured by mere
                                                         # notification events; a gap is honest, a zero is a
                                                         # claim about the owner's day the data can't support
        key, partial = d.isoformat(), d >= today
        prior = hist["days"].get(key)
        if prior and not prior.get("partial") and partial:
            continue                                     # never overwrite a final day with a partial
        top = dict(sorted(rec["apps"].items(), key=lambda kv: -kv[1])[:20])
        hist["days"][key] = {
            "total_min": round(rec["total_min"], 1),
            "unlocks": rec["unlocks"],
            "notifications": rec["notifications"],
            "first": rec["first"].strftime("%H:%M") if rec["first"] else None,
            "last": rec["last"].strftime("%H:%M") if rec["last"] else None,
            "longest_block_min": round(rec["longest_block_min"], 1),
            "apps": {b: round(m, 1) for b, m in top.items()},
            "partial": partial,
        }
        saved += 1
    if not saved:
        return 0, len(hist["days"])                      # nothing to upsert → don't touch the file
    hist["days"] = dict(sorted(hist["days"].items()))
    hist["updated"] = dt.datetime.now().isoformat(timespec="seconds")
    hist["host"] = socket.gethostname()
    _atomic_write(HISTORY, hist)
    return saved, len(hist["days"])


def _atomic_write(path, obj):
    path.parent.mkdir(parents=True, exist_ok=True)       # first run creates Agent/scripts/screentime/
    tmp = path.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(obj, indent=1) + "\n")
    os.replace(tmp, path)


def screen_on_header():
    """Header stamped into a new history-screen-on.json (an existing file's header
    is preserved as-is)."""
    return {
        "metric": ("screen-on minutes from /display/isBacklit (ZVALUEINTEGER=1 intervals) — "
                   "FALLBACK series, not /app/usage"),
        "why": ("knowledgeC's /app/usage stream can silently stop being donated; once its rows age "
                "out (~4 weeks) the app-usage series in history.json honestly ends, and this series "
                "carries the timeline. Screen-on tracks app-usage within a few percent where the "
                "streams overlap, slightly generous (counts idle-with-display-lit)."),
        "semantics": {
            "boundary_hour": BOUNDARY_HOUR,
            "gap_merge_min": GAP_MIN,
            "hourly_index": "local hour 0–23 of the event's own GMT offset",
            "after_midnight_min": f"minutes in 00:00–{BOUNDARY_HOUR:02d}:00, credited to the evening's logical day",
            "partial": "logical day not over at write time — a later run may overwrite",
        },
        "no_per_app_data": True,
    }


def load_history_screen_on():
    if not HISTORY_SCREEN_ON.exists():
        return dict(screen_on_header(), days={})
    try:
        return json.loads(HISTORY_SCREEN_ON.read_text())
    except Exception:
        return dict(screen_on_header(), days={})


def save_history_screen_on(screen_days, today):
    """Upsert screen-on days into the fallback series — same rules as app-usage:
    zero days are never written, a final day is never overwritten by a partial,
    and the file is untouched when there is nothing to upsert. The header
    (metric / why / semantics) of an existing file is preserved as-is."""
    hist = load_history_screen_on()
    hist.setdefault("days", {})
    saved = 0
    for d, rec in screen_days.items():
        if rec["total_min"] <= 0:
            continue
        key, partial = d.isoformat(), d >= today
        prior = hist["days"].get(key)
        if prior and not prior.get("partial") and partial:
            continue                                     # never overwrite a final day with a partial
        hist["days"][key] = {
            "weekday": rec["weekday"],
            "total_min": round(rec["total_min"], 1),
            "blocks": len(rec["blocks"]),
            "longest_block_min": round(rec["longest_block_min"], 1),
            "first": rec["first"].strftime("%H:%M"),
            "last": rec["last"].strftime("%H:%M"),
            "after_midnight_min": round(rec["after_midnight_min"], 1),
            "hourly": [round(x) for x in rec["hourly"]],
            "blocks_detail": [[s.strftime("%H:%M"), e.strftime("%H:%M"),
                               round((e - s).total_seconds() / 60)] for s, e in rec["blocks"]],
            "partial": partial,
        }
        saved += 1
    if not saved:
        return 0, len(hist["days"])
    hist["days"] = dict(sorted(hist["days"].items()))
    hist["updated"] = dt.datetime.now().isoformat(timespec="seconds")
    hist["host"] = socket.gethostname()
    _atomic_write(HISTORY_SCREEN_ON, hist)
    return saved, len(hist["days"])


# ── output ────────────────────────────────────────────────────────────────────
SPARK = "▁▂▃▄▅▆▇█"


def sparkline(values):
    top = max(values) or 1
    return "".join(SPARK[min(int(v / top * (len(SPARK) - 1) + 0.5), len(SPARK) - 1)] if v else "·" for v in values)


def print_human(report, top_n):
    w = report["window"]
    print(f"Screen Time — {w['start']} → {w['end']} (days end {BOUNDARY_HOUR:02d}:00) · {report['host']}")
    status = report.get("app_usage_status", "ok")
    if status != "ok":
        print(f"⚠ /app/usage: {status}")
    if report.get("headline_metric") == "screen-on":
        print_screen_on_human(report)
        return
    print(f"Coverage: knowledgeC since {report['coverage']['earliest']} (~4-week rolling window)\n")

    days = report["days"]
    if not days:
        print("No usage recorded in this window.")
        return
    peak = max(d["total_min"] for d in days) or 1
    for d in days:
        bar = "█" * int(d["total_min"] / peak * 18 + 0.5)
        span = f"{d['first']}→{d['last']}" if d["first"] else "—"
        tag = "  (so far)" if d["partial"] else ""
        print(f"  {d['weekday']} {d['date'][5:]}  {fmt_h(d['total_min']):>6}  {bar:<18}  "
              f"{span:<12} longest block {fmt_h(d['longest_block_min'])}{tag}")

    total = sum(d["total_min"] for d in days)
    full = [d for d in days if not d["partial"]]
    avg = (sum(d["total_min"] for d in full) / len(full)) if full else total
    print(f"\n  Total {fmt_h(total)} · avg {fmt_h(avg)}/day over {len(full)} full day(s)")

    apps = report["apps"][:top_n]
    if apps:
        print("\nTop apps:")
        for a in apps:
            share = a["minutes"] / total * 100 if total else 0
            print(f"  {fmt_h(a['minutes']):>6}  {a['name']:<18} {share:4.0f}%   "
                  f"{a['sessions']} sessions · longest {fmt_h(a['longest_min'])}")
    cats = " · ".join(f"{c['category']} {fmt_h(c['minutes'])} ({c['share']:.0f}%)" for c in report["categories"])
    print(f"\nCategories:  {cats}")
    print(f"Hourly:      {sparkline(report['hourly_min'])}  (00→23, minutes on screen)")
    if report.get("notifications", {}).get("apps"):
        top_notif = sorted(report["notifications"]["apps"].items(), key=lambda kv: -kv[1])[:5]
        line = " · ".join(f"{app_name(b)} {n}" for b, n in top_notif)
        print(f"Notifs:      {report['notifications']['total']} ({line})")
    unl = sum(d["unlocks"] for d in days)
    if unl:                                               # modern macOS logs /device/isLocked sporadically
        print(f"Unlocks:     {unl} (OS logs these sporadically — treat as a floor, not a count)")
    if report.get("history"):
        h = report["history"]
        print(f"\nHistory: {h['saved']} day(s) upserted → {HISTORY.name} ({h['total_days']} days retained)")


def print_screen_on_human(report):
    """Fallback headline: screen-on (display lit) — no per-app breakdown."""
    so = report["screen_on"]
    cov = so["coverage"]
    print("Headline metric: SCREEN-ON (display lit) — no per-app breakdown; slightly generous vs app-usage.")
    print(f"Coverage: /display/isBacklit {cov['earliest']} → {cov['latest']} (~4-week rolling window)\n")
    days = so["days"]
    if not days:
        print("No screen-on time recorded in this window.")
        return
    peak = max(d["total_min"] for d in days) or 1
    for d in days:
        bar = "█" * int(d["total_min"] / peak * 18 + 0.5)
        span = f"{d['first']}→{d['last']}"
        tag = "  (so far)" if d["partial"] else ""
        am = f" · past-midnight {d['after_midnight_min']:.0f}m" if d["after_midnight_min"] >= 1 else ""
        print(f"  {d['weekday']} {d['date'][5:]}  {fmt_h(d['total_min']):>6}  {bar:<18}  "
              f"{span:<12} {d['blocks']} block(s) · longest {fmt_h(d['longest_block_min'])}{am}{tag}")
    total = sum(d["total_min"] for d in days)
    full = [d for d in days if not d["partial"]]
    avg = (sum(d["total_min"] for d in full) / len(full)) if full else total
    print(f"\n  Total {fmt_h(total)} · avg {fmt_h(avg)}/day over {len(full)} full day(s)")
    print(f"\nHourly:      {sparkline(so['hourly_min'])}  (00→23, minutes display-lit)")
    late = [d for d in days if d["after_midnight_min"] >= 1]
    if late:
        line = " · ".join(f"{d['weekday']} {d['date'][5:]} {d['after_midnight_min']:.0f}m" for d in late[-6:])
        print(f"Late nights: {len(late)} day(s) past midnight ({line})")
    print("Caveats:     idle-with-display-lit counts; the running lit block appears only once the display "
          "next sleeps; a slow machine clock can mis-bucket late-night days.")
    if so.get("history"):
        h = so["history"]
        line = f"\nHistory: {h['saved']} day(s) upserted → {HISTORY_SCREEN_ON.name} ({h['total_days']} days retained)"
        ah = report.get("history")
        if ah and ah.get("saved"):
            line += f" · {ah['saved']} app-usage day(s) → {HISTORY.name}"
        else:
            line += f" · {HISTORY.name} untouched"
        print(line)


def print_history():
    hist = load_history()
    days = hist.get("days", {})
    if not days:
        print(f"No app-usage history yet — run the reader once to seed {HISTORY}.")
    else:
        print(f"App-usage daily history — {len(days)} day(s) · {HISTORY}\n")
        for key, rec in days.items():
            top = max(rec.get("apps", {}).items(), key=lambda kv: kv[1], default=(None, 0))
            top_s = f"top {app_name(top[0])} {fmt_h(top[1])}" if top[0] else ""
            tag = "  (partial)" if rec.get("partial") else ""
            print(f"  {key}  {fmt_h(rec['total_min']):>6}  block {fmt_h(rec['longest_block_min']):>5}  {top_s}{tag}")

    shist = load_history_screen_on()
    sdays = shist.get("days", {})
    print()
    if not sdays:
        print(f"No screen-on fallback history yet — {HISTORY_SCREEN_ON.name} is empty.")
        return
    print(f"Screen-on fallback daily history (display lit, no per-app data) — {len(sdays)} day(s) · {HISTORY_SCREEN_ON}\n")
    for key, rec in sdays.items():
        tag = "  (partial)" if rec.get("partial") else ""
        am = f"  past-midnight {rec['after_midnight_min']:.0f}m" if rec.get("after_midnight_min", 0) >= 1 else ""
        print(f"  {key}  {fmt_h(rec['total_min']):>6}  block {fmt_h(rec['longest_block_min']):>5}  "
              f"{rec['first']}→{rec['last']}{am}{tag}")


# ── coverage reporting (stderr, so --json stdout stays parseable) ────────────
def warn(msg):
    print("screentime_read: %s" % msg, file=sys.stderr)


def report_coverage(report):
    """State what the run actually covered — window, days with data, and the
    /app/usage stream status — so a dead stream can never pass for a quiet week."""
    w = report["window"]
    fallback = report["headline_metric"] == "screen-on"
    days = report["screen_on"]["days"] if fallback else report["days"]
    n = sum(1 for d in days if d["total_min"] > 0)
    warn("window %s → %s · %d day(s) with data · headline %s · /app/usage %s"
         % (w["start"], w["end"], n, report["headline_metric"], report["app_usage_status"]))
    if fallback:
        warn("FALLBACK — totals are screen-on (display lit): no per-app breakdown, "
             "slightly generous vs app-usage.")


# ── main ──────────────────────────────────────────────────────────────────────
def main():
    ap = argparse.ArgumentParser(description="Read macOS Screen Time (knowledgeC.db), read-only.")
    ap.add_argument("--since", default=DEFAULT_SINCE,
                    help="7d / 2w / 30d / today / yesterday / YYYY-MM-DD / all (default %s)" % DEFAULT_SINCE)
    ap.add_argument("--json", action="store_true", help="machine-readable output")
    ap.add_argument("--history", action="store_true",
                    help="print both saved series (app-usage + screen-on fallback) and exit")
    ap.add_argument("--no-save", action="store_true", help="don't upsert the history files this run")
    ap.add_argument("--top", type=int, default=8, help="apps to list in the human summary (default 8)")
    ap.add_argument("--db", type=Path, default=DB_PATH, help="override knowledgeC.db path")
    args = ap.parse_args()

    if args.history:
        if args.json:
            print(json.dumps({"app_usage": load_history(), "screen_on": load_history_screen_on()}, indent=1))
        else:
            print_history()
        return

    now_ts = dt.datetime.now().timestamp()
    today = logical_date(dt.datetime.now())
    start, end = parse_since(args.since, today)

    con, tmp = snapshot_and_open(args.db)
    try:
        events = load_events(con)
        cov = con.execute(
            "SELECT MIN(ZSTARTDATE), MAX(ZENDDATE) FROM ZOBJECT WHERE ZSTREAMNAME='/app/usage'"
        ).fetchone()
        cov_so = con.execute(
            "SELECT MIN(ZSTARTDATE), MAX(ZENDDATE) FROM ZOBJECT "
            "WHERE ZSTREAMNAME='/display/isBacklit' AND ZVALUEINTEGER=1"
        ).fetchone()
    finally:
        con.close()
        shutil.rmtree(tmp, ignore_errors=True)

    all_days = aggregate(events, now_ts)
    screen_all = aggregate_screen_on(events["backlit"], now_ts)
    if start is None:
        known = list(all_days) + list(screen_all)
        start = min(known) if known else today

    status, usage_ok, usage_latest_day = app_usage_state(cov, screen_all)
    effective_end = min(end, today)
    headline = "app-usage" if usage_ok or (usage_latest_day and effective_end <= usage_latest_day) \
        else "screen-on"

    hist_info = hist_info_so = None
    if not args.no_save:
        saved, total_days = save_history(all_days, today)
        hist_info = {"saved": saved, "total_days": total_days, "path": str(HISTORY)}
        saved_so, total_so = save_history_screen_on(screen_all, today)
        hist_info_so = {"saved": saved_so, "total_days": total_so, "path": str(HISTORY_SCREEN_ON)}

    window_days_all = {d: r for d, r in sorted(all_days.items()) if start <= d <= end}
    window_days = window_days_all if usage_ok else \
        {d: r for d, r in window_days_all.items() if r["total_min"] > 0}   # a dead stream manufactures
                                                                           # zero-days from notifications
    window_so = {d: r for d, r in sorted(screen_all.items()) if start <= d <= end}
    apps, cats = {}, {}
    hourly = [0.0] * 24
    notif_total, notif_apps = 0, {}
    for d, rec in window_days_all.items():
        for b, m in rec["apps"].items():
            a = apps.setdefault(b, {"minutes": 0.0, "sessions": 0, "longest_min": 0.0, "per_day_min": {}})
            a["minutes"] += m
            a["sessions"] += rec["sessions"][b]
            a["longest_min"] = max(a["longest_min"], rec["longest"][b])
            a["per_day_min"][d.isoformat()] = round(m, 1)
            cats[BUNDLE_CATEGORY.get(b, "Other")] = cats.get(BUNDLE_CATEGORY.get(b, "Other"), 0.0) + m
        hourly = [h + rh for h, rh in zip(hourly, rec["hourly"])]
        notif_total += rec["notifications"]
        for b, n in rec["notif_apps"].items():
            notif_apps[b] = notif_apps.get(b, 0) + n

    total_min = sum(r["total_min"] for r in window_days.values())
    report = {
        "generated": dt.datetime.now().isoformat(timespec="seconds"),
        "host": socket.gethostname(),
        "db": str(args.db),
        "app_usage_status": status,
        "headline_metric": headline,
        "coverage": {
            "earliest": dt.datetime.fromtimestamp(cov[0] + COCOA_EPOCH).strftime("%Y-%m-%d %H:%M") if cov[0] else None,
            "latest": dt.datetime.fromtimestamp(cov[1] + COCOA_EPOCH).strftime("%Y-%m-%d %H:%M") if cov[1] else None,
        },
        "window": {"start": start.isoformat(), "end": end.isoformat(), "boundary_hour": BOUNDARY_HOUR},
        "days": [
            {
                "date": d.isoformat(),
                "weekday": d.strftime("%a"),
                "total_min": round(rec["total_min"], 1),
                "first": rec["first"].strftime("%H:%M") if rec["first"] else None,
                "last": rec["last"].strftime("%H:%M") if rec["last"] else None,
                "unlocks": rec["unlocks"],
                "notifications": rec["notifications"],
                "longest_block_min": round(rec["longest_block_min"], 1),
                "partial": d >= today,
                "apps": {b: round(m, 1) for b, m in sorted(rec["apps"].items(), key=lambda kv: -kv[1])},
            }
            for d, rec in window_days.items()
        ],
        "apps": sorted(
            [
                {"bundle": b, "name": app_name(b), "category": BUNDLE_CATEGORY.get(b, "Other"),
                 "minutes": round(a["minutes"], 1), "sessions": a["sessions"],
                 "longest_min": round(a["longest_min"], 1), "per_day_min": a["per_day_min"]}
                for b, a in apps.items()
            ],
            key=lambda a: -a["minutes"],
        )[:40],
        "categories": sorted(
            [{"category": c, "minutes": round(m, 1), "share": round(m / total_min * 100, 1) if total_min else 0}
             for c, m in cats.items()],
            key=lambda c: -c["minutes"],
        ),
        "hourly_min": [round(h, 1) for h in hourly],
        "notifications": {"total": notif_total, "apps": notif_apps},
        "web_domains": [{"domain": d, "minutes": round(m, 1)} for d, m in events["web"] if d and m and m >= 1],
        "screen_on": {
            "label": "screen-on (display lit) — no per-app breakdown; slightly generous vs app-usage",
            "coverage": {
                "earliest": dt.datetime.fromtimestamp(cov_so[0] + COCOA_EPOCH).strftime("%Y-%m-%d %H:%M") if cov_so[0] else None,
                "latest": dt.datetime.fromtimestamp(cov_so[1] + COCOA_EPOCH).strftime("%Y-%m-%d %H:%M") if cov_so[1] else None,
            },
            "days": [
                {
                    "date": d.isoformat(),
                    "weekday": rec["weekday"],
                    "total_min": round(rec["total_min"], 1),
                    "blocks": len(rec["blocks"]),
                    "longest_block_min": round(rec["longest_block_min"], 1),
                    "first": rec["first"].strftime("%H:%M"),
                    "last": rec["last"].strftime("%H:%M"),
                    "after_midnight_min": round(rec["after_midnight_min"], 1),
                    "hourly_min": [round(h, 1) for h in rec["hourly"]],
                    "blocks_detail": [[s.strftime("%H:%M"), e.strftime("%H:%M"),
                                       round((e - s).total_seconds() / 60)] for s, e in rec["blocks"]],
                    "partial": d >= today,
                }
                for d, rec in window_so.items()
            ],
            "total_min": round(sum(r["total_min"] for r in window_so.values()), 1),
            "hourly_min": [round(sum(r["hourly"][h] for r in window_so.values()), 1) for h in range(24)],
            "history": hist_info_so,
        },
        "history": hist_info,
    }

    if args.json:
        print(json.dumps(report, indent=1))
    else:
        print_human(report, args.top)
    report_coverage(report)


if __name__ == "__main__":
    main()
