attempt better archiving of quote tweets.

scripts/twitter/README.md

@@ -5,6 +5,8 @@ Front door: `tw.py`. Run from the vault root.
 ```
 python scripts/twitter/tw.py graph --id <snowflake>
 python scripts/twitter/tw.py refresh --id <spine-tip> --tip
+python scripts/twitter/tw.py refresh --id <spine-tip> --tip --preserve-existing
+python scripts/twitter/tw.py refresh --id <spine-tip> --tip --no-quotes
 python scripts/twitter/tw.py refresh --id <spine-tip> --tip --branch <branch-tip>
 python scripts/twitter/tw.py add-branch --id <original-spine-tip> --from <reply-node>
 python scripts/twitter/tw.py emit --id <spine-tip> --tip --preserve-existing
@@ -27,6 +29,11 @@ python scripts/twitter/tw.py relabel --all --apply
 
 `refresh` is refetch + ingest + emit `--force` + media merge. `--tip` treats `--id` as the tip and walks back to the opening post. If that tip is the OP, same-handle self-replies stay on the spine and foreign replies are branches. Repeat `--branch <tip-id>` to capture extra tips into the same `thread_data.json`. `--attach <child>:<parent>` sets one hop before the walk. A walk that stops on a missing parent aborts unless `--allow-broken-walk`. A leftover dir under another handle aborts unless `--retire-old-dir`. It does not invent replies. Captures run one at a time. gallery-dl retries are off. Extractor and request sleeps are five seconds. Re-emit deletes branch notes that are no longer roots.
 
+`refresh` also captures each quote tweet on the quoter spine as
+its own root archive (`refresh --id <quoted>` without `--tip`).
+One hop. `--no-quotes` skips that pass. `emit` does not fetch.
+Established threads: `refresh --id <quoter-tip> --tip --preserve-existing`.
+
 Each thread writes one archive folder. The owner is the opening post (`reply_to_id` is empty). Other authors stay in that folder. New notes set `draft: false`. The command does not commit. Frozen ids abort.
 
 For a thread already on disk, stage with `refetch`, compare ids, then:

scripts/twitter/emit_archive.py

@@ -48,6 +48,7 @@ try:
         descendants,
         spine_from_tip,
         spine_ids,
+        spine_quote_ids,
     )
 except ImportError:  # pragma: no cover - script-mode import
     from frozen import frozen_match, load_frozen_ids
@@ -79,6 +80,7 @@ except ImportError:  # pragma: no cover - script-mode import
         descendants,
         spine_from_tip,
         spine_ids,
+        spine_quote_ids,
     )
 
 
@@ -309,18 +311,23 @@ def empty_text_ids(thread: ThreadData) -> list[str]:
     return [p.post_id for p in thread.posts if not (p.text or "").strip()]
 
 
-def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root") -> str:
-    """Render a human-readable gaps report (quote_of, missing_reply_to, empty_text) for the thread."""
+def render_gaps(
+    thread: ThreadData,
+    spine: list[str],
+    *,
+    input_kind: str = "root",
+    missing_quote_of: list[str] | None = None,
+) -> str:
+    """Render a gaps report. ``missing_quote_of`` is quoted ids with no archive."""
     ids = by_id(thread)
     suggested = spine[-1] if spine else ""
-    quote_ids = [p.post_id for p in thread.posts if p.quote_of_id]
+    quote_ids = list(missing_quote_of) if missing_quote_of is not None else []
     missing = [
         p.post_id
         for p in thread.posts
         if p.reply_to_id is not None and p.reply_to_id not in ids
     ]
     empty = empty_text_ids(thread)
-
     lines = [
         "# gaps",
         f"input: {input_kind}",
@@ -338,6 +345,49 @@ def render_gaps(thread: ThreadData, spine: list[str], *, input_kind: str = "root
     return "\n".join(lines) + "\n"
 
 
+def archive_for_post(vault: Path, post_id: str) -> tuple[Path | None, Path | None]:
+    """Return ``(asset_dir, note_dir)`` for the thread that contains ``post_id``."""
+    assets_root = vault / "assets" / "threads"
+    notes_root = vault / "archive" / "threads"
+    if not assets_root.is_dir():
+        return None, None
+    for td in assets_root.rglob("thread_data.json"):
+        try:
+            data = json.loads(td.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            continue
+        posts = data.get("posts") or []
+        hit = str(data.get("root_post_id") or "") == post_id
+        if not hit:
+            hit = any(str(p.get("post_id") or "") == post_id for p in posts)
+        if not hit:
+            continue
+        rel = td.parent.relative_to(assets_root)
+        notes = notes_root / rel
+        return td.parent, notes if notes.is_dir() else None
+    return None, None
+
+
+def quote_ref_for(vault: Path, quote_of_id: str) -> tuple[str, str | None]:
+    """Return ``(status_url, wikilink_target_or_none)`` for a quoted id."""
+    assets, notes = archive_for_post(vault, quote_of_id)
+    if assets is None or not (assets / "thread_data.json").is_file():
+        return f"https://x.com/i/status/{quote_of_id}", None
+    quoted_thread = load_thread(assets / "thread_data.json")
+    quoted_ids = by_id(quoted_thread)
+    quoted_post = quoted_ids.get(quote_of_id)
+    handle = quoted_post.handle if quoted_post is not None else "i"
+    url = f"https://x.com/{handle}/status/{quote_of_id}"
+    if notes is None:
+        return url, None
+    wiki = f"archive/threads/{notes.parent.name}/{notes.name}"
+    ops = [p.post_id for p in quoted_thread.posts if p.reply_to_id is None]
+    op_id = ops[0] if ops else quoted_thread.root_post_id
+    if quote_of_id != op_id:
+        wiki = f"{wiki}#^{quote_of_id}"
+    return url, wiki
+
+
 def _frontmatter_block(text: str) -> str:
     """Return the raw YAML block between the ``---`` fences, or ``""`` if not parseable."""
     if not text.startswith("---"):
@@ -778,6 +828,11 @@ def emit(
         rid: wiki_branch(handle, dir_name, name)
         for rid, name in branch_names.items()
     }
+    quote_refs: dict[str, tuple[str, str | None]] = {}
+    for pid in spine:
+        quote_id = ids[pid].quote_of_id
+        if quote_id:
+            quote_refs[pid] = quote_ref_for(vault, quote_id)
 
     spine_text = render_spine(
         thread,
@@ -786,6 +841,7 @@ def emit(
         branch_links,
         archived,
         media_by_post,
+        quote_refs,
     )
     spine_path = note_dir / "index.md"
     if spine_path.is_file():
@@ -829,7 +885,17 @@ def emit(
         "mirrors": [],
     }
     atomic_write_json(asset_dir / "media.json", media_doc)
-    gaps_text = render_gaps(thread, spine, input_kind=input_kind)
+    missing_quotes = [
+        qid
+        for qid in spine_quote_ids(thread, spine)
+        if archive_for_post(vault, qid)[0] is None
+    ]
+    gaps_text = render_gaps(
+        thread,
+        spine,
+        input_kind=input_kind,
+        missing_quote_of=missing_quotes,
+    )
     if missing_keys:
         gaps_text += "\nmissing_media_local:\n" + "\n".join(missing_keys) + "\n"
     (asset_dir / "gaps.md").write_text(

scripts/twitter/render.py

@@ -148,12 +148,19 @@ def _post_block(
     post: PostData,
     media_by_post: dict[str, tuple[str, ...]] | None,
     branch_links: list[str] | None = None,
+    quote_ref: tuple[str, str | None] | None = None,
 ) -> str:
-    """Render one post: number, bold speaker, block id, mentions, text, media."""
+    """Render one post: number, bold speaker, block id, mentions, text, quote, media."""
     if not post.handle:
         raise ValueError(f"empty handle on post {post.post_id}")
     head = f"**{n}/** **@{post.handle}** ^{post.post_id}"
     parts = [head, "", format_post_text(post.text)]
+    if quote_ref is not None:
+        url, wiki = quote_ref
+        parts.append("")
+        parts.append(url)
+        if wiki:
+            parts.append(f"[[{wiki}]]")
     media = (media_by_post or {}).get(post.post_id) or ()
     if media:
         parts.append("")
@@ -189,6 +196,7 @@ def render_spine(
     branch_names: dict[str, str],
     archived: str,
     media_by_post: dict[str, tuple[str, ...]] | None = None,
+    quote_refs: dict[str, tuple[str, str | None]] | None = None,
 ) -> str:
     """branch_names maps branch-root post_id -> wikilink target (filename without .md)."""
     ids = by_id(thread)
@@ -227,7 +235,15 @@ def render_spine(
 
     for n, pid in enumerate(spine, 1):
         post = ids[pid]
-        chunks.append(_post_block(n, post, media_by_post, branches_under.get(pid)))
+        chunks.append(
+            _post_block(
+                n,
+                post,
+                media_by_post,
+                branches_under.get(pid),
+                (quote_refs or {}).get(pid),
+            )
+        )
 
     return "\n".join(chunks).rstrip() + "\n"
 

scripts/twitter/tests/test_quote_spine.py

@@ -0,0 +1,418 @@
+from __future__ import annotations
+
+import json
+import unittest
+from argparse import Namespace
+from dataclasses import asdict
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from twitter.emit_archive import (
+    archive_for_post,
+    quote_ref_for,
+    render_gaps,
+)
+from twitter.models import PostData, PostMetrics, ThreadData
+from twitter.render import _post_block, render_spine
+from twitter.tree import spine_quote_ids
+from twitter.tw import (
+    _merge_existing_capture,
+    build_parser,
+    chase_spine_quotes,
+    cmd_refresh,
+    overlay_quote_of_ids,
+)
+
+
+def post(
+    post_id: str,
+    handle: str,
+    text: str,
+    reply_to_id: str | None,
+    quote_of_id: str | None = None,
+) -> PostData:
+    return PostData(
+        post_id=post_id,
+        author=handle,
+        handle=handle,
+        text=text,
+        timestamp="2025-02-09 02:35:43",
+        media_urls=(),
+        reply_to_id=reply_to_id,
+        quote_of_id=quote_of_id,
+        metrics=PostMetrics(0, 0, 0, None),
+    )
+
+
+class PostBlockQuoteRefTests(unittest.TestCase):
+    def test_quote_ref_url_and_wikilink_before_media(self) -> None:
+        quoted = post(
+            "1888416459132395792",
+            "AgileJebrim",
+            "Check your compiler.\n\nstep() isn’t necessary however. https://t.co/ianYUdKGcc",
+            None,
+            "1888409333182218691",
+        )
+        block = _post_block(
+            1,
+            quoted,
+            {"1888416459132395792": ("https://pbs.twimg.com/media/AAA.jpg",)},
+            None,
+            (
+                "https://x.com/iquilezles/status/1888409333182218691",
+                "archive/threads/iquilezles/2025-02-09-slug",
+            ),
+        )
+        self.assertIn("https://t.co/ianYUdKGcc", block)
+        url_at = block.index(
+            "https://x.com/iquilezles/status/1888409333182218691"
+        )
+        wiki_at = block.index(
+            "[[archive/threads/iquilezles/2025-02-09-slug]]"
+        )
+        media_at = block.index("https://pbs.twimg.com/media/AAA.jpg")
+        self.assertLess(url_at, wiki_at)
+        self.assertLess(wiki_at, media_at)
+
+    def test_quote_ref_url_only_when_wiki_missing(self) -> None:
+        quoted = post("1", "AgileJebrim", "body https://t.co/ianYUdKGcc", None, "9")
+        block = _post_block(
+            1,
+            quoted,
+            None,
+            None,
+            ("https://x.com/i/status/9", None),
+        )
+        self.assertIn("https://x.com/i/status/9\n", block)
+        self.assertNotIn("[[", block)
+        self.assertIn("https://t.co/ianYUdKGcc", block)
+
+    def test_no_quote_ref_leaves_body_unchanged(self) -> None:
+        quoted = post("1", "AgileJebrim", "body", None, "9")
+        block = _post_block(1, quoted, None)
+        self.assertNotIn("x.com", block)
+
+
+class SpineQuoteIdTests(unittest.TestCase):
+    def test_spine_only_unique_order(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(
+                post("1", "A", "qt", None, "9"),
+                post("2", "B", "reply", "1"),
+                post("3", "A", "qt2", "2", "8"),
+                post("4", "C", "branch qt", "1", "7"),
+            ),
+            source_url="https://x.com/i/status/1",
+        )
+        self.assertEqual(spine_quote_ids(thread, ["1", "2", "3"]), ["9", "8"])
+
+    def test_duplicate_quote_id_once(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(
+                post("1", "A", "a", None, "9"),
+                post("2", "A", "b", "1", "9"),
+            ),
+            source_url="https://x.com/i/status/1",
+        )
+        self.assertEqual(spine_quote_ids(thread, ["1", "2"]), ["9"])
+
+
+class RenderSpineQuoteTests(unittest.TestCase):
+    def test_render_spine_passes_quote_ref(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "AgileJebrim", "Check your compiler.", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        text = render_spine(
+            thread,
+            ["1"],
+            [],
+            {},
+            "2026-08-26",
+            None,
+            {"1": ("https://x.com/iquilezles/status/9", "archive/threads/iquilezles/slug")},
+        )
+        self.assertIn("https://x.com/iquilezles/status/9", text)
+        self.assertIn("[[archive/threads/iquilezles/slug]]", text)
+
+
+class MergeOmittedOpTests(unittest.TestCase):
+    def test_retains_op_when_fresh_omits_root(self) -> None:
+        existing = ThreadData(
+            root_post_id="1",
+            posts=(
+                post("1", "A", "op qt", None, "9"),
+                post("2", "B", "tip", "1"),
+            ),
+            source_url="https://x.com/i/status/1",
+        )
+        fresh = ThreadData(
+            root_post_id="2",
+            posts=(post("2", "B", "tip-fresh", "1"),),
+            source_url="https://x.com/i/status/2",
+        )
+        merged, retained = _merge_existing_capture(fresh, existing, "2")
+        ids = {p.post_id: p for p in merged.posts}
+        self.assertIn("1", ids)
+        self.assertEqual(ids["1"].quote_of_id, "9")
+        self.assertEqual(ids["2"].text, "tip-fresh")
+        self.assertEqual(retained, ("1",))
+
+
+class OverlayQuoteOfIdTests(unittest.TestCase):
+    def test_keeps_on_disk_when_fresh_empty(self) -> None:
+        previous = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "old", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        fresh = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "new", None, None),),
+            source_url="https://x.com/i/status/1",
+        )
+        merged = overlay_quote_of_ids(fresh, previous)
+        self.assertEqual(merged.posts[0].text, "new")
+        self.assertEqual(merged.posts[0].quote_of_id, "9")
+
+    def test_fresh_nonempty_wins(self) -> None:
+        previous = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "old", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        fresh = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "new", None, "8"),),
+            source_url="https://x.com/i/status/1",
+        )
+        merged = overlay_quote_of_ids(fresh, previous)
+        self.assertEqual(merged.posts[0].quote_of_id, "8")
+
+    def test_missing_previous_post_unchanged(self) -> None:
+        previous = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "old", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        fresh = ThreadData(
+            root_post_id="2",
+            posts=(post("2", "A", "only-fresh", None, None),),
+            source_url="https://x.com/i/status/2",
+        )
+        merged = overlay_quote_of_ids(fresh, previous)
+        self.assertIsNone(merged.posts[0].quote_of_id)
+
+
+def _write_thread(path: Path, thread: ThreadData) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(asdict(thread), indent=2) + "\n",
+        encoding="utf-8",
+        newline="\n",
+    )
+
+
+class ArchiveForPostTests(unittest.TestCase):
+    def test_finds_post_in_thread_data(self) -> None:
+        with TemporaryDirectory() as tmp:
+            vault = Path(tmp)
+            assets = vault / "assets" / "threads" / "iquilezles" / "2025-02-09-slug"
+            notes = vault / "archive" / "threads" / "iquilezles" / "2025-02-09-slug"
+            notes.mkdir(parents=True)
+            thread = ThreadData(
+                root_post_id="9",
+                posts=(post("9", "iquilezles", "quoted root", None),),
+                source_url="https://x.com/iquilezles/status/9",
+            )
+            _write_thread(assets / "thread_data.json", thread)
+            found_assets, found_notes = archive_for_post(vault, "9")
+            self.assertEqual(found_assets, assets)
+            self.assertEqual(found_notes, notes)
+            url, wiki = quote_ref_for(vault, "9")
+            self.assertEqual(url, "https://x.com/iquilezles/status/9")
+            self.assertEqual(wiki, "archive/threads/iquilezles/2025-02-09-slug")
+
+    def test_block_id_when_not_op(self) -> None:
+        with TemporaryDirectory() as tmp:
+            vault = Path(tmp)
+            assets = vault / "assets" / "threads" / "H" / "slug"
+            notes = vault / "archive" / "threads" / "H" / "slug"
+            notes.mkdir(parents=True)
+            thread = ThreadData(
+                root_post_id="1",
+                posts=(
+                    post("1", "H", "op", None),
+                    post("2", "H", "child", "1"),
+                ),
+                source_url="https://x.com/H/status/1",
+            )
+            _write_thread(assets / "thread_data.json", thread)
+            url, wiki = quote_ref_for(vault, "2")
+            self.assertEqual(url, "https://x.com/H/status/2")
+            self.assertEqual(wiki, "archive/threads/H/slug#^2")
+
+    def test_miss_uses_i_status(self) -> None:
+        with TemporaryDirectory() as tmp:
+            url, wiki = quote_ref_for(Path(tmp), "9")
+            self.assertEqual(url, "https://x.com/i/status/9")
+            self.assertIsNone(wiki)
+
+
+class RenderGapsQuoteTests(unittest.TestCase):
+    def test_lists_missing_quoted_ids(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "qt", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        text = render_gaps(
+            thread, ["1"], input_kind="tip", missing_quote_of=["9"]
+        )
+        self.assertIn("quote_of:\n9\n", text)
+        self.assertNotIn("quote_of:\n1\n", text)
+
+    def test_unset_when_none_missing(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "qt", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        text = render_gaps(
+            thread, ["1"], input_kind="tip", missing_quote_of=[]
+        )
+        self.assertIn("quote_of: unset\n", text)
+
+
+class ParserNoQuotesTests(unittest.TestCase):
+    def test_refresh_accepts_no_quotes(self) -> None:
+        args = build_parser().parse_args(
+            ["refresh", "--id", "1", "--tip", "--no-quotes"]
+        )
+        self.assertTrue(args.no_quotes)
+
+    def test_refresh_default_chases(self) -> None:
+        args = build_parser().parse_args(["refresh", "--id", "1", "--tip"])
+        self.assertFalse(args.no_quotes)
+
+
+class ChaseSpineQuotesTests(unittest.TestCase):
+    def test_inner_refresh_is_root_no_quotes(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(
+                post("1", "A", "qt", None, "9"),
+                post("2", "B", "reply", "1"),
+                post("4", "C", "branch qt", "1", "7"),
+            ),
+            source_url="https://x.com/i/status/1",
+        )
+        calls: list[Namespace] = []
+
+        def fake_refresh(args: Namespace) -> int:
+            calls.append(args)
+            return 0
+
+        with (
+            patch("twitter.tw.locate", return_value=(None, None)),
+            patch("twitter.tw.check_frozen"),
+            patch("twitter.tw.cmd_refresh", fake_refresh),
+        ):
+            chase_spine_quotes(thread, tip=True, tip_id="2")
+        self.assertEqual(len(calls), 1)
+        self.assertEqual(calls[0].id, "9")
+        self.assertFalse(calls[0].tip)
+        self.assertTrue(calls[0].no_quotes)
+        self.assertEqual(calls[0].branch, [])
+
+    def test_locate_hit_skips_refresh(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "qt", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        calls: list[Namespace] = []
+
+        def fake_refresh(args: Namespace) -> int:
+            calls.append(args)
+            return 0
+
+        with (
+            patch(
+                "twitter.tw.locate",
+                return_value=(Path("assets"), Path("notes")),
+            ),
+            patch("twitter.tw.cmd_refresh", fake_refresh),
+        ):
+            chase_spine_quotes(thread, tip=True, tip_id="1")
+        self.assertEqual(calls, [])
+
+    def test_frozen_skips_refresh(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "qt", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+        calls: list[Namespace] = []
+
+        def fake_refresh(args: Namespace) -> int:
+            calls.append(args)
+            return 0
+
+        def fake_frozen(post_id: str) -> None:
+            raise SystemExit(f"frozen: skipped ({post_id})")
+
+        with (
+            patch("twitter.tw.locate", return_value=(None, None)),
+            patch("twitter.tw.check_frozen", fake_frozen),
+            patch("twitter.tw.cmd_refresh", fake_refresh),
+        ):
+            chase_spine_quotes(thread, tip=True, tip_id="1")
+        self.assertEqual(calls, [])
+
+    def test_inner_failure_does_not_raise(self) -> None:
+        thread = ThreadData(
+            root_post_id="1",
+            posts=(post("1", "A", "qt", None, "9"),),
+            source_url="https://x.com/i/status/1",
+        )
+
+        def fake_refresh(args: Namespace) -> int:
+            raise SystemExit("gallery-dl failed")
+
+        with (
+            patch("twitter.tw.locate", return_value=(None, None)),
+            patch("twitter.tw.check_frozen"),
+            patch("twitter.tw.cmd_refresh", fake_refresh),
+        ):
+            chase_spine_quotes(thread, tip=True, tip_id="1")
+
+    def test_cmd_refresh_no_quotes_skips_chase(self) -> None:
+        args = Namespace(
+            id="1",
+            branch=[],
+            tip=True,
+            slug=None,
+            preserve_existing=False,
+            attach=[],
+            allow_broken_walk=False,
+            retire_old_dir=False,
+            no_quotes=True,
+        )
+        with (
+            patch("twitter.tw.check_frozen"),
+            patch("twitter.tw.cmd_refetch", return_value=0),
+            patch("twitter.tw.cmd_graph", return_value=0),
+            patch("twitter.tw.cmd_emit", return_value=0),
+            patch("twitter.tw.chase_spine_quotes") as chase,
+            patch("twitter.tw.scratch_dir") as scratch,
+            patch("twitter.tw.locate", return_value=(None, None)),
+        ):
+            scratch.return_value = Path("missing-scratch")
+            result = cmd_refresh(args)
+        self.assertEqual(result, 0)
+        chase.assert_not_called()

scripts/twitter/tree.py

@@ -123,3 +123,17 @@ def descendants(start_id: str, kids: dict[str, list[str]]) -> list[str]:
         out.append(n)
         stack.extend(reversed(kids.get(n, [])))
     return out
+
+
+def spine_quote_ids(thread: ThreadData, spine: list[str]) -> list[str]:
+    """Return unique ``quote_of_id`` values on ``spine``, first-seen order."""
+    ids = by_id(thread)
+    out: list[str] = []
+    seen: set[str] = set()
+    for pid in spine:
+        quote_id = ids[pid].quote_of_id
+        if not quote_id or quote_id in seen:
+            continue
+        seen.add(quote_id)
+        out.append(quote_id)
+    return out

scripts/twitter/tw.py

@@ -20,7 +20,7 @@ import shutil
 import subprocess
 import sys
 from collections.abc import Callable
-from dataclasses import asdict
+from dataclasses import asdict, replace
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -31,6 +31,12 @@ if str(_HERE) not in sys.path:
 from paths import COOKIES, DUMPS, FROZEN, HERE, SCRATCH, VAULT
 from models import PostData, ThreadData, load_thread
 
+try:
+    from twitter.tree import spine_from_tip, spine_ids, spine_quote_ids
+except ImportError:
+    sys.path.insert(0, str(_HERE.parent))
+    from twitter.tree import spine_from_tip, spine_ids, spine_quote_ids
+
 try:
     from frozen import frozen_match, load_frozen_ids, require_writable
     from backup_assets import backup_thread, load_destination_root
@@ -342,13 +348,9 @@ def _merge_existing_capture(
     """Keep fresh post data while retaining archived posts it omitted."""
     if tip_id not in {post.post_id for post in fresh.posts}:
         raise SystemExit(f"spine tip {tip_id} missing from fresh capture")
-    fresh_roots = {
-        post.post_id for post in fresh.posts if post.reply_to_id is None
-    }
-    existing_roots = {
-        post.post_id for post in existing.posts if post.reply_to_id is None
-    }
-    if not fresh_roots or fresh_roots.isdisjoint(existing_roots):
+    fresh_ids = {post.post_id for post in fresh.posts}
+    existing_ids = {post.post_id for post in existing.posts}
+    if fresh_ids.isdisjoint(existing_ids):
         raise SystemExit("fresh and existing captures are a different conversation")
     return _merge_branch_posts(fresh, [existing.posts])
 
@@ -624,11 +626,85 @@ def cmd_merge(args: argparse.Namespace) -> int:
     return 0
 
 
+def overlay_quote_of_ids(current: ThreadData, previous: ThreadData) -> ThreadData:
+    """Keep previous ``quote_of_id`` when the current post has none."""
+    prev_by = {post.post_id: post for post in previous.posts}
+    posts: list[PostData] = []
+    for post in current.posts:
+        old = prev_by.get(post.post_id)
+        if (not post.quote_of_id) and old is not None and old.quote_of_id:
+            post = replace(post, quote_of_id=old.quote_of_id)
+        posts.append(post)
+    return replace(current, posts=tuple(posts))
+
+
+def chase_spine_quotes(
+    thread: ThreadData,
+    *,
+    tip: bool,
+    tip_id: str,
+) -> None:
+    """Refresh each missing spine ``quote_of_id`` as a root. One hop."""
+    spine = spine_from_tip(thread, tip_id) if tip else spine_ids(thread)
+    for qid in spine_quote_ids(thread, spine):
+        assets, _notes = locate(qid)
+        if assets is not None:
+            continue
+        try:
+            check_frozen(qid)
+        except SystemExit:
+            continue
+        inner = argparse.Namespace(
+            id=qid,
+            branch=[],
+            tip=False,
+            slug=None,
+            preserve_existing=False,
+            attach=[],
+            allow_broken_walk=False,
+            retire_old_dir=False,
+            no_quotes=True,
+        )
+        try:
+            cmd_refresh(inner)
+        except (SystemExit, subprocess.CalledProcessError) as exc:
+            print(f"quote-capture failed {qid}: {exc}")
+
+
 def cmd_refresh(args: argparse.Namespace) -> int:
-    """Convenience: ``refetch`` + ``graph`` + ``emit --force`` for ``args.id``."""
+    """Convenience: ``refetch`` + quote chase + ``graph`` + ``emit --force``."""
     check_frozen(args.id)
     cmd_refetch(args)
+    src = scratch_dir(args.id) / "thread_data.json"
+    if getattr(args, "preserve_existing", False):
+        assets, _notes = locate(args.id)
+        if assets is not None and (assets / "thread_data.json").is_file():
+            fresh = load_thread(src)
+            existing = load_thread(assets / "thread_data.json")
+            merged, retained_ids = _merge_existing_capture(
+                fresh,
+                existing,
+                args.id,
+            )
+            _write_thread(src, merged)
+            retained = ",".join(retained_ids) if retained_ids else "-"
+            print(f"preserve-existing retained={retained}")
+    assets, _notes = locate(args.id)
+    if (
+        assets is not None
+        and (assets / "thread_data.json").is_file()
+        and src.is_file()
+    ):
+        current = load_thread(src)
+        previous = load_thread(assets / "thread_data.json")
+        _write_thread(src, overlay_quote_of_ids(current, previous))
     cmd_graph(args)
+    if not getattr(args, "no_quotes", False) and src.is_file():
+        chase_spine_quotes(
+            load_thread(src),
+            tip=bool(getattr(args, "tip", False)),
+            tip_id=args.id,
+        )
     cmd_emit(args)
     return 0
 
@@ -893,6 +969,12 @@ def build_parser() -> argparse.ArgumentParser:
                 action="store_true",
                 help="delete this thread's archive under another handle",
             )
+        if name == "refresh":
+            p.add_argument(
+                "--no-quotes",
+                action="store_true",
+                help="do not capture quoted tweets on the spine",
+            )
         if name == "lift":
             p.add_argument("--orig", action="store_true")
         if name == "ocr":