#!/usr/bin/env python3
"""
sweep.py — the hourly vault agent.

One launchd job runs this every hour. It computes the owner's *local* time from the
travel schedule (location.py), then runs whichever tasks are due:
  - daily_digest  : the two 9am-local vault digests (email) — a Acme-only brief
                    to the Acme inbox and an everything-but-Acme brief to the
                    Acme Summer inbox, both off the same digest.py engine.
  - team_digests : per-member briefs (email) for the owner's team — each person's delegated
                    tasks, the follow-ups the owner's waiting on, and an optional area they can
                    help with (team_digest.py). The roster (who, scope, timezone, send
                    hour, reply on/off) lives in the vault note Agent/Team Digests.md; each
                    member fires on THEIR OWN tz + hour, not the owner's.
  - team_inbox   : the PROTECTED reverse path — reads team members' email REPLIES to those
                    briefs and applies their progress to THEIR items only (append a note /
                    mark done / flag blocked), confirms back, and rolls the update into
                    the owner's next digest (team_inbox.py). Untrusted input behind
                    identity/auth/thread gates + a per-member allowlist + a deterministic
                    applier; off unless TEAM_INBOX_ENABLE=1.
  - vault_hygiene : a weekly hygiene/triage report (email)

Safety (the "full auto-cleanup" posture, with rails):
  - SWEEP_SHADOW=1 (default): never writes to the vault; logs/reports what it would do.
  - There are currently no autonomous vault writes (the daily-note scaffold was
    retired with Daily/). Judgment cleanups are reported, not executed. Nothing is
    ever hard-deleted.
  - Kill switch: env SWEEP_DISABLE=1, or a file ~/Obsidian/secrets/SWEEP_OFF.
  - Every task is isolated in try/except so one failure can't stop the others.
  - All writes go through the working tree -> vault-sync commits -> git history = undo.

Flags: --plan (show location + which tasks are due; no side effects)
       --once (run all due tasks now)   [default]
Test hooks: SWEEP_FAKE_LOCAL=ISO8601 forces "now" (naive = local).
"""
import os
import re
import sys
import json
import html as _html
import tempfile
import unicodedata
import datetime as dt
from collections import defaultdict
from pathlib import Path

from alexandria_vault.engine import consume_vault_arg, load_fresh
from alexandria_vault.models import AnthropicProvider
from alexandria_vault.vault_core import analyze_links, fs, note_index

# `--vault` must land in VAULT_DIR before digest.py's import-time locator runs.
consume_vault_arg()

SHADOW   = os.environ.get("SWEEP_SHADOW", "1") not in ("0", "false", "")
AUTOFIX  = os.environ.get("SWEEP_AUTOFIX", "1") not in ("0", "false", "")
# Todo/hygiene auto-fix knobs.
ARCHIVE_AFTER_DAYS = int(os.environ.get("ARCHIVE_AFTER_DAYS", "0"))   # 0 = archive [x] immediately
DEFAULT_AUTHOR     = os.environ.get("SWEEP_DEFAULT_AUTHOR", "claude") # value backfilled into missing author:
STALE_DAYS_REPORT  = int(os.environ.get("SWEEP_STALE_DAYS", "21"))    # undated open item older than this → stale
# Vault-wide scans skip Git/Obsidian internals, the Archive, and the script tree.
_SKIP_DIRS = ("/Archive/", "/.git/", "/.obsidian/", "/Agent/scripts/")
DIGEST_HOUR = int(os.environ.get("DIGEST_HOUR", "9"))
# Team-member briefs fire on each member's OWN timezone + hour (from the Team Digests
# roster note), independent of where the owner is — unlike the entity-split digests above,
# which track the owner's local 9am.
# How many hours past DIGEST_HOUR the digest may still go out (catch-up window for a
# run that was missed all morning — host asleep / transient send failure). Past it,
# the digest waits for the next local day instead of mailing a "morning" brief at night.
DIGEST_CATCHUP_HOURS = int(os.environ.get("DIGEST_CATCHUP_HOURS", "6"))
# The digest is split by entity into scoped sends — one per `owner_digests` entry,
# falling back to a single $DIGEST_TO digest on a bare vault. That split (filters,
# recipients, scope, the "the agent" sender) is defined ONCE in digest.py.digest_specs()
# and shared with the standalone digest.py entry point; task_daily_digest just adds
# the 9am once/day rails.
STATE_PATH = Path(os.environ.get(
    "SWEEP_STATE", str(Path.home() / "Obsidian" / "secrets" / "sweep-state.json")))
# The "is it alive?" heartbeat — last run time + per-task outcomes — beside the state file.
HEARTBEAT_PATH = STATE_PATH.parent / "runner-status.json"
# Where the server-macos launchd jobs write their logs (see install-launchd's __LOGDIR__);
# capped below so an always-on host can't fill its disk over months.
LOG_DIR = Path.home() / "Library" / "Logs" / "alexandria"
LOG_CAP_BYTES = 5 * 1024 * 1024


def log(msg):
    print(f"[{dt.datetime.now().isoformat(timespec='seconds')}] {msg}", flush=True)


def _load_digest():
    """Import the tested digest module (engine.digest) for its vault parser +
    email/Claude helpers, so there's one source of truth. Re-loaded fresh each
    call so DIGEST_TODAY/VAULT_DIR pinned since the last load are re-frozen."""
    return load_fresh("digest")


def _load_team_digest():
    """Import engine.team_digest (the roster-driven per-member morning briefs).
    Mirrors _load_digest; team_digest itself reuses digest.py."""
    return load_fresh("team_digest")


def _load_team_inbox():
    """Import engine.team_inbox (the PROTECTED processor for team members' email
    replies). Self-contained; reuses digest.py + team_state.py."""
    return load_fresh("team_inbox")


def _load_will_inbox():
    """Import engine.principal_inbox (the PRINCIPAL/higher-trust processor for the
    owner's own emails). Self-contained; reuses digest.py + team_state.py + team_inbox.
    (The module was renamed from will_inbox.py → principal_inbox.py; the loader/task keep the
    historical name.)"""
    return load_fresh("principal_inbox")


def _load_attention_digest():
    """Import engine.attention_digest (the automatic email->todo engine).
    Self-contained; reuses digest.py + email_body.py. Mirrors _load_will_inbox."""
    return load_fresh("attention_digest")


def _module_enabled(vault, module_id):
    """Config-driven module gate: is `module_id` in the vault's modules.enabled?

    Replaces the old "engine file rendered beside sweep = module on" check — the
    engine no longer renders into the vault (it always exists app-side), so file
    presence can't distinguish an enabled module any more. Fails closed: no config
    (or an unreadable one) means no optional-module task fires."""
    try:
        cfg = json.loads((Path(vault) / "Agent" / "config.json").read_text(encoding="utf-8"))
        return module_id in (((cfg.get("modules") or {}).get("enabled")) or [])
    except Exception:
        return False


def _now_in_tz(tzname):
    """Real wall-clock time in `tzname` (a team member may not be where the owner is). The
    hourly sweep runs on the owner's resolved local clock; each member's brief is gated on
    their own zone instead. TEAM_FAKE_NOW (naive ISO8601) overrides for tests."""
    fake = os.environ.get("TEAM_FAKE_NOW")
    if fake:
        return dt.datetime.fromisoformat(fake)
    from zoneinfo import ZoneInfo
    return dt.datetime.now(dt.timezone.utc).astimezone(ZoneInfo(tzname)).replace(tzinfo=None)


def load_state():
    try:
        return json.loads(STATE_PATH.read_text())
    except Exception:
        return {}


def save_state(state):
    # Cadence/dedup bookkeeping lives off-repo in ~/Obsidian/secrets/, so it is NOT a
    # vault write — persist it ALWAYS, even in shadow mode. Shadow gates *vault* writes
    # (enforced at each task's write site), not the agent's own state, and emails DO
    # send in shadow. If this flag didn't persist, every once-per-day task would think
    # it had never run and re-fire on every hourly sweep — that was the duplicate-digest
    # bug (sent 10am, again 11am, …).
    try:
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        _atomic_write(STATE_PATH, json.dumps(state, indent=2))   # temp+rename: never a half-written state
    except Exception as e:
        log(f"state save failed: {e}")


def write_heartbeat(hb):
    """Persist runner-status.json next to the state file — the "is it alive?" signal.
    Written every real sweep so `alexandria doctor` / a health check can see the last run,
    its per-task outcomes, and any errors. Unlike the launchd /tmp logs, this survives."""
    try:
        HEARTBEAT_PATH.parent.mkdir(parents=True, exist_ok=True)
        _atomic_write(HEARTBEAT_PATH, json.dumps(hb, indent=2))
    except Exception as e:
        log(f"heartbeat write failed: {e}")


def cap_logs():
    """Keep the launchd logs (append-only files launchd holds open) from growing without
    bound. When one exceeds the cap, copy its tail to <log>.1 and truncate it in place — an
    O_APPEND writer (the always-running listener) keeps appending from 0, so the inode stays
    valid. No-op where those logs don't exist (e.g. the GitHub-Actions runner)."""
    try:
        for f in LOG_DIR.glob("*.log"):
            if f.stat().st_size <= LOG_CAP_BYTES:
                continue
            (f.parent / (f.name + ".1")).write_bytes(f.read_bytes()[-(LOG_CAP_BYTES // 2):])
            os.truncate(f, 0)
            log(f"rotated {f.name} (exceeded {LOG_CAP_BYTES // (1024 * 1024)}MB)")
    except Exception as e:
        log(f"log cap failed: {e}")


def _check_env():
    """Surface a missing core credential up front (a clear log line) instead of letting the
    digest fail cryptically mid-run. Doesn't abort — the non-digest sweep work still runs and
    the per-task error is captured in the heartbeat for the health check to alert on."""
    missing = [v for v in ("ANTHROPIC_API_KEY", "SMTP_HOST", "SMTP_PASS") if not os.environ.get(v)]
    if missing:
        log(f"MISSING ENV: {', '.join(missing)} — the digest can't send until these are set "
            "(run: alexandria secrets --set).")


def killed():
    if os.environ.get("SWEEP_DISABLE", "0") in ("1", "true"):
        return True
    return (Path.home() / "Obsidian" / "secrets" / "SWEEP_OFF").exists()


# ---- cadence helpers -------------------------------------------------------
def once_today(state, key, now):
    return state.get(key) != now.date().isoformat()


def mark_today(state, key, now):
    state[key] = now.date().isoformat()


def resolve_location(state, vault):
    """Resolve the owner's location/tz from the travel schedule on EVERY run, anchored to the
    local date *wherever he is* (location.resolve tests each segment against that
    segment's own timezone), NOT the host's the home timezone clock.

    This replaced an earlier once-per-CR-day cache, which silently broke on big eastward
    jumps: keyed on the host's the home timezone date, it kept resolving the OLD zone until CR
    midnight rolled over — so on the New York→Manila move the agent stayed on Eastern
    (~12h behind Manila) and fired the '9am' digest at ~9pm Manila. Re-parsing one
    schedule note hourly is cheap, and resolving fresh flips the zone the instant it
    becomes a new day where the owner actually is, so the 9am-local digest lands on his
    morning. The cache block is kept only for visibility/debugging in sweep-state.json."""
    from alexandria_vault.engine import location
    loc = location.resolve(vault)
    state["location:date"] = dt.date.today().isoformat()   # host clock (CR), for reference
    state["location:cache"] = {"location": loc["location"], "tz": loc["tz"],
                               "confident": loc["confident"],
                               "tz_source": loc.get("tz_source"),
                               "lat": loc.get("lat"), "lon": loc.get("lon")}
    loc["cached"] = False
    return loc


# ---- tasks -----------------------------------------------------------------
def _render_habits(habits):
    if not habits:
        return "_No daily habits configured (edit Daily/Daily.md)._"
    return "\n".join(f"- [ ] {h}" for h in habits)


def _render_workout(workout):
    if not workout:
        return "_Rotation hasn't started yet._"
    out = [f"**Day {workout['day']} — {workout['session']}**"]
    rows = workout.get("exercises") or []
    if rows:
        out += ["", "| Exercise | Sets × Reps |", "|---|---|"]
        out += [f"| {name} | {sr} |" for name, sr in rows]
    elif workout.get("prose"):
        out += ["", workout["prose"]]
    elif workout.get("is_rest"):
        out += ["", "Rest day."]
    return "\n".join(out)


def _build_daily_note(day, weekday, habits, workout):
    return (
        "---\n"
        "type: daily-note\n"
        f"date: {day.isoformat()}\n"
        f"author: {DEFAULT_AUTHOR}\n"
        "tags: [daily]\n"
        "---\n\n"
        f"# {day.isoformat()} {weekday}\n\n"
        "## Priorities\n\n\n"
        "## On my mind\n\n\n"
        f"## Daily Habits\n\n{_render_habits(habits)}\n\n"
        f"## Workout\n\n{_render_workout(workout)}\n\n"
        "## Debrief\n"
    )


def _ensure_daily_sections(text, habits, workout):
    """Add only the routine sections (Daily Habits, Workout) that are missing —
    never touch Priorities / On my mind / Debrief or existing checkmarks."""
    add = []
    if not re.search(r"(?im)^##\s+Daily Habits\b", text):
        add.append(f"## Daily Habits\n\n{_render_habits(habits)}\n")
    if not re.search(r"(?im)^##\s+Workout\b", text):
        add.append(f"## Workout\n\n{_render_workout(workout)}\n")
    if not add:
        return text
    block = "\n".join(add)
    m = re.search(r"(?im)^##\s+Debrief\b", text)
    if m:                                  # keep routine sections above the Debrief
        return text[:m.start()] + block + "\n" + text[m.start():]
    return text.rstrip("\n") + "\n\n" + block


def _ensure_location(text, location, tz):
    """Stamp the day's location + timezone so the note is a canonical daily record:
    `location:`/`timezone:` in frontmatter (queryable) + a visible 📍 line under the
    title. Idempotent — adds each only if absent; no-op when location is unknown."""
    if not location:
        return text
    tz = tz or ""
    if re.search(r"(?m)^location:", text) is None:
        text = re.sub(r"(?m)^(date:.*\n)",
                      lambda m: f"{m.group(1)}location: {location}\ntimezone: {tz}\n",
                      text, count=1)
    if "📍" not in text:
        stamp = f"\n📍 {location}" + (f" · {tz}" if tz else "") + "\n"
        text = re.sub(r"(?m)^(#\s.*\n)", lambda m: f"{m.group(1)}{stamp}", text, count=1)
    return text


def task_daily_seed(ctx):
    """Auto-safe: ensure TODAY's Daily/ note exists with the habit checklist + the
    day's workout pre-filled, BEFORE the digest builds. Idempotent: creates the
    skeleton if absent; if present, only inserts missing routine sections, never
    touching the owner's Priorities / On my mind / Debrief or his checkmarks. Writes
    only when content changes (like the Action Board), atomically."""
    d, vault, now = ctx["digest"](), ctx["vault"], ctx["now"]
    loc = ctx.get("loc") or {}
    day = now.date()
    weekday = day.strftime("%A")
    path = vault / "Daily" / f"{day.isoformat()} {weekday}.md"
    habits = d.canonical_habits()
    workout = d.workout_detail(day)
    if path.exists():
        text = path.read_text(encoding="utf-8")
        new = _ensure_daily_sections(text, habits, workout)
    else:
        text, new = None, _build_daily_note(day, weekday, habits, workout)
    new = _ensure_location(new, loc.get("location"), loc.get("tz"))   # canonical daily location/tz
    if new == text:
        return None                        # nothing missing — no write, no churn
    done, would = ("seeded", "seed") if text is None else ("updated", "update")
    if not _writes_ok(ctx):
        return f"would {would} daily note {day.isoformat()}"
    _atomic_write(path, new)
    return f"{done} daily note {day.isoformat()}"


_AB_TIME_STAMP = re.compile(r" at \d{2}:\d{2}")
def task_action_board(ctx):
    """Regenerate the top-level Action Board.md from the collated todos. Auto-safe:
    a deterministic, idempotent overwrite (no Claude) — only writes when the
    rendered content actually changed, so it doesn't churn the git history. The
    `generated_at` time stamp is excluded from the diff: same todos ⇒ no write ⇒
    stamp stays at the last *actual* regeneration (the truthful semantics —
    'when did this content last change'), instead of clicking forward every hour
    and committing the file 24 times a day."""
    d, vault = ctx["digest"](), ctx["vault"]
    payload = d.build_payload()
    md = d.render_action_board(
        payload, location=ctx["loc"].get("location"), generated_at=ctx["now"])
    n = sum(payload["counts"].values())
    target = vault / "Action Board.md"
    if ctx["plan"] or SHADOW or not AUTOFIX:
        return f"would regenerate Action Board ({n} items)"
    if target.exists() and (
            _AB_TIME_STAMP.sub("", target.read_text(encoding="utf-8"))
            == _AB_TIME_STAMP.sub("", md)):
        return None                            # unchanged — no write, no commit
    target.write_text(md, encoding="utf-8")
    return f"regenerated Action Board ({n} items)"
def task_daily_digest(ctx):
    """Send the entity-split morning digests once per LOCAL day, inside a bounded
    catch-up window. The digests themselves \u2014 the Acme-only vs everything-but-
    Acme split, recipients, area scoping, the "the agent" sender, and the actual
    build/render/send \u2014 all live in digest.py (`digest_specs()` + `run_digests()`), the
    single sender that the standalone `digest.sh` also calls. This task is just the
    *scheduler*: it owns the cadence and delegates delivery to `run_digests()`, passing
    its 9am-local window + per-digest once/day gate in as callbacks. So `digest.sh` and
    the sweep send through identical code \u2014 there is no second send path to drift.

    Window = [DIGEST_HOUR, DIGEST_HOUR + DIGEST_CATCHUP_HOURS) in the owner's local time.
    Starting "at or after 9" means an eastward flight (whose local clock jumps past 9)
    still triggers it; the upper bound means a morning the sweep missed entirely (host
    asleep, transient send error) catches up later that morning but never mails a
    "morning" brief at night \u2014 past the window it waits for the next local day.

    Each digest has its OWN once-per-day key (`digest:<spec key>:sent`), persisted
    immediately after it sends, so the two are independent: a missing recipient or a
    send error on one never blocks or re-fires the other, and neither re-sends across
    the window's several hourly runs.
    """
    now, state = ctx["now"], ctx["state"]
    in_window = DIGEST_HOUR <= now.hour < DIGEST_HOUR + DIGEST_CATCHUP_HOURS
    if not in_window:
        return None

    def _record(key):
        # Persist the per-digest "sent" flag the instant it sends \u2014 a crash on the
        # other digest must never let this one re-send across the window's hourly runs.
        mark_today(state, f"digest:{key}:sent", now)
        save_state(state)

    # Delegate the actual build/render/send to digest.run_digests() (the one sender);
    # this task supplies only the cadence: the 9am window (above) + the once/day gate.
    results = ctx["digest"]().run_digests(
        mode=("plan" if ctx["plan"] else "send"),
        should_send=lambda key: once_today(state, f"digest:{key}:sent", now),
        after_send=_record,
        sink=None,
    )
    return "; ".join(results) or None


def task_team_digests(ctx):
    """Send each configured team member their own morning brief, once per their-local day,
    inside the same bounded catch-up window as the main digests. UNLIKE task_daily_digest
    (which fires at 9am wherever WILL is), each member is pinned to THEIR OWN timezone +
    hour from the Team Digests roster note — so a member's window is checked against
    `_now_in_tz(member.timezone)` and the once/day key is keyed on that member + their
    local date. The roster (who, scope, tz, hour, prompt, reply) and the content build live
    in team_digest.py; this task owns only the per-member cadence. Emails send in shadow
    mode like the other digests; only --plan is side-effect-free."""
    # Skip cleanly when the team-digests module isn't enabled for this vault. (The gate
    # used to be "team_digest.py rendered beside sweep"; the engine file always exists
    # app-side now, so the config carries the enablement.)
    if not _module_enabled(ctx["vault"], "team-digests"):
        return None
    state = ctx["state"]
    tm = _load_team_digest()
    done = []
    for member in tm.load_members(ctx["vault"]):
        now = _now_in_tz(member["timezone"])
        hour = member["hour"]
        if not (hour <= now.hour < hour + DIGEST_CATCHUP_HOURS):
            continue
        key = f"digest:team:{member['key']}:sent"
        if not once_today(state, key, now):
            continue
        mode = "plan" if ctx["plan"] else "send"
        results = tm.run(member, mode=mode, today=now.date(), sink=None)
        if not ctx["plan"] and any(r.startswith("sent") for r in results):
            # Persist the once/day flag the instant it sends, keyed on this member + their
            # local date, so it never re-fires across the window's remaining hourly runs.
            mark_today(state, key, now)
            save_state(state)
        done += results
    return "; ".join(done) or None


def task_team_inbox(ctx):
    """Process team members' email REPLIES into vault progress updates — the protected
    return path for their briefs. Runs every sweep (replies can arrive any time, not just
    at send time); idempotency is per-Message-ID inside team_inbox, not a once/day gate.
    Disabled unless TEAM_INBOX_ENABLE=1. All the hard parts — the identity/auth/thread
    gates, the tool-less LLM mapper, the per-member allowlist, and the deterministic
    applier that can ONLY append a note or tick off one of THAT member's items — live in
    team_inbox.py, which iterates every reply-enabled member in the roster. In --plan it
    runs a read-only dry-run (no vault writes, no confirmation email)."""
    if os.environ.get("TEAM_INBOX_ENABLE", "0") in ("0", "false", ""):
        return None
    mode = "dryrun" if ctx["plan"] else "apply"
    results = _load_team_inbox().process(mode=mode, sink=None)
    # Surface only the meaningful lines (processed / skipped / errors), not the empty case.
    keep = [r for r in results if r and ("processed" in r or "skipped" in r or "failed" in r
                                         or "dry-run" in r or "disabled" in r)]
    return "; ".join(keep) or None


def task_will_inbox(ctx):
    """Process the owner's own PRINCIPAL emails into vault changes — the higher-trust path
    (freeform note+attachment capture into Daily + todo ops + new todos + digest feedback).
    Same cadence/role as task_team_inbox (every sweep, per-Message-ID idempotent, before the
    board/archive run) but gated on PRINCIPAL_INBOX_ENABLE and authenticated by DKIM/SPF + a
    passphrase (From is spoofable). The IDLE daemon runs this too for low latency; the shared
    flock keeps them from racing. All logic — gates, parser, capture, applier — lives in
    principal_inbox.py. --plan → read-only dry-run."""
    # PRINCIPAL_INBOX_ENABLE is the current knob (matches the module + the inbox workflow);
    # WILL_INBOX_ENABLE is the historical name, still honored for back-compat.
    if os.environ.get("PRINCIPAL_INBOX_ENABLE", os.environ.get("WILL_INBOX_ENABLE", "0")) in ("0", "false", ""):
        return None
    mode = "dryrun" if ctx["plan"] else "apply"
    results = _load_will_inbox().process(mode=mode, sink=None)
    keep = [r for r in results if r and ("processed" in r or "skipped" in r or "failed" in r
                                         or "dry-run" in r or "disabled" in r)]
    return "; ".join(keep) or None


def task_attention_digest(ctx):
    """File todos from the owner's inbound EMAIL, once per local day inside the morning
    window — the automatic counterpart to the on-demand attention-digest skill. Like
    task_daily_digest this task owns only the cadence (the same 9am-local window + a
    dedicated once/day key); the read/extract/apply/notify all live in attention_digest.py,
    which is READ-ONLY over the mailbox and files git-revertible `- [ ]` items into the right
    area TODO. Placed before the board/digest run so fresh todos surface the same morning.

    Skips cleanly when the attention-digest module isn't enabled for this vault (the
    same config gate as task_team_digests — the engine file always exists app-side now,
    so the config carries the enablement). It writes to the vault, so — unlike the
    digests, whose point is the email — it stands down in shadow/--plan rather than
    half-acting."""
    if not _module_enabled(ctx["vault"], "attention-digest"):
        return None
    if os.environ.get("ATTENTION_DIGEST_ENABLE", "0") in ("0", "false", ""):
        return None
    now, state = ctx["now"], ctx["state"]
    if not (DIGEST_HOUR <= now.hour < DIGEST_HOUR + DIGEST_CATCHUP_HOURS):
        return None
    if not once_today(state, "attention:filed", now):
        return None
    if not _writes_ok(ctx):
        return "would file inbox todos (attention digest)"
    results = _load_attention_digest().run(mode="apply", force=True, sink=None)
    mark_today(state, "attention:filed", now)     # persist immediately so it won't re-fire this window
    save_state(state)
    return "; ".join(results) if isinstance(results, list) else (results or None)


def task_vault_hygiene(ctx):
    """Weekly (Mon): email a deterministic hygiene + todo-triage report.
    All scans are read-only. (Daily notes were retired, so there is no longer a
    daily-note scaffold; this task currently has no autonomous writes.)"""
    now, state, vault = ctx["now"], ctx["state"], ctx["vault"]
    done = []

    # -- weekly report (Monday, at/after digest hour) --
    week = "%d-W%02d" % now.isocalendar()[:2]
    if now.weekday() == 0 and now.hour >= DIGEST_HOUR and state.get("hygiene:week") != week:
        report = _hygiene_report(ctx)
        if ctx["plan"]:
            done.append("would email weekly hygiene report")
        else:
            ctx["digest"]().send_email(report, "\U0001F9F9 Vault hygiene \u2014 " + week)
            state["hygiene:week"] = week
            done.append("emailed weekly hygiene report")
    return "; ".join(done) or None


def _hygiene_report(ctx):
    vault, d = ctx["vault"], ctx["digest"]()
    notes = list(fs.iter_paths(vault, "**/*.md"))
    oversize, no_fm, stub = [], [], []
    # Broken links come from the one analyzer (vault_core.graph), not a local
    # regex. The regex this replaced was not fence-aware and knew nothing about
    # aliases, attachments, or placeholders, so it reported every `[[link]]`
    # inside a code fence — and every alias-resolved link — as broken.
    broken = set(analyze_links(vault)["content_broken"])
    for p in notes:
        if "/Archive/" in str(p):
            continue
        try:
            t = p.read_text(encoding="utf-8")
        except Exception:
            continue
        if len(t.encode()) > 60000:
            oversize.append(p.stem)
        if not t.lstrip().startswith("---"):
            no_fm.append(p.stem)
        if len(re.sub(r"^---.*?---", "", t, flags=re.S).strip()) < 60:
            stub.append(p.stem)
    todos = d.collect_todos()
    today_iso = ctx["now"].date().isoformat()
    overdue = [it for it in todos if it.get("due") and it["due"] < today_iso]
    xcount = sum(len(re.findall(r"(?m)^- \[x\]", p.read_text(encoding="utf-8", errors="ignore")))
                 for p in fs.iter_paths(vault, "Areas/**/*TODO.md") if p.name != "Completed Todos.md")

    # -- #2 report-first: near-duplicate todos + stale undated items --
    groups = defaultdict(list)
    for it in todos:
        key = _normalize(it.get("text", ""))[:80]
        if key:
            groups[key].append(it)
    dup_groups = [v for v in groups.values() if len(v) > 1]
    stale_cut = (ctx["now"].date() - dt.timedelta(days=STALE_DAYS_REPORT)).isoformat()
    stale = [it for it in todos
             if not it.get("due") and not it.get("check")
             and (not it.get("added") or it["added"] < stale_cut)]

    def section(title, items, n=15):
        if not items:
            return ""
        shown = sorted(set(items))[:n]
        extra = f" <i>(+{len(set(items)) - n} more)</i>" if len(set(items)) > n else ""
        lis = "".join(f"<li>{_html.escape(s)}</li>" for s in shown)
        return f"<h3>{title} ({len(set(items))})</h3><ul>{lis}</ul>{extra}"

    def dup_section():
        if not dup_groups:
            return ""
        lis = "".join(
            f"<li>{_html.escape(g[0].get('text', '')[:120])} "
            f"<i>({', '.join(sorted({_html.escape(it['area']) for it in g}))})</i></li>"
            for g in sorted(dup_groups, key=lambda g: g[0].get('text', ''))[:15])
        return f"<h3>Possible duplicate todos ({len(dup_groups)})</h3><ul>{lis}</ul>"

    def stale_section():
        if not stale:
            return ""
        shown = stale[:15]
        extra = f" <i>(+{len(stale) - 15} more)</i>" if len(stale) > 15 else ""
        lis = "".join(f"<li><b>{_html.escape(it['area'])}:</b> {_html.escape(it.get('text','')[:120])}</li>"
                      for it in shown)
        return (f"<h3>Stale — no date, untouched &gt;{STALE_DAYS_REPORT}d ({len(stale)})</h3>"
                f"<ul>{lis}</ul>{extra}")

    parts = [
        f"<p>{len(notes)} notes scanned.</p>",
        section("Broken wikilinks", broken),
        section("Missing frontmatter", no_fm),
        section("Stub / near-empty notes", stub),
        section("Oversized notes (&gt;60KB)", oversize),
        dup_section(),
        stale_section(),
        f"<h3>Todos</h3><ul><li>{len(overdue)} overdue</li>"
        f"<li>{xcount} completed [x] not yet archived</li></ul>",
    ]
    return "".join(p for p in parts if p)


# ---- auto-fix helpers ------------------------------------------------------
def _writes_ok(ctx):
    """Auto-safe vault writes happen only on a live, non-plan, autofix-enabled run.
    In --plan or shadow they are reported ('would …') but never applied."""
    return not ctx["plan"] and not SHADOW and AUTOFIX


def _atomic_write(path, text):
    """Write atomically (tmp + os.replace) so vault-sync, which commits this tree
    every ~10 min, can never capture a half-written file."""
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(text)
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp):
            try:
                os.remove(tmp)
            except OSError:
                pass


def _iter_notes(vault):
    """Every markdown note that is real vault content. Skips the Archive and script
    tree (_SKIP_DIRS), any dot-directory (.git/.obsidian/.claude/.trash), and
    `_`-prefixed files (templates/partials) — none of which should be auto-edited."""
    for p in fs.iter_paths(vault, "**/*.md"):
        rel = p.relative_to(vault).parts
        if any(seg.startswith(".") for seg in rel) or p.name.startswith("_"):
            continue
        if any(d in str(p) for d in _SKIP_DIRS):
            continue
        yield p


def _normalize(s):
    """Casefold + strip accents + collapse non-alphanumerics to single spaces — for
    fuzzy todo/link matching (NOT Obsidian's own resolution, which is only
    case-insensitive)."""
    s = unicodedata.normalize("NFKD", s or "")
    s = "".join(c for c in s if not unicodedata.combining(c))
    return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip()


def _frontmatter_block(text):
    m = re.match(r"^---\n(.*?)\n---", text, re.S)
    return m.group(1) if m else None


# ---- task: archive completed [x] todos -------------------------------------
_DONE_LINE  = re.compile(r"^- \[[xX]\] ")
_COMPLETED  = re.compile(r"\bcompleted (\d{4}-\d{2}-\d{2})")
_DONE_DATE  = re.compile(r"\b[Dd]one (\d{4}-\d{2}-\d{2})")
_ADDED_SUB  = re.compile(r"\badded \d{4}-\d{2}-\d{2}")


def _area_of(text, path):
    m = re.search(r'area:\s*"?\[\[([^\]|]+)', text)
    return (m.group(1) if m else path.stem.replace(" TODO", "")).strip()


def _ensure_completed_stamp(block, date):
    """Make sure an archived block records a completed date (matching the existing
    'added …; completed YYYY-MM-DD.' shape) — append to a metadata sub-bullet, else
    add one. Leaves blocks that already carry a completed date untouched."""
    if _COMPLETED.search("\n".join(block)):
        return block
    for i in range(len(block) - 1, 0, -1):
        if _ADDED_SUB.search(block[i]):
            block[i] = block[i].rstrip().rstrip(".") + f"; completed {date}."
            return block
    indent = "    "
    block.append(f"{indent}- completed {date}.")
    return block


def _archive_from_file(path, today, cutoff):
    """Return (new_source_text, [block, …]) for the [x] items eligible to archive
    (completed on/before `cutoff`), or None if there are none. Open-Items sections
    are left alone — resolved gaps belong in Resolved Open Items, not here."""
    lines = path.read_text(encoding="utf-8").split("\n")
    kept, moved, i, in_open = [], [], 0, False
    while i < len(lines):
        line = lines[i]
        h = re.match(r"^#{2,4}\s+(.*)", line)
        if h:
            # Sticky across ### subsections (mirrors digest.collect_todos): only a
            # level-2 `## ` heading enters/leaves Open Items, so resolved [x] gaps in a
            # `### …` subsection of Open Items stay put instead of being archived.
            if line[:3] != "###":
                in_open = h.group(1).strip().lower().startswith("open items")
            kept.append(line); i += 1; continue
        if not in_open and _DONE_LINE.match(line):
            block, j = [line], i + 1
            while j < len(lines) and lines[j][:1] in (" ", "\t") and lines[j].strip():
                block.append(lines[j]); j += 1
            blob = "\n".join(block)
            m = _COMPLETED.search(blob) or _DONE_DATE.search(blob)
            cdate = m.group(1) if m else today
            if cdate <= cutoff:
                moved.append(_ensure_completed_stamp(block, cdate))
            else:
                kept.extend(lines[i:j])           # grace window not elapsed — keep
            i = j; continue
        kept.append(line); i += 1
    if not moved:
        return None
    new_text = re.sub(r"\n{3,}", "\n\n", "\n".join(kept))   # tidy gaps left by removals
    return new_text, moved


def _heading_matches_area(heading, area):
    base = re.split(r"\s[—-]\s", heading, 1)[0]              # "the owner — personal" → "the owner"
    return _normalize(base) == _normalize(area) or _normalize(heading).startswith(_normalize(area))


def _split_sections(text):
    pre, sections, cur = [], [], None
    for l in text.split("\n"):
        if l.startswith("## "):
            if cur is not None:
                sections.append(cur)
            cur = [l]
        elif cur is None:
            pre.append(l)
        else:
            cur.append(l)
    if cur is not None:
        sections.append(cur)
    return pre, sections


def _append_completed(text, by_area, today):
    """Append archived blocks under each area's `## <Area>` section in Completed
    Todos, creating the section if absent; bump last_reviewed."""
    text = re.sub(r"(?m)^last_reviewed:.*$", f"last_reviewed: {today}", text, count=1)
    pre, sections = _split_sections(text)
    remaining = dict(by_area)
    for sec in sections:
        htext = sec[0][3:].strip()
        for area in list(remaining):
            if _heading_matches_area(htext, area):
                while sec and sec[-1].strip() == "":
                    sec.pop()
                for block in remaining.pop(area):
                    sec.extend(block)
                sec.append("")
                break
    for area, blocks in remaining.items():                   # areas with no section yet
        sec = [f"## {area}", ""]
        for block in blocks:
            sec.extend(block)
        sec.append("")
        sections.append(sec)
    body = [l for sec in sections for l in sec]
    return "\n".join(pre + body).rstrip("\n") + "\n"


def _bump_updated(text, today):
    return re.sub(r"(?m)^last_updated:.*$", f"last_updated: {today}", text, count=1)


def task_archive_completed(ctx):
    """Auto-safe: move completed `- [x]` action items (with their sub-bullets) from
    each `<Area> TODO.md` to [[Completed Todos]], stamped `completed YYYY-MM-DD`,
    under the matching area heading. Idempotent (a clean file yields nothing),
    reversible (a move, never a delete), atomic per file. ARCHIVE_AFTER_DAYS sets an
    optional grace window (default 0 = archive on the next run)."""
    vault, now = ctx["vault"], ctx["now"]
    today = now.date().isoformat()
    cutoff = (now.date() - dt.timedelta(days=ARCHIVE_AFTER_DAYS)).isoformat()
    sources, by_area = {}, {}
    for path in fs.iter_paths(vault, "Areas/**/*TODO.md"):
        if path.name == "Completed Todos.md":
            continue
        res = _archive_from_file(path, today, cutoff)
        if not res:
            continue
        new_text, moved = res
        sources[path] = new_text
        by_area.setdefault(_area_of(path.read_text(encoding="utf-8"), path), []).extend(moved)
    total = sum(len(v) for v in by_area.values())
    if not total:
        return None
    if not _writes_ok(ctx):
        return f"would archive {total} completed item(s) → Completed Todos"
    comp = vault / "Archive" / "Completed Todos.md"
    _atomic_write(comp, _append_completed(comp.read_text(encoding="utf-8"), by_area, today))
    for path, new_text in sources.items():
        _atomic_write(path, _bump_updated(new_text, today))
    return f"archived {total} completed item(s) → Completed Todos"


# ---- task: frontmatter backfill (author) -----------------------------------
def task_frontmatter_backfill(ctx):
    """Auto-safe: add `author: {DEFAULT_AUTHOR}` to notes that have a frontmatter
    block but no author line. Notes with NO frontmatter at all are left for the
    weekly report (fabricating a whole block is a judgment call, not a safe fix)."""
    targets = []
    for p in _iter_notes(ctx["vault"]):
        try:
            text = p.read_text(encoding="utf-8")
        except Exception:
            continue
        fm = _frontmatter_block(text)
        if fm is None or re.search(r"(?m)^author:", fm):
            continue
        end = text.index("\n---", 3)                        # end of the frontmatter body
        targets.append((p, text[:end] + f"\nauthor: {DEFAULT_AUTHOR}" + text[end:]))
    if not targets:
        return None
    if not _writes_ok(ctx):
        return f"would backfill author on {len(targets)} note(s)"
    for p, new_text in targets:
        _atomic_write(p, new_text)
    return f"backfilled author on {len(targets)} note(s)"


# ---- task: single-candidate broken-link repair -----------------------------
_LINK = re.compile(r"\[\[([^\]\n|#]+)((?:#[^\]\n|]+)?)(\|[^\]\n]+)?\]\]")


def task_link_repair(ctx):
    """Auto-safe: repair a broken `[[wikilink]]` only when exactly one note matches
    after accent/punct normalization (e.g. [[Guacimo]]→[[San Francisco]]). Links that
    already resolve are untouched; 0-candidate or ambiguous ones are left for the
    weekly report. Display text and #anchors are preserved."""
    # The note index lives in vault_core.graph now (the absorbed sweep-side
    # variant — accent-stripping normalize, naive comma aliases). The repair-time
    # _normalize below must agree with the index's; a lockstep test enforces it.
    resolve, norm_to_stem = note_index(ctx["vault"])
    edits, total = {}, 0
    for p in _iter_notes(ctx["vault"]):
        try:
            text = p.read_text(encoding="utf-8")
        except Exception:
            continue
        changed = []

        def repl(m):
            target = m.group(1).strip()
            # A pipe escaped for a Markdown table cell ([[Note\|alias]]) leaves a
            # trailing backslash on the captured target. Those links resolve fine in
            # Obsidian — never rewrite them (doing so would unescape the pipe and
            # corrupt the table).
            if target.endswith("\\"):
                return m.group(0)
            if not target or target.lower().startswith("http") or target.lower() in resolve:
                return m.group(0)
            cand = norm_to_stem.get(_normalize(target))
            if cand and cand.lower() != target.lower():
                changed.append((target, cand))
                return f"[[{cand}{m.group(2)}{m.group(3) or ''}]]"
            return m.group(0)

        new_text = _LINK.sub(repl, text)
        if changed:
            edits[p] = new_text
            total += len(changed)
    if not total:
        return None
    if not _writes_ok(ctx):
        return f"would repair {total} single-candidate link(s) in {len(edits)} note(s)"
    for p, new_text in edits.items():
        _atomic_write(p, new_text)
    return f"repaired {total} link(s) in {len(edits)} note(s)"


def task_inbox_drain(ctx):
    """Count what is pending in ``Agent/inbox/`` and surface it — the obligation that
    makes a staging surface safe to have.

    ``Agent/inbox/`` exists so bulk import has somewhere to put what it genuinely
    cannot route (docs/INGESTION.md §7). The reason the vault refused an inbox
    for so long is real: one nobody empties is a slower way to lose things. So
    the backlog is counted every sweep and reported with the oldest item's age —
    a pile that is growing, or old, is visible instead of quietly permanent.

    Read-only. Draining is a judgment call about where each note belongs, which
    is the agent's job with the routing table, not a cron's.
    """
    vault = ctx["vault"]
    inbox = vault / "Agent" / "inbox"
    if not inbox.is_dir():
        return None
    staged = []
    for p in sorted(inbox.glob("*.md")):
        if p.name.lower() == "inbox.md":              # the folder note, not an item
            continue
        try:
            head = p.read_text(encoding="utf-8", errors="replace").split("\n---", 1)[0]
        except OSError:
            continue                                  # a broken item can't break the sweep
        # Glob on `status: pending` the way the drain does, so an item already
        # filed but not yet deleted stops being counted as a backlog.
        if re.search(r"^status:\s*pending\s*$", head, re.M):
            staged.append(p)
    if not staged:
        ctx["state"].pop("inbox:last_count", None)
        return None
    oldest = min(p.stat().st_mtime for p in staged)
    now_ts = dt.datetime.now(dt.timezone.utc).timestamp()
    age_days = max(0, int((now_ts - oldest) // 86400))
    ctx["state"]["inbox:last_count"] = len(staged)
    ctx["state"]["inbox:oldest_days"] = age_days
    return (f"Agent/inbox/ holds {len(staged)} pending item(s), oldest {age_days}d — "
            f"drain them ('process my inbox')")


TASKS = [
    ("team_inbox", task_team_inbox),   # apply team members' emailed updates BEFORE the board/archive run
    ("will_inbox", task_will_inbox),   # apply the owner's own emailed changes (principal path) — same, before board/archive
    ("attention_digest", task_attention_digest),  # file todos from inbound third-party email — before board/digest so they surface
    ("daily_seed", task_daily_seed),   # ensure today's Daily/ note (habits + workout) exists before the board/digest read it
    ("action_board", task_action_board),
    ("archive_completed", task_archive_completed),
    ("frontmatter_backfill", task_frontmatter_backfill),
    ("link_repair", task_link_repair),
    ("daily_digest", task_daily_digest),
    ("team_digests", task_team_digests),
    ("vault_hygiene", task_vault_hygiene),
    ("inbox_drain", task_inbox_drain),   # count the staging backlog so it can't rot
]


def main():
    plan = "--plan" in sys.argv
    if killed():
        log("kill switch active (SWEEP_DISABLE / SWEEP_OFF) — exiting")
        return
    if not plan:
        cap_logs()
    _check_env()
    d = _load_digest()
    vault = d.VAULT_DIR
    state = load_state()
    loc = resolve_location(state, vault)       # re-resolved every run, anchored to the owner's local date
    now = loc["now"]
    fake = os.environ.get("SWEEP_FAKE_LOCAL")
    if fake:
        now = dt.datetime.fromisoformat(fake)
    # Pin the digest engine's notion of "today" to the owner's resolved LOCAL date. digest.py
    # otherwise defaults to this machine's clock (the home timezone on the host), which would
    # bucket "today/tomorrow" todos — and stamp the date header — against the wrong
    # calendar day whenever the owner is abroad (e.g. Manila, +14h). Re-read on every
    # _load_digest()/build_payload(), so setting the env here is enough.
    os.environ["DIGEST_TODAY"] = now.date().isoformat()
    log(f"location={loc['location']} tz={loc['tz']} tz_source={loc.get('tz_source')} "
        f"local={now.isoformat(timespec='minutes')} confident={loc['confident']} "
        f"cached={loc.get('cached')} shadow={SHADOW} plan={plan}")
    ctx = {"vault": Path(vault), "loc": loc, "now": now, "state": state,
           "plan": plan, "digest": _load_digest}
    results, errors = {}, []
    for name, fn in TASKS:
        try:
            res = fn(ctx)
            if res:
                log(f"{name}: {res}")
            results[name] = "ok"
        except Exception as e:
            log(f"{name}: ERROR {type(e).__name__}: {e}")
            results[name] = "error"
            errors.append(f"{name}: {type(e).__name__}: {e}")
    if not plan:                      # --plan has no side effects; never persist state
        save_state(state)
        write_heartbeat({
            "kind": "sweep",
            "at": dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
            "local": now.isoformat(timespec="minutes"),
            "location": loc.get("location"),
            "ok": not errors,
            "tasks": results,
            "errors": errors,
            "shadow": SHADOW,
        })


if __name__ == "__main__":
    main()
