directory name fix

scripts/twitter/README.md

@@ -18,6 +18,8 @@ python C:\projects\Threadwell\scripts\twitter\tw.py fallback --id <snowflake> --
 python C:\projects\Threadwell\scripts\twitter\tw.py restore-origin --id <snowflake> --media-id <media-id>
 python C:\projects\Threadwell\scripts\twitter\tw.py migrate-media --id <snowflake>
 python C:\projects\Threadwell\scripts\twitter\tw.py migrate-media --all --apply
+python C:\projects\Threadwell\scripts\twitter\tw.py reslug --all
+python C:\projects\Threadwell\scripts\twitter\tw.py reslug --all --apply
 ```
 
 `paths.py` sets vault = two parents above this file, dumps = `../manual_slop/docs/twitter`, scratch = `../Threadwell-ai/scratch`.
@@ -45,6 +47,12 @@ The `lift` command is retired. Use `fallback` for a single media item after expl
 
 Frozen threads (ids listed in `do_not_refetch.txt`, matched against every captured post id) are read-only across refetch, refresh, emit, manifest migration, note rewrite, fallback, publication selection, and tracked backup operations. Read-only `audit-media` may still inspect them.
 
+## Canonical thread directories
+
+Each thread archive directory is named from the rendered title that appears in `index.md` (`<YYYY-MM-DD>-<slugified rendered title>`). `tw.py reslug --all` reads each existing `index.md` frontmatter date and title, derives the expected directory, and reports every mapping without writing. `tw.py reslug --all --apply` preflights every pair, refuses the entire apply on any non-frozen conflict, and moves assets first with the archive directory immediately after; if the archive move fails, assets roll back. If a destination is occupied, the apply aborts without `-2`, `-3`, or any other suffix. Frozen ids in `do_not_refetch.txt` are reported as `frozen` and skipped.
+
+Future emits use the same rendered title for the directory and `index.md` frontmatter, so normal emit does this automatically — no flag is needed on emit, refresh, or add-branch. Hand-move only the archive or only the asset is not supported; both must move as a pair. Each old archive prefix is replaced exactly once in mutable `*.md` and `*.canvas` outside `.git/`, `site/`, `assets/`, `secrets/`, `node_modules/`, and frozen archive dirs, so mappings do not cascade. No aliases, redirect pages, compatibility folders, suffix fallback, or network contact.
+
 ## Pieces
 
 | Script | Job |

scripts/twitter/emit_archive.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import argparse
 import json
 import logging
+import os
 import re
 import shutil
 import sys
@@ -18,6 +19,7 @@ logger = logging.getLogger(__name__)
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
 try:
+    from twitter.frozen import frozen_match, load_frozen_ids
     from twitter.media_manifest import (
         _from_wire_dict,
         _item_to_wire,
@@ -30,7 +32,8 @@ try:
     )
     from twitter.media_refs import remote_markup
     from twitter.models import MediaItem, ThreadData, load_thread
-    from twitter.render import render_branch, render_spine
+    from twitter.paths import FROZEN, SCRATCH
+    from twitter.render import render_branch, render_spine, title_text
     from twitter.slug import branch_file_name, date_prefix, thread_dir_name
     from twitter.tree import (
         by_id,
@@ -41,6 +44,7 @@ try:
         spine_ids,
     )
 except ImportError:  # pragma: no cover - script-mode import
+    from frozen import frozen_match, load_frozen_ids
     from media_manifest import (
         _from_wire_dict,
         _item_to_wire,
@@ -53,7 +57,8 @@ except ImportError:  # pragma: no cover - script-mode import
     )
     from media_refs import remote_markup
     from models import MediaItem, ThreadData, load_thread
-    from render import render_branch, render_spine
+    from paths import FROZEN, SCRATCH
+    from render import render_branch, render_spine, title_text
     from slug import branch_file_name, date_prefix, thread_dir_name
     from tree import (
         by_id,
@@ -360,16 +365,6 @@ def collect_existing_ids(vault: Path) -> tuple[set[str], dict[str, Path]]:
     return ids, spines
 
 
-def unique_dir_name(parent: Path, base: str) -> str:
-    """Return ``base`` if unused, else ``base-2``, ``base-3``, ... until a free name is found."""
-    if not (parent / base).exists():
-        return base
-    n = 2
-    while (parent / f"{base}-{n}").exists():
-        n += 1
-    return f"{base}-{n}"
-
-
 def _split_frontmatter(text: str) -> tuple[str, str]:
     """Split an archive note into (frontmatter_with_closing_fence, body). Returns ``("", text)`` if no fence."""
     if not text.startswith("---"):
@@ -547,6 +542,8 @@ def emit(
     force: bool = False,
     reuse_dir: Path | None = None,
     tip: str | None = None,
+    reconcile_scratch: Path = SCRATCH,
+    frozen_ids: set[str] | None = None,
 ) -> EmitResult:
     src = input_dir / "thread_data.json"
     if not src.is_file():
@@ -567,7 +564,8 @@ def emit(
     kids = children_map(thread)
     first = ids[spine[0]]
     handle = first.handle
-    first_line = (first.text or "").split("\n", 1)[0]
+    rendered_title = title_text(first.text)
+    desired_dir_name = thread_dir_name(date_prefix(first.timestamp), rendered_title, slug)
 
     if reuse_dir is None and force:
         _existing_ids, spines = collect_existing_ids(vault)
@@ -590,19 +588,47 @@ def emit(
                 reuse_dir = candidate
 
     if reuse_dir is not None:
-        dir_name = reuse_dir.name
         handle = reuse_dir.parent.name
+        dir_name = reuse_dir.name
         note_dir = reuse_dir
         asset_dir = vault / "assets" / "threads" / handle / dir_name
-    else:
-        dir_name = thread_dir_name(date_prefix(first.timestamp), first_line, slug)
-        if slug is None:
-            dir_name = unique_dir_name(
-                vault / "archive" / "threads" / handle,
-                dir_name,
+        if dir_name != desired_dir_name:
+            active_frozen_ids = (
+                frozen_ids if frozen_ids is not None else load_frozen_ids(FROZEN)
             )
+            match = frozen_match(asset_dir, active_frozen_ids)
+            if match is not None:
+                raise SystemExit(f"frozen: skipped ({match})")
+            new_archive = vault / "archive" / "threads" / handle / desired_dir_name
+            new_asset = vault / "assets" / "threads" / handle / desired_dir_name
+            move = ReslugMove(
+                handle=handle,
+                old_dir_name=dir_name,
+                new_dir_name=desired_dir_name,
+                archive_old=reuse_dir,
+                archive_new=new_archive,
+                asset_old=asset_dir,
+                asset_new=new_asset,
+            )
+            rename_thread_pair(move)
+            frozen_dirs = plan_reslug(vault, active_frozen_ids).frozen_dirs
+            old_prefix = f"archive/threads/{handle}/{dir_name}"
+            new_prefix = f"archive/threads/{handle}/{desired_dir_name}"
+            rewrite_archive_paths(
+                vault,
+                reconcile_scratch,
+                {old_prefix: new_prefix},
+                frozen_dirs,
+            )
+            dir_name = desired_dir_name
+            note_dir = new_archive
+            asset_dir = new_asset
+    else:
+        dir_name = desired_dir_name
         note_dir = vault / "archive" / "threads" / handle / dir_name
         asset_dir = vault / "assets" / "threads" / handle / dir_name
+        if note_dir.exists() or asset_dir.exists():
+            raise SystemExit(f"destination occupied: {note_dir}")
 
     branch_names: dict[str, str] = {}
     for rid in roots:
@@ -754,6 +780,361 @@ def emit_all(
     return len(dumps), skipped, written, errors
 
 
+# --- reslug (one-time bulk title/path reconciliation) ---
+
+
+@dataclass(frozen=True)
+class ReslugMove:
+    """One paired archive/asset directory rename."""
+    handle: str
+    old_dir_name: str
+    new_dir_name: str
+    archive_old: Path
+    archive_new: Path
+    asset_old: Path
+    asset_new: Path
+
+
+@dataclass(frozen=True)
+class ReslugConflict:
+    """One bulk preflight failure surfaced before any rename."""
+    path: Path
+    reason: str
+
+
+@dataclass(frozen=True)
+class ReslugPlan:
+    """Small immutable one-time corpus repair plan."""
+    vault: Path
+    moves: tuple[ReslugMove, ...]
+    noops: tuple[Path, ...]
+    frozen: tuple[Path, ...]
+    frozen_dirs: tuple[Path, ...]
+    conflicts: tuple[ReslugConflict, ...]
+
+    @property
+    def can_apply(self) -> bool:
+        return not self.conflicts
+
+
+_TITLE_LINE: re.Pattern[str] = re.compile(r"^title:\s*(.+?)\s*$")
+_DATE_LINE: re.Pattern[str] = re.compile(r"^date:\s*(\d{4}-\d{2}-\d{2})\s*$")
+
+
+def _yaml_scalar_value(raw: str) -> str:
+    """Decode a single-line YAML scalar that is either JSON-quoted or unquoted."""
+    s = raw.strip()
+    if len(s) >= 2 and s.startswith('"') and s.endswith('"'):
+        body = s[1:-1]
+        body = body.replace("\\\\", "\x00").replace('\\"', '"').replace("\x00", "\\")
+        return body
+    return s
+
+
+def _frontmatter_title_date(text: str) -> tuple[str, str] | None:
+    """Return (title, date) from an archive note's frontmatter or ``None`` when unparseable."""
+    block = _frontmatter_block(text)
+    if not block:
+        return None
+    title: str | None = None
+    date_value: str | None = None
+    for line in block.splitlines():
+        stripped = line.strip()
+        if title is None:
+            match = _TITLE_LINE.match(line)
+            if match:
+                title = _yaml_scalar_value(match.group(1))
+                continue
+        if date_value is None:
+            match = _DATE_LINE.match(line)
+            if match:
+                date_value = match.group(1)
+                continue
+    if title is None or date_value is None:
+        return None
+    return title, date_value
+
+
+_RESLUG_EXCLUDED_TOP_LEVEL: tuple[str, ...] = (
+    ".git",
+    "site",
+    "assets",
+    "secrets",
+    "node_modules",
+)
+
+
+def plan_reslug(vault: Path, frozen_ids: set[str]) -> ReslugPlan:
+    """Build an immutable repair plan for ``archive/threads/<handle>/<thread>`` directories."""
+    moves: list[ReslugMove] = []
+    noops: list[Path] = []
+    frozen_entries: list[Path] = []
+    frozen_dirs: list[Path] = []
+    conflicts: list[ReslugConflict] = []
+
+    threads_root = vault / "archive" / "threads"
+    assets_root = vault / "assets" / "threads"
+
+    if not threads_root.is_dir():
+        return ReslugPlan(vault, (), (), (), (), ())
+
+    for handle_dir in sorted(threads_root.iterdir(), key=lambda p: p.name):
+        if not handle_dir.is_dir():
+            continue
+        handle = handle_dir.name
+        for thread_dir in sorted(handle_dir.iterdir(), key=lambda p: p.name):
+            if not thread_dir.is_dir():
+                continue
+            old_dir_name = thread_dir.name
+            index_path = thread_dir / "index.md"
+            asset_dir = assets_root / handle / old_dir_name
+            try:
+                if not index_path.is_file():
+                    conflicts.append(
+                        ReslugConflict(thread_dir, "missing index.md")
+                    )
+                    continue
+                try:
+                    text = index_path.read_text(encoding="utf-8")
+                except OSError as exc:
+                    conflicts.append(
+                        ReslugConflict(index_path, f"unreadable frontmatter: {exc}")
+                    )
+                    continue
+                parsed = _frontmatter_title_date(text)
+                if parsed is None:
+                    conflicts.append(
+                        ReslugConflict(
+                            index_path,
+                            "missing or malformed frontmatter title/date",
+                        )
+                    )
+                    continue
+                title, date_value = parsed
+                if not asset_dir.is_dir():
+                    conflicts.append(
+                        ReslugConflict(
+                            index_path,
+                            f"missing asset pair at {asset_dir}",
+                        )
+                    )
+                    continue
+                thread_data = asset_dir / "thread_data.json"
+                media_json = asset_dir / "media.json"
+                if not thread_data.is_file():
+                    conflicts.append(
+                        ReslugConflict(index_path, "missing thread_data.json")
+                    )
+                    continue
+                if not media_json.is_file():
+                    conflicts.append(
+                        ReslugConflict(index_path, "missing media.json")
+                    )
+                    continue
+                try:
+                    json.loads(thread_data.read_text(encoding="utf-8"))
+                except (OSError, ValueError) as exc:
+                    conflicts.append(
+                        ReslugConflict(
+                            index_path,
+                            f"invalid thread_data.json: {exc}",
+                        )
+                    )
+                    continue
+                try:
+                    json.loads(media_json.read_text(encoding="utf-8"))
+                except (OSError, ValueError) as exc:
+                    conflicts.append(
+                        ReslugConflict(
+                            index_path,
+                            f"invalid media.json: {exc}",
+                        )
+                    )
+                    continue
+                expected = thread_dir_name(date_value, title, None)
+                match = frozen_match(asset_dir, frozen_ids)
+                if old_dir_name == expected:
+                    noops.append(thread_dir)
+                    if match is not None:
+                        frozen_dirs.append(thread_dir)
+                    continue
+                if match is not None:
+                    frozen_entries.append(thread_dir)
+                    frozen_dirs.append(thread_dir)
+                    continue
+                new_archive = threads_root / handle / expected
+                new_asset = assets_root / handle / expected
+                if new_archive.exists():
+                    conflicts.append(
+                        ReslugConflict(
+                            thread_dir,
+                            f"destination occupied: {new_archive}",
+                        )
+                    )
+                    continue
+                if new_asset.exists():
+                    conflicts.append(
+                        ReslugConflict(
+                            thread_dir,
+                            f"destination occupied: {new_asset}",
+                        )
+                    )
+                    continue
+                moves.append(
+                    ReslugMove(
+                        handle=handle,
+                        old_dir_name=old_dir_name,
+                        new_dir_name=expected,
+                        archive_old=thread_dir,
+                        archive_new=new_archive,
+                        asset_old=asset_dir,
+                        asset_new=new_asset,
+                    )
+                )
+            except OSError as exc:
+                conflicts.append(
+                    ReslugConflict(thread_dir, f"filesystem error: {exc}")
+                )
+                continue
+
+    moves.sort(key=lambda m: str(m.archive_old))
+    noops.sort(key=str)
+    frozen_entries.sort(key=str)
+    frozen_dirs.sort(key=str)
+    conflicts.sort(key=lambda c: str(c.path))
+    return ReslugPlan(
+        vault,
+        tuple(moves),
+        tuple(noops),
+        tuple(frozen_entries),
+        tuple(frozen_dirs),
+        tuple(conflicts),
+    )
+
+
+def rename_thread_pair(move: ReslugMove) -> None:
+    """Validate, rename assets first, then archive; roll the asset back on archive failure."""
+    if not move.archive_old.is_dir() or not move.asset_old.is_dir():
+        raise RuntimeError(f"source pair missing: {move.archive_old}")
+    if move.archive_new.exists() or move.asset_new.exists():
+        raise RuntimeError(f"destination occupied: {move.archive_new}")
+    move.asset_old.rename(move.asset_new)
+    try:
+        move.archive_old.rename(move.archive_new)
+    except OSError:
+        move.asset_new.rename(move.asset_old)
+        raise
+
+
+def rewrite_archive_paths(
+    vault: Path,
+    scratch: Path,
+    replacements: dict[str, str],
+    frozen_dirs: tuple[Path, ...],
+) -> tuple[Path, ...]:
+    """Recursively rewrite exact vault-root prefixes in mutable Markdown and Canvas files."""
+    if not replacements:
+        return ()
+
+    keys = sorted(replacements.keys(), key=len, reverse=True)
+    pattern = re.compile("|".join(re.escape(key) for key in keys))
+
+    changed: list[Path] = []
+    stage_dir = scratch / "reslug_text"
+    sequence = 0
+    for path in sorted(vault.rglob("*")):
+        if not path.is_file():
+            continue
+        if path.suffix.lower() not in (".md", ".canvas"):
+            continue
+        try:
+            rel_parts = path.relative_to(vault).parts
+        except ValueError:
+            continue
+        if rel_parts and rel_parts[0] in _RESLUG_EXCLUDED_TOP_LEVEL:
+            continue
+        skip = False
+        for frozen in frozen_dirs:
+            try:
+                path.relative_to(frozen)
+                skip = True
+                break
+            except ValueError:
+                continue
+        if skip:
+            continue
+        try:
+            text = path.read_text(encoding="utf-8")
+        except (OSError, UnicodeDecodeError):
+            continue
+        new_text = pattern.sub(lambda m: replacements[m.group(0)], text)
+        if new_text == text:
+            continue
+        stage_dir.mkdir(parents=True, exist_ok=True)
+        stage_path = stage_dir / f"{sequence}.tmp"
+        sequence += 1
+        stage_path.write_text(new_text, encoding="utf-8", newline="\n")
+        os.replace(stage_path, path)
+        changed.append(path)
+
+    try:
+        stage_dir.rmdir()
+    except OSError:
+        pass
+
+    changed.sort(key=str)
+    return tuple(changed)
+
+
+def apply_reslug_plan(plan: ReslugPlan, scratch: Path) -> tuple[Path, ...]:
+    """Run every rename, then recursively rewrite mutable paths; refuse on conflict."""
+    if plan.conflicts:
+        reasons = ", ".join(c.reason for c in plan.conflicts)
+        raise RuntimeError(
+            f"reslug blocked by {len(plan.conflicts)} conflict(s): {reasons}"
+        )
+    replacements: dict[str, str] = {}
+    for move in plan.moves:
+        rename_thread_pair(move)
+        old_prefix = f"archive/threads/{move.handle}/{move.old_dir_name}"
+        new_prefix = f"archive/threads/{move.handle}/{move.new_dir_name}"
+        replacements[old_prefix] = new_prefix
+        print(f"renamed -> {move.archive_new}")
+    rewritten = (
+        rewrite_archive_paths(
+            plan.vault,
+            scratch,
+            replacements,
+            plan.frozen_dirs,
+        )
+        if plan.moves
+        else ()
+    )
+    for path in rewritten:
+        print(f"rewrote -> {path}")
+    return rewritten
+
+
+def format_reslug_plan(plan: ReslugPlan) -> str:
+    """Return a deterministic human-readable summary of a repair plan."""
+    lines: list[str] = []
+    for path in plan.noops:
+        lines.append(f"[noop] {path.as_posix()}")
+    for move in plan.moves:
+        lines.append(
+            f"[rename] {move.archive_old.as_posix()} -> {move.archive_new.as_posix()}"
+        )
+    for path in plan.frozen:
+        lines.append(f"[frozen] {path.as_posix()}")
+    for conflict in plan.conflicts:
+        lines.append(f"[conflict] {conflict.path.as_posix()}: {conflict.reason}")
+    lines.append(
+        f"summary: moves={len(plan.moves)} noops={len(plan.noops)} "
+        f"frozen={len(plan.frozen)} conflicts={len(plan.conflicts)}"
+    )
+    return "\n".join(lines) + "\n"
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(
         description=(

scripts/twitter/render.py

@@ -29,8 +29,8 @@ def _first_line(text: str) -> str:
     return (text or "").split("\n", 1)[0].rstrip("\r")
 
 
-def _title_text(text: str) -> str:
-    """Derive a title from the first line, truncating at the first sentence end (``. ``)."""
+def title_text(text: str) -> str:
+    """Derive the rendered title from the first sentence of the first line."""
     line = _first_line(text)
     idx = line.find(". ")
     if idx >= 0:
@@ -149,7 +149,7 @@ def render_spine(
     first = ids[spine[0]]
     spine_handle = first.handle
     in_reply_to, parent = _foreign_parent(first, ids)
-    title = _title_text(first.text)
+    title = title_text(first.text)
     source_url = _status_url(first.handle, first.post_id)
     chunks = [
         _frontmatter(
@@ -211,7 +211,7 @@ def render_per_author_spine(
             "render_per_author_spine requires a non-empty author_posts list"
         )
     first = author_posts[0]
-    title = _title_text(first.text)
+    title = title_text(first.text)
     source_url = _status_url(first.handle, first.post_id)
     chunks: list[str] = [
         _frontmatter(
@@ -260,7 +260,7 @@ def render_branch(
     """descendant_ids are posts after the root, in display order (tree.descendants)."""
     ids = by_id(thread)
     root = ids[branch_root_id]
-    title = _title_text(root.text)
+    title = title_text(root.text)
     source_url = _status_url(root.handle, root.post_id)
     chunks = [
         _frontmatter(

scripts/twitter/tests/test_emit_media.py

@@ -6,7 +6,7 @@ from pathlib import Path
 
 from twitter.emit_archive import missing_local_media, preserve_review_state
 from twitter.media_manifest import new_original_item
-from twitter.render import render_media
+from twitter.render import render_media, title_text
 
 
 class EmitMediaTests(unittest.TestCase):
@@ -62,6 +62,13 @@ class EmitMediaTests(unittest.TestCase):
             )
             self.assertEqual(missing_local_media([item]), ["100/AAA/orig"])
 
+    def test_title_text_is_shared_by_frontmatter_and_slug(self) -> None:
+        self.assertEqual(
+            title_text("Root title. Extra first-line detail"),
+            "Root title.",
+        )
+        self.assertEqual(title_text("One uninterrupted title"), "One uninterrupted title")
+
 
 if __name__ == "__main__":
     unittest.main()

scripts/twitter/tests/test_emit_op_owned.py

@@ -13,6 +13,7 @@ import shutil
 import tempfile
 import unittest
 from pathlib import Path
+from tempfile import TemporaryDirectory
 
 from twitter.emit_archive import emit
 from twitter.models import PostData, PostMetrics, ThreadData
@@ -812,5 +813,388 @@ class TipIsOpTests(unittest.TestCase):
         self.assertEqual(spine, ["2000", "2001", "2002"])
 
 
+class ReslugReuseTests(unittest.TestCase):
+    """Compact same-handle integration tests for the emit-side rename flow.
+
+    Each test pre-populates one or more archive/asset pairs under the
+    vault, invokes ``emit`` with explicit ``reconcile_scratch`` and
+    ``frozen_ids`` so nothing is written to the workspace scratch and
+    no frozen list is read from secrets.
+    """
+
+    HANDLE = "alice"
+    ROOT_ID = "1000"
+    TIMESTAMP = "2026-08-24 12:00:00"
+    ROOT_TEXT = "Root title. Extra first-line detail that should be dropped."
+    EXPECTED_CANONICAL = "2026-08-24-root-title"
+
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp(prefix="reslug_reuse_"))
+        self.input_dir = self.tmp / "input"
+        self.input_dir.mkdir()
+        self.vault = self.tmp / "vault"
+        (self.vault / "archive" / "threads").mkdir(parents=True)
+        (self.vault / "assets" / "threads").mkdir(parents=True)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _write_thread(self) -> None:
+        thread = {
+            "root_post_id": self.ROOT_ID,
+            "source_url": f"https://x.com/{self.HANDLE}/status/{self.ROOT_ID}",
+            "posts": [
+                {
+                    "post_id": self.ROOT_ID,
+                    "author": self.HANDLE,
+                    "handle": self.HANDLE,
+                    "text": self.ROOT_TEXT,
+                    "timestamp": self.TIMESTAMP,
+                    "media_urls": [],
+                    "reply_to_id": None,
+                    "quote_of_id": None,
+                    "metrics": {
+                        "reply_count": 0,
+                        "repost_count": 0,
+                        "like_count": 0,
+                        "view_count": 0,
+                    },
+                }
+            ],
+        }
+        (self.input_dir / "thread_data.json").write_text(
+            json.dumps(thread, indent=2),
+            encoding="utf-8",
+            newline="\n",
+        )
+
+    def _write_existing_pair(
+        self,
+        *,
+        dir_name: str,
+        post_id: str,
+        index_body: str = "",
+        editorial_tag: str | None = None,
+        draft: str = "true",
+        timestamp: str | None = None,
+    ) -> tuple[Path, Path]:
+        archive = self.vault / "archive" / "threads" / self.HANDLE / dir_name
+        asset = self.vault / "assets" / "threads" / self.HANDLE / dir_name
+        archive.mkdir(parents=True)
+        asset.mkdir(parents=True)
+        body_timestamp = timestamp or self.TIMESTAMP
+        date_value = body_timestamp[:10]
+        tags_lines = (
+            "tags:\n"
+            "  - archive\n"
+            "  - twitter\n"
+            f"  - {self.HANDLE}\n"
+        )
+        if editorial_tag:
+            tags_lines += f"  - {editorial_tag}\n"
+        (archive / "index.md").write_text(
+            "---\n"
+            'title: "Old title."\n'
+            f"handle: {self.HANDLE}\n"
+            f'post_id: "{post_id}"\n'
+            f"date: {date_value}\n"
+            f"draft: {draft}\n"
+            f"{tags_lines}---\n\n{index_body}",
+            encoding="utf-8",
+            newline="\n",
+        )
+        (asset / "thread_data.json").write_text(
+            json.dumps(
+                {
+                    "root_post_id": post_id,
+                    "source_url": f"https://x.com/{self.HANDLE}/status/{post_id}",
+                    "posts": [
+                        {
+                            "post_id": post_id,
+                            "author": self.HANDLE,
+                            "handle": self.HANDLE,
+                            "text": "Old body.",
+                            "timestamp": body_timestamp,
+                            "media_urls": [],
+                            "reply_to_id": None,
+                            "quote_of_id": None,
+                            "metrics": {},
+                        }
+                    ],
+                },
+                indent=2,
+            ),
+            encoding="utf-8",
+            newline="\n",
+        )
+        (asset / "media.json").write_text(
+            json.dumps(
+                {
+                    "schema_version": 2,
+                    "root_post_id": post_id,
+                    "items": [],
+                    "mirrors": [],
+                },
+                indent=2,
+            ),
+            encoding="utf-8",
+            newline="\n",
+        )
+        return archive, asset
+
+    def test_emit_renames_reused_pair_to_canonical_dir(self) -> None:
+        """Existing archive/asset pair found by post identity is renamed
+        to date + rendered-title canonical directory."""
+        old_dir = "2026-08-24-stale-slug"
+        archive_old, asset_old = self._write_existing_pair(
+            dir_name=old_dir,
+            post_id=self.ROOT_ID,
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-24",
+                force=True,
+                reconcile_scratch=Path(scratch_raw),
+                frozen_ids=set(),
+            )
+
+        self.assertFalse(archive_old.exists())
+        self.assertFalse(asset_old.exists())
+        archive_new = (
+            self.vault
+            / "archive"
+            / "threads"
+            / self.HANDLE
+            / self.EXPECTED_CANONICAL
+        )
+        asset_new = (
+            self.vault
+            / "assets"
+            / "threads"
+            / self.HANDLE
+            / self.EXPECTED_CANONICAL
+        )
+        self.assertTrue(archive_new.is_dir())
+        self.assertTrue(asset_new.is_dir())
+        self.assertTrue((asset_new / "thread_data.json").is_file())
+        self.assertTrue((asset_new / "media.json").is_file())
+        self.assertTrue((archive_new / "index.md").is_file())
+
+    def test_emit_handle_index_links_to_new_path_after_rename(self) -> None:
+        """The handle index wikilink is updated from the old prefix to
+        the new canonical prefix after rename."""
+        old_dir = "2026-08-24-stale-slug"
+        self._write_existing_pair(
+            dir_name=old_dir,
+            post_id=self.ROOT_ID,
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-24",
+                force=True,
+                reconcile_scratch=Path(scratch_raw),
+                frozen_ids=set(),
+            )
+
+        handle_index = self.vault / "archive" / "threads" / self.HANDLE / "index.md"
+        self.assertTrue(handle_index.is_file())
+        text = handle_index.read_text(encoding="utf-8")
+        self.assertIn(
+            f"archive/threads/{self.HANDLE}/{self.EXPECTED_CANONICAL}",
+            text,
+        )
+        self.assertNotIn(f"archive/threads/{self.HANDLE}/{old_dir}", text)
+
+    def test_emit_preserves_draft_false_and_editorial_tags_across_rename(self) -> None:
+        """Existing ``draft: false`` and a non-mechanical editorial tag
+        survive the rename."""
+        old_dir = "2026-08-24-stale-slug"
+        self._write_existing_pair(
+            dir_name=old_dir,
+            post_id=self.ROOT_ID,
+            editorial_tag="editorial",
+            draft="false",
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-24",
+                force=True,
+                reconcile_scratch=Path(scratch_raw),
+                frozen_ids=set(),
+            )
+
+        archive_new = (
+            self.vault
+            / "archive"
+            / "threads"
+            / self.HANDLE
+            / self.EXPECTED_CANONICAL
+        )
+        text = (archive_new / "index.md").read_text(encoding="utf-8")
+        self.assertIn("draft: false", text)
+        self.assertIn("- editorial", text)
+
+    def test_emit_slug_override_uses_explicit_suffix(self) -> None:
+        """An explicit ``--slug`` override produces a ``date-manual-name``
+        directory even when a reused old pair exists."""
+        old_dir = "2026-08-24-stale-slug"
+        self._write_existing_pair(
+            dir_name=old_dir,
+            post_id=self.ROOT_ID,
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug="manual-name",
+                archived="2026-08-24",
+                force=True,
+                reconcile_scratch=Path(scratch_raw),
+                frozen_ids=set(),
+            )
+
+        expected = "2026-08-24-manual-name"
+        self.assertTrue(
+            (self.vault / "archive" / "threads" / self.HANDLE / expected).is_dir()
+        )
+        self.assertTrue(
+            (self.vault / "assets" / "threads" / self.HANDLE / expected).is_dir()
+        )
+        self.assertFalse(
+            (self.vault / "archive" / "threads" / self.HANDLE / old_dir).exists()
+        )
+
+    def test_emit_occupied_canonical_target_refuses_before_move(self) -> None:
+        """A new-thread emit whose desired canonical target is already
+        occupied by another pair must abort with ``SystemExit`` before any
+        write, with no ``-2`` suffix fallback."""
+        canonical_archive, canonical_asset = self._write_existing_pair(
+            dir_name=self.EXPECTED_CANONICAL,
+            post_id="9999",
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            with self.assertRaisesRegex(SystemExit, "destination occupied"):
+                emit(
+                    input_dir=self.input_dir,
+                    vault=self.vault,
+                    slug=None,
+                    archived="2026-08-24",
+                    force=True,
+                    reconcile_scratch=Path(scratch_raw),
+                    frozen_ids=set(),
+                )
+
+        self.assertFalse(
+            (
+                self.vault
+                / "archive"
+                / "threads"
+                / self.HANDLE
+                / f"{self.EXPECTED_CANONICAL}-2"
+            ).exists()
+        )
+        self.assertFalse(
+            (
+                self.vault
+                / "assets"
+                / "threads"
+                / self.HANDLE
+                / f"{self.EXPECTED_CANONICAL}-2"
+            ).exists()
+        )
+        self.assertTrue(canonical_archive.is_dir())
+        self.assertTrue(canonical_asset.is_dir())
+
+    def test_emit_frozen_mismatch_refuses_before_move(self) -> None:
+        """A frozen mismatch on the existing asset dir must abort with
+        ``SystemExit`` before either directory is moved."""
+        old_dir = "2026-08-24-stale-slug"
+        archive_old, asset_old = self._write_existing_pair(
+            dir_name=old_dir,
+            post_id=self.ROOT_ID,
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            with self.assertRaisesRegex(SystemExit, "frozen"):
+                emit(
+                    input_dir=self.input_dir,
+                    vault=self.vault,
+                    slug=None,
+                    archived="2026-08-24",
+                    force=True,
+                    reconcile_scratch=Path(scratch_raw),
+                    frozen_ids={self.ROOT_ID},
+                )
+
+        self.assertTrue(archive_old.is_dir())
+        self.assertTrue(asset_old.is_dir())
+        self.assertFalse(
+            (
+                self.vault
+                / "archive"
+                / "threads"
+                / self.HANDLE
+                / self.EXPECTED_CANONICAL
+            ).exists()
+        )
+        self.assertFalse(
+            (
+                self.vault
+                / "assets"
+                / "threads"
+                / self.HANDLE
+                / self.EXPECTED_CANONICAL
+            ).exists()
+        )
+
+    def test_emit_matching_existing_name_emits_normally(self) -> None:
+        """If the existing reused pair already matches the desired name,
+        emit reuses it without rename and writes the fresh note + media."""
+        archive_canonical, asset_canonical = self._write_existing_pair(
+            dir_name=self.EXPECTED_CANONICAL,
+            post_id=self.ROOT_ID,
+        )
+        self._write_thread()
+
+        with TemporaryDirectory() as scratch_raw:
+            emit(
+                input_dir=self.input_dir,
+                vault=self.vault,
+                slug=None,
+                archived="2026-08-24",
+                force=True,
+                reconcile_scratch=Path(scratch_raw),
+                frozen_ids=set(),
+            )
+
+        self.assertTrue(archive_canonical.is_dir())
+        self.assertTrue(asset_canonical.is_dir())
+        self.assertTrue((archive_canonical / "index.md").is_file())
+        self.assertTrue((asset_canonical / "thread_data.json").is_file())
+        handle_dir = self.vault / "archive" / "threads" / self.HANDLE
+        dirs = sorted(d.name for d in handle_dir.iterdir() if d.is_dir())
+        self.assertEqual(dirs, [self.EXPECTED_CANONICAL])
+
+
 if __name__ == "__main__":
     unittest.main()

scripts/twitter/tests/test_reslug.py

@@ -0,0 +1,494 @@
+"""Focused tests for the one-time bulk thread-directory reslug plumbing.
+
+These tests exercise the small planning, pair-rename, recursive-rewrite,
+apply, and format helpers added to ``emit_archive``. They use compact local
+fixtures under a ``TemporaryDirectory`` and never touch the real vault.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from twitter.emit_archive import (
+    ReslugConflict,
+    ReslugMove,
+    ReslugPlan,
+    apply_reslug_plan,
+    format_reslug_plan,
+    plan_reslug,
+    rename_thread_pair,
+    rewrite_archive_paths,
+)
+
+
+def _write_pair(
+    vault: Path,
+    *,
+    handle: str = "example",
+    directory: str = "2026-08-24-wrong",
+    title: str = "Canonical title.",
+    date: str = "2026-08-24",
+    post_id: str = "100",
+) -> tuple[Path, Path]:
+    """Build a paired archive/asset directory with a thread_data + media json."""
+    archive = vault / "archive" / "threads" / handle / directory
+    assets = vault / "assets" / "threads" / handle / directory
+    archive.mkdir(parents=True, exist_ok=True)
+    assets.mkdir(parents=True, exist_ok=True)
+    (archive / "index.md").write_text(
+        "---\n"
+        f"title: {json.dumps(title)}\n"
+        f"handle: {handle}\n"
+        f'post_id: "{post_id}"\n'
+        f"date: {date}\n"
+        "draft: true\n"
+        "tags:\n"
+        "  - archive\n"
+        "  - twitter\n"
+        f"  - {handle}\n"
+        "---\n\n**1/**\n\nFixture body.\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    (assets / "thread_data.json").write_text(
+        json.dumps(
+            {
+                "root_post_id": post_id,
+                "source_url": f"https://x.com/{handle}/status/{post_id}",
+                "posts": [
+                    {
+                        "post_id": post_id,
+                        "author": handle,
+                        "handle": handle,
+                        "text": title,
+                        "timestamp": f"{date} 12:00:00",
+                        "media_urls": [],
+                        "reply_to_id": None,
+                        "quote_of_id": None,
+                        "metrics": {},
+                    }
+                ],
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    (assets / "media.json").write_text(
+        json.dumps(
+            {
+                "schema_version": 2,
+                "root_post_id": post_id,
+                "items": [],
+                "mirrors": [],
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+    return archive, assets
+
+
+class ReslugPlanTests(unittest.TestCase):
+    def test_plan_computes_expected_dir_from_frontmatter(self) -> None:
+        """Title/date frontmatter must yield the canonical dir name."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault)
+            plan = plan_reslug(vault, set())
+            self.assertFalse(plan.conflicts)
+            self.assertEqual(len(plan.moves), 1)
+            self.assertEqual(plan.moves[0].new_dir_name, "2026-08-24-canonical-title")
+            self.assertEqual(plan.moves[0].old_dir_name, "2026-08-24-wrong")
+            self.assertEqual(plan.moves[0].handle, "example")
+
+    def test_exact_match_is_noop(self) -> None:
+        """A directory whose name already matches must be reported as a no-op."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault, directory="2026-08-24-canonical-title")
+            plan = plan_reslug(vault, set())
+            self.assertEqual(len(plan.noops), 1)
+            self.assertFalse(plan.moves)
+            self.assertFalse(plan.conflicts)
+            self.assertTrue(plan.can_apply)
+
+    def test_occupied_destination_blocks_entire_plan(self) -> None:
+        """An occupied archive or asset destination must surface as a conflict."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault)
+            _write_pair(vault, directory="2026-08-24-canonical-title", post_id="200")
+            plan = plan_reslug(vault, set())
+            self.assertTrue(plan.conflicts)
+            self.assertFalse(plan.can_apply)
+            self.assertFalse(plan.moves)
+
+    def test_mismatched_frozen_pair_is_skipped_and_tracked(self) -> None:
+        """A mismatched frozen pair is skipped and its archive dir is excluded."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault, post_id="300")
+            plan = plan_reslug(vault, {"300"})
+            self.assertEqual(len(plan.frozen), 1)
+            self.assertEqual(len(plan.frozen_dirs), 1)
+            self.assertFalse(plan.moves)
+            self.assertFalse(plan.conflicts)
+            self.assertTrue(plan.can_apply)
+
+    def test_missing_asset_pair_becomes_conflict(self) -> None:
+        """An archive dir with no asset partner is a blocking conflict."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            archive = vault / "archive" / "threads" / "example" / "2026-08-24-wrong"
+            archive.mkdir(parents=True)
+            (archive / "index.md").write_text(
+                "---\n"
+                'title: "Canonical title."\n'
+                'handle: example\n'
+                'post_id: "100"\n'
+                "date: 2026-08-24\n"
+                "draft: true\n"
+                "tags:\n"
+                "  - archive\n"
+                "---\n\n**1/**\n\nFixture.\n",
+                encoding="utf-8",
+                newline="\n",
+            )
+            plan = plan_reslug(vault, set())
+            self.assertTrue(plan.conflicts)
+            self.assertFalse(plan.moves)
+
+
+class ReslugRenameTests(unittest.TestCase):
+    def test_rename_moves_both_dirs(self) -> None:
+        """rename_thread_pair must move assets first and archive second."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault)
+            plan = plan_reslug(vault, set())
+            move = plan.moves[0]
+            rename_thread_pair(move)
+            self.assertFalse(move.archive_old.exists())
+            self.assertFalse(move.asset_old.exists())
+            self.assertTrue(move.archive_new.is_dir())
+            self.assertTrue(move.asset_new.is_dir())
+
+    def test_archive_failure_rolls_asset_back(self) -> None:
+        """An injected archive rename failure must move the asset back and re-raise."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault)
+            move = plan_reslug(vault, set()).moves[0]
+            original_rename = Path.rename
+
+            def fail_archive(path: Path, destination: Path) -> Path:
+                if path == move.archive_old:
+                    raise OSError("injected archive failure")
+                return original_rename(path, destination)
+
+            with patch.object(Path, "rename", fail_archive):
+                with self.assertRaisesRegex(OSError, "injected"):
+                    rename_thread_pair(move)
+            self.assertTrue(move.archive_old.is_dir())
+            self.assertTrue(move.asset_old.is_dir())
+            self.assertFalse(move.asset_new.exists())
+
+
+class ReslugRewriteTests(unittest.TestCase):
+    def test_recursive_rewrite_updates_markdown_canvas_and_branch_paths(self) -> None:
+        """Markdown and Canvas files inside and outside the thread dir must update."""
+        with TemporaryDirectory() as raw:
+            root = Path(raw)
+            vault = root / "vault"
+            _write_pair(vault)
+            old = "archive/threads/example/2026-08-24-wrong"
+            new = "archive/threads/example/2026-08-24-canonical-title"
+            inbound = vault / "notes" / "inbound.md"
+            canvas = vault / "canvases" / "map.canvas"
+            inbound.parent.mkdir(parents=True)
+            canvas.parent.mkdir(parents=True)
+            inbound.write_text(
+                f"[[{old}/branch]] and [[{old}|alias]]\n",
+                encoding="utf-8",
+            )
+            canvas.write_text(
+                json.dumps({"edges": [{"file": f"{old}/branch.md"}]}),
+                encoding="utf-8",
+            )
+            archive_branch = vault / "archive" / "threads" / "example" / "2026-08-24-wrong" / "branch.md"
+            archive_branch.write_text(f"link: [[{old}/index]]\n", encoding="utf-8")
+            changed = rewrite_archive_paths(vault, root / "scratch", {old: new}, ())
+            self.assertIn(new + "/branch", inbound.read_text(encoding="utf-8"))
+            self.assertIn(new + "/branch.md", canvas.read_text(encoding="utf-8"))
+            self.assertIn(new + "/index", archive_branch.read_text(encoding="utf-8"))
+            self.assertIn(canvas, changed)
+            self.assertIn(inbound, changed)
+            self.assertIn(archive_branch, changed)
+
+    def test_replacement_is_one_pass_and_non_cascading(self) -> None:
+        """Both keys must be applied in one pass; longest key wins at overlap."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            inbound = vault / "notes" / "inbound.md"
+            inbound.parent.mkdir(parents=True)
+            inbound.write_text(
+                "[[archive/threads/example/old]] and [[archive/threads/example/old-x]]\n",
+                encoding="utf-8",
+            )
+            rewrite_archive_paths(
+                vault,
+                Path(raw) / "scratch",
+                {
+                    "archive/threads/example/old": "archive/threads/example/new",
+                    "archive/threads/example/old-x": "archive/threads/example/new-x",
+                },
+                (),
+            )
+            self.assertEqual(
+                inbound.read_text(encoding="utf-8"),
+                "[[archive/threads/example/new]] and [[archive/threads/example/new-x]]\n",
+            )
+
+    def test_excluded_top_level_dirs_and_frozen_dirs_are_byte_identical(self) -> None:
+        """site/, assets/, secrets/, .git/, node_modules/, and frozen dirs must not change."""
+        with TemporaryDirectory() as raw:
+            root = Path(raw)
+            vault = root / "vault"
+            _write_pair(vault)
+            old = "archive/threads/example/2026-08-24-wrong"
+            new = "archive/threads/example/2026-08-24-canonical-title"
+            frozen_archive, _ = _write_pair(
+                vault,
+                handle="frozen_author",
+                directory="2024-01-01-locked",
+                title="Locked title.",
+                date="2024-01-01",
+                post_id="999",
+            )
+            frozen_index = frozen_archive / "index.md"
+            frozen_bytes = frozen_index.read_bytes()
+            for excluded in ("site", "assets", "secrets", "node_modules", ".git"):
+                excluded_dir = vault / excluded / "nested"
+                excluded_dir.mkdir(parents=True, exist_ok=True)
+                (excluded_dir / "note.md").write_text(
+                    f"see {old} for context\n",
+                    encoding="utf-8",
+                )
+            rewrite_archive_paths(
+                vault,
+                root / "scratch",
+                {old: new},
+                (frozen_archive,),
+            )
+            self.assertEqual(frozen_index.read_bytes(), frozen_bytes)
+            for excluded in ("site", "assets", "secrets", "node_modules", ".git"):
+                txt = (vault / excluded / "nested" / "note.md").read_text(encoding="utf-8")
+                self.assertIn(old, txt)
+                self.assertNotIn(new, txt)
+
+
+class ReslugApplyTests(unittest.TestCase):
+    def test_apply_refuses_conflicts_before_any_move(self) -> None:
+        """apply_reslug_plan must raise before touching disk when conflicts exist."""
+        with TemporaryDirectory() as raw:
+            root = Path(raw)
+            vault = root / "vault"
+            old_archive, old_assets = _write_pair(vault)
+            _write_pair(vault, directory="2026-08-24-canonical-title", post_id="200")
+            plan = plan_reslug(vault, set())
+            with self.assertRaisesRegex(RuntimeError, "conflict"):
+                apply_reslug_plan(plan, root / "scratch")
+            self.assertTrue(old_archive.is_dir())
+            self.assertTrue(old_assets.is_dir())
+
+    def test_apply_moves_pairs_and_rewrites_inbound_links(self) -> None:
+        """apply_reslug_plan must move both dirs and rewrite inbound links."""
+        with TemporaryDirectory() as raw:
+            root = Path(raw)
+            vault = root / "vault"
+            _write_pair(vault)
+            old = "archive/threads/example/2026-08-24-wrong"
+            new = "archive/threads/example/2026-08-24-canonical-title"
+            inbound = vault / "notes" / "inbound.md"
+            inbound.parent.mkdir(parents=True)
+            inbound.write_text(f"see [[{old}/index]]\n", encoding="utf-8")
+            plan = plan_reslug(vault, set())
+            rewritten = apply_reslug_plan(plan, root / "scratch")
+            self.assertTrue((vault / "archive" / "threads" / "example" / "2026-08-24-canonical-title").is_dir())
+            self.assertTrue((vault / "assets" / "threads" / "example" / "2026-08-24-canonical-title").is_dir())
+            self.assertIn(new, inbound.read_text(encoding="utf-8"))
+            self.assertIn(inbound, rewritten)
+
+
+class ReslugFormatTests(unittest.TestCase):
+    def test_format_output_is_deterministic_and_complete(self) -> None:
+        """format_reslug_plan must enumerate noop/rename/conflict lines plus counts deterministically."""
+        with TemporaryDirectory() as raw:
+            vault = Path(raw) / "vault"
+            _write_pair(vault, directory="2026-08-24-canonical-title")
+            _write_pair(
+                vault,
+                handle="other",
+                directory="2024-01-01-old",
+                title="Other title.",
+                date="2024-01-01",
+                post_id="500",
+            )
+            _write_pair(
+                vault,
+                handle="busy",
+                directory="2024-05-05-old",
+                title="Busy title.",
+                date="2024-05-05",
+                post_id="600",
+            )
+            _write_pair(
+                vault,
+                handle="busy",
+                directory="2024-05-05-busy-title",
+                title="Busy title.",
+                date="2024-05-05",
+                post_id="700",
+            )
+            plan = plan_reslug(vault, set())
+            first = format_reslug_plan(plan)
+            second = format_reslug_plan(plan)
+            self.assertEqual(first, second, "format must be deterministic")
+            lines = first.splitlines()
+            noop_lines = [line for line in lines if line.startswith("[noop]")]
+            rename_lines = [line for line in lines if line.startswith("[rename]")]
+            conflict_lines = [line for line in lines if line.startswith("[conflict]")]
+            self.assertGreaterEqual(len(noop_lines), 1)
+            self.assertEqual(len(rename_lines), 1)
+            self.assertEqual(len(conflict_lines), 1)
+            noop_text = "\n".join(noop_lines)
+            self.assertIn(
+                "archive/threads/example/2026-08-24-canonical-title",
+                noop_text,
+            )
+            self.assertIn("archive/threads/other/2024-01-01-old", rename_lines[0])
+            self.assertIn(
+                "archive/threads/other/2024-01-01-other-title",
+                rename_lines[0],
+            )
+            self.assertIn("->", rename_lines[0])
+            self.assertIn("archive/threads/busy", conflict_lines[0])
+            self.assertIn("2024-05-05-old", conflict_lines[0])
+            self.assertTrue(lines[-1].startswith("summary: "))
+            self.assertIn("moves=", lines[-1])
+            self.assertIn("noops=", lines[-1])
+            self.assertIn("conflicts=", lines[-1])
+            self.assertTrue(first.endswith("\n"))
+
+
+class ReslugModuleOrderTests(unittest.TestCase):
+    def test_reslug_definitions_appear_before_main_guard(self) -> None:
+        """The reslug block must sit above ``if __name__ == "__main__"``."""
+        module_path = Path(__file__).resolve().parent.parent / "emit_archive.py"
+        tree = ast.parse(module_path.read_text(encoding="utf-8"))
+        names: set[str] = {
+            "ReslugMove",
+            "ReslugConflict",
+            "ReslugPlan",
+            "plan_reslug",
+            "format_reslug_plan",
+            "rename_thread_pair",
+            "rewrite_archive_paths",
+            "apply_reslug_plan",
+        }
+        main_line: int | None = None
+        for node in tree.body:
+            if not isinstance(node, ast.If):
+                continue
+            test = node.test
+            if (
+                isinstance(test, ast.Compare)
+                and len(test.ops) == 1
+                and isinstance(test.ops[0], ast.Eq)
+                and isinstance(test.left, ast.Name)
+                and test.left.id == "__name__"
+            ):
+                main_line = node.lineno
+                break
+        self.assertIsNotNone(main_line, "if __name__ == '__main__' block not found")
+        first_def_line: dict[str, int] = {}
+        for node in ast.walk(tree):
+            if isinstance(
+                node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
+            ) and node.name in names:
+                first_def_line.setdefault(node.name, node.lineno)
+        for name in sorted(names):
+            self.assertIn(name, first_def_line, f"{name} not defined in module")
+            self.assertLess(
+                first_def_line[name],
+                main_line,
+                f"{name} defined at line {first_def_line[name]} "
+                f"is after main guard at line {main_line}",
+            )
+
+
+class ReslugNoopRewriteTests(unittest.TestCase):
+    def test_nonfrozen_noop_is_not_frozen_and_links_still_rewrite(self) -> None:
+        """A canonical nonfrozen noop must NOT enter frozen_dirs and its links must still rewrite."""
+        with TemporaryDirectory() as raw:
+            root = Path(raw)
+            vault = root / "vault"
+            # Thread A: nonfrozen mismatched -> rename
+            archive_a, _ = _write_pair(
+                vault,
+                handle="author_a",
+                directory="2024-01-01-old",
+                title="A title.",
+                date="2024-01-01",
+                post_id="111",
+            )
+            # Thread B: canonical nonfrozen noop with a cross-thread link to A's old prefix
+            archive_b, _ = _write_pair(
+                vault,
+                handle="author_b",
+                directory="2024-02-02-b-title",
+                title="B title.",
+                date="2024-02-02",
+                post_id="222",
+            )
+            (archive_b / "index.md").write_text(
+                "---\n"
+                'title: "B title."\n'
+                "handle: author_b\n"
+                'post_id: "222"\n'
+                "date: 2024-02-02\n"
+                "draft: true\n"
+                "tags:\n"
+                "  - archive\n"
+                "  - twitter\n"
+                "  - author_b\n"
+                "---\n\n"
+                "See [[archive/threads/author_a/2024-01-01-old/index]] for context.\n",
+                encoding="utf-8",
+                newline="\n",
+            )
+            plan = plan_reslug(vault, set())
+            self.assertIn(archive_b, plan.noops)
+            self.assertNotIn(archive_b, plan.frozen_dirs)
+            self.assertNotIn(archive_b, plan.frozen)
+            rewritten = apply_reslug_plan(plan, root / "scratch")
+            new_text = (archive_b / "index.md").read_text(encoding="utf-8")
+            self.assertIn(
+                "[[archive/threads/author_a/2024-01-01-a-title/index]]",
+                new_text,
+            )
+            self.assertNotIn("2024-01-01-old", new_text)
+            self.assertIn(archive_b / "index.md", rewritten)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/twitter/tests/test_tw_media_commands.py

@@ -11,6 +11,7 @@ from tempfile import TemporaryDirectory
 from unittest.mock import patch
 
 from twitter import tw as tw_module
+from twitter.emit_archive import ReslugConflict, ReslugPlan
 from twitter.models import PostData, PostMetrics, ThreadData, load_thread
 from twitter.tw import (
     _capture_ids,
@@ -24,6 +25,7 @@ from twitter.tw import (
     build_parser,
     cmd_add_branch,
     cmd_emit,
+    cmd_reslug,
 )
 
 
@@ -458,5 +460,91 @@ class TwMediaCommandTests(unittest.TestCase):
         self.assertTrue(args.confirm_origin_unavailable)
 
 
+class ReslugCommandParserTests(unittest.TestCase):
+    def test_reslug_parser_dry_run_defaults(self) -> None:
+        args = build_parser().parse_args(["reslug", "--all"])
+        self.assertEqual(args.cmd, "reslug")
+        self.assertTrue(args.all_threads)
+        self.assertFalse(args.apply)
+
+    def test_reslug_parser_apply_flag(self) -> None:
+        args = build_parser().parse_args(["reslug", "--all", "--apply"])
+        self.assertEqual(args.cmd, "reslug")
+        self.assertTrue(args.all_threads)
+        self.assertTrue(args.apply)
+
+
+class ReslugCommandDispatchTests(unittest.TestCase):
+    def test_reslug_dry_run_does_not_apply(self) -> None:
+        plan = ReslugPlan(
+            vault=Path("/tmp"),
+            moves=(),
+            noops=(),
+            frozen=(),
+            frozen_dirs=(),
+            conflicts=(),
+        )
+        with (
+            patch.object(tw_module, "plan_reslug", return_value=plan) as plan_fn,
+            patch.object(tw_module, "format_reslug_plan", return_value="") as fmt,
+            patch.object(tw_module, "apply_reslug_plan") as apply,
+            redirect_stdout(StringIO()),
+        ):
+            result = cmd_reslug(Namespace(all_threads=True, apply=False))
+        self.assertEqual(result, 0)
+        plan_fn.assert_called_once()
+        fmt.assert_called_once()
+        apply.assert_not_called()
+
+    def test_reslug_conflicts_return_two_without_apply(self) -> None:
+        plan = ReslugPlan(
+            vault=Path("/tmp"),
+            moves=(),
+            noops=(),
+            frozen=(),
+            frozen_dirs=(),
+            conflicts=(ReslugConflict(Path("/tmp/x"), "bad"),),
+        )
+        with (
+            patch.object(tw_module, "plan_reslug", return_value=plan),
+            patch.object(tw_module, "format_reslug_plan", return_value=""),
+            patch.object(tw_module, "apply_reslug_plan") as apply,
+            redirect_stdout(StringIO()),
+        ):
+            result = cmd_reslug(Namespace(all_threads=True, apply=False))
+        self.assertEqual(result, 2)
+        apply.assert_not_called()
+
+    def test_reslug_clean_apply_calls_apply_once_with_plan_and_scratch(self) -> None:
+        plan = ReslugPlan(
+            vault=Path("/tmp"),
+            moves=(),
+            noops=(),
+            frozen=(),
+            frozen_dirs=(),
+            conflicts=(),
+        )
+        with (
+            patch.object(tw_module, "plan_reslug", return_value=plan),
+            patch.object(tw_module, "format_reslug_plan", return_value=""),
+            patch.object(tw_module, "apply_reslug_plan", return_value=()) as apply,
+            redirect_stdout(StringIO()),
+        ):
+            result = cmd_reslug(Namespace(all_threads=True, apply=True))
+        self.assertEqual(result, 0)
+        apply.assert_called_once()
+        positional = apply.call_args.args
+        self.assertEqual(positional[0], plan)
+        self.assertEqual(positional[1], tw_module.SCRATCH)
+
+
+class ReslugParserNoCrossContaminationTests(unittest.TestCase):
+    def test_emit_and_refresh_parser_have_no_reslug_attr(self) -> None:
+        emit_args = build_parser().parse_args(["emit", "--id", "100"])
+        refresh_args = build_parser().parse_args(["refresh", "--id", "100"])
+        self.assertFalse(hasattr(emit_args, "reslug"))
+        self.assertFalse(hasattr(refresh_args, "reslug"))
+
+
 if __name__ == "__main__":
     unittest.main()

scripts/twitter/tw.py

@@ -36,6 +36,7 @@ try:
     from frozen import frozen_match, load_frozen_ids, require_writable
     from backup_assets import backup_thread, load_destination_root
     from catbox_client import load_userhash, upload_file
+    from emit_archive import apply_reslug_plan, format_reslug_plan, plan_reslug
     from fallback_media import activate_fallback, find_existing_fallback, restore_origin
     from media_audit import audit_thread
     from media_migrate import migrate_legacy_thread
@@ -43,6 +44,7 @@ except ImportError:  # pragma: no cover - script-mode import
     from frozen import frozen_match, load_frozen_ids, require_writable
     from backup_assets import backup_thread, load_destination_root
     from catbox_client import load_userhash, upload_file
+    from emit_archive import apply_reslug_plan, format_reslug_plan, plan_reslug
     from fallback_media import activate_fallback, find_existing_fallback, restore_origin
     from media_audit import audit_thread
     from media_migrate import migrate_legacy_thread
@@ -809,6 +811,19 @@ def cmd_restore_origin(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_reslug(args: argparse.Namespace) -> int:
+    """Audit or apply the one-time corpus-wide thread-directory rename."""
+    plan = plan_reslug(VAULT, load_frozen_ids(FROZEN))
+    print(format_reslug_plan(plan), end="")
+    if plan.conflicts:
+        return 2
+    if not args.apply:
+        print("dry-run: explicit --apply required to write")
+        return 0
+    apply_reslug_plan(plan, SCRATCH)
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     sub = parser.add_subparsers(dest="cmd", required=True)
@@ -907,6 +922,13 @@ def build_parser() -> argparse.ArgumentParser:
         help="capture this reply node's visible subtree; repeatable",
     )
 
+    reslug = sub.add_parser(
+        "reslug",
+        help="audit or apply canonical thread directory names",
+    )
+    reslug.add_argument("--all", dest="all_threads", action="store_true", required=True)
+    reslug.add_argument("--apply", action="store_true")
+
     return parser
 
 
@@ -927,6 +949,7 @@ _COMMANDS: dict[str, Callable[[argparse.Namespace], int]] = {
     "backup": cmd_backup,
     "fallback": cmd_fallback,
     "restore-origin": cmd_restore_origin,
+    "reslug": cmd_reslug,
 }