# Ephemera — Media archive consolidation (local macOS + CLI)

> Self-executing Markdown. The **local** binding of the *archive* intent — extract media out of an
> opaque application store (worked binding: macOS Messages `chat.db` + `Attachments/`), enrich each
> file with a real date and a human sender name, consolidate it into a dated review tree, fold in
> other loose folders with exact-duplicate detection, and verify every byte before deleting anything.
> The **filesystem is the source of truth**; this file is intent + ledger + audit.

> **Provides / Requires**: **Provides** `media-archive(<DEST>)` + `manifest(<DEST>/manifest.csv)` —
> the manifest is the reusable index other passes consume (see §6, which reads it instead of
> re-walking the tree). **Requires** nothing but read access to the source home folder and Full Disk
> Access for the terminal (§1 probes it).

---

## 🤖 Director prompt

You are the Director. Execute this plan:
- Observe-before-act; verify each step before advancing
- The source store is **read-only, always**. Copy, never move. Never open the source DB read-write
- **🔴 GATE before every 💥.** Deletion is the only irreversible act here, and on a network volume
  there is no Trash — a `rm` is final. No file is ever deleted until §7 has hash-verified a
  byte-identical counterpart exists in `$DEST`
- Never fabricate a date. A file with no recoverable timestamp goes to `Undated/` — filing it under a
  guessed year is a silent lie that survives into the user's photo library
- Write realized values + verify results back into Live State after each step
- On any failure, annotate Live State and stop

```
Legend  🟢 create · 🟡 config · 🔴 GATE (human go) · 💥 destructive (human go) · ⏳ wait · ✔ verify
```

> **Candor:** the learnings folded into this plan — the copy-timestamp trap, the size-index-then-hash
> reduction, the stale-AFP-fork failure, the epoch ambiguity, the orphan-attachment gap — come from a
> dogfooded run against a real ~7 GB Messages store over an AFP mount, observed live. This contributed
> copy ships with Live State reset to `not-created` per the contribution contract: treat every ✔ row
> as unverified until you re-run it against YOUR OWN source. A `photos-library` binding is **designed
> but not included here** (see *Deliberately not included*) — this file is the `messages` binding only.
>
> **Review-hardened at promotion 2026-07-23:** the teardown-path blockers were fixed (a recorded
> `LOOSE_DIRS` list, and the §7 gate re-proven before every 💥) and the two embedded scripts
> (`plan.py`, `verify_gate.py`) were **reconstructed from this plan's own spec** — the contributor's
> originals were not shipped; treat both as unverified until a live run proves them.

---

## Intent

Recover media that an application has locked inside an opaque store — filenames stripped of meaning,
dates buried in a database, senders reduced to phone numbers — and turn it into a plain folder tree a
non-technical human can review, keep, and delete from. The worked binding is macOS Messages
(`chat.db` + content-addressed `Attachments/`), where 6,000 files named `IMG_2784.JPG` sit in hashed
subdirectories with no date and no sender on disk.

The output is deliberately **dumb**: dated folders, descriptive filenames, no index, no app. It must
stay useful after every tool that made it is gone. That constraint is why the plan writes
`manifest.csv` in plain CSV and why sender and date live *in the filename* rather than in metadata.

**Shared acceptance contract** (every binding of the *archive* intent must pass — the honest-equivalence test):
1. **Nothing silently dropped** — every source file is copied, or classified as duplicate/filtered, and
   the counts reconcile three ways (script tally, `find`, manifest)
2. **No fabricated dates** — every file's date provenance is recorded, and a file with no recoverable
   date is marked undated rather than guessed
3. **Source path-set unchanged after the run** — every source path present at survey is still present
   (new files on a live source are allowed and listed), proven by a negative path-diff, not asserted
4. **Nothing deleted without proof** — no in-class source file is removed until a hash-identical
   counterpart is verified present elsewhere, or it is **exempt-by-class** (filtered as junk), and every
   exempt-by-class file is ENUMERATED at the gate for the human

## What you need, and why

| Prerequisite | Why this plan needs it | How to tell if you have it |
|---|---|---|
| **Full Disk Access** for your terminal | `~/Library/Messages/` is TCC-protected. Without it every read returns `Operation not permitted` — which looks like an empty archive, not an error | §1 probe; System Settings → Privacy & Security → Full Disk Access |
| **Read access to the source home** | A mounted volume, a Time Machine restore, or the live `~` | `ls "$SRC_HOME/Library/Messages/chat.db"` |
| **Free space ≥ source size** | This is a **copy**. Same-volume consolidation transiently doubles the footprint | §1 measures both and refuses to proceed if short |
| **`sqlite3` + `python3`** | Both ship with macOS | `sqlite3 -version && python3 -V` |

---

## Provisioning Inputs

Resolve **once**, up front, before any mutation. Director walks top-down, accepts the **default** on
silence, writes `resolved_inputs` into Live State. Same answers ⇒ same tree.

| # | Question | Options (closed enum) | Default | Sets | Gates |
|---|---|---|---|---|---|
| 1 | What kind of store are we extracting? | `messages` | `messages` | `SOURCE_KIND` | §2 index step |
| 2 | How should the review tree be organized? | `by-year` / `by-sender` / `flat` | `by-year` | `ORGANIZE_BY` | §4 naming |
| 3 | Include video? | `yes` / `no` | `yes` | `INCLUDE_VIDEO` | §0 `KEEP_EXT` |
| 4 | Drop files below the junk floor? | `yes` / `no` | `yes` | `APPLY_MIN_SIZE` | §0 `MIN_BYTES` |
| 5 | Resolve phone numbers to contact names? | `yes` / `no` | `yes` | `RESOLVE_NAMES` | §2 contact map |
| 6 | After a verified merge, the source folder… | `keep` / `delete` | `keep` | `TEARDOWN_MODE` | §8 |

> **Row 1 is a single-member enum for now** — a `photos-library` binding is designed but not authored
> here (see *Deliberately not included*); a future binding restores the enum member.

```yaml
resolved_inputs:
  source_kind:    'messages'
  organize_by:    'by-year'
  include_video:  'yes'      # quote — bare yes/no are YAML booleans
  apply_min_size: 'yes'
  resolve_names:  'yes'
  teardown_mode:  'keep'
  resolved_by:    <human who confirmed>
  resolved_at:    <timestamp>
```

> **Why `teardown_mode` defaults to `keep`:** the consolidation is valuable on its own; deleting the
> source reclaims a rounding error of disk in exchange for the only irreversible step in the plan.
> Make the human ask for it.

---

## Live State

```yaml
status:        not-created      # published template - run it to realize state
last_action:   —
last_verified: —
resolved_inputs: { ... }
```

| key | value (filled on apply) |
|---|---|
| SRC_HOME | `—` |
| DEST | `—` |
| SOURCE_BYTES | `—` |
| SOURCE_FILES | `—` |
| COPIED | `—` |
| SKIPPED_DUPLICATE | `—` |
| UNDATED | `—` |
| ORPHANS | `—` |
| MANIFEST_ROWS | `—` |
| LOOSE_DIRS | `—` (newline-separated; recorded by §6, consumed by teardown) |

| ✔ check | expected | observed | result |
|---|---|---|---|
| source readable | `chat.db` opens; attachment count > 0 | — | — |
| free space sufficient | `avail > SOURCE_BYTES` | — | — |
| files on disk == manifest rows | three-way agreement (script tally, `find`, manifest) | — | — |
| media integrity | `file(1)` on a random sample reports real image/video types, no truncation | — | — |
| **source unchanged** (negative) | source path-set identical to §1 baseline (new files listed & allowed; none removed/moved) | — | — |
| **no fabricated dates** (negative) | every `by-year` file has `date_source != copy-mtime` | — | — |
| **pre-teardown gate** (negative) | zero source files lack a hash-identical counterpart | — | — |

> Assert the **negatives**. "Source unchanged" is what proves the plan was non-destructive; a plan
> that only checks its output can't tell you it ate the input.

---

## 0. Variables

```bash
# ── source / destination (genericize before contributing) ──
export SRC_HOME="/Volumes/<source-home>"          # mounted home folder, Time Machine restore, or "$HOME"
export DEST="$SRC_HOME/Pictures/<Review Folder>"  # review tree
export WORK="$(mktemp -d)"                        # local scratch — DB copies, never on the network volume

export MSG_DIR="$SRC_HOME/Library/Messages"
export ATT_DIR="$MSG_DIR/Attachments"
export AB_DIR="$SRC_HOME/Library/Application Support/AddressBook"

# ── filters ──
export KEEP_EXT_IMG=".jpg .jpeg .png .gif .heic"
export KEEP_EXT_VID=".mov .mp4"
export MIN_BYTES=$((50 * 1024))   # junk floor: stickers, link-preview icons, UI thumbnails

# ── copy-cluster detector (§3 rung-4 guard) ──
# ≥ CLUSTER_N files whose mtimes span ≤ CLUSTER_WINDOW s = a folder that was COPIED
# (not photos taken then); such files are barred from mtime and drop to Undated.
export CLUSTER_N=5
export CLUSTER_WINDOW=60           # seconds

# ── loose folders to fold in (§6) and, ONLY after the §7 gate passes, delete (§8) ──
# Newline-separated list. §6 records the realized set into Live State LOOSE_DIRS;
# teardown deletes EXACTLY this set and refuses to run on an empty/unset value.
export LOOSE_DIRS=""              # e.g. LOOSE_DIRS=$'/Volumes/x/Old iPhone\n/Volumes/x/Camera Dump'

# ── resolved Provisioning Inputs — EXPORT post-interview so §2–§7's python inherits them ──
# (plan.py / verify_gate.py read these from os.environ; without the export they'd silently
#  fall back to defaults and, e.g., a `no` answer would be ignored.)
export SOURCE_KIND="${SOURCE_KIND:-messages}"
export ORGANIZE_BY="${ORGANIZE_BY:-by-year}"
export INCLUDE_VIDEO="${INCLUDE_VIDEO:-yes}"
export APPLY_MIN_SIZE="${APPLY_MIN_SIZE:-yes}"
export RESOLVE_NAMES="${RESOLVE_NAMES:-yes}"
export TEARDOWN_MODE="${TEARDOWN_MODE:-keep}"

export PLAN_SOURCE="archive.local.md"
export PLAN_VERSION="<YYYY-MM-DD>"
```

> **No TAGS section.** Local filesystem artifacts have no tag API; provenance rides in `manifest.csv`
> (every row records source path, date, and *how the date was derived*) and in the `- <Source> -`
> filename infix. Record this as a portability-ledger asymmetry against the cloud bindings.

---

## Dependency frontier

```
§1 survey ─┬─> §2 index (chat.db + contacts) ─┐
           └─> §3 dry-run plan ───────────────┼─> §4 extract ─> ✔ §5 verify ─┐
                                              │                              ├─> §7 🔴 GATE ─> §8 💥 teardown
                                    §6 fold-in loose folders ────────────────┘
```

Non-negotiable edges:
- **§3 before §4** — the dry-run is what catches a systematically wrong date source *before* you write
  thousands of misnamed files. Skipping it is how you find out at file 4,000.
- **§5 before §6** — the merge reads `manifest.csv` as its index. An unverified manifest poisons it.
- **§7 before §8, always** — the gate is the only thing standing between a bug and permanent loss.

---

## 1. Survey  🟢 ✔

```bash
mkdir -p "$WORK"
# TCC probe — distinguishes "no permission" from "nothing there"; then assert a real count (>0)
ATT_COUNT="$(sqlite3 "file:$MSG_DIR/chat.db?mode=ro" "select count(*) from attachment;")" \
  || { echo "FAIL: cannot read chat.db — grant Full Disk Access to this terminal"; exit 1; }
[ "${ATT_COUNT:-0}" -gt 0 ] || { echo "FAIL: attachment table empty — wrong store or unreadable"; exit 1; }
echo "attachments in chat.db: $ATT_COUNT"

find "$ATT_DIR" -type f > "$WORK/source_files.txt"
wc -l < "$WORK/source_files.txt"                       # → SOURCE_FILES (baseline for the negative check)
awk -F. 'NF>1{print tolower($NF)}' "$WORK/source_files.txt" | sort | uniq -c | sort -rn

# free-space gate — fail CLOSED; check dirname($DEST) so it works BEFORE $DEST exists
SOURCE_KB="$(du -sk "$ATT_DIR" | cut -f1)"             # → SOURCE_BYTES
AVAIL_KB="$(df -k "$(dirname "$DEST")" | awk 'NR==2{print $4}')"
echo "source ${SOURCE_KB}KB  avail ${AVAIL_KB}KB"
[ "${AVAIL_KB:-0}" -gt "${SOURCE_KB:-0}" ] || { echo "FAIL: insufficient free space for the copy"; exit 1; }
```

```bash
# ✔ time it — a "slow network drive" is a hypothesis until it's a number
/usr/bin/time cp "$MSG_DIR/chat.db" "$WORK/chat.db"
```

> → Live State: `SOURCE_FILES`, `SOURCE_BYTES`
>
> **Expect junk in the extension histogram.** `.pluginpayloadattachment` (link previews, Apple Pay),
> `.vcf`, and audio are not photos. In the dogfooded run they were **1,376 of 6,264 files — 22%**.
> Filtering by extension is not a nicety; it is most of the noise.

## 2. Index — dates and names  🟢

**Copy the DB before reading it.** A live Messages process holds `chat.db` open with a populated
`-wal`; querying it in place risks a lock and reads a torn view. Copy `chat.db`, `-wal`, and `-shm`
together — the `-wal` holds recent messages not yet checkpointed into the main file.

```bash
cp "$MSG_DIR"/chat.db* "$WORK/"
# if/then/fi, not `[ … ] && …`: with RESOLVE_NAMES=no the test is a non-zero exit that
# reads as a step failure (and aborts under `set -e`) — the guard must simply skip.
if [ "${RESOLVE_NAMES:-yes}" = yes ]; then
  find "$AB_DIR" -name '*.abcddb' -print0 | while IFS= read -r -d '' db; do
    cp "$db" "$WORK/ab-$(basename "$(dirname "$db")").db"
  done
fi
```

> **`find … -print0 | while read -r -d ''` is not optional.** `Application Support` contains a space;
> an unquoted `for f in $(find …)` word-splits the path and silently produces zero-byte copies that
> fail later as "no such table". Ask how this plan knows.

```python
# attachment relpath -> (datetime, sender handle, is_from_me)
rows = db.execute("""
    SELECT a.filename, m.date, a.created_date, m.is_from_me, h.id
    FROM attachment a
    LEFT JOIN message_attachment_join maj ON maj.attachment_id = a.ROWID
    LEFT JOIN message m ON m.ROWID = maj.message_id
    LEFT JOIN handle  h ON h.ROWID = m.handle_id
""")
```

**Two traps in that join:**

- **Epoch ambiguity.** `message.date` is Apple-epoch (2001-01-01), but in **seconds** on older
  macOS and **nanoseconds** on newer. Normalize or your dates land in 1970 or the far future:
  ```python
  if not value:                       # 0 / NULL date must NOT become 2001-01-01 — next rung
      dt = None
  else:
      secs = value / 1e9 if value > 1e11 else value
      dt = datetime.fromtimestamp(secs + 978307200)
  ```
- **Orphans.** Files on disk outnumber `attachment` rows — threads were deleted, the blob wasn't. The
  dogfooded run had **646 orphans (14%)**. They are real photos; do not drop them. They fall through
  to the §3 date ladder and a sender of `Unknown`.

Contact resolution — normalize **both** sides to the last 10 digits. `chat.db` stores `+1XXXXXXXXXX`;
AddressBook stores `(XXX) XXX-XXXX`. Nothing matches without it.

```python
def digits10(s):
    d = re.sub(r"\D", "", s or "")
    return d[-10:] if len(d) >= 10 else None
```

Merge **every** `.abcddb` under `Sources/` — accounts are sharded across them, and the top-level file
is often the empty one. Label `is_from_me=1` as `Me`; fall back to the raw handle, never to a blank.

> → Live State: `ORPHANS`

## 3. Dry-run plan  ✔

Produce the full rename plan and **print it without copying a byte**. This step exists to catch a
systematically wrong date source. It reconstructs `plan.py` from this plan's spec (the contributor's
original was not shipped — treat as unverified until this run exercises it) and writes it to `$WORK`:

```bash
cat > "$WORK/plan.py" <<'PY'
# reconstructed from the plan spec at promotion — contributor's originals not
# included; treat as unverified until a live run.
#
# plan.py — §2 index (chat.db join + contact map) + §3 dry-run planner. Emits
# the rename plan to $WORK/plan.tsv and a human report to stdout; copies nothing.
# stdlib only (sqlite3, csv, struct, re, datetime).
import csv, os, re, sqlite3, struct
from collections import defaultdict
from datetime import datetime

WORK      = os.environ["WORK"]
DEST      = os.environ["DEST"]
ATT       = os.environ["ATT_DIR"]
ORG       = os.environ.get("ORGANIZE_BY", "by-year")
RESOLVE   = os.environ.get("RESOLVE_NAMES", "yes") == "yes"
APPLY_MIN = os.environ.get("APPLY_MIN_SIZE", "yes") == "yes"
MIN_BYTES = int(os.environ.get("MIN_BYTES", 50 * 1024)) if APPLY_MIN else 0
CLUSTER_N      = int(os.environ.get("CLUSTER_N", 5))
CLUSTER_WINDOW = int(os.environ.get("CLUSTER_WINDOW", 60))

KEEP = set(e.lower() for e in os.environ.get("KEEP_EXT_IMG", ".jpg .jpeg .png .gif .heic").split())
if os.environ.get("INCLUDE_VIDEO", "yes") == "yes":
    KEEP |= set(e.lower() for e in os.environ.get("KEEP_EXT_VID", ".mov .mp4").split())

APPLE_EPOCH = 978307200          # 2001-01-01 in Unix seconds
THIS_YEAR   = datetime.now().year


def sanitize(s):
    return (re.sub(r"[/:\x00-\x1f]", "", s or "").strip()[:80]) or "Unknown"


def norm_epoch(value):
    # n1: 0 / NULL must NOT become 2001-01-01 — fall through to the next rung.
    if not value:
        return None
    secs = value / 1e9 if value > 1e11 else value      # ns on newer macOS, s on older
    try:
        return datetime.fromtimestamp(secs + APPLE_EPOCH)
    except (OverflowError, OSError, ValueError):
        return None


def digits10(s):
    d = re.sub(r"\D", "", s or "")
    return d[-10:] if len(d) >= 10 else None


def load_contacts():
    # Merge EVERY ab-*.abcddb copy in $WORK — accounts are sharded across them.
    m = {}
    for name in sorted(os.listdir(WORK)):
        if not (name.startswith("ab-") and name.endswith(".db")):
            continue
        try:
            c = sqlite3.connect("file:%s?mode=ro" % os.path.join(WORK, name), uri=True)
            for first, last, num in c.execute(
                    "SELECT ZFIRSTNAME, ZLASTNAME, ZFULLNUMBER "
                    "FROM ZABCDRECORD r JOIN ZABCDPHONENUMBER p ON p.ZOWNER = r.Z_PK"):
                k = digits10(num)
                if not k:
                    continue
                nm = " ".join(x for x in (first, last) if x).strip()
                if nm:
                    m.setdefault(k, nm)
            c.close()
        except sqlite3.Error:
            continue
    return m


def att_relkey(path):
    # M4: key by the path relative to Attachments/, NOT basename — recycled camera
    # names (IMG_0001.JPG) collide on basename and mis-assign dates/senders. The
    # content-addressed layout makes the relpath unique. attachment.filename carries
    # the full path (often "~/…/Attachments/ab/12/GUID/IMG.JPG"); split on the marker.
    marker = "/Attachments/"
    i = path.rfind(marker)
    tail = path[i + len(marker):] if i >= 0 else os.path.basename(path)
    return tail.strip("/")


def load_messages():
    # attachment Attachments-relative key -> (datetime, sender handle) from chat.db.
    idx = {}
    db = sqlite3.connect("file:%s?mode=ro" % os.path.join(WORK, "chat.db"), uri=True)
    for filename, mdate, cdate, is_from_me, handle in db.execute("""
            SELECT a.filename, m.date, a.created_date, m.is_from_me, h.id
            FROM attachment a
            LEFT JOIN message_attachment_join maj ON maj.attachment_id = a.ROWID
            LEFT JOIN message m ON m.ROWID = maj.message_id
            LEFT JOIN handle  h ON h.ROWID = m.handle_id"""):
        if not filename:
            continue
        dt = norm_epoch(mdate) or norm_epoch(cdate)
        sender = "Me" if is_from_me else (handle or None)
        idx[att_relkey(filename)] = (dt, sender)
    db.close()
    return idx


def exif_datetime(path):
    # Rung 2 — EXIF DateTimeOriginal (0x9003) via a minimal stdlib JPEG walk.
    try:
        with open(path, "rb") as f:
            data = f.read(65536)
    except OSError:
        return None
    if data[0:2] != b"\xff\xd8":               # not a JPEG
        return None
    i, n = 2, len(data)
    while i + 4 < n and data[i] == 0xFF:
        marker = data[i + 1]
        size = struct.unpack(">H", data[i + 2:i + 4])[0]
        if marker == 0xE1 and data[i + 4:i + 8] == b"Exif":
            return _parse_tiff(data[i + 10:i + 2 + size])
        i += 2 + size
    return None


def _parse_tiff(tiff):
    if len(tiff) < 8:
        return None
    en = "<" if tiff[0:2] == b"II" else ">"
    u16 = lambda o: struct.unpack(en + "H", tiff[o:o + 2])[0]
    u32 = lambda o: struct.unpack(en + "I", tiff[o:o + 4])[0]
    try:
        ifd0 = u32(4)
        exif_off = None
        for k in range(u16(ifd0)):
            e = ifd0 + 2 + k * 12
            if u16(e) == 0x8769:               # ExifIFD pointer
                exif_off = u32(e + 8)
        if exif_off is None:
            return None
        for k in range(u16(exif_off)):
            e = exif_off + 2 + k * 12
            if u16(e) == 0x9003:               # DateTimeOriginal
                s = tiff[u32(e + 8):u32(e + 8) + 19].decode("ascii", "ignore")
                return datetime.strptime(s, "%Y:%m:%d %H:%M:%S")
    except (struct.error, ValueError):
        return None
    return None


# Rung 3 — date embedded in the filename. Two named formats; validate by
# CONSTRUCTING datetime() so a hex hash (which can look like digits) is rejected.
RE_DASHED  = re.compile(r"(?<!\d)(20\d{2})[-_.](\d{2})[-_.](\d{2})(?!\d)")   # Screenshot_2015-06-03
RE_COMPACT = re.compile(r"(?<!\d)(20\d{2})(\d{2})(\d{2})(?!\d)")            # 20131124_192916

def filename_date(name):
    for rx in (RE_DASHED, RE_COMPACT):         # dashed first — less ambiguous
        m = rx.search(name)
        if not m:
            continue
        y, mo, d = (int(x) for x in m.groups())
        if not (2000 <= y <= THIS_YEAR):
            continue
        try:                                   # n2: ValueError => next rung
            return datetime(y, mo, d)
        except ValueError:
            continue
    return None


def find_clusters(mtimes):
    # M2 — a bucket of >= CLUSTER_N files whose mtimes span <= CLUSTER_WINDOW s is
    # a copy cluster: those files may use rung 4 (mtime) only if NOT in a cluster.
    clustered = set()
    ordered = sorted((t, p) for p, t in mtimes.items() if t is not None)
    lo = 0
    for hi in range(len(ordered)):
        while ordered[hi][0] - ordered[lo][0] > CLUSTER_WINDOW:
            lo += 1
        if hi - lo + 1 >= CLUSTER_N:
            for j in range(lo, hi + 1):
                clustered.add(ordered[j][1])
    return clustered


def main():
    files = [l.rstrip("\n") for l in open(os.path.join(WORK, "source_files.txt")) if l.strip()]
    contacts = load_contacts() if RESOLVE else {}
    messages = load_messages()

    mtimes = {}
    for p in files:
        try:
            mtimes[p] = os.stat(p).st_mtime
        except OSError:
            mtimes[p] = None
    clustered = find_clusters(mtimes)

    folder_counts = defaultdict(int)
    src_hist = defaultdict(int)
    unresolved = 0
    samples, plan_rows = [], []

    for p in files:
        ext = os.path.splitext(p)[1].lower()
        if ext not in KEEP:
            src_hist["filtered-ext"] += 1
            continue
        try:
            size = os.path.getsize(p)
        except OSError:
            continue
        if size < MIN_BYTES:
            src_hist["filtered-small"] += 1
            continue

        try:
            relkey = os.path.relpath(p, ATT).strip("/")
        except ValueError:
            relkey = os.path.basename(p)
        dt, sender = messages.get(relkey, (None, None))
        date_source = "chat.db" if dt else None
        if dt is None:                                     # rung 2
            dt = exif_datetime(p)
            date_source = "exif" if dt else date_source
        if dt is None:                                     # rung 3
            dt = filename_date(os.path.basename(p))
            date_source = "filename" if dt else date_source
        if dt is None and p not in clustered:              # rung 4 (mtime, if not clustered)
            t = mtimes.get(p)
            if t is not None:
                dt = datetime.fromtimestamp(t)
                date_source = "mtime"
        if dt is None:                                     # rung 5
            date_source = "undated"

        if sender:
            k10 = digits10(sender)
            name = contacts.get(k10) if k10 else None
            sender_name = name or sender
            if not name and sender != "Me":
                unresolved += 1
        else:
            sender_name = "Unknown"
            unresolved += 1

        if dt is None:
            folder = "Undated"
        elif ORG == "by-year":
            folder = str(dt.year)
        elif ORG == "by-sender":
            folder = sanitize(sender_name)
        else:
            folder = ""                                    # flat

        datestr = dt.strftime("%Y-%m-%d") if dt else "0000-00-00"
        stem, e = os.path.splitext(os.path.basename(p))
        target = "%s - %s - %s%s" % (datestr, sanitize(sender_name), sanitize(stem), e)
        rel = os.path.join(folder, target) if folder else target

        folder_counts[folder or "flat"] += 1
        src_hist[date_source] += 1
        if len(samples) < 20:
            samples.append(rel)
        plan_rows.append((p, os.path.join(DEST, rel), datestr, sender_name, size, date_source))

    with open(os.path.join(WORK, "plan.tsv"), "w", newline="") as fh:
        w = csv.writer(fh, delimiter="\t")
        for r in plan_rows:
            w.writerow(r)

    print("=== dry-run: %d files planned (nothing copied) ===" % len(plan_rows))
    print("\n-- per-folder counts --")
    for k in sorted(folder_counts):
        print("  %-14s %d" % (k, folder_counts[k]))
    print("\n-- date-source histogram --")
    for k in sorted(src_hist):
        print("  %-14s %d" % (k, src_hist[k]))
    print("\nunresolved senders: %d" % unresolved)
    print("copy-cluster files barred from rung 4 (mtime): %d" % len(clustered))
    print("\n-- 20 sample names --")
    for s in samples:
        print("  " + s)


if __name__ == "__main__":
    main()
PY

set -o pipefail                                # m-a: tee must not mask a plan.py crash
python3 "$WORK/plan.py" | tee "$WORK/dryrun.txt"
# report: per-folder counts, date-source histogram, unresolved-sender count, 20 sample names.
# Also writes the rename plan to $WORK/plan.tsv for §4 to copy from — no byte copied here.
```

> **Inspect the date histogram here — this is a dry run, so it carries no gate.** The human go that
> authorizes writing files is the **🔴 at §4 entry**; approve there only after reading this histogram.
> (Gating the dry run would be gate-fatigue: it spends nothing and reverses freely.)
>
> **The copy-timestamp trap.** If a cluster of files share an mtime within seconds of each other
> (`2015-12-26 10:20:27`, `:29`, `:32`, …), that is when the folder was **copied**, not when the
> photos were taken. `plan.py` operationalizes this: **≥ `CLUSTER_N` files whose mtimes span
> ≤ `CLUSTER_WINDOW` s** (defaults 5 / 60 s, tunable in §0) are a copy cluster, and any such file that
> missed rungs 1–3 drops to **`Undated/` (rung 5), never mtime (rung 4)** — filing it by that mtime
> fabricates a date. The §3 histogram is the human **override** channel, not the detector: in the
> dogfooded run the cluster hit **15 of 30** files in a loose folder, invisible unless you look.

**Date ladder** — first hit wins, and the rung used is recorded in `manifest.csv`:

| Rung | Source | Trust |
|---|---|---|
| 1 | `chat.db` message date | authoritative |
| 2 | EXIF `DateTimeOriginal` (tag `0x9003`) | authoritative |
| 3 | Date embedded in filename (`20131124_192916`, `Screenshot_2015-06-03`) | good — accepted only if `datetime(y,m,d)` **constructs** and `2000 ≤ y ≤ this_year`; a `ValueError` ⇒ next rung, so a hex hash that looks like digits is rejected (two named-format regexes in `plan.py`) |
| 4 | File mtime — **only if not in a copy cluster** (§3: ≥`CLUSTER_N` files within `CLUSTER_WINDOW` s) | weak |
| 5 | `Undated/` | honest |

## 4. Extract  🔴 🟢

> **🔴 GATE — human go before the first write.** Proceed only on explicit human approval of the §3
> date-source histogram; nothing has been copied yet, so this is the first mutation. Record
> `approved_by` + `approved_at` in Live State at this gate. The copy is reversible (delete `$DEST`),
> but writing thousands of misnamed files on a systematically wrong date source is the expensive
> mistake §3 exists to prevent — so the gate rides the first write, not the dry run.

```
$DEST/
  2015/   2016/   …          # $ORGANIZE_BY
  Undated/
  manifest.csv               # destination, source, date, sender, bytes, date_source
```

Naming: `YYYY-MM-DD - <Sender> - <original>.ext` — sorts chronologically inside each folder and keeps
the original filename so the manifest can round-trip. §4 copies from `$WORK/plan.tsv` (written by §3).

> **Pinned manifest contract.** `manifest.csv` **line 1 MUST be the header**
> `destination,source,date,sender,bytes,date_source` and every row follows that order — §5's negative
> checks and §6's size-index read it by name (`csv.DictReader`), so a drift here silently breaks both.
> Note the intermediate `$WORK/plan.tsv` uses a **different** column order
> (`source, destination, date, sender, bytes, date_source`) and carries **no header**; the copier must
> **reorder to the pinned layout and write the header row** as it streams `manifest.csv`.
>
> **Candor:** the §4 copier itself is **Director-synthesized** against that pinned contract this cycle —
> `extract.py` is deliberately **not** authored until the first live dogfood exercises the copy path
> (controller decision). §3's `plan.py` and §7's `verify_gate.py` are the two reconstructed scripts;
> the copier is prose + contract until proven.

Non-negotiables:
- `shutil.copy2` (preserves mtime), **never** `move`
- **Resumable**: if the target exists at the **same size**, skip — re-running after an interruption is free
- **Collision rule is content-decided:** a same-name target is a true collision **only when its content
  differs** (hash-compare). Same content ⇒ it's a resume, skip it; different content ⇒ suffix `-2`,
  `-3`, never overwrite
- **No truncated-partial window:** copy to `$DEST/.tmp.<name>` then atomically `mv` it over the final
  name — an interrupted copy can never leave a short file a later run mistakes for a same-size skip
- `sanitize()` the sender and stem: strip `/`, `:`, control chars; cap length
- Flush the manifest periodically — an interrupted run should leave a usable index

> → Live State: `COPIED`, `UNDATED`, `MANIFEST_ROWS`
>
> **Expect it to be slow, and know why.** Consolidating *within* one network volume pushes every byte
> over the wire **twice** (read + write). Run it detached with progress every N files. Hardlinking
> would make it instant but AFP/SMB do not support it reliably — copying is the honest option.

## 5. Verify  ✔

```bash
# m-g: two of the three ways machine-compared (the third is the script's own §4 COPIED tally).
# Exclude manifest, .DS_Store, and .tmp.* partials (m-c). $DEST reaches python via os.environ (m-e).
FIND_N="$(find "$DEST" -type f ! -name manifest.csv ! -name '.DS_Store' ! -name '.tmp.*' | wc -l | tr -d ' ')"
MANI_N="$(python3 - <<'PY'
import csv, os
with open(os.environ["DEST"] + "/manifest.csv") as f:
    print(sum(1 for _ in csv.reader(f)) - 1)
PY
)"
echo "on-disk=$FIND_N  manifest=$MANI_N   (both must also == the §4 COPIED tally)"
[ "$FIND_N" = "$MANI_N" ] || { echo "FAIL: on-disk count != manifest rows"; exit 1; }

# media integrity — random sample reports real image/video types, no truncation
find "$DEST" -type f ! -name manifest.csv ! -name '.DS_Store' ! -name '.tmp.*' | sort -R | head -20 | while read -r f; do file -b "$f"; done
```

```bash
# ✔ NEGATIVE — no fabricated dates: zero manifest rows are filed under a year dir on a copy-mtime date.
# $DEST via os.environ (m-e, never string-interpolated); csv module honors quoted sender commas.
python3 - <<'PY' || { echo "FAIL: mtime-dated files filed under a year — fabricated dates"; exit 1; }
import csv, os, sys
dest = os.environ["DEST"]
bad = [r for r in csv.DictReader(open(dest + "/manifest.csv"))
       if r["date_source"] == "mtime" and os.path.basename(os.path.dirname(r["destination"])).isdigit()]
print("fabricated-date rows:", len(bad))
sys.exit(1 if bad else 0)
PY
```

```bash
# ✔ NEGATIVE — source path-set unchanged. New files on a live source are expected and listed ('>' lines);
# a source path present at §1 that is now gone/moved ('<' lines) is the failure.
find "$ATT_DIR" -type f | sort > "$WORK/source_now.txt"
diff <(sort "$WORK/source_files.txt") "$WORK/source_now.txt" > "$WORK/source_diff.txt" || true
if grep -q '^<' "$WORK/source_diff.txt"; then
  echo "FAIL: a source file present at §1 is gone or moved — source was mutated"; exit 1
fi
grep '^>' "$WORK/source_diff.txt" && echo "(new source files since §1 — expected on a live machine)" || true
```

Three-way agreement (script tally, `find`, manifest) is the check. Any two agreeing proves nothing —
the script's own count and its manifest share a bug.

> → Live State: set `status: live`

## 6. Fold in loose folders  🟢 🟡

Other folders hold overlapping copies. Merge uniques, detect exact duplicates — **without** re-reading
the archive.

```python
# Size index from the MANIFEST — zero re-scan of $DEST
by_size = defaultdict(list)
for row in csv.DictReader(open(MANIFEST)):
    by_size[int(row["bytes"])].append(row["destination"])

for src in loose_files:
    if by_size.get(size(src)):        # only a size collision can possibly be a duplicate
        h = sha256(src)               # hash ONLY the handful of candidates
```

> **This is the load-bearing idea in the plan.** Naive dedupe hashes both trees. Size is a free,
> perfect *pre-filter*: two files of different sizes cannot be identical. In the dogfooded run, 66
> loose files against a 4,639-file / 6.8 GB archive collided on size at only 39 candidates —
> **56 MB read instead of 6,800 MB, a 120× reduction.** The manifest made the index free.
>
> Never dedupe on filename or mtime. Same-name-different-photo and same-photo-different-name are both
> routine; only content hashing is sound.

Uniques flow through the **same §3 date ladder** and are tagged with a distinct sender infix
(`- iPhone -`) so provenance survives. Append them to the manifest.

> → Live State: `SKIPPED_DUPLICATE`, `COPIED` (updated), `MANIFEST_ROWS` (updated); and **record the
> realized fold-in folders into `LOOSE_DIRS`** (newline-separated). This recorded list is the *only*
> thing teardown will delete — a folder not recorded here is never a deletion candidate.

## 7. 🔴 Pre-teardown verification gate  ✔

**Only reached if `TEARDOWN_MODE=delete`.** The gate's universe is the **folders slated for deletion
(`$LOOSE_DIRS`) — never the Messages store**, which is never deleted (see Teardown). For every
**in-class** file in those folders (`KEEP_EXT` + ≥ `MIN_BYTES`), independently re-hash it and prove a
byte-identical counterpart exists in `$DEST`. Files a loose folder holds that are **excluded by
`KEEP_EXT` / `MIN_BYTES` are EXEMPT-BY-CLASS**: the `rm` removes them, but they carry no counterpart
requirement — they were filtered as junk and never copied. Do not trust §6's report — index `$DEST`
by walking the **disk**, not by reading the manifest the same code wrote. This step reconstructs
`verify_gate.py` from the plan spec (contributor's original not shipped — unverified until this run):

```bash
cat > "$WORK/verify_gate.py" <<'PY'
# reconstructed from the plan spec at promotion — contributor's originals not
# included; treat as unverified until a live run.
#
# verify_gate.py — §7 pre-teardown gate. Independently indexes $DEST by walking
# the DISK (never the manifest the copier wrote), then proves every in-class file
# in each $LOOSE_DIRS folder has a byte-identical counterpart in $DEST. Files
# excluded by KEEP_EXT / MIN_BYTES are EXEMPT-BY-CLASS: the rm removes them but
# they carry no counterpart requirement — they are ENUMERATED at the gate. FAILS
# CLOSED: unreadable dir/file, DEST/LOOSE overlap, or zero in-class files verified
# all block deletion. Exits nonzero unless every in-class file is accounted for.
# stdlib only.
import hashlib, os, sys
from collections import defaultdict

WORK      = os.environ["WORK"]
DEST      = os.environ["DEST"]
APPLY_MIN = os.environ.get("APPLY_MIN_SIZE", "yes") == "yes"
MIN_BYTES = int(os.environ.get("MIN_BYTES", 50 * 1024)) if APPLY_MIN else 0

KEEP = set(e.lower() for e in os.environ.get("KEEP_EXT_IMG", ".jpg .jpeg .png .gif .heic").split())
if os.environ.get("INCLUDE_VIDEO", "yes") == "yes":
    KEEP |= set(e.lower() for e in os.environ.get("KEEP_EXT_VID", ".mov .mp4").split())

EXEMPT_NAMES = {".DS_Store", "manifest.csv"}


def loose_dirs():
    raw = os.environ.get("LOOSE_DIRS", "").strip()
    if not raw:
        sys.stderr.write("FAIL: LOOSE_DIRS unset — no recorded folders to verify\n")
        sys.exit(1)
    return [os.path.realpath(d.strip()) for d in raw.split("\n") if d.strip()]


def is_under(child, parent):
    # True if realpath(child) is at or below realpath(parent). Uses inode identity, not string
    # prefix, so a case-variant recording ('/pictures' vs '/Pictures') on a case-insensitive
    # filesystem (default macOS APFS) can't evade the disjointness check (BLOCKER-2 hardening).
    child = os.path.realpath(child)
    parent = os.path.realpath(parent)
    try:
        cs, ps = os.stat(child), os.stat(parent)
        if (cs.st_dev, cs.st_ino) == (ps.st_dev, ps.st_ino):
            return True
    except OSError:
        # a path that can't be stat'd during a safety check is not provably disjoint — fail closed
        sys.stderr.write("FAIL: cannot stat path during disjointness check: %s / %s\n" % (child, parent))
        sys.exit(1)
    # walk child upward, comparing inode identity at each ancestor (case-fold-proof)
    cur = child
    while True:
        parent_of = os.path.dirname(cur)
        if parent_of == cur:
            return False  # reached filesystem root without matching parent
        try:
            ps2 = os.stat(parent)
            cur_stat = os.stat(parent_of)
        except OSError:
            sys.stderr.write("FAIL: cannot stat ancestor during disjointness check: %s\n" % parent_of)
            sys.exit(1)
        if (cur_stat.st_dev, cur_stat.st_ino) == (ps2.st_dev, ps2.st_ino):
            return True
        cur = parent_of


def check_disjoint(dest, loose):
    # BLOCKER-2: $DEST and every loose dir must be mutually disjoint, and no two
    # loose dirs may nest — otherwise a file can be its own counterpart => false PASS.
    for ld in loose:
        if is_under(dest, ld) or is_under(ld, dest):
            sys.stderr.write("FAIL: $DEST and a loose dir overlap: %s <> %s\n" % (dest, ld))
            sys.exit(1)
    for i, a in enumerate(loose):
        for j, b in enumerate(loose):
            if i != j and is_under(a, b):
                sys.stderr.write("FAIL: loose dirs nest: %s under %s\n" % (a, b))
                sys.exit(1)


def _walk_err(err):
    # BLOCKER-1: a subdir we cannot read must FAIL the gate, never be skipped silently
    # (the plan's own AFP stale-fork gotcha makes this the EXPECTED failure).
    sys.stderr.write("FAIL: cannot walk %s: %s\n" % (getattr(err, "filename", "?"), err))
    sys.exit(1)


def sha256(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()


def index_dest(loose):
    # size -> [dest paths]; walk the DISK, not the manifest.
    by_size = defaultdict(list)
    for root, _dirs, names in os.walk(DEST, onerror=_walk_err):
        for nm in names:
            if nm in EXEMPT_NAMES or nm.startswith(".tmp."):
                continue                              # m-c: skip partial-copy transients
            p = os.path.join(root, nm)
            if any(is_under(p, ld) for ld in loose):  # BLOCKER-2 defence-in-depth (symlinks)
                continue
            try:
                by_size[os.path.getsize(p)].append(p)
            except OSError:
                pass                                  # a dest file we can't stat isn't offered
    return by_size


def in_class(path, size):
    if os.path.basename(path) in EXEMPT_NAMES:
        return False
    if os.path.splitext(path)[1].lower() not in KEEP:
        return False
    return size >= MIN_BYTES


def main():
    dest = os.path.realpath(DEST)
    loose = loose_dirs()
    check_disjoint(dest, loose)
    by_size = index_dest(loose)
    hash_cache = {}                       # dest path -> sha256 (size-collisions only)
    checked = accounted = 0
    exempt_paths = []
    unaccounted = []

    # a recorded entry that EXISTS but is not a directory (a misrecorded file) must never slide
    # into the "absent" no-op bucket — the bash rm -rf would delete it unverified.
    for d in loose:
        if os.path.exists(d) and not os.path.isdir(d):
            print(f"GATE FAIL: recorded loose entry is not a directory: {d}")
            sys.exit(1)
    present = [d for d in loose if os.path.isdir(d)]

    for d in present:
        for root, _dirs, names in os.walk(d, onerror=_walk_err):
            for nm in names:
                if nm.startswith(".tmp."):
                    continue                          # m-c: transient — neither exempt nor checked
                p = os.path.join(root, nm)
                try:
                    size = os.path.getsize(p)
                except OSError:
                    # BLOCKER-1: an unstat-able file would be deleted unchecked — fail closed
                    unaccounted.append(p + "  (cannot stat)")
                    continue
                if not in_class(p, size):
                    exempt_paths.append(p)            # EXEMPT-BY-CLASS: no counterpart required
                    continue
                checked += 1
                candidates = by_size.get(size, [])
                if not candidates:                    # no same-size file in $DEST
                    unaccounted.append(p)
                    continue
                src_h = sha256(p)
                hit = False
                for c in candidates:
                    if c not in hash_cache:
                        try:
                            hash_cache[c] = sha256(c)
                        except OSError:
                            hash_cache[c] = None
                    if hash_cache[c] == src_h:
                        hit = True
                        break
                if hit:
                    accounted += 1
                else:
                    unaccounted.append(p)

    # BLOCKER-3: enumerate EXEMPT-BY-CLASS at the gate — full list to $WORK/exempt.txt
    # plus a per-extension histogram and the names, so the human approves what the rm sweeps.
    exempt_file = os.path.join(WORK, "exempt.txt")
    with open(exempt_file, "w") as f:
        for p in exempt_paths:
            f.write(p + "\n")
    ext_hist = defaultdict(int)
    for p in exempt_paths:
        ext_hist[os.path.splitext(p)[1].lower() or "<none>"] += 1

    print("gate: in-class checked=%d accounted=%d exempt-by-class=%d unaccounted=%d"
          % (checked, accounted, len(exempt_paths), len(unaccounted)))
    if exempt_paths:
        print("\nEXEMPT-BY-CLASS (deleted with the folder, no counterpart required) — histogram:")
        for ext in sorted(ext_hist):
            print("  %-16s %d" % (ext, ext_hist[ext]))
        print("  full list: %s" % exempt_file)
        for p in exempt_paths:
            print("  exempt: " + p)

    if unaccounted:
        print("\nUNACCOUNTED — would be deleted WITHOUT a verified counterpart:")
        for p in unaccounted:
            print("  " + p)
        print("\nGATE FAIL — %d file(s) unaccounted; nothing may be deleted." % len(unaccounted))
        sys.exit(1)

    if not present:
        # All recorded loose dirs already gone: a completed teardown re-entry. Nothing to
        # verify or delete — pass so the (no-op) rm loop may run, without a vacuous claim.
        print("\nGATE PASS (no-op) — all recorded loose dirs already absent; teardown resume-complete.")
        sys.exit(0)

    if checked == 0:
        # BLOCKER-3: real dirs present but zero in-class files => nothing was ever proven
        # duplicated. A vacuous PASS here would delete a folder nothing was copied from.
        print("\nGATE FAIL — loose dirs are present but ZERO in-class files were verified. Nothing "
              "was proven duplicated, so nothing may be deleted. Override only by turning "
              "TEARDOWN_MODE off or acting manually — there is no silent-pass path.")
        sys.exit(1)

    print("\nGATE PASS — all %d in-class file(s) have a byte-identical counterpart in $DEST." % checked)
    sys.exit(0)


if __name__ == "__main__":
    main()
PY

# exits non-zero if ANY in-class file is unaccounted for; teardown is conditional on this.
# It ENUMERATES every to-be-deleted-without-counterpart file, so the human approves NAMES, not a bare PASS.
python3 "$WORK/verify_gate.py" || { echo "GATE FAIL — nothing will be deleted"; exit 1; }
```

> → Live State: `pre-teardown gate` row. **A partial pass is a fail** — one unaccounted in-class file
> fails the whole gate and lists itself by name at the 🔴.

## Update (idempotent reconcile) — incremental re-run  🟡

Re-running §2–§5 is the **drift check**. The resume rule (target exists at same size ⇒ skip) makes it
cheap: only messages received since the last run are copied. A source that is a **live machine** keeps
accruing messages, so the archive is a *snapshot, not a sync* — say so to the user, or they will
assume it keeps itself current.

## Teardown (observe-first, resumable)  💥

> Reverse of create. **Never** deletes the source application store — only the loose folders §6 proved
> redundant, and **only** those recorded in `LOOSE_DIRS`. `$DEST` itself is the deliverable; removing
> it is the human's business, not this plan's.
>
> **The recorded loose dirs must be quiescent during teardown** — no other process writing into them.
> A file that appears between the §7 gate and the `rm` would be deleted unverified (a TOCTOU window);
> quiesce the source (close the app, unmount other writers) before proceeding.

```bash
# 💥 On a network volume there is no Trash — this is FINAL. Refuse on an empty/unset list, then
# RE-PROVE the §7 gate on every teardown entry (resumable, observes-first) before any delete.
[ -n "${LOOSE_DIRS:-}" ] || { echo "FAIL: no recorded loose dirs — refusing to delete"; exit 1; }
# m-b: $WORK is a fresh mktemp each session — re-run §7 to regenerate the gate script before trusting
# it; never run a stale copy from a prior session's path.
[ -f "$WORK/verify_gate.py" ] || { echo "FAIL: $WORK/verify_gate.py missing — re-run §7 first to regenerate it"; exit 1; }
python3 "$WORK/verify_gate.py" || { echo "GATE FAIL — nothing deleted"; exit 1; }

# Record the human go (approver + timestamp) in Live State, then iterate the RECORDED list only.
# here-string (not a pipe) so each rm's exit stays observable in this shell. Entries are trimmed so
# a whitespace-padded recording can't be gate-verified as the real dir yet no-op'd by rm (false "gone").
while IFS= read -r d; do
  d="$(printf '%s' "$d" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
  [ -n "$d" ] || continue
  echo "💥 deleting recorded loose folder: $d"
  rm -rf "$d"
done <<< "$LOOSE_DIRS"
```

```bash
# ✔ teardown verify — rm prints only failures, so a partial success is SILENT: re-count, don't assume.
while IFS= read -r d; do
  d="$(printf '%s' "$d" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
  [ -n "$d" ] || continue
  if [ ! -d "$d" ]; then echo "gone: $d"; else echo "STILL PRESENT — re-count: $d"; fi
done <<< "$LOOSE_DIRS"
```

> **Known failure: `Resource busy` / stale AFP forks.** On AFP/SMB, bulk-hashing a directory (which
> §6 and §7 both do) can leave open forks on the *server* that block `unlink` afterward. Symptoms:
> every file returns `Resource busy`, `lsof` shows **no local** handle, and yet creating and deleting
> a *new* file in the same directory succeeds — which rules out permissions and file flags.
>
> `rm` prints only failures, so a partial success is silent — **re-count, don't assume**.
>
> Fix: drop the session (`diskutil unmount /Volumes/<vol>`, reconnect), or delete from the source
> machine. Closing Finder windows is **not** sufficient — dogfooded and disproven.
>
> This is cosmetic. §7 already proved the data is duplicated. Do not burn the user's time on it.

> → Live State: set `status: live+source-pruned` — the recorded loose folders are gone, the archive
> itself stays `live`. (Named intermediate, not `gone`: the deliverable is intact; only redundant
> sources were pruned.)

---

## Deliberately not included

- **Perceptual / near-duplicate detection** — only exact SHA-256 matches are removed. A resized or
  re-compressed copy is a *different file*, and "probably the same photo" is not a standard to delete
  on. Reviewing near-dupes is the human's job.
- **Whole-archive hashing** — deliberately avoided via the §6 size pre-filter. Hashing 6.8 GB to find
  a handful of duplicates is the naive shape this plan exists to demonstrate an alternative to.
- **Writing to the source store** — no cleanup, no compaction, no deletion from `Attachments/`. The
  application owns it.
- **Photos/iPhoto library ingestion** — a `photos-library` binding is **designed but not included
  here**; no such movement exists in this file. Two unresolved problems keep it out: libraries are
  package bundles that hold locks even when the app is closed, and photos may be *referenced* rather
  than copied in — deleting a referenced original leaves broken thumbnails. Authoring it means solving
  both, then restoring the enum member in Provisioning Inputs row 1.
- **HEIC transcoding** — `.heic` is copied through untouched. The dogfooded archive predated HEIC, so
  this path is **unexercised**; a modern source will be HEIC-heavy and may need `sips` for EXIF.
- **Group-chat and thread reconstruction** — sender attribution is per-attachment. Rebuilding
  conversations is a different intent.
- **iOS-side backups** (`Manifest.db`) — this plan reads the *macOS* store only.