#!/usr/bin/env python3
"""mhs-thread — the MHS reader behind `memory ls | tree | cat | run PATH` (not a verb itself).

Ported from Pip's print-thread script (CID-164, 2026-08-24). Agent from
CLAUDE_CODE_AGENT / SESSION_AGENT; root hint from $<AGENT>_ROOT; --box NAME
for a shared box. Originally: context-conserving view of a mail thread.

Charter: [MAIL-0811] branch [7]. Subjects as a tree, superseded filtered,
bodies only on request. Ends with its own token cost so you know what you paid.

Usage:
  print-thread QUERY...          notmuch query (e.g. 'subject:"MAIL-0811"' or thread:ID)
  print-thread --last-session    the predecessor's registry thread (SESSION UP + replies)
  print-thread --wake            boot readout: delegates to `memory wake`
                                 (BOOT+RECENT+COUNTS), then adds RULES + LAST SESSION
  print-thread --index           the canonical root: every memory category, one page each
  print-thread --ls              `ls /` of memory: the [ROOT] and its labels only (depth 1)
  print-thread --boot            wake in one verb: --ls, then the newest signed item under
                                 [BOOT] with its body (sig-verified; refuses if unsigned)
  --depth N                      cut the tree N levels below the top (0 = top only)
  print-thread --path /A/B/10    MHS path (item 70): walk from [ROOT] by subject token.
                                 A segment is matched against the children's subjects by
                                 the first rule that hits (SESH-1098): the EXACT printed
                                 name ("/ASKS/[session]"), then the bracket form
                                 ("/ASKS" -> "[ASKS]"), then CANONICAL (case/spacing, both
                                 sides: "/ASKS/session"), and for a NUMBERED segment only,
                                 the "10 ..." prefix. Every name the tree prints is
                                 therefore a valid segment. A canonical hit spanning two
                                 spellings refuses as ambiguous; /today and /yesterday map
                                 to an ISO date first. Renders the subtree (use --depth);
                                 --resolve prints the Message-ID only
  --under MSGID                  render only the SUBTREE under MSGID (that message
                                 and its descendants), not the whole thread
  -b / --bodies                  include message bodies
  -a / --all                     don't filter superseded messages
  -d / --dates                   show [sender date] suffixes (default: sender only)
  -n / --dry-run                 show the notmuch commands, don't render
                                 (read-only script — every line copy-pasteable)
  -h / --help                    this text
"""
import json, os, re, shlex, subprocess, sys

DRY = False   # -n/--dry-run: show the notmuch commands instead of rendering.
              # The script never mutates anything (S01), so dry-run is about
              # VISIBILITY: black-box users learn the underlying invocations,
              # and every printed line is copy-pasteable into a shell.

def sh(args, resolve=False):
    if DRY:
        tag = "resolve (ran)" if resolve else "would run    "
        print(f"  {tag}: {shlex.join(args)}")
        if not resolve:
            return ""
    return subprocess.run(args, capture_output=True, text=True).stdout

def short_from(h):
    # SM-398: a named From (`Cora <cora@m5>`) is its ADDRESS, not its name -
    # the innermost <...> first, as session-mail-verify's local_part() does.
    # Reading the name here judged a named memory item NOT the owner's (MHS
    # 170) and hid it. A bare From reduces exactly as before.
    v = h.get("From", "?")
    i = v.rfind("<")
    j = v.find(">", i + 1) if i != -1 else -1
    if i != -1 and j != -1:
        v = v[i + 1:j]
    m = re.match(r'([^<@\s]+)', v.strip())
    return m.group(1) if m else "?"

def flatten(node, depth, out):
    msg, children = node[0], node[1]
    if msg:
        out.append((depth, msg))
    for c in children:
        flatten(c, depth + (1 if msg else 0), out)

def body_text(msg):
    parts = []
    def walk(p):
        if isinstance(p, list):
            for x in p: walk(x)
        elif isinstance(p, dict):
            if p.get("content-type", "").startswith("text/plain") and isinstance(p.get("content"), str):
                parts.append(p["content"])
            walk(p.get("content", []))
    walk(msg.get("body", []))
    return "\n".join(parts).strip()

def supersede_pairs(flats):
    """{superseded_id: [superseder_id, ...]} from Supersedes:/X-Supersedes:
    headers. notmuch's JSON omits nonstandard headers, so read each message
    file's header block directly (filenames are in the JSON; local IO, capped)."""
    pairs = {}
    pat = re.compile(r'(?im)^(?:X-)?Supersedes:\s*<?([^>\s]+)>?')
    for flat in flats:
        for _, m in flat:
            for fn in m.get("filename", []):
                try:
                    with open(fn, "rb") as fh:
                        head = fh.read(8192).split(b"\n\n", 1)[0]
                    for tid in pat.findall(head.decode("utf-8", "replace")):
                        pairs.setdefault(tid, []).append(m.get("id", "").strip("<>"))
                    break
                except OSError:
                    continue
    return pairs

def superseded_ids(flats):
    """Collect message-ids named in Supersedes:/X-Supersedes: headers."""
    return set(supersede_pairs(flats))

# ── MHS 170 membership — ONE test, applied wherever a mail takes effect ─────
# TARS found (Kith 13, 18:48 Mon 2026-09-28) that `ls /RULES`, `tree` and
# `cat` read an unsigned or forged item as live memory: only `ls /` and
# `--boot` passed verify=True. Reading this file for it (Pip, 18:49) found
# the same gap wherever a mail can TAKE EFFECT without being printed: the
# walk's child pick (newest of one spelling wins), `latest`, a Supersedes:
# header (honoured from any mail in the thread, the wake included), the
# subject-tag shadows, a fork twin, and the [ROOT] anchor (newest [ROOT] by
# subject). Headers are not the test — any local sender writes From and To.

PRIMED = ""   # the root hint's Message-ID once root_thread() has primed it

_SIG = {}     # filename -> True (verified) / False (refused) / None (timed out)
# A verify is ~80 ms; one that has not answered in this long is a hang, and
# a hang is NOT a pass (review, 19:52 Mon 2026-09-28: without it one
# stuck verifier hung every reader).
VERIFY_TIMEOUT = float(os.environ.get("SESSION_MAIL_VERIFY_TIMEOUT") or 10)
# The verifier is the sibling session-mail-verify, which is what `session
# mail-verify` dispatched to: this reader never leaves its own directory.
# $SESSION_MAIL_VERIFY (a command, shlex-split, the file appended) replaces
# it; the test suites point it at their stub.
_HERE = os.path.dirname(os.path.realpath(__file__))
VERIFY = (shlex.split(os.environ.get("SESSION_MAIL_VERIFY") or "")
          or [os.path.join(_HERE, "session-mail-verify")])

def _sig_ok(m):
    fn = next((f for f in m.get("filename", []) if os.path.exists(f)), None)
    if not fn:
        return False
    if fn not in _SIG:
        try:
            _SIG[fn] = subprocess.run([*VERIFY, fn], capture_output=True,
                                      timeout=VERIFY_TIMEOUT).returncode == 0
        except subprocess.TimeoutExpired:
            # UNKNOWN, not False: a mail we could not check is not a member,
            # and a Supersedes: it carries still RETIRES its target
            # (MemberPairs.blocked) — a timeout never hands a step back to
            # the version it replaced (review, 20:26 Mon 2026-09-28).
            _SIG[fn] = None
    return _SIG[fn]

class Unverifiable(Exception):
    """A mail that decides the answer could not be verified (timed out)."""

class Members:
    """MHS 170: a mail is memory iff From the owner (any sender, for a shared
    box) AND To/Cc the recipient at any plus-tag AND its signature verifies.
    Call it on one notmuch message; the verdict is kept per Message-ID.

    LAZY on purpose: a real root thread is ~1600 mails, ~670 of them
    superseders, and one verify is ~80 ms (measured 18:55 Mon 2026-09-28), so
    checking the thread up front is ~55 s. Only a mail that could take effect
    is verified, and prefetch() runs a known batch (a node's children, a
    rendered subtree) in parallel."""
    def __init__(self, owner, recip=None, shared=False):
        self.owner, self.recip, self.shared = owner, recip or owner, shared
        self._memo = {}

    def headers_ok(self, m):
        # Plus-tag is NOT the test; owner-as-recipient is. The pre-standard
        # tree used +rules/+convo/… (110 of 168 mails as of 2026-08-23).
        # Case-blind, as the verifier's local_part() is.
        h = m["headers"]
        return ((self.shared or short_from(h).lower() == self.owner.lower())
                and re.search(r'(?<![\w.-])' + re.escape(self.recip) + r'(\+[\w-]+)?@',
                              h.get("To", "") + " " + h.get("Cc", ""), re.I) is not None)

    def __call__(self, m):
        mid = m.get("id", "")
        if mid not in self._memo:
            self._memo[mid] = bool(self.headers_ok(m) and _sig_ok(m) is True)
        return self._memo[mid]

    def unknown(self, m):
        """Would be a member, but its verify timed out: neither in nor out."""
        return bool(self.headers_ok(m)) and _sig_ok(m) is None

    def counts(self, m):
        """May m TAKE EFFECT — retire, shadow, make a fork? A member may, and
        so may an unknown one: whatever it might change is withheld, never
        answered without it (review, 22:03 Mon 2026-09-28). Only a member
        is SHOWN or RUN."""
        return self(m) or self.unknown(m)

    def prefetch(self, msgs):
        todo = [m for m in msgs if m.get("id", "") not in self._memo and self.headers_ok(m)]
        if len(todo) > 1:
            from concurrent.futures import ThreadPoolExecutor
            with ThreadPoolExecutor(max_workers=8) as ex:
                list(ex.map(_sig_ok, todo))

class AssumedMembers:
    """Members with some verdicts ASSUMED: each id in `assign` answers True
    or False as given and is never unknown; every other mail keeps the
    verdict it already has. The walk weighs mails whose verify timed out by
    trying each one both ways (review low, 22:46 Mon 2026-09-28)."""
    def __init__(self, base, assign):
        self.base, self.assign = base, assign

    def __call__(self, m):
        mid = m.get("id", "")
        return self.assign[mid] if mid in self.assign else self.base(m)

    def unknown(self, m):
        return False if m.get("id", "") in self.assign else self.base.unknown(m)

    def counts(self, m):
        return self(m) or self.unknown(m)

    def prefetch(self, msgs):
        pass    # the verdicts it weighs are already in; it never verifies

class MemberPairs:
    """supersede_pairs() as MHS 170 reads it: a Supersedes: counts only when
    the mail CARRYING it is a member. Serves as both a reader's `pairs` map
    (.get) and its `dead` set (`in`), resolved per id on first use, so a
    superseder nobody walks past is never verified."""
    def __init__(self, flats, member):
        self.raw = supersede_pairs(flats)
        self.member = member
        self.by_id = {m.get("id", "").strip("<>"): m for flat in flats for _, m in flat}
        self._memo = {}

    def get(self, mid, default=None):
        if mid not in self._memo:
            self._memo[mid] = [s for s in self.raw.get(mid, [])
                               if s in self.by_id and self.member(self.by_id[s])]
        return self._memo[mid] or (default if default is not None else [])

    def blocked(self, mid):
        """Superseders of mid whose verify timed out. They retire mid (a
        superseder we cannot check must not un-supersede it) but replace
        it with nothing: the step is gone, and the readers say why."""
        return [s for s in self.raw.get(mid, [])
                if s in self.by_id and self.member.unknown(self.by_id[s])]

    def __contains__(self, mid):
        return bool(self.get(mid)) or bool(self.blocked(mid))

    def assuming(self, member):
        """These pairs as `member` (an AssumedMembers) reads them: the same
        headers and index, a memo of its own."""
        other = object.__new__(MemberPairs)
        other.raw, other.by_id, other.member, other._memo = self.raw, self.by_id, member, {}
        return other

    def __iter__(self):
        return (k for k in self.raw if k in self)

    def prefetch(self, ids):
        """Verify, in parallel, every superseder reachable from `ids` through
        Supersedes: chains. A chain is walked one link at a time (v1 is dead
        only if v2 is a member, v2 only if v3 is…), so without this the
        ~190-letter [BOOT] chain ran its verifies in series: 18 s for
        `cat /BOOT/latest` (measured 19:05 Mon 2026-09-28)."""
        seen, todo, msgs = set(), [i.strip("<>") for i in ids], []
        while todo:
            for sid in self.raw.get(todo.pop(), []):
                if sid not in seen and sid in self.by_id:
                    seen.add(sid)
                    msgs.append(self.by_id[sid])
                    todo.append(sid)
        self.member.prefetch(msgs)

def member_pairs(flats, member):
    """The supersede pairs `member` honours; member None = every mail (a plain
    notmuch query, where there is no owner to be a member of)."""
    return MemberPairs(flats, member) if member else supersede_pairs(flats)

def is_anchor(m, member):
    """May m be the [ROOT] a walk or `ls /` starts from? The primed hint
    always may (MHS 90 vetted it); any other [ROOT] only as a member."""
    return (not member or m.get("id", "").strip("<>") == PRIMED or member(m))

def effective_children(flat, i, dead, pairs, only=None):
    """Indices of node i's children AS THE TREE MEANS THEM (MHS: a supersede
    is a replacement IN PLACE). A live direct child is itself; a dead direct
    child is replaced by the surviving end(s) of its Supersedes chain, which
    sit deeper (the superseder is a reply to the item it replaces). Two
    survivors of one item = a FORK: both are returned, so a renderer shows
    both flagged and never silently picks one (Cora, 2026-08-23 22:22).
    only: resolve just these direct-child indices — a walk opening ONE
    segment need not resolve, and so verify, every chain beside it."""
    d = flat[i][0]
    by_id = {m.get("id", "").strip("<>"): j for j, (_, m) in enumerate(flat)}
    def survivors(j, seen):
        mid = flat[j][1].get("id", "").strip("<>")
        if mid not in dead:
            return [j]
        out = []
        for sid in pairs.get(mid, []):
            k = by_id.get(sid)
            if k is None or k in seen:
                continue
            seen.add(k)
            out += survivors(k, seen)
        return out
    out = []
    for j in range(i + 1, len(flat)):
        d2, m2 = flat[j]
        if d2 <= d:
            break
        if d2 == d + 1 and (only is None or j in only):
            out += survivors(j, {j})
    # SESH-994 (a): survivors are a SET of messages. Two dead siblings whose
    # chains CONVERGE on one superseder (v1 -> v3 and v2 -> v3) reached v3
    # once per dead sibling, so `run` executed item 15 three times and item
    # 17 three times on Cora's wake (measured 23:18 Wed 2026-09-09: 18 items
    # run, 11 distinct). A FORK is two DIFFERENT survivors under one number;
    # one survivor named twice is not a fork. Keep first occurrence, in order.
    seen_out = set()
    out = [j for j in out if not (j in seen_out or seen_out.add(j))]
    # Numbered items sort by number (the number IS the filename); labels and
    # unnumbered suggestions keep thread order, ahead of the items.
    def key(j):
        s = flat[j][1]["headers"].get("Subject", "").strip()
        mm = re.match(r"^(\d+(?:\.\d+)*)\s", s)
        return (1, [int(x) for x in mm.group(1).split(".")], j) if mm else (0, [], j)
    return sorted(out, key=key)

def canon_name(s):
    """One label name, canonicalised — for BOTH sides of a comparison.

    MHS §Kinds: "Trailing or differing spaces/case are the same label —
    resolvers canonicalise." SESH-1098's root cause was canonicalising only the
    TYPED side (`"[%s]" % seg.upper()`) against a RAW subject, which made every
    label the walker prints but the standard does not spell — `[session]`,
    `2026-09-06` — unreachable by any spelling."""
    return re.sub(r"\s+", " ", s.strip()).casefold()

def match_child(seg, kids, is_item):
    """Resolve ONE path segment against a node's children (SESH-1098 fix-001).

    kids: [(index, subject)] — subjects exactly as the walker PRINTS them.
    Returns (hits, rule): the indices matched by the first tier that matched
    anything, and that tier's name; ([], None) if no tier matched.

    The tiers, in order. Literal first, so every name the walker prints
    round-trips to itself before any convenience rule can run — the walker's
    output language must be a subset of this resolver's input language, which
    is what "one resolver" means:

      exact        the printed name typed straight back: `/ASKS/[session]`
      bracket      the MHS bare label word: `/ASKS` -> "[ASKS]" (docstring above)
      canonical    case- and whitespace-insensitive, both sides, with and
                   without brackets: `/ASKS/session`, `/ASKS/SESSION`
      item-prefix  "10" -> "10 subject" — NUMBERED ITEMS ONLY

    item-prefix is gated on is_item deliberately. Measured in rca-001: a prefix
    rule on a LABEL segment turns a refusal into a wrong answer with rc 0 —
    `/WEEK/today`, on a day whose label was never written, opens whichever
    sibling was posted last. A number is an item's identity by construction; a
    label's name is not a prefix of its meaning."""
    seg_c = canon_name(seg)
    bracket = "[%s]" % seg
    bracket_c = canon_name(bracket)
    tiers = [("exact",     lambda sj: sj == seg),
             ("bracket",   lambda sj: sj == bracket),
             ("canonical", lambda sj: canon_name(sj) in (seg_c, bracket_c))]
    if is_item:
        tiers.append(("item-prefix", lambda sj: sj.startswith(seg + " ")))
    for rule, test in tiers:
        hits = [j for j, sj in kids if test(sj)]
        if hits:
            return hits, rule
    return [], None

def near_misses(seg, kids):
    """Children a LOOSER rule would have opened, and this one deliberately did not.

    Named in the refusal (charter 2) so a reader can see what was left shut and
    type the exact name: these are precisely the siblings that a prefix rule on
    a label segment would have handed back silently."""
    seg_c = canon_name(seg)
    return [sj for _j, sj in kids
            if canon_name(sj) != seg_c and canon_name(sj).startswith(seg_c + " ")]

def render(query, bodies=False, show_all=False, dates=False, under=None, depth=None, under_subject=None,
           owner=None, recip=None, shared=False):
    """Render one query. Returns (text, n_shown, n_hidden).

    owner: MHS 170 membership (Members: From, To AND signature) on every edge
    below the anchor, else the node and its whole subtree are not edges
    (hidden, counted, reported) — and only a member supersedes, shadows, or
    makes a fork. There is no unverified owner render: a path render without
    the signature was TARS's finding (Kith 13, 2026-09-28).
    Owner tags are printed only when the sender is NOT the owner — under
    membership that never happens, so a tag is a warning, not decoration."""
    n_unmember = 0
    member = Members(owner, recip, shared) if owner else None
    raw = sh(["notmuch", "show", "--format=json", "--body=true"] + query)
    if DRY:
        return ("", 0, 0)
    if not raw.strip():
        return ("(no messages match: %s)" % " ".join(query), 0, 0)
    threads = json.loads(raw)
    out_lines, n_shown, n_hidden = [], 0, 0
    flats = []
    for thread in threads:
        flat = []
        for top in thread:
            flatten(top, 0, flat)
        flats.append(flat)
    if under_subject and not under:
        # Anchor on a SUBJECT inside the matched thread (e.g. "[ROOT]") — the
        # FHS view: the root is a directory, find it by name, then `ls`.
        # Newest match wins so a superseding [ROOT] beats the original.
        # Only a MEMBER may be the anchor (or the primed hint itself): an
        # unsigned "[ROOT]" reply, newest in the thread, took over `ls /`.
        best = None
        for flat in flats:
            for _, m in flat:
                if m["headers"].get("Subject", "").strip() == under_subject:
                    if under_subject == "[ROOT]" and not is_anchor(m, member):
                        continue
                    if under_subject != "[ROOT]" and member and not member(m):
                        continue
                    if best is None or m.get("timestamp", 0) > best.get("timestamp", 0):
                        best = m
        if member:
            bts = best.get("timestamp", 0) if best else -1
            late = [m for flat in flats for _, m in flat
                    if m["headers"].get("Subject", "").strip() == under_subject
                    and m.get("timestamp", 0) > bts and member.unknown(m)]
            if late:
                sys.exit("memory: a newer %s could not be verified (verify timed out after %gs) — "
                         "refusing rather than anchor on an older one" % (under_subject, VERIFY_TIMEOUT))
        if best is None:
            sys.exit("memory: no message with subject %r in the matched thread(s)" % under_subject)
        under = best.get("id", "")
    if under:
        # Subtree scoping (Mike's ask, voice 2026-08-12: "see the tree with
        # subjects only from one message down"). notmuch has no subtree query —
        # `id:X` returns the WHOLE thread — so we render the thread and cut.
        # A message's descendants are exactly the entries that follow it in the
        # flattened walk while depth stays GREATER than its own; the first
        # entry back at its depth is its next sibling. Depths are re-based so
        # the chosen message prints flush left.
        want = under if under.startswith("<") else "<%s>" % under
        want = want.strip("<>")
        cut = []
        for flat in flats:
            for i, (d, m) in enumerate(flat):
                if m.get("id", "").strip("<>") == want:
                    sub = [(0, m)]
                    for d2, m2 in flat[i + 1:]:
                        if d2 <= d:
                            break
                        sub.append((d2 - d, m2))
                    cut.append(sub)
                    break
        if not cut:
            # Refuse loudly: silently rendering the whole thread would answer a
            # narrow question with a wide, expensive lie.
            sys.exit("memory: --under %s not found in the matched thread(s)" % under)
        flats = cut
    # Anchored = the top of the render was CHOSEN (a walk, --under, or an
    # anchor subject) and checked there. Unanchored (--index), depth 0 is
    # whatever the thread's top-level nodes are — an unsigned mail threaded
    # in by References sits there — so depth 0 gets the anchor test too.
    anchored = bool(under)
    if depth is not None:
        # `ls` not `find`: Mike's FHS rule (voice 2026-08-23) — the root holds
        # only directories, so a wake lists them and descends on demand.
        flats = [[(d, m) for d, m in flat if d <= depth] for flat in flats]
    if member:
        member.prefetch([m for flat in flats for _, m in flat])
    dead = member_pairs(flats, member) if not show_all else set()
    glatest = {}
    gtags = {}   # leading tag like [R01] / R03 -> newest timestamp. Same tag,
                 # latest wins: replying with the SAME rule tag supersedes it
                 # (R02 culture applied to rules; Mike asked for this by voice).
    TAG = re.compile(r'^\[?([A-Z]{1,3}\d{1,3})\]?(?=[\s\u2014:.\-])')
    if not show_all:
        for flat in flats:
            for _, m in flat:
                if member and not member.counts(m):
                    continue    # an unsigned "R05 …" or "NAME v9" shadows nothing;
                                # an unknown one still does (it may be real)
                s = m["headers"].get("Subject", "")
                vm = re.match(r'([A-Z][A-Z-]+)\s+v(\d+)', s)
                if vm:
                    glatest[vm.group(1)] = max(glatest.get(vm.group(1), 0), int(vm.group(2)))
                tm = TAG.match(s)
                if tm:
                    ts = m.get("timestamp", 0)
                    gtags[tm.group(1)] = max(gtags.get(tm.group(1), 0), ts)
    # FORK detection (MHS 110; [PROPOSALS] 40, Mike ruled 2026-08-23): two
    # SURVIVING items with the same number under one parent. Refuse to pick
    # — report the tie and hide both, never silently take the newest.
    NUM = re.compile(r'^(\d+(?:\.\d+)*)\s')
    forks = {}      # (parent_id, number) -> count of survivors
    lagging = {}    # (parent_id, number) -> how many of them timed out
    if not show_all:
        for flat in flats:
            stack = {}  # depth -> id
            for depth, m in flat:
                stack[depth] = m.get("id")
                s = m["headers"].get("Subject", "")
                if m.get("id") in dead:
                    continue
                if member and depth > 0 and not member.counts(m):
                    continue    # an unsigned twin is not a survivor: no FORK
                                # (an unknown one is: the pick is withheld)
                nm = NUM.match(s)
                if nm:
                    key = (stack.get(depth - 1), nm.group(1))
                    forks[key] = forks.get(key, 0) + 1
                    if member and member.unknown(m):
                        lagging[key] = lagging.get(key, 0) + 1
        forks = {k: v for k, v in forks.items() if v > 1}
    fork_said = set()
    for flat in flats:
        # Only a member's "Superseded by …" retires a listing (review of the member walk,
        # 19:52 Mon 2026-09-28: an unsigned one hid all of `ls /`).
        # An unknown one (verify timed out) withholds it too, and says so.
        sup = next((m for d, m in flat if d <= 1
                    and re.match(r'(?i)superseded by', m["headers"].get("Subject", ""))
                    and (not member or member.counts(m))), None)
        root_subj = sup["headers"].get("Subject", "") if sup else ""
        if not show_all and root_subj:
            if member and member.unknown(sup):
                out_lines.append(f"⊘ {root_subj}  — UNVERIFIED (verify timed out after {VERIFY_TIMEOUT:g}s): "
                                 f"{len(flat)} msgs withheld, since it may retire them")
            else:
                out_lines.append(f"⊘ {root_subj}  ({len(flat)} msgs skipped — superseded)")
            n_hidden += len(flat)
            continue
        prune = None
        stack = {}
        for depth, m in flat:
            s = m["headers"].get("Subject", "")
            stack[depth] = m.get("id")
            nm = NUM.match(s)
            if nm and (stack.get(depth - 1), nm.group(1)) in forks:
                key = (stack.get(depth - 1), nm.group(1))
                n_hidden += 1
                if key not in fork_said:
                    fork_said.add(key)
                    # A twin whose verify timed out still withholds the pick,
                    # and the line says so (review low, 22:46 Mon 2026-09-28:
                    # "FORK ... supersede one" hid the timeout).
                    num, n, lag = nm.group(1), forks[key], lagging.get(key, 0)
                    why = f"{lag} UNVERIFIED (verify timed out after {VERIFY_TIMEOUT:g}s)"
                    if not lag:
                        line = f"FORK: {num} x{n} — refusing to pick (MHS 110); supersede one"
                    elif n - lag >= 2:
                        line = f"FORK: {num} x{n}, {why} — refusing to pick (MHS 110); supersede one"
                    else:
                        line = f"{num} x{n} withheld — {why} may make a FORK (MHS 110); re-run"
                    out_lines.append(f"{'  ' * depth}└ ⊘ {line}")
                prune = depth
                continue
            vm = re.match(r'([A-Z][A-Z-]+)\s+v(\d+)', s)
            if vm and int(vm.group(2)) < glatest.get(vm.group(1), 0):
                n_hidden += 1
                continue
            tm = TAG.match(s)
            if tm and m.get("timestamp", 0) < gtags.get(tm.group(1), 0):
                n_hidden += 1
                continue
            if m.get("id") in dead:
                n_hidden += 1
                continue
            if prune is not None:
                if depth > prune:
                    n_hidden += 1
                    continue
                prune = None
            if member and (depth > 0 or not anchored):
                h = m["headers"]
                ok = member(m) if depth > 0 else is_anchor(m, member)
                if not ok:
                    n_hidden += 1
                    n_unmember += 1
                    if member.unknown(m):
                        out_lines.append(f"{'  ' * depth}└ ⊘ UNVERIFIED (verify timed out): {s[:60]}   [{short_from(h)}] — subtree hidden; what it supersedes stays retired")
                    else:
                        out_lines.append(f"{'  ' * depth}└ ⊘ NOT A MEMBER: {s[:60]}   [{short_from(h)}] — subtree hidden (MHS 170)")
                    prune = depth
                    continue
            n_shown += 1
            indent = "  " * depth + ("└ " if depth else "")
            # Date suffix off by default (Mike: "the date's in the email") —
            # -d restores. Short sender stays: cheap, and who-said-it matters.
            who = short_from(m["headers"])
            tag = "" if owner and who == owner and not shared else f"   [{who}]"
            if dates:
                dm = re.search(r'(\d{1,2}\s+\w{3}\s+\d{4})', m["headers"].get("Date", ""))
                tag = f"   [{who} {dm.group(1) if dm else ''}]"
            out_lines.append(f"{indent}{s}{tag}")
            if bodies:
                bt = body_text(m)
                if bt:
                    pad = "  " * (depth + 1)
                    out_lines.extend(pad + l for l in bt.splitlines())
                    out_lines.append("")
    render.unmember = n_unmember
    return ("\n".join(out_lines), n_shown, n_hidden)

def last_session_query(agent, cur):
    lines = sh(["notmuch", "search", "--sort=newest-first", "--output=summary",
                f'to:sessions and subject:"SESSION UP" and from:{agent}'], resolve=True).splitlines()
    for ln in lines:
        if cur and cur in ln:
            continue
        return [ln.split()[0]]
    return None

def main():
    argv = sys.argv[1:]
    # Unix verbs over MHS paths (Mike, voice 2026-08-23 19:45): ls = one
    # level, tree = the subtree (--depth N), cat = one item with its body.
    # `/` alone is the root. boot and wake stop being verbs: they are
    # `ls /` + `cat /BOOT/latest` and `run /BOOT/WAKE` in the agent MD.
    _boxpair = []
    if "--box" in argv:
        _i = argv.index("--box"); _boxpair = argv[_i:_i + 2]; del argv[_i:_i + 2]
    if argv and argv[0] in ("ls", "tree", "cat"):
        verb = argv.pop(0)
        target = argv.pop(0) if argv and not argv[0].startswith("-") else "/"
        if verb == "ls" and target == "/":
            argv = ["--ls"] + argv
        else:
            argv = ["--path", target] + argv
            if verb == "ls" and "--depth" not in argv:
                argv += ["--depth", "1"]
            if verb == "cat":
                argv += ["--depth", "0", "-b"]
    argv = _boxpair + argv
    bodies = any(a in ("-b", "--bodies") for a in argv)
    show_all = any(a in ("-a", "--all") for a in argv)
    dates = any(a in ("-d", "--dates") for a in argv)
    global DRY
    DRY = any(a in ("-n", "--dry-run") for a in argv)
    if any(a in ("-h", "--help") for a in argv):
        sys.exit(__doc__)
    argv = [a for a in argv if a not in ("-b", "--bodies", "-a", "--all", "-d", "--dates", "-n", "--dry-run")]
    # --under takes a VALUE, so it is consumed as a pair before the unknown-flag
    # guard below (which would otherwise reject the bare msgid as a stray arg).
    # --box NAME (Mike, voice 2026-08-23 19:13 — "commons"): render a SHARED
    # box. Root hint is $<NAME>_ROOT; membership per edge is To the box (any
    # plus-tag) AND signed — From is any crew member, not one owner. The
    # personal rule (From the owner) is the default when --box is absent.
    box = None
    if "--box" in argv:
        i = argv.index("--box")
        box = argv[i + 1]
        del argv[i:i + 2]
    path = None
    if "--path" in argv:
        i = argv.index("--path")
        if i + 1 >= len(argv):
            sys.exit("memory: --path needs a PATH like /CONVO/today/ASKS\n" + __doc__)
        path = argv[i + 1]
        del argv[i:i + 2]
    resolve_only = "--resolve" in argv
    argv = [a for a in argv if a != "--resolve"]
    depth = None
    if "--depth" in argv:
        i = argv.index("--depth")
        if i + 1 >= len(argv) or not argv[i + 1].isdigit():
            sys.exit("memory: --depth needs a non-negative integer\n" + __doc__)
        depth = int(argv[i + 1])
        del argv[i:i + 2]
    under = None
    if "--under" in argv:
        i = argv.index("--under")
        if i + 1 >= len(argv):
            sys.exit("memory: --under needs a MSGID\n" + __doc__)
        under = argv[i + 1]
        del argv[i:i + 2]
        if not argv:
            # No query given: the message's own id pulls its whole thread
            # (notmuch show is entire-thread by default), which we then cut.
            argv = ["id:" + under.strip("<>")]
    # Unknown dash-args must fail LOUD, not become notmuch queries: "-h" once
    # leaked through as a query and printed 317 messages of Mike's comms
    # (~11k tokens). Substrate's first law: a tool that cannot fail loudly lies.
    for a in argv:
        if a.startswith("-") and a not in ("--wake", "--last-session", "--index", "--ls", "--boot"):
            sys.exit(f"memory: unknown flag {a!r} — refusing to treat it as a query\n{__doc__}")
    # Identity from the environment, the way `session mail arm` resolves it
    # (CA-209): CLAUDE_CODE_AGENT first, SESSION_AGENT second. No default —
    # a thread reader that silently assumes an owner reads the wrong memory.
    agent = (os.environ.get("CLAUDE_CODE_AGENT") or os.environ.get("SESSION_AGENT") or "").strip()
    if not agent and not box:
        sys.exit("memory: no agent — set CLAUDE_CODE_AGENT (or SESSION_AGENT), or pass --box NAME")
    cur = os.environ.get("CLAUDE_SESSION_ID", "")[:8]
    # Recipient principal for membership and the root-hint variable.
    recip = box or agent
    root_var = (box or agent).upper().replace("-", "_").replace(".", "_") + "_ROOT"

    def root_thread():
        # The canonical root (Mike's design): categories hang off one message.
        # ROOT HINT (Mike, voice 2026-08-23): $PIP_ROOT, minted by the launcher
        # from the agent MD's environment block — the layer with the fewest
        # writers. Priming check (MHS 90): the hint must resolve to a message
        # whose subject is [ROOT] and From: the agent, or we refuse loudly.
        # Fallback when the env var is absent (old launcher, shipwreck): the
        # subject search, and it says so.
        raw = os.environ.get(root_var)
        hint = (raw or "").strip().strip("<>").strip()
        if raw is not None and not hint:
            # SET BUT EMPTY ('<>', '  ', '< >') is a launcher that tried and
            # failed — not an old launcher. The shipwreck search here printed
            # "$VAR = thread:… ✓", a tick for a hint never used (Pip, 20:02
            # Sat 2026-09-26). Refuse; unset is the announced fallback.
            sys.exit(f"memory: PRIMING FAILED — ${root_var} is set but empty ({raw!r}). "
                     "Refusing rather than guessing the root by subject (MHS 90); "
                     f"unset ${root_var} to take the announced shipwreck search.")
        if hint:
            raw = sh(["notmuch", "show", "--format=json", "--body=false",
                      "--entire-thread=false", f"id:{hint}"], resolve=True)
            try:
                msgs = json.loads(raw) if raw.strip() else []
                flat = []
                for t in msgs:
                    for top in t:
                        flatten(top, 0, flat)
                m = flat[0][1] if flat else None
            except Exception:
                m = None
            if not m:
                sys.exit(f"memory: PRIMING FAILED — ${root_var} <{hint}> is not in the store. "
                         "Not rendering an empty tree (MHS 90).")
            subj = m["headers"].get("Subject", "").strip()
            frm = short_from(m["headers"])
            if subj != "[ROOT]" or (not box and frm != agent):
                sys.exit(f"memory: PRIMING FAILED — ${root_var} resolves to {subj!r} from {frm!r}, "
                         f"not [ROOT] from {agent!r}. Refusing (MHS 90).")
            global PRIMED
            PRIMED = hint
            return f"id:{hint}"
        sys.stderr.write(f"memory: no ${root_var} in the environment — falling back to the subject search (shipwreck path)\n")
        tid = sh(["notmuch", "search", "--output=threads", "--limit=1",
                  "--sort=oldest-first",
                  (f'subject:"[ROOT]" and to:{box}' if box else f'subject:"{agent} ROOT" and from:{agent}')], resolve=True).strip()
        if not tid:
            sys.exit(f"no canonical root found (subject:\"{agent} ROOT\")")
        return tid

    if argv and argv[0] == "--index":
        text, shown, hidden = render([root_thread()], bodies=bodies, show_all=show_all,
                                     dates=dates, depth=depth, owner=agent, recip=recip, shared=bool(box))
        if DRY:
            return
        print(text)
        print(f"\n— {shown} categories+root shown, {hidden} filtered · ≈{len(text) // 4} tok")
        return

    if path is not None:
        # MHS path resolution (item 70, Mike 2026-08-23): traversal, no table.
        # Name = subject token, address = Message-ID, edge = In-Reply-To.
        # Newest surviving child wins at each step (supersede-aware).
        import datetime as _dt
        tid = root_thread()
        raw = sh(["notmuch", "show", "--format=json", "--body=true", tid])
        if DRY:
            return
        threads = json.loads(raw)
        flats = []
        for thread in threads:
            flat = []
            for top in thread:
                flatten(top, 0, flat)
            flats.append(flat)
        member = Members(agent, recip, bool(box))
        pairs = member_pairs(flats, member)
        dead = pairs
        _by = {}
        def index_of(flat):
            if id(flat) not in _by:
                _by[id(flat)] = {m.get("id", "").strip("<>"): k for k, (_d, m) in enumerate(flat)}
            return _by[id(flat)]
        def direct(flat, i):
            d, out = flat[i][0], []
            for j in range(i + 1, len(flat)):
                if flat[j][0] <= d:
                    break
                if flat[j][0] == d + 1:
                    out.append(j)
            return out
        def chain(flat, j):
            # j and every mail a Supersedes: chain from j could hand the walk:
            # raw headers, nothing verified — it only says which chains matter.
            by, out, todo = index_of(flat), [j], [j]
            while todo:
                for sid in pairs.raw.get(flat[todo.pop()][1].get("id", "").strip("<>"), []):
                    k = by.get(sid)
                    if k is not None and k not in out:
                        out.append(k)
                        todo.append(k)
            return out
        def children(flat, i, want=None):
            # A walk step is an edge: only a member is a child (MHS 170).
            # want(subject): resolve only the chains holding a subject it
            # accepts. Every link of a chain must be verified (v1 is dead only
            # if v2 is a member…), so resolving all ~190 [BOOT] letters to
            # open /BOOT/WAKE cost 262 verifies (measured 19:08 Mon 2026-09-28).
            ds = direct(flat, i)
            if want is not None:
                ds = [j for j in ds if any(want(subj(k)) for k in chain(flat, j))]
            pairs.prefetch([flat[j][1].get("id", "") for j in ds])
            eff = effective_children(flat, i, dead, pairs, only=set(ds))
            member.prefetch([flat[j][1] for j in eff])
            return [j for j in eff if member(flat[j][1])]
        def latest(flat, i):
            # The newest SURVIVING numbered child — the max() over children()
            # it replaces, found newest-first so [BOOT] costs a verify or two.
            # A candidate survives iff it is a member, not dead, and reached
            # from a direct child through member superseders: the walk
            # effective_children makes, run backwards.
            ds = set(direct(flat, i))
            by = index_of(flat)
            cands = {k for j in ds for k in chain(flat, j)}
            targets = {}
            for t, sids in pairs.raw.items():
                for sid in sids:
                    targets.setdefault(sid, []).append(t)
            def reached(k, seen):
                if k in ds:
                    return True
                for t in targets.get(flat[k][1].get("id", "").strip("<>"), []):
                    kt = by.get(t)
                    if kt is None or kt in seen:
                        continue
                    seen.add(kt)
                    if kt in ds or (member(flat[kt][1]) and reached(kt, seen)):
                        return True
                return False
            order = sorted((k for k in cands if re.match(r'^\d', subj(k))),
                           key=lambda k: flat[k][1].get("timestamp", 0), reverse=True)
            for n, k in enumerate(order):
                if n % 16 == 0:     # verify the next few in parallel, not one by one
                    member.prefetch([flat[x][1] for x in order[n:n + 16]])
                m = flat[k][1]
                if member.unknown(m):
                    raise Unverifiable(subj(k))     # newer than any answer: refuse
                mid_ = m.get("id", "").strip("<>")
                if member(m) and not pairs.get(mid_) and pairs.blocked(mid_):
                    # Retired only by superseders nobody could check: it may be
                    # the latest still, so there is no answer to fall back to
                    # (review, 22:03 Mon 2026-09-28: `latest` gave "05 older").
                    raise Unverifiable(pairs.by_id[pairs.blocked(mid_)[0]]["headers"].get("Subject", "").strip())
                if not member(m) or mid_ in dead:
                    continue
                # Its ancestry, verified in parallel before reached() walks it
                # link by link: /CONVO/latest's answer hangs off a 56-link
                # reply chain (3.4 s in series, measured 19:13 Mon 2026-09-28).
                anc, todo = set(), [m.get("id", "").strip("<>")]
                while todo:
                    for t in targets.get(todo.pop(), []):
                        kt = by.get(t)
                        if kt is not None and kt not in ds and kt not in anc:
                            anc.add(kt)
                            todo.append(t)
                member.prefetch([flat[x][1] for x in anc])
                for x in anc:
                    if member.unknown(flat[x][1]):
                        raise Unverifiable(subj(x))
                if reached(k, {k}):
                    return k
            return None
        def subj(j):
            return flat[j][1]["headers"].get("Subject", "").strip()
        def step_pick(flat, i, seg, is_item, ds, mem, prs):
            # What one walk step opens under these verdicts. A refusal is an
            # answer too: ("none",) for no child, ("ambiguous", spellings).
            eff = effective_children(flat, i, prs, prs, only=ds)
            kids = [(j, subj(j)) for j in eff if mem(flat[j][1])]
            hits, rule = match_child(seg, kids, is_item)
            if not hits:
                return ("none",)
            spellings = frozenset(sj for j, sj in kids if j in hits)
            if rule == "canonical" and len(spellings) > 1:
                return ("ambiguous", spellings)
            return ("pick", max(hits, key=lambda j: flat[j][1].get("timestamp", 0)))
        LAG_CAP = 8     # at most 2^8 assignments to weigh; past that, refuse
        def weigh_step(flat, i, seg, is_item, want, lag):
            # None when no verdict for the timed-out mails in `lag` can change
            # what this step opens; otherwise why the walk refuses. Every
            # true/false assignment is tried, not just all-in and all-out: a
            # mixed one can open what neither corner does ([LAGMIX]).
            if len(lag) > LAG_CAP:
                return ("there are too many to weigh (%d > %d) — refusing rather than open "
                        "anything without them" % (len(lag), LAG_CAP))
            ds = {j for j in direct(flat, i) if any(want(subj(k)) for k in chain(flat, j))}
            ids = [flat[k][1].get("id", "") for k in lag]
            picks = set()
            for bits in range(1 << len(ids)):
                mem = AssumedMembers(member, {mid: bool(bits >> n & 1) for n, mid in enumerate(ids)})
                picks.add(step_pick(flat, i, seg, is_item, ds, mem, pairs.assuming(mem)))
                if len(picks) > 1:
                    return "may change which item this opens — refusing"
            if step_pick(flat, i, seg, is_item, ds, member, pairs) not in picks:
                # Every verdict gives one answer and the reader without them
                # gives another: fail closed rather than trust either.
                return ("every verdict for them gives one answer, but the walk without them "
                        "gives another — refusing (fail closed)")
            return None
        # locate [ROOT]
        cur = None
        for flat in flats:
            for j, (d, m) in enumerate(flat):
                if m["headers"].get("Subject", "").strip() == "[ROOT]" and is_anchor(m, member):
                    if cur is None or m.get("timestamp", 0) > flat[cur][1].get("timestamp", 0):
                        cur = j
            if cur is not None:
                break
        if cur is None:
            sys.exit("memory: no [ROOT] in the canonical thread (priming check failed, item 90)")
        for flat in flats:
            for j, (d, m) in enumerate(flat):
                if (m["headers"].get("Subject", "").strip() == "[ROOT]"
                        and m.get("timestamp", 0) > flat[cur][1].get("timestamp", 0)
                        and member.unknown(m)):
                    sys.exit("memory: a newer [ROOT] could not be verified (verify timed out "
                             "after %gs) — refusing rather than walk from an older one" % VERIFY_TIMEOUT)
        segs = [x for x in path.strip().split("/") if x]
        walked = "/"
        for seg in segs:
            if seg == "today":
                seg = _dt.date.today().isoformat()
            elif seg == "yesterday":
                seg = (_dt.date.today() - _dt.timedelta(days=1)).isoformat()
            is_item = bool(re.match(r'^\d+(\.\d+)*$', seg))   # 10, 10.10 = items; dates = labels
            # `latest` (MHS 70 magic name): the newest SURVIVING numbered item
            # under this label — `cat /BOOT/latest` is the letter.
            if seg == "latest":
                try:
                    nxt = latest(flat, cur)
                except Unverifiable as e:
                    sys.exit("memory: %s/latest: %r could not be verified (verify timed out "
                             "after %gs) — refusing rather than open an older item"
                             % (walked.rstrip("/"), str(e), VERIFY_TIMEOUT))
                if nxt is None:
                    sys.exit("memory: %s has no numbered items for 'latest'" % walked)
                cur = nxt
                walked = walked.rstrip("/") + "/" + subj(cur).split()[0]
                continue
            want = lambda sj: bool(match_child(seg, [(0, sj)], is_item)[0])
            kids = [(j, subj(j)) for j in children(flat, cur, want=want)]
            hits, rule = match_child(seg, kids, is_item)
            # Any mail in a matching chain whose verify timed out may be the
            # answer (a newer duplicate) or retire it (review, 22:03 Mon
            # 2026-09-28). Not every one can: a v1 already retired by a signed
            # v2, or an older duplicate beside a newer signed one, changes
            # nothing, and refusing there declined the newest item (review
            # low, 22:46 Mon 2026-09-28). So weigh them, and refuse only when
            # a verdict could change what this step opens.
            lag = []
            for j in direct(flat, cur):
                ch = chain(flat, j)
                if any(want(subj(k)) for k in ch):
                    lag += [k for k in ch if k not in lag and member.unknown(flat[k][1])]
            if lag:
                why = weigh_step(flat, cur, seg, is_item, want, lag)
                if why:
                    sys.exit("memory: mhs-thread resolver: %s/%s: %s could not be verified (verify "
                             "timed out after %gs) and %s"
                             % (walked.rstrip("/"), seg, [subj(k) for k in lag], VERIFY_TIMEOUT, why))
            if not hits or (rule == "canonical" and len({sj for j, sj in kids if j in hits}) > 1):
                # A refusal names EVERY child and every near miss, so it
                # resolves them all — the slow path, taken only to say no.
                kids = [(j, subj(j)) for j in children(flat, cur)]
            names = [sj[:40] for _j, sj in kids]
            if not hits:
                # Charter 2: name the resolver, the rules it tried, and what it
                # deliberately did NOT open — a refusal a reader can act on.
                missed = near_misses(seg, kids)
                rules = "exact, bracket, canonical" + (", item-prefix" if is_item
                                                       else " (item-prefix is for numbered items only)")
                sys.exit("memory: mhs-thread resolver: %s has no child %r.\n"
                         "  rules tried: %s\n%s"
                         "  children: %s"
                         % (walked, seg, rules,
                            ("  near misses NOT opened (type the exact name): %s\n" % missed) if missed else "",
                            names))
            spellings = sorted({sj for j, sj in kids if j in hits})
            if rule == "canonical" and len(spellings) > 1:
                # A fork of spellings (MHS 110): both are equally the label, so
                # picking one would be a guess wearing rc 0. Refuse by name.
                sys.exit("memory: mhs-thread resolver: %s child %r is AMBIGUOUS — the "
                         "canonical rule matched more than one spelling: %s.\n"
                         "  type the exact name (the exact rule is tried first, and never guesses).\n"
                         "  children: %s" % (walked, seg, spellings, names))
            # Duplicates of ONE spelling: newest wins, as before (supersede-aware).
            cur = max(hits, key=lambda j: flat[j][1].get("timestamp", 0))
            # The trail echoes what you TYPED while the match was literal, and
            # the child's REAL name when the resolver had to canonicalise — so
            # `/ASKS/session` reports back that it opened `[session]`, and every
            # path that worked before still reads exactly as it did.
            walked = walked.rstrip("/") + "/" + (subj(cur) if rule == "canonical" else seg)
        # HOIST the walked node through its Supersedes chain (MHS: a supersede
        # replaces IN PLACE; the superseder is a REPLY so it sits deeper and the
        # raw walk lands on the anchor). Mirrors effective_children/survivors —
        # without this, `cat /BOOT/latest` reads a stale letter (found 12:43
        # Mon 2026-08-24: cat showed v172 while --path showed v174).
        if not show_all:
            by_id = {m.get("id", "").strip("<>"): j for j, (_d, m) in enumerate(flat)}
            hseen = set()
            while True:
                cid = flat[cur][1].get("id", "").strip("<>")
                cands = [by_id[s] for s in pairs.get(cid, []) if s in by_id and by_id[s] not in hseen]
                if not cands:
                    break
                cur = max(cands, key=lambda j: flat[j][1].get("timestamp", 0))
                hseen.add(cur)
        mid = flat[cur][1].get("id", "")
        if resolve_only:
            print("<%s>" % mid)
            return
        text, shown, hidden = render([tid], bodies=bodies, show_all=show_all, dates=dates,
                                     under=mid, depth=depth, owner=agent, recip=recip, shared=bool(box))
        print(text)
        print(f"— {walked} = <{mid}> · {shown} shown, {hidden} filtered · ≈{len(text) // 4} tok")
        return

    if argv and argv[0] in ("--ls", "--boot"):
        # `ls /` — FHS view of memory (Mike, voice 2026-08-23): the root holds
        # only labels (directories), never items (files). Depth 1 lists them.
        tid = root_thread()
        text, shown, hidden = render([tid], show_all=show_all, dates=dates,
                                     under_subject="[ROOT]", depth=depth if depth is not None else 1,
                                     owner=agent, recip=recip, shared=bool(box))
        if DRY and argv[0] == "--ls":
            return
        if not DRY:
            # One line says where / comes from and that it primed (Mike, voice
            # 2026-08-23 18:54): the VARIABLE so a reader knows the layer it
            # lives in, the ID so the tree is self-identifying, and the
            # verdict. Per-node owner tags are gone — every edge below is
            # membership-checked and signed, so "all pip" is one word here.
            src = f"${root_var}" if os.environ.get(root_var) else f"subject search (no ${root_var} — shipwreck)"
            print(f"═══ ls /{' ' + box if box else ''}   {src} = {tid[3:] if tid.startswith('id:') else tid} · [ROOT] {'to ' + box if box else 'from ' + agent} ✓ ═══")
            print(text)
            bad = getattr(render, "unmember", 0)
            verdict = (f"all to {box}, signed" if box else f"all {agent}, signed") if not bad else f"{bad} NOT MEMBERS (see ⊘)"
            if shown == 0:
                # Nothing shown is not "all signed": say what hid it.
                verdict = f"NOTHING SHOWN, {hidden} hidden (see ⊘)"
            print(f"— {max(shown - 1, 0)} labels · {verdict} · ≈{len(text) // 4} tok")
        if argv[0] == "--ls":
            return
        # --boot: the newest SURVIVING item under [BOOT] is the letter to self.
        # Trust before read (pip-mail): the file must verify under
        # `session mail-verify` — signature AND principal — or we refuse.
        if DRY:
            # Dry-run only: show the command a [BOOT] render would run. Its
            # output was never used otherwise, so it no longer runs.
            render([tid], show_all=show_all, dates=dates, under_subject="[BOOT]", depth=1)
            return
        print("\n═══ [BOOT] — the letter to self (newest surviving item) ═══")
        # Re-walk for the newest surviving child of [BOOT] to get its file.
        raw = sh(["notmuch", "show", "--format=json", "--body=true", tid])
        threads = json.loads(raw)
        flats = []
        for thread in threads:
            flat = []
            for top in thread:
                flatten(top, 0, flat)
            flats.append(flat)
        # Membership on the anchor, the items and their supersedes (MHS 170):
        # an unsigned "[BOOT]" or a newer unsigned item used to be what this
        # picked, and the verify below then refused the whole wake.
        member = Members(agent, recip, bool(box))
        boot = None
        late = []
        for flat in flats:
            for i, (d, m) in enumerate(flat):
                if m["headers"].get("Subject", "").strip() == "[BOOT]":
                    if member(m):
                        boot = (flat, i, d)
                    elif member.unknown(m):
                        late.append(m)
        if late and (boot is None or any(u.get("timestamp", 0) > boot[0][boot[1]][1].get("timestamp", 0)
                                         for u in late)):
            sys.exit("memory: REFUSING — a newer [BOOT] label could not be verified (verify timed "
                     "out after %gs); not reading an older one" % VERIFY_TIMEOUT)
        if boot is None:
            sys.exit("memory: no signed [BOOT] label under the root — create it: reply to [ROOT] "
                     "with subject '[BOOT]', no body (R15)")
        flat, i, d = boot
        kids = []
        for d2, m2 in flat[i + 1:]:
            if d2 <= d:
                break
            if d2 == d + 1:
                kids.append(m2)
        dead = member_pairs([[(0, k) for k in kids]], member)
        # Only ITEMS (leading number) can be the letter; a sub-label like
        # [WAKE] is a directory, never the letter (MHS kinds, item 140).
        # Newest first, and stop at the first member: one verify, not ~190.
        items = [k for k in kids if re.match(r'^\d', k["headers"].get("Subject", "").strip())]
        items.sort(key=lambda m: m.get("timestamp", 0), reverse=True)
        item = None
        for k in items:
            if member.unknown(k):
                sys.exit("memory: REFUSING — the newest [BOOT] item %r could not be verified "
                         "(verify timed out after %gs); not falling back to an older letter"
                         % (k["headers"].get("Subject", "").strip(), VERIFY_TIMEOUT))
            kid_ = k.get("id", "").strip("<>")
            if member(k) and not dead.get(kid_) and dead.blocked(kid_):
                # Retired only by superseders nobody could check (an
                # unnumbered one included): it may still be the letter.
                sys.exit("memory: REFUSING — the newest [BOOT] letter %r is retired only by a mail "
                         "that could not be verified (verify timed out after %gs); not reading an older one"
                         % (k["headers"].get("Subject", "").strip(), VERIFY_TIMEOUT))
            if member(k) and k.get("id") not in dead:
                item = k
                break
        if item is None:
            member.prefetch(kids)
            lagging = [x["headers"].get("Subject", "").strip() for x in kids if member.unknown(x)]
            if lagging:
                sys.exit("memory: REFUSING — no signed, surviving letter under [BOOT], and %s could "
                         "not be verified (verify timed out after %gs)" % (lagging, VERIFY_TIMEOUT))
            if items:
                sys.exit("memory: REFUSING — [BOOT] holds %d item(s) and none is a signed, "
                         "surviving letter (MHS 170)" % len(items))
            print("(the [BOOT] label is empty — no letter to self yet)")
            return
        live = [item]
        fn = next((f for f in item.get("filename", []) if os.path.exists(f)), None)
        if not fn:
            sys.exit("memory: boot item has no readable file")
        try:
            v = subprocess.run([*VERIFY, fn], capture_output=True, text=True,
                               timeout=VERIFY_TIMEOUT)
        except subprocess.TimeoutExpired:
            sys.exit("memory: REFUSING boot item — session-mail-verify timed out after %gs"
                     % VERIFY_TIMEOUT)
        if v.returncode != 0:
            sys.exit("memory: REFUSING boot item — session-mail-verify rc=%d\n%s%s"
                     % (v.returncode, v.stdout, v.stderr))
        hdr = item["headers"]
        print(f"{hdr.get('Subject','')}   [{short_from(hdr)} · {hdr.get('Date','')} · sig OK]")
        print(f"  id: {item.get('id','')}")
        for l in body_text(item).splitlines():
            print("  " + l)
        n_old = len(items) - 1
        print(f"— {n_old} other boot item(s): superseded, older, or not members · ≈{len(body_text(item)) // 4} tok")
        return

    if argv and argv[0] == "--wake":
        # Delegation (Mike's ruling by voice, 2026-08-14): the base readout —
        # BOOT (the +boot shelf) + RECENT + COUNTS — has ONE implementation
        # now, `session mail wake` (ported FROM this script's RECENT, then
        # generalised; session repo 294b4b4). This script no longer carries a
        # second copy: it delegates, then adds the two pip-specific sections
        # the generic verb rightly doesn't know: RULES and LAST SESSION.
        # No legacy fallback on purpose — one implementation, two callers;
        # a fallback would quietly become the second implementation again.
        cmd = [os.path.join(_HERE, "session-mail-wake"), agent] + (["-n"] if DRY else [])
        if DRY:
            print(f"  delegate     : {shlex.join(cmd)}")
        # Flush before handing stdout to the child: the parent is block-
        # buffered when piped, the child writes straight through — without
        # this the delegate line (and any earlier print) lands AFTER the
        # child's entire readout (observed on this branch's first dry run).
        sys.stdout.flush()
        try:
            r = subprocess.run(cmd)
        except FileNotFoundError:
            sys.exit(f"memory: delegation failed — {cmd[0]} is missing "
                     "(a broken sessionmail install)")
        if r.returncode != 0:
            sys.exit(f"memory: delegation failed — {cmd[0]} exited {r.returncode}")
        total_cost = 0
        # RULES: resolve the rules thread by its root subject. (NOT to:{agent}+rules
        # — notmuch tokenizes on '+', which matches ALL the agent's mail. Found the
        # hard way: 29k tokens.)
        tids = []
        # oldest-first: the TRUE root predates commentary that quotes its
        # phrase (tonight's RULES index page quotes "PIP MAIL RULES" and
        # newest-first resolved BOTH phrases to the how-to thread).
        for phrase in (f"{agent} MAIL HOW-TO", f"{agent} MAIL RULES",
                       f"{agent} RULES"):
            t = sh(["notmuch", "search", "--output=threads", "--limit=1",
                    "--sort=oldest-first",
                    f'subject:"{phrase}" and from:{agent}'], resolve=True).strip()
            if t and t not in tids:
                tids.append(t)
        for title, q in [("HOW-TO + RULES — index pages, then the law",
                          [" or ".join(tids)] if tids else None),
                         ("LAST SESSION — predecessor's registry thread",
                          last_session_query(agent, cur))]:
            print(f"═══ {title} ═══")
            if not q:
                print("(none found)\n")
                continue
            text, shown, hidden = render(q, bodies=False, show_all=show_all, dates=dates)
            if DRY:
                continue
            print(text)
            cost = len(text) // 4
            total_cost += cost
            print(f"— {shown} shown, {hidden} filtered · ≈{cost} tok\n")
        print(f"═══ agent extras ≈{total_cost} tok — session wake total above ═══")
        return

    if argv and argv[0] == "--last-session":
        query = last_session_query(agent, cur)
        if not query:
            sys.exit("no previous session thread found in registry")
    elif argv:
        query = argv
    else:
        sys.exit(__doc__)

    text, n_shown, n_hidden = render(query, bodies, show_all, dates, under=under)
    if DRY:
        return
    print(text)
    cost = len(text) // 4
    note = "" if bodies else "; -b for bodies"
    print(f"\n— {n_shown} msgs shown, {n_hidden} filtered · ≈{cost} tok{note}")

if __name__ == "__main__":
    main()
