Update convention for op handle.

scripts/twitter/README.md

@@ -19,7 +19,7 @@ python C:\projects\Threadwell\scripts\twitter\tw.py migrate-media --all --apply
 
 `paths.py` sets vault = two parents above this file, dumps = `../manual_slop/docs/twitter`, scratch = `../Threadwell-ai/scratch`.
 
-`refresh` = refetch (cookies) + ingest + emit `--force` + media_merge. `--tip` treats `--id` as the tip (climb same-handle `reply_to`). Does not flip `draft`. Does not commit. Frozen ids abort.
+`refresh` = refetch (cookies) + ingest + emit `--force` + media_merge. `--tip` treats `--id` as the tip and walks back to the OP regardless of handle. Each thread emits exactly one archive directory owned by the OP (the post with `reply_to_id == None`); cross-author responders do not get a directory. Does not flip `draft`. Does not commit. Frozen ids abort.
 
 Usual publish path after you like the note:
 

scripts/twitter/audit_handle_index.py

@@ -0,0 +1,141 @@
+"""Audit every handle index in the vault against real subdirectories.
+
+Reports per handle:
+  real       — set of subdirectory basenames on disk
+  links      — set of basenames extracted from index.md wikilinks
+  missing    — subdirectories with no wikilink in index.md
+  stale      — wikilinks in index.md whose subdirectory does not exist
+
+Usage:
+  python scripts/twitter/audit_handle_index.py            # text report
+  python scripts/twitter/audit_handle_index.py --json     # JSON report
+  python scripts/twitter/audit_handle_index.py --vault <root>
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+DEFAULT_VAULT = HERE.parent.parent
+
+_WIKILINK_ITEM = re.compile(r"^- \[\[([^\]|]+)\]\]\s*$")
+
+
+def _split_frontmatter(text: str) -> tuple[str, str]:
+    if not text.startswith("---"):
+        return "", text
+    end = text.find("\n---", 3)
+    if end < 0:
+        return "", text
+    return text[: end + 4], text[end + 4 :]
+
+
+def _index_links(idx_path: Path) -> set[str]:
+    if not idx_path.is_file():
+        return set()
+    text = idx_path.read_text(encoding="utf-8")
+    _, body = _split_frontmatter(text)
+    out: set[str] = set()
+    for line in body.splitlines():
+        match = _WIKILINK_ITEM.match(line)
+        if not match:
+            continue
+        target = match.group(1)
+        # vault-root path: archive/threads/<handle>/<dir-name>
+        # compare basename to the set of real_dirs
+        name = target.rsplit("/", 1)[-1]
+        if name:
+            out.add(name)
+    return out
+
+
+def audit_one(handle_dir: Path) -> dict[str, object]:
+    idx_path = handle_dir / "index.md"
+    real = {d.name for d in handle_dir.iterdir() if d.is_dir()}
+    links = _index_links(idx_path)
+    missing = sorted(real - links)
+    stale = sorted(links - real)
+    return {
+        "handle": handle_dir.name,
+        "has_index": idx_path.is_file(),
+        "real": sorted(real),
+        "links": sorted(links),
+        "missing": missing,
+        "stale": stale,
+    }
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--json", action="store_true", help="emit JSON instead of text"
+    )
+    parser.add_argument(
+        "--vault",
+        default=str(DEFAULT_VAULT),
+        help=f"vault root (default: {DEFAULT_VAULT})",
+    )
+    args = parser.parse_args(argv)
+
+    vault = Path(args.vault).resolve()
+    threads_root = vault / "archive" / "threads"
+    if not threads_root.is_dir():
+        raise SystemExit(f"missing {threads_root}")
+
+    reports: list[dict[str, object]] = []
+    for entry in sorted(threads_root.iterdir()):
+        if not entry.is_dir() or entry.name.startswith("."):
+            continue
+        reports.append(audit_one(entry))
+
+    if args.json:
+        print(json.dumps(reports, indent=2, ensure_ascii=False))
+        return 0
+
+    if not reports:
+        print("no handles found")
+        return 0
+
+    max_handle = max(len(str(r["handle"])) for r in reports)
+    flagged = 0
+    for r in reports:
+        handle = str(r["handle"]).ljust(max_handle)
+        real_n = len(r["real"])
+        link_n = len(r["links"])
+        miss_n = len(r["missing"])
+        stale_n = len(r["stale"])
+        marker = ""
+        if miss_n or stale_n:
+            marker = "  ***"
+            flagged += 1
+        print(
+            f"{handle}  real={real_n:<3}  links={link_n:<3}  "
+            f"missing={miss_n:<3}  stale={stale_n}{marker}"
+        )
+
+    print()
+    if flagged:
+        print(f"flagged {flagged} handle(s):")
+        for r in reports:
+            if not (r["missing"] or r["stale"]):
+                continue
+            if r["missing"]:
+                missing_sample = ", ".join(r["missing"][:5])
+                if len(r["missing"]) > 5:
+                    missing_sample += f", ... (+{len(r['missing']) - 5} more)"
+                print(f"  {r['handle']}: missing_in_index = {missing_sample}")
+            if r["stale"]:
+                stale_sample = ", ".join(r["stale"])
+                print(f"  {r['handle']}: stale_in_index = {stale_sample}")
+    else:
+        print("clean: no handle has missing or stale wikilinks")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/twitter/emit_archive.py

@@ -401,7 +401,8 @@ def _upsert_wikilink(path: Path, target: str) -> None:
         match = _WIKILINK_ITEM.match(line)
         if match:
             name = match.group(1)
-            if name in real_dirs or not real_dirs:
+            # wikilinks are vault-root paths; compare the basename to real_dirs
+            if name.rsplit("/", 1)[-1] in real_dirs or not real_dirs:
                 if name not in names:
                     names.append(name)
         else:

scripts/twitter/render.py

@@ -178,6 +178,68 @@ def render_spine(
     return "\n".join(chunks).rstrip() + "\n"
 
 
+def render_per_author_spine(
+    handle: str,
+    author_posts: list[PostData],
+    ids: dict[str, PostData],
+    archived: str,
+    conversation_links: list[str],
+    media_by_post: dict[str, tuple[str, ...]] | None = None,
+) -> str:
+    """Render a per-author archive note.
+
+    `author_posts` is the author's posts in chronological order.
+    Cross-author parents appear as blockquotes immediately before the
+    post that replies to them. `conversation_links` is a list of
+    vault-root wikilink targets for the other authors' directories
+    in the same conversation; if non-empty, a `## Conversation`
+    section is appended.
+
+    The first post in `author_posts` is treated as the spine root for
+    the frontmatter. Its title is derived from its first line.
+    """
+    if not author_posts:
+        raise ValueError(
+            "render_per_author_spine requires a non-empty author_posts list"
+        )
+    first = author_posts[0]
+    title = _title_text(first.text)
+    source_url = _status_url(first.handle, first.post_id)
+    chunks: list[str] = [
+        _frontmatter(
+            title=title,
+            source_url=source_url,
+            author=first.author,
+            handle=first.handle,
+            post_id=first.post_id,
+            date=date_prefix(first.timestamp),
+            archived=archived,
+            spine_handle=first.handle,
+            in_reply_to="",
+        ),
+        "",
+        _source_block(first),
+        "",
+        "## Thread",
+        "",
+    ]
+    for n, post in enumerate(author_posts, 1):
+        parent_id = post.reply_to_id
+        if parent_id:
+            parent = ids.get(parent_id)
+            if parent is not None and parent.handle != handle:
+                chunks.append(_blockquote(parent.text))
+                chunks.append("")
+        chunks.append(_post_block(n, post, media_by_post))
+    if conversation_links:
+        chunks.append("## Conversation")
+        chunks.append("")
+        for link in conversation_links:
+            chunks.append(f"- [[{link}]]")
+        chunks.append("")
+    return "\n".join(chunks).rstrip() + "\n"
+
+
 def render_branch(
     thread: ThreadData,
     branch_root_id: str,

scripts/twitter/tests/test_emit_op_owned.py

@@ -0,0 +1,343 @@
+"""End-to-end tests for OP-owned emit on cross-author fixtures.
+
+The OP-owned model emits one archive dir per thread, owned by the OP
+(the post with `reply_to_id == None`). All posts in the conversation
+are rendered in chronological order. Cross-handle chains are preserved.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.emit_archive import emit
+from twitter.models import PostData, PostMetrics
+
+
+def make_post(
+    post_id: str,
+    handle: str,
+    text: str,
+    timestamp: str,
+    reply_to: str | None,
+) -> PostData:
+    return PostData(
+        post_id=post_id,
+        author=handle,
+        handle=handle,
+        text=text,
+        timestamp=timestamp,
+        media_urls=(),
+        reply_to_id=reply_to,
+        quote_of_id=None,
+        metrics=PostMetrics(0, 0, 0, None),
+    )
+
+
+SEVEN_POST: tuple[PostData, ...] = (
+    make_post(
+        "1597661267845865474",
+        "rianflo",
+        '"The road to 16-bit floats GPU is paved with our blood" :-/ https://example.com/road',
+        "2022-11-29 18:38:24",
+        None,
+    ),
+    make_post(
+        "1597717009369735169",
+        "NOTimothyLottes",
+        "@rianflo Explicit packed 16-bit works on AMD Vulkan Vega and up.",
+        "2022-11-29 22:19:53",
+        "1597661267845865474",
+    ),
+    make_post(
+        "1597718146663690240",
+        "rianflo",
+        "@NOTimothyLottes Oh I know the benefits. Just no simple clear way to write it in GLSL for vulkan.",
+        "2022-11-29 22:24:24",
+        "1597717009369735169",
+    ),
+    make_post(
+        "1597718560511385600",
+        "NOTimothyLottes",
+        "@rianflo Sure there is. CAS/FSR1/etc all shipped with fantastic GLSL versions using 16-bit packed math.",
+        "2022-11-29 22:26:03",
+        "1597718146663690240",
+    ),
+    make_post(
+        "1597720097753542656",
+        "rianflo",
+        "@NOTimothyLottes What GLSL extension did you use?",
+        "2022-11-29 22:32:10",
+        "1597718560511385600",
+    ),
+    make_post(
+        "1597720454000541696",
+        "rianflo",
+        "@NOTimothyLottes Oh wait, you're saying you wrote the fp16 math manually?",
+        "2022-11-29 22:33:35",
+        "1597720097753542656",
+    ),
+    make_post(
+        "1597798161665253376",
+        "NOTimothyLottes",
+        "@rianflo Explicit packed 16-bit code. FSR1 example: https://example.com/fsr1",
+        "2022-11-30 03:42:22",
+        "1597720454000541696",
+    ),
+)
+
+
+def _wire_dump(posts: tuple[PostData, ...]) -> dict:
+    return {
+        "root_post_id": "1597798161665253376",
+        "source_url": "https://x.com/i/status/1597798161665253376",
+        "posts": [
+            {
+                "post_id": p.post_id,
+                "author": p.author,
+                "handle": p.handle,
+                "text": p.text,
+                "timestamp": p.timestamp,
+                "media_urls": list(p.media_urls),
+                "reply_to_id": p.reply_to_id,
+                "quote_of_id": p.quote_of_id,
+                "metrics": {
+                    "reply_count": p.metrics.reply_count,
+                    "repost_count": p.metrics.repost_count,
+                    "like_count": p.metrics.like_count,
+                    "view_count": p.metrics.view_count,
+                },
+            }
+            for p in posts
+        ],
+    }
+
+
+def _wikilink_names(text: str) -> list[str]:
+    names: list[str] = []
+    for line in text.splitlines():
+        m = re.match(r"^\s*-\s+\[\[([^\]|]+)\]\]\s*$", line)
+        if m:
+            names.append(m.group(1).rsplit("/", 1)[-1])
+    return sorted(names)
+
+
+class OpOwnedEmitTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp(prefix="op_owned_emit_"))
+        self.input_dir = self.tmp / "input"
+        self.input_dir.mkdir()
+        (self.input_dir / "thread_data.json").write_text(
+            json.dumps(_wire_dump(SEVEN_POST), indent=2),
+            encoding="utf-8",
+            newline="\n",
+        )
+        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 test_one_archive_dir_owned_by_op(self) -> None:
+        """The thread produces exactly one archive dir, owned by the
+        OP (rianflo). lottes does not get a separate dir."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        archive_root = self.vault / "archive" / "threads"
+        handles = sorted(
+            d.name for d in archive_root.iterdir() if d.is_dir()
+        )
+        self.assertEqual(handles, ["rianflo"])
+
+    def test_one_asset_dir_with_all_seven_posts(self) -> None:
+        """The asset dir holds the full thread_data.json (all 7
+        posts), not a per-author subset."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        asset_root = self.vault / "assets" / "threads"
+        handles = sorted(
+            d.name for d in asset_root.iterdir() if d.is_dir()
+        )
+        self.assertEqual(handles, ["rianflo"])
+
+        thread_dirs = sorted(
+            d for d in (asset_root / "rianflo").iterdir() if d.is_dir()
+        )
+        self.assertEqual(len(thread_dirs), 1)
+        thread_data = json.loads(
+            (thread_dirs[0] / "thread_data.json").read_text(encoding="utf-8")
+        )
+        self.assertEqual(len(thread_data["posts"]), 7)
+
+    def test_index_md_renders_all_seven_posts_in_order(self) -> None:
+        """The index.md shows all 7 posts in chain order with N/
+        numbering. The OP (rianflo) is the frontmatter handle and
+        source_url."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        thread_dir = next(
+            d for d in (self.vault / "archive" / "threads" / "rianflo").iterdir()
+            if d.is_dir()
+        )
+        text = (thread_dir / "index.md").read_text(encoding="utf-8")
+        self.assertIn("handle: rianflo", text)
+        self.assertIn(
+            'source_url: "https://x.com/rianflo/status/1597661267845865474"',
+            text,
+        )
+        # All 7 N/ markers in chronological chain order
+        for n in range(1, 8):
+            self.assertIn(f"**{n}/**", text)
+        # Posts appear in chain order. Use a short, unique fragment
+        # of each post's text to locate its position in the file.
+        fragments = [
+            "The road to 16-bit floats GPU",        # OP post 1
+            "Explicit packed 16-bit works on AMD",   # lottes post 1 (2/)
+            "Oh I know the benefits",                # rianflo (3/)
+            "CAS/FSR1/etc all shipped",              # lottes (4/)
+            "What GLSL extension",                   # rianflo (5/)
+            "Oh wait, you're saying",                # rianflo (6/)
+            "FSR1 example",                          # lottes tip (7/)
+        ]
+        positions = [text.find(f) for f in fragments]
+        self.assertTrue(
+            all(p >= 0 for p in positions),
+            f"missing fragments in index.md: "
+            f"{[fragments[i] for i, p in enumerate(positions) if p < 0]}",
+        )
+        self.assertEqual(
+            positions,
+            sorted(positions),
+            f"posts out of order in index.md: {positions}",
+        )
+
+    def test_lottes_handle_index_does_not_link_to_thread(self) -> None:
+        """lottes is a responder, not the OP. Her handle index must
+        not wikilink this thread."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        archive_root = self.vault / "archive" / "threads"
+        # lottes's handle dir should not exist (she's not the OP)
+        lottes_handle = archive_root / "NOTimothyLottes"
+        self.assertFalse(
+            lottes_handle.exists(),
+            "lottes is a responder, not the OP; no handle dir expected",
+        )
+
+    def test_rianflo_handle_index_links_to_thread(self) -> None:
+        """rianflo is the OP. His handle index lists the new thread."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        archive_root = self.vault / "archive" / "threads"
+        rianflo_index = (
+            archive_root / "rianflo" / "index.md"
+        ).read_text(encoding="utf-8")
+        rianflo_thread_dirs = sorted(
+            d.name
+            for d in (archive_root / "rianflo").iterdir()
+            if d.is_dir()
+        )
+        self.assertEqual(_wikilink_names(rianflo_index), rianflo_thread_dirs)
+
+    def test_no_conversation_section(self) -> None:
+        """The OP-owned model does not emit a ## Conversation
+        section (the thread is the conversation)."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+        )
+        thread_dir = next(
+            d for d in (self.vault / "archive" / "threads" / "rianflo").iterdir()
+            if d.is_dir()
+        )
+        text = (thread_dir / "index.md").read_text(encoding="utf-8")
+        self.assertNotIn("## Conversation", text)
+
+
+class OpOwnedEmitTipModeTests(unittest.TestCase):
+    """--tip mode climbs back to the OP and renders the full chain."""
+
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp(prefix="op_owned_tip_"))
+        self.input_dir = self.tmp / "input"
+        self.input_dir.mkdir()
+        (self.input_dir / "thread_data.json").write_text(
+            json.dumps(_wire_dump(SEVEN_POST), indent=2),
+            encoding="utf-8",
+            newline="\n",
+        )
+        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 test_tip_mode_emits_full_chain(self) -> None:
+        """Passing --tip with the tip's post_id still produces one
+        OP-owned dir with all 7 posts."""
+        emit(
+            input_dir=self.input_dir,
+            vault=self.vault,
+            slug=None,
+            archived="2026-08-24",
+            force=True,
+            tip="1597798161665253376",
+        )
+        archive_root = self.vault / "archive" / "threads"
+        handles = sorted(
+            d.name for d in archive_root.iterdir() if d.is_dir()
+        )
+        self.assertEqual(handles, ["rianflo"])
+
+        thread_dirs = [
+            d for d in (archive_root / "rianflo").iterdir() if d.is_dir()
+        ]
+        self.assertEqual(len(thread_dirs), 1)
+
+        asset_root = self.vault / "assets" / "threads"
+        thread_dirs = [
+            d for d in (asset_root / "rianflo").iterdir() if d.is_dir()
+        ]
+        self.assertEqual(len(thread_dirs), 1)
+        thread_data = json.loads(
+            (thread_dirs[0] / "thread_data.json").read_text(encoding="utf-8")
+        )
+        self.assertEqual(len(thread_data["posts"]), 7)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/twitter/tests/test_upsert_wikilink.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import re
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from twitter.emit_archive import _upsert_wikilink
+
+
+_WIKILINK_LINE = re.compile(r"^\s*-\s+\[\[([^\]|]+)\]\]\s*$")
+
+
+def _read_wikilink_lines(idx: Path) -> list[str]:
+    text = idx.read_text(encoding="utf-8")
+    out: list[str] = []
+    for line in text.splitlines():
+        m = _WIKILINK_LINE.match(line)
+        if m:
+            out.append("- [[" + m.group(1) + "]]")
+    return out
+
+
+class UpsertWikilinkTests(unittest.TestCase):
+    def _build(self, subdirs: list[str], wikilinks: list[str]) -> Path:
+        """Build a temp vault with one handle containing `subdirs` and an
+        index.md whose body is the given wikilinks (one per line)."""
+        raw = tempfile.mkdtemp(prefix="upsert_wikilink_")
+        self.addCleanup(shutil.rmtree, raw, ignore_errors=True)
+        root = Path(raw)
+        handle_dir = root / "archive" / "threads" / "handle"
+        handle_dir.mkdir(parents=True)
+        for name in subdirs:
+            (handle_dir / name).mkdir()
+        idx = handle_dir / "index.md"
+        body = "\n".join(wikilinks)
+        idx.write_text(
+            "---\ntitle: handle\ndraft: false\n---\n\n" + body + "\n",
+            encoding="utf-8",
+            newline="\n",
+        )
+        return idx
+
+    def test_preserves_existing_wikilink_whose_dir_still_exists(self) -> None:
+        """Upserting a target whose dir already exists must keep every
+        other real-directory wikilink in the index."""
+        idx = self._build(
+            ["2024-01-01-foo", "2024-02-02-bar"],
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+                "- [[archive/threads/handle/2024-02-02-bar]]",
+            ],
+        )
+        _upsert_wikilink(idx, "archive/threads/handle/2024-01-01-foo")
+        self.assertEqual(
+            _read_wikilink_lines(idx),
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+                "- [[archive/threads/handle/2024-02-02-bar]]",
+            ],
+        )
+
+    def test_drops_wikilink_whose_dir_no_longer_exists(self) -> None:
+        """Upserting must drop a wikilink whose subdirectory is gone and
+        leave the new target in its place."""
+        idx = self._build(
+            ["2024-01-01-foo"],
+            [
+                "- [[archive/threads/handle/9999-99-99-missing]]",
+            ],
+        )
+        _upsert_wikilink(idx, "archive/threads/handle/2024-01-01-foo")
+        self.assertEqual(
+            _read_wikilink_lines(idx),
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+            ],
+        )
+
+    def test_adds_new_target_when_missing_from_index(self) -> None:
+        """Upserting a target that is not in the index must add it while
+        preserving the existing real-directory wikilinks."""
+        idx = self._build(
+            ["2024-01-01-foo", "2024-02-02-bar"],
+            [
+                "- [[archive/threads/handle/2024-02-02-bar]]",
+            ],
+        )
+        _upsert_wikilink(idx, "archive/threads/handle/2024-01-01-foo")
+        self.assertEqual(
+            _read_wikilink_lines(idx),
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+                "- [[archive/threads/handle/2024-02-02-bar]]",
+            ],
+        )
+
+    def test_does_not_duplicate_target_already_in_index(self) -> None:
+        """Upserting a target that is already in the index must not
+        produce a duplicate wikilink line."""
+        idx = self._build(
+            ["2024-01-01-foo"],
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+            ],
+        )
+        _upsert_wikilink(idx, "archive/threads/handle/2024-01-01-foo")
+        self.assertEqual(
+            _read_wikilink_lines(idx),
+            [
+                "- [[archive/threads/handle/2024-01-01-foo]]",
+            ],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/twitter/tree.py

@@ -19,18 +19,24 @@ def children_map(thread: ThreadData) -> dict[str, list[str]]:
 
 
 def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
+    """Walk back from `tip_id` through `reply_to_id` to the OP (the
+    post with `reply_to_id == None` or a missing parent). Returns
+    all post ids on the linear chain in chronological order: OP at
+    index 0, tip at last.
+
+    The walk crosses handle boundaries. For a chain that alternates
+    authors, every post is included. The archive dir for the
+    thread is owned by the OP at index 0.
+    """
     ids = by_id(thread)
     if tip_id not in ids:
         raise ValueError(f"tip {tip_id} not in dump")
-    handle = ids[tip_id].handle
     chain = [tip_id]
     cur = tip_id
     while True:
         parent = ids[cur].reply_to_id
         if not parent or parent not in ids:
             break
-        if ids[parent].handle != handle:
-            break
         chain.append(parent)
         cur = parent
     chain.reverse()
@@ -38,17 +44,33 @@ def spine_from_tip(thread: ThreadData, tip_id: str) -> list[str]:
 
 
 def spine_ids(thread: ThreadData) -> list[str]:
+    """Walk the linear reply chain from the OP (the post with
+    `reply_to_id == None`) down through `children_map`, picking the
+    first child at each level (already sorted by `(timestamp,
+    post_id)`).
+
+    The walk crosses handle boundaries. For a chain that alternates
+    authors (e.g., rianflo ↔ NOTimothyLottes), every post on the
+    chain is included. The archive dir for the thread is owned by
+    the OP — the first post in the returned list.
+
+    If no post has `reply_to_id == None`, the first post in
+    `thread.posts` is used as a fallback start.
+    """
     ids = by_id(thread)
     kids = children_map(thread)
-    start = thread.root_post_id if thread.root_post_id in ids else thread.posts[0].post_id
-    handle = ids[start].handle
+    ops = [p.post_id for p in thread.posts if p.reply_to_id is None]
+    if ops:
+        start = ops[0]
+    else:
+        start = thread.posts[0].post_id
     out = [start]
     cur = start
     while True:
-        same = [c for c in kids[cur] if ids[c].handle == handle]
-        if not same:
+        children = kids.get(cur, [])
+        if not children:
             break
-        cur = same[0]
+        cur = children[0]
         out.append(cur)
     return out
 

scripts/twitter/tw.py

@@ -105,6 +105,46 @@ def locate(post_id: str) -> tuple[Path | None, Path | None]:
     return None, None
 
 
+def locate_all(post_id: str) -> list[tuple[Path, Path | None]]:
+    """Return every (asset_dir, note_dir) pair whose media.json,
+    thread_data.json, or captured posts contain post_id. Used by emit
+    to find per-author asset dirs after a cross-author refresh."""
+    assets_root = VAULT / "assets" / "threads"
+    notes_root = VAULT / "archive" / "threads"
+    if not assets_root.is_dir():
+        return []
+    out: list[tuple[Path, Path | None]] = []
+    for media in assets_root.rglob("media.json"):
+        try:
+            data = json.loads(media.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            continue
+        hit = data.get("root_post_id") == post_id
+        if not hit:
+            for item in data.get("items") or []:
+                if str(item.get("post_id") or "") == post_id:
+                    hit = True
+                    break
+        if not hit:
+            td = media.parent / "thread_data.json"
+            if td.is_file():
+                try:
+                    t = json.loads(td.read_text(encoding="utf-8"))
+                except json.JSONDecodeError:
+                    t = {}
+                if t.get("root_post_id") == post_id:
+                    hit = True
+                else:
+                    hit = any(
+                        p.get("post_id") == post_id for p in (t.get("posts") or [])
+                    )
+        if hit:
+            rel = media.parent.relative_to(assets_root)
+            notes = notes_root / rel
+            out.append((media.parent, notes if notes.is_dir() else None))
+    return out
+
+
 def run(cmd: list[str]) -> None:
     print(">", " ".join(cmd))
     subprocess.check_call(cmd)
@@ -208,9 +248,34 @@ def cmd_emit(post_id: str, tip: bool, slug: str | None) -> int:
     if slug:
         cmd.extend(["--slug", slug])
     run(cmd)
-    assets, _notes = locate(post_id)
-    if assets is not None:
-        run([sys.executable, str(HERE / "media_merge.py"), "--thread", str(assets)])
+    # Cross-author threads emit one asset dir per author. The tip's
+    # post_id is only in the tip's author's asset dir; the other
+    # authors' per-author dirs are missed by a tip-only lookup.
+    # Iterate every post in the source dump and union the matches.
+    seen: set[Path] = set()
+    try:
+        thread_data = json.loads(
+            (src / "thread_data.json").read_text(encoding="utf-8")
+        )
+        post_ids = {
+            str(p.get("post_id"))
+            for p in thread_data.get("posts") or []
+        }
+    except (json.JSONDecodeError, OSError):
+        post_ids = {post_id}
+    for pid in post_ids:
+        for asset_dir, _ in locate_all(pid):
+            if asset_dir in seen:
+                continue
+            seen.add(asset_dir)
+            run(
+                [
+                    sys.executable,
+                    str(HERE / "media_merge.py"),
+                    "--thread",
+                    str(asset_dir),
+                ]
+            )
     return 0