fix(storage): enforce orphan-ness inside the purge DELETE + chunk IN-lists

Review feedback on the purge's race window: the pre-SELECT re-verify
left a statement-to-statement gap where a concurrent registration could
still lose rows — and the pre-counted refcount release could underflow
when it didn't. Orphan-ness now rides the DELETE itself (correlated
NOT EXISTS) with refcounts released from its RETURNING, so refs are
released for exactly the rows that were deleted. Input is de-duplicated,
IN-lists chunk at the storage layer's 500 convention, and the scan's
per-workstream ref-count loop is now one anti-join pass.

tests/test_orphan_cleanup.py

@@ -134,6 +134,31 @@ class TestOrphanPurge:
             ).scalar()
         assert left == 0
 
+    def test_purge_dedupes_input(self, backend):
+        """Duplicate ws_ids must not inflate counts or bind params."""
+        _orphan(backend, "ghost1", n=2)
+        result = backend.delete_orphan_conversations(["ghost1", "ghost1", "ghost1"])
+        assert result["workstreams"] == 1
+        assert result["rows"] == 2
+        assert result["skipped"] == 0
+
+    def test_purge_unknown_ws_counts_skipped(self, backend):
+        """An input with no rows and no workstream is reported, not purged."""
+        result = backend.delete_orphan_conversations(["nope-never-existed"])
+        assert result == {"workstreams": 0, "rows": 0, "released_refs": 0, "skipped": 1}
+
+    def test_purge_chunks_large_input(self, backend, monkeypatch):
+        """IN-lists are chunked (SQLite bind-parameter limits) without losing rows."""
+        import turnstone.core.storage._utils as storage_utils
+
+        monkeypatch.setattr(storage_utils, "_PURGE_CHUNK", 2)
+        for i in range(5):
+            _orphan(backend, f"ghost{i}", n=1)
+        result = backend.delete_orphan_conversations([f"ghost{i}" for i in range(5)])
+        assert result["workstreams"] == 5
+        assert result["rows"] == 5
+        assert backend.list_orphan_conversations() == []
+
     def test_purge_empty_list_is_noop(self, backend):
         assert backend.delete_orphan_conversations([]) == {
             "workstreams": 0,

turnstone/core/storage/_protocol.py

@@ -646,11 +646,13 @@ class StorageBackend(Protocol):
     def delete_orphan_conversations(self, ws_ids: list[str]) -> dict[str, int]:
         """Purge conversation rows for the *ws_ids* that are STILL orphaned.
 
-        Re-verifies against ``workstreams`` in-transaction (a re-registered
-        ws_id is skipped, never deleted), releases the deleted rows'
-        attachment refcounts, and sweeps matching ``workstream_config`` /
-        ``workstream_overrides`` rows.  Returns counts keyed ``workstreams``,
-        ``rows``, ``released_refs``, ``skipped``.
+        Orphan-ness is enforced inside the DELETE itself (correlated
+        ``NOT EXISTS`` against ``workstreams``) and refcounts are released
+        from its ``RETURNING`` — a ws_id registered before or during the
+        purge keeps both its rows and its refcounts.  Sweeps the purged
+        ws_ids' ``workstream_config`` / ``workstream_overrides`` rows.
+        Returns counts keyed ``workstreams``, ``rows``, ``released_refs``,
+        ``skipped`` (distinct inputs not purged).
         """
         ...
 

turnstone/core/storage/_utils.py

@@ -189,6 +189,7 @@ def find_orphan_conversations(conn: Any) -> list[dict[str, Any]]:
     entry carries the attachment-ref count so a purge's refcount release is
     visible before it happens.
     """
+    anti_join = conversations.outerjoin(workstreams, conversations.c.ws_id == workstreams.c.ws_id)
     rows = conn.execute(
         sa.select(
             conversations.c.ws_id,
@@ -196,80 +197,98 @@ def find_orphan_conversations(conn: Any) -> list[dict[str, Any]]:
             sa.func.min(conversations.c.timestamp).label("first"),
             sa.func.max(conversations.c.timestamp).label("last"),
         )
-        .select_from(
-            conversations.outerjoin(workstreams, conversations.c.ws_id == workstreams.c.ws_id)
-        )
+        .select_from(anti_join)
         .where(workstreams.c.ws_id.is_(None))
         .group_by(conversations.c.ws_id)
         .order_by(sa.func.min(conversations.c.timestamp))
     ).fetchall()
-    orphans: list[dict[str, Any]] = []
-    for ws_id, row_count, first, last in rows:
-        referenced = conn.execute(
-            sa.select(conversations.c.attachments).where(
-                sa.and_(
-                    conversations.c.ws_id == ws_id,
-                    conversations.c.attachments.is_not(None),
-                )
+    # Ref counts in ONE pass over the orphan rows that carry attachments —
+    # not a query per orphan workstream, so the scan stays proportional to
+    # orphan ROW count.
+    ref_counts: dict[str, int] = {}
+    ref_rows = conn.execute(
+        sa.select(conversations.c.ws_id, conversations.c.attachments)
+        .select_from(anti_join)
+        .where(
+            sa.and_(
+                workstreams.c.ws_id.is_(None),
+                conversations.c.attachments.is_not(None),
             )
-        ).fetchall()
-        ref_count = sum(len(parse_attachment_refs(refs)) for (refs,) in referenced)
-        orphans.append(
-            {
-                "ws_id": ws_id,
-                "rows": int(row_count),
-                "first": first,
-                "last": last,
-                "attachment_refs": ref_count,
-            }
         )
-    return orphans
+    ).fetchall()
+    for ws_id, refs in ref_rows:
+        ref_counts[ws_id] = ref_counts.get(ws_id, 0) + len(parse_attachment_refs(refs))
+    return [
+        {
+            "ws_id": ws_id,
+            "rows": int(row_count),
+            "first": first,
+            "last": last,
+            "attachment_refs": ref_counts.get(ws_id, 0),
+        }
+        for ws_id, row_count, first, last in rows
+    ]
+
+
+# IN-list chunk size for the purge statements — mirrors the storage layer's
+# existing bulk chunking (SQLite bind-parameter limits).
+_PURGE_CHUNK = 500
 
 
 def purge_orphan_conversations(conn: Any, ws_ids: list[str]) -> dict[str, int]:
     """Delete conversation rows for the *ws_ids* that are STILL orphans.
 
-    Orphan-ness is re-verified here, inside the caller's transaction — a
-    ws_id that gained a ``workstreams`` row between scan and purge is counted
-    in ``skipped`` and left untouched, so a stale scan can never delete a
-    live workstream's history.  Mirrors ``delete_workstream``'s cascade for
-    rows with no owning workstream: release the deleted rows' attachment
-    refcounts, then sweep the matching ``workstream_config`` /
-    ``workstream_overrides`` rows.  Caller owns commit.
+    Orphan-ness is enforced INSIDE the DELETE itself (a correlated
+    ``NOT EXISTS`` against ``workstreams``), and the refcounts to release
+    come from the DELETE's ``RETURNING`` — so refs are released for exactly
+    the rows that were deleted.  A ws_id registered at any point before the
+    DELETE statement keeps both its rows AND its refcounts; there is no
+    pre-count/delete window to underflow.  (Needs ``DELETE .. RETURNING``:
+    PostgreSQL, or SQLite ≥ 3.35.)
+
+    Input is de-duplicated and all IN-lists are chunked.  ``skipped`` =
+    distinct input ws_ids not purged (registered before/during the purge, or
+    no rows).  The purged ws_ids' ``workstream_config`` /
+    ``workstream_overrides`` rows are swept.  Caller owns commit.
     """
-    if not ws_ids:
+    distinct = list(dict.fromkeys(ws_ids))
+    if not distinct:
         return {"workstreams": 0, "rows": 0, "released_refs": 0, "skipped": 0}
-    registered = {
-        row[0]
-        for row in conn.execute(
-            sa.select(workstreams.c.ws_id).where(workstreams.c.ws_id.in_(ws_ids))
-        ).fetchall()
-    }
-    targets = [w for w in ws_ids if w not in registered]
-    if not targets:
-        return {"workstreams": 0, "rows": 0, "released_refs": 0, "skipped": len(registered)}
-    referenced = conn.execute(
-        sa.select(conversations.c.attachments).where(
-            sa.and_(
-                conversations.c.ws_id.in_(targets),
-                conversations.c.attachments.is_not(None),
-            )
-        )
-    ).fetchall()
     ref_ids: list[str] = []
-    for (refs,) in referenced:
-        ref_ids.extend(parse_attachment_refs(refs))
+    purged_ws: set[str] = set()
+    rows_deleted = 0
+    for i in range(0, len(distinct), _PURGE_CHUNK):
+        chunk = distinct[i : i + _PURGE_CHUNK]
+        returned = conn.execute(
+            sa.delete(conversations)
+            .where(
+                sa.and_(
+                    conversations.c.ws_id.in_(chunk),
+                    ~sa.exists(
+                        sa.select(workstreams.c.ws_id).where(
+                            workstreams.c.ws_id == conversations.c.ws_id
+                        )
+                    ),
+                )
+            )
+            .returning(conversations.c.ws_id, conversations.c.attachments)
+        ).fetchall()
+        for ws_id, refs in returned:
+            purged_ws.add(ws_id)
+            rows_deleted += 1
+            if refs:
+                ref_ids.extend(parse_attachment_refs(refs))
     release_attachment_refs(conn, ref_ids)
-    deleted = conn.execute(
-        sa.delete(conversations).where(conversations.c.ws_id.in_(targets))
-    ).rowcount
-    conn.execute(sa.delete(workstream_config).where(workstream_config.c.ws_id.in_(targets)))
-    conn.execute(sa.delete(workstream_overrides).where(workstream_overrides.c.ws_id.in_(targets)))
+    swept = sorted(purged_ws)
+    for i in range(0, len(swept), _PURGE_CHUNK):
+        chunk = swept[i : i + _PURGE_CHUNK]
+        conn.execute(sa.delete(workstream_config).where(workstream_config.c.ws_id.in_(chunk)))
+        conn.execute(sa.delete(workstream_overrides).where(workstream_overrides.c.ws_id.in_(chunk)))
     return {
-        "workstreams": len(targets),
-        "rows": int(deleted or 0),
+        "workstreams": len(purged_ws),
+        "rows": rows_deleted,
         "released_refs": len(ref_ids),
-        "skipped": len(registered),
+        "skipped": len(distinct) - len(purged_ws),
     }