Sinity · Sinity · Aug 9, 2026 · Aug 8, 2026 · Aug 8, 2026 · Aug 8, 2026
diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl
diff --git a/docs/plans/reindex-incident-coverage.json b/docs/plans/reindex-incident-coverage.json
@@ -69,7 +69,8 @@
     "polylogue-byte-supersession-live-proof": {"kind": "named-child-bead"},
     "polylogue-hook-authority-conflict-proof": {"kind": "named-child-bead"},
     "polylogue-excluded-cursor-live-proof": {"kind": "named-child-bead"},
-    "polylogue-chatgpt-content-live-proof": {"kind": "named-child-bead"}
+    "polylogue-chatgpt-content-live-proof": {"kind": "named-child-bead"},
+    "polylogue-xselt": {"kind": "named-child-bead"}
   },
   "rows": [
     {"bead_id": "polylogue-0qfy", "bead_status": "closed", "incident": {"incident_id": "incident-0qfy", "bead_id": "polylogue-0qfy", "forcing_class": "content-fidelity"}, "route": {"kind": "registry", "entrypoint": "reindex-campaign"}, "schedule": {"phase": "preflight", "order": 1}, "expected_snapshot": {"snapshot_id": "reindex-baseline-2026-08-03", "state": "blocking"}, "registry_checks": ["content-fidelity", "campaign-coverage"], "red_mutation": {"fixture_id": "vintage-reorder", "mutation_id": "mutation-0qfy"}, "receipts": [], "residual_successor": {"bead_id": "polylogue-claude-vintage-live-proof", "kind": "live-proof"}},
@@ -105,7 +106,7 @@
     {"bead_id": "polylogue-rrxe4", "bead_status": "open", "incident": {"incident_id": "incident-rrxe4", "bead_id": "polylogue-rrxe4", "forcing_class": "properties"}, "route": {"kind": "campaign", "entrypoint": "reindex-campaign"}, "schedule": {"phase": "preflight", "order": 31}, "expected_snapshot": {"snapshot_id": "derived-model-candidate", "state": "blocking"}, "registry_checks": ["convergence-properties"], "red_mutation": {"fixture_id": "campaign-corpus", "mutation_id": "mutation-rrxe4"}, "receipts": [], "residual_successor": null},
     {"bead_id": "polylogue-rrxe4.1", "bead_status": "open", "incident": {"incident_id": "incident-rrxe4-1", "bead_id": "polylogue-rrxe4.1", "forcing_class": "properties"}, "route": {"kind": "campaign", "entrypoint": "reindex-campaign"}, "schedule": {"phase": "preflight", "order": 32}, "expected_snapshot": {"snapshot_id": "derived-model-candidate", "state": "blocking"}, "registry_checks": ["convergence-properties"], "red_mutation": {"fixture_id": "campaign-corpus", "mutation_id": "mutation-rrxe4-1"}, "receipts": [], "residual_successor": null},
     {"bead_id": "polylogue-s8s54", "bead_status": "open", "incident": {"incident_id": "incident-s8s54", "bead_id": "polylogue-s8s54", "forcing_class": "origin-repair"}, "route": {"kind": "registry", "entrypoint": "origin-capability"}, "schedule": {"phase": "preflight", "order": 33}, "expected_snapshot": {"snapshot_id": "live-preflight-2026-08-04", "state": "blocking"}, "registry_checks": ["origin-repair"], "red_mutation": {"fixture_id": "origin-matrix", "mutation_id": "mutation-s8s54"}, "receipts": [], "residual_successor": null},
-    {"bead_id": "polylogue-slshy", "bead_status": "in_progress", "incident": {"incident_id": "incident-slshy", "bead_id": "polylogue-slshy", "forcing_class": "parser-stamps"}, "route": {"kind": "registry", "entrypoint": "reindex-final-proof"}, "schedule": {"phase": "preflight", "order": 34}, "expected_snapshot": {"snapshot_id": "post-reindex-acceptance", "state": "blocking"}, "registry_checks": ["parser-stamps"], "red_mutation": {"fixture_id": "vintage-reorder", "mutation_id": "mutation-slshy"}, "receipts": [], "residual_successor": null},
+    {"bead_id": "polylogue-slshy", "bead_status": "closed", "incident": {"incident_id": "incident-slshy", "bead_id": "polylogue-slshy", "forcing_class": "parser-stamps"}, "route": {"kind": "registry", "entrypoint": "reindex-final-proof"}, "schedule": {"phase": "preflight", "order": 34}, "expected_snapshot": {"snapshot_id": "post-reindex-acceptance", "state": "blocking"}, "registry_checks": ["parser-stamps"], "red_mutation": {"fixture_id": "vintage-reorder", "mutation_id": "mutation-slshy"}, "receipts": [], "residual_successor": {"bead_id": "polylogue-xselt", "kind": "named-child-bead"}},
     {"bead_id": "polylogue-sp72", "bead_status": "open", "incident": {"incident_id": "incident-sp72", "bead_id": "polylogue-sp72", "forcing_class": "revision-lineage"}, "route": {"kind": "registry", "entrypoint": "raw-authority-ledger"}, "schedule": {"phase": "preflight", "order": 35}, "expected_snapshot": {"snapshot_id": "live-preflight-2026-08-04", "state": "blocking"}, "registry_checks": ["revision-lineage"], "red_mutation": {"fixture_id": "drive-revision", "mutation_id": "mutation-sp72"}, "receipts": [], "residual_successor": null},
     {"bead_id": "polylogue-t0m73", "bead_status": "open", "incident": {"incident_id": "incident-t0m73", "bead_id": "polylogue-t0m73", "forcing_class": "archive-invariants"}, "route": {"kind": "registry", "entrypoint": "archive-verification"}, "schedule": {"phase": "promotion", "order": 36}, "expected_snapshot": {"snapshot_id": "post-reindex-acceptance", "state": "blocking"}, "registry_checks": ["archive-invariants"], "red_mutation": {"fixture_id": "derived-model", "mutation_id": "mutation-t0m73"}, "receipts": [], "residual_successor": null},
     {"bead_id": "polylogue-uqwd", "bead_status": "in_progress", "incident": {"incident_id": "incident-uqwd", "bead_id": "polylogue-uqwd", "forcing_class": "lifecycle-anchor"}, "route": {"kind": "registry", "entrypoint": "reindex-campaign"}, "schedule": {"phase": "preflight", "order": 37}, "expected_snapshot": {"snapshot_id": "reindex-baseline-2026-08-03", "state": "blocking"}, "registry_checks": ["lifecycle-anchor"], "red_mutation": {"fixture_id": "lifecycle-anchor-drift", "mutation_id": "mutation-uqwd"}, "receipts": [], "residual_successor": null},

diff --git a/polylogue/api/archive.py b/polylogue/api/archive.py
@@ -32,7 +32,7 @@
     DEFAULT_CONTEXT_IMAGE_MAX_CHARS_PER_MESSAGE,
     DEFAULT_CONTEXT_IMAGE_MAX_MESSAGES_PER_SESSION,
 )
-from polylogue.core.enums import AssertionKind, AssertionStatus, MaterialOrigin, Origin, TitleSource
+from polylogue.core.enums import AssertionKind, AssertionStatus, MaterialOrigin, Origin, Provider, TitleSource
 from polylogue.core.errors import PolylogueError
 from polylogue.core.json import JSONDocument, JSONValue
 from polylogue.core.refs import (
@@ -44,6 +44,7 @@
     parse_delegation_subtree_object_id,
     parse_public_ref,
 )
+from polylogue.core.sources import origin_from_provider
 from polylogue.core.timestamps import parse_archive_datetime
 from polylogue.core.types import SessionId
 from polylogue.core.user_state_targets import TARGET_MESSAGE, TARGET_SESSION
@@ -296,6 +297,62 @@ class SessionNotFoundError(PolylogueError):
     http_status_code = 404
 
 
+def _resolve_durable_user_state_session_id(archive_root: Path, token: str) -> str | None:
+    """Resolve a session alias from canonical mark/annotation owners.
+
+    The index is rebuildable, while mark and annotation ownership is durable.
+    When the index cannot resolve a token, match it against the canonical
+    session ids already persisted in those user-state rows. Every accepted
+    alias shape is checked against the complete durable owner set, and an
+    ambiguous prefix fails closed instead of selecting an arbitrary owner.
+    """
+    if not token:
+        return None
+    user_db = archive_root / "user.db"
+    if not user_db.exists():
+        return None
+    try:
+        with closing(open_readonly_connection(user_db)) as conn:
+            rows = conn.execute(
+                """
+                SELECT target_ref, scope_ref
+                FROM assertions
+                WHERE kind IN (?, ?)
+                  AND COALESCE(status, 'active') != 'deleted'
+                """,
+                (AssertionKind.MARK.value, AssertionKind.ANNOTATION.value),
+            ).fetchall()
+    except sqlite3.Error:
+        return None
+
+    canonical_ids: set[str] = set()
+    for row in rows:
+        for value in (row[0], row[1]):
+            if not isinstance(value, str):
+                continue
+            if value.startswith("session:"):
+                canonical_id = value[len("session:") :]
+                if canonical_id:
+                    canonical_ids.add(canonical_id)
+
+    matches = {canonical_id for canonical_id in canonical_ids if _durable_session_alias_matches(token, canonical_id)}
+    if len(matches) > 1:
+        raise ValueError(f"session id alias {token!r} is ambiguous")
+    return next(iter(matches), None)
+
+
+def _durable_session_alias_matches(token: str, canonical_id: str) -> bool:
+    """Apply the same exact/provider/prefix/suffix alias shapes as the index."""
+    if token == canonical_id or canonical_id.startswith(token):
+        return True
+    if ":" in token:
+        provider_token, native_id = token.split(":", 1)
+        provider_origin = origin_from_provider(Provider.from_string(provider_token)).value
+        return canonical_id == f"{provider_origin}:{native_id}"
+    _, separator, native_id = canonical_id.partition(":")
+    return bool(separator and (native_id == token or native_id.startswith(token)))
-def _durable_session_alias_matches(token: str, canonical_id: str) -> bool:
-    """Apply the same exact/provider/prefix/suffix alias shapes as the index."""
-    if token == canonical_id or canonical_id.startswith(token):
-        return True
-    if ":" in token:
-        provider_token, native_id = token.split(":", 1)
-        provider_origin = origin_from_provider(Provider.from_string(provider_token)).value
-        return canonical_id == f"{provider_origin}:{native_id}"
-    _, separator, native_id = canonical_id.partition(":")
-    return bool(separator and (native_id == token or native_id.startswith(token)))
+def _durable_session_alias_match_rank(token: str, canonical_id: str) -> int | None:
+    """Rank one alias match: 0 is exact evidence, 1 is prefix-widened evidence.
+
+    Mirrors ``ArchiveStore.resolve_session_id``: an exact canonical id, an
+    exact provider-qualified id, or an exact native id outranks any
+    prefix-widened match, so a sibling native id cannot make an exact token
+    ambiguous.
+    """
+    if token == canonical_id:
+        return 0
+    if ":" in token:
+        provider_token, native_id = token.split(":", 1)
+        provider_origin = origin_from_provider(Provider.from_string(provider_token)).value
+        if canonical_id == f"{provider_origin}:{native_id}":
+            return 0
+        return 1 if canonical_id.startswith(token) else None
+    _, separator, canonical_native_id = canonical_id.partition(":")
+    if separator and canonical_native_id == token:
+        return 0
+    if canonical_id.startswith(token):
+        return 1
+    if separator and canonical_native_id.startswith(token):
+        return 1
+    return None
-def _durable_session_alias_matches(token: str, canonical_id: str) -> bool:
-    """Apply the same exact/provider/prefix/suffix alias shapes as the index."""
-    if token == canonical_id or canonical_id.startswith(token):
-        return True
-    if ":" in token:
-        provider_token, native_id = token.split(":", 1)
-        provider_origin = origin_from_provider(Provider.from_string(provider_token)).value
-        return canonical_id == f"{provider_origin}:{native_id}"
-    _, separator, native_id = canonical_id.partition(":")
-    return bool(separator and (native_id == token or native_id.startswith(token)))
+def _durable_session_alias_match_rank(token: str, canonical_id: str) -> int | None:
+    """Rank one alias match: 0 is exact evidence, 1 is prefix-widened evidence.
+
+    Mirrors ``ArchiveStore.resolve_session_id``: an exact canonical id, an
+    exact provider-qualified id, or an exact native id outranks any
+    prefix-widened match, so a sibling native id cannot make an exact token
+    ambiguous.
+    """
+    if token == canonical_id:
+        return 0
+    if ":" in token:
+        provider_token, native_id = token.split(":", 1)
+        provider_origin = origin_from_provider(Provider.from_string(provider_token)).value
+        if canonical_id == f"{provider_origin}:{native_id}":
+            return 0
+        return 1 if canonical_id.startswith(token) else None
+    _, separator, canonical_native_id = canonical_id.partition(":")
+    if separator and canonical_native_id == token:
+        return 0
+    if canonical_id.startswith(token):
+        return 1
+    if separator and canonical_native_id.startswith(token):
+        return 1
+    return None
+
+
 def _archive_query_date_ms(field: str, value: str | None) -> int | None:
     parsed = parse_query_date(field, value)
     if parsed is None:
@@ -6761,10 +6818,19 @@ async def _resolve_user_state_target(
         }
 
     async def _resolve_user_state_session_id(self, session_id: str) -> str:
-        archive_resolved = await self._archive_resolve_session_id(session_id)
-        if archive_resolved is None:
-            raise SessionNotFoundError(session_id)
-        return archive_resolved
+        try:
+            archive_resolved = await self._archive_resolve_session_id(session_id)
+        except SessionNotFoundError:
+            archive_resolved = None
+        if archive_resolved is not None:
+            return archive_resolved
+        durable_resolved = _resolve_durable_user_state_session_id(
+            _active_archive_root(self.config),
+            session_id,
+        )
+        if durable_resolved is not None:
+            return durable_resolved
+        raise SessionNotFoundError(session_id)
 
     async def _user_state_message_exists(self, session_id: str, message_id: str) -> bool:
         return bool(await self._archive_message_exists(session_id, message_id))
@@ -6847,6 +6913,7 @@ async def add_mark(
                 target_type=str(target["target_type"]),
                 target_id=str(target["target_id"]),
                 mark_type=mark_type,
+                owner_session_id=str(target["session_id"]) if target.get("session_id") else None,
             ),
             capability="archive.add_mark",
         )
@@ -6883,6 +6950,7 @@ async def remove_mark(
                 target_type=str(target["target_type"]),
                 target_id=str(target["target_id"]),
                 mark_type=mark_type,
+                owner_session_id=str(target["session_id"]) if target.get("session_id") else None,
             ),
             capability="archive.remove_mark",
         )
@@ -6900,23 +6968,32 @@ async def list_marks(
         """List marks, optionally filtered by type, target, session, or message."""
         resolved_target_type = target_type
         resolved_target_id = target_id
+        scope_session_id: str | None = None
         if message_id is not None:
             resolved_target_type = TARGET_MESSAGE
             resolved_target_id = message_id
         elif session_id is not None and target_id is None:
             try:
-                resolved_target_id = await self._resolve_user_state_session_id(session_id)
-                resolved_target_type = TARGET_SESSION
+                scope_session_id = await self._resolve_user_state_session_id(session_id)
             except SessionNotFoundError:
-                return []
+                # A durable user assertion can outlive the rebuildable
+                # session row. Keep the caller's canonical token so the
+                # archive read can use its durable message owner scope.
+                scope_session_id = session_id
         return await run_archive_read(
             _active_archive_root(self.config),
             operation="user_state.marks.list",
-            arguments={"mark_type": mark_type, "target_type": resolved_target_type, "target_id": resolved_target_id},
+            arguments={
+                "mark_type": mark_type,
+                "target_type": resolved_target_type,
+                "target_id": resolved_target_id,
+                "session_id": scope_session_id,
+            },
             work=lambda archive: archive.list_marks(
                 mark_type=mark_type,
                 target_type=resolved_target_type,
                 target_id=resolved_target_id,
+                session_id=scope_session_id,
             ),
             projection="marks",
             stable_order="created_at,target_id",
@@ -6960,6 +7037,7 @@ async def save_annotation(
                 target_type=str(target["target_type"]),
                 target_id=str(target["target_id"]),
                 note_text=note_text,
+                owner_session_id=str(target["session_id"]) if target.get("session_id") else None,
             ),
             capability="archive.save_annotation",
         )
@@ -6994,7 +7072,9 @@ async def list_annotations(
             try:
                 scope_session_id = await self._resolve_user_state_session_id(session_id)
             except SessionNotFoundError:
-                return []
+                # See list_marks: the user tier remains authoritative after
+                # the rebuildable index row is gone.
+                scope_session_id = session_id
         return await run_archive_read(
             _active_archive_root(self.config),
             operation="user_state.annotations.list",

diff --git a/polylogue/archive/session_revision_membership.py b/polylogue/archive/session_revision_membership.py
@@ -89,24 +89,57 @@ def _message_identities(contents: frozenset[MessageContent]) -> frozenset[bytes]
     return frozenset(identity for identity, _content, _multiplicity in contents)
 
 
-def _message_axis_relation(contents_a: frozenset[MessageContent], contents_b: frozenset[MessageContent]) -> _Relation:
+def _message_axis_relation(
+    contents_a: frozenset[MessageContent],
+    contents_b: frozenset[MessageContent],
+    *,
+    mutable_identities: frozenset[bytes] = frozenset(),
+) -> _Relation:
     """Compare message content as an unordered multiset.
 
     Reordering remains equivalent, while a repeated id-less message is an
     additional persisted turn rather than a duplicate that set semantics can
-    erase. A shared identity carrying different content remains a conflict.
+    erase. Native shared identities carrying different content remain a
+    conflict. Timestamped id-less identities are deliberately mutable: their
+    content hash still triggers archive replacement, while their revision
+    membership axis must not turn an edit into a false fork merely because a
+    sibling was added or changed.
     """
     counts_a = {(identity, content): multiplicity for identity, content, multiplicity in contents_a}
     counts_b = {(identity, content): multiplicity for identity, content, multiplicity in contents_b}
+    identity_counts_a: dict[bytes, int] = {}
+    identity_counts_b: dict[bytes, int] = {}
+    for (identity, _content), multiplicity in counts_a.items():
+        identity_counts_a[identity] = identity_counts_a.get(identity, 0) + multiplicity
+    for (identity, _content), multiplicity in counts_b.items():
+        identity_counts_b[identity] = identity_counts_b.get(identity, 0) + multiplicity
     identities_a = _message_identities(contents_a)
     identities_b = _message_identities(contents_b)
     for identity in identities_a & identities_b:
         content_values_a = {content for candidate_identity, content in counts_a if candidate_identity == identity}
         content_values_b = {content for candidate_identity, content in counts_b if candidate_identity == identity}
+        if identity in mutable_identities:
+            continue
         if content_values_a != content_values_b:
             return "conflict"
-    a_richer = bool(identities_a - identities_b) or any(count > counts_b.get(key, 0) for key, count in counts_a.items())
-    b_richer = bool(identities_b - identities_a) or any(count > counts_a.get(key, 0) for key, count in counts_b.items())
+
+    def _has_extra(
+        side: dict[tuple[bytes, bytes], int],
+        other: dict[tuple[bytes, bytes], int],
+        side_identity_counts: dict[bytes, int],
+        other_identity_counts: dict[bytes, int],
+    ) -> bool:
+        for key, count in side.items():
+            identity = key[0]
+            if identity in mutable_identities:
+                if side_identity_counts.get(identity, 0) > other_identity_counts.get(identity, 0):
+                    return True
+            elif count > other.get(key, 0):
+                return True
+        return False
+
+    a_richer = bool(identities_a - identities_b) or _has_extra(counts_a, counts_b, identity_counts_a, identity_counts_b)
+    b_richer = bool(identities_b - identities_a) or _has_extra(counts_b, counts_a, identity_counts_b, identity_counts_a)
     if a_richer and b_richer:
         return "conflict"
     if a_richer:
@@ -183,7 +216,11 @@ def _relation(a: SessionRevisionProjection, b: SessionRevisionProjection) -> _Re
     any single axis already is.
     """
     axes = (
-        _message_axis_relation(a.message_contents, b.message_contents),
+        _message_axis_relation(
+            a.message_contents,
+            b.message_contents,
+            mutable_identities=a.mutable_message_identities | b.mutable_message_identities,
+        ),
         _axis_relation(a.attachment_identities, a.attachment_contents, b.attachment_identities, b.attachment_contents),
         _axis_relation(
             _identities(a.event_contents), a.event_contents, _identities(b.event_contents), b.event_contents
@@ -259,6 +296,16 @@ def _equal_content_representative(
             if candidate_time.timestamp() > incumbent_time.timestamp()
             else (incumbent, candidate)
         )
+    if (
+        incumbent.observed_at_ms is not None
+        and candidate.observed_at_ms is not None
+        and candidate.observed_at_ms != incumbent.observed_at_ms
+    ):
+        # Timestamped id-less messages intentionally share a mutable revision
+        # axis.  When their content changes but provider_updated_at is absent
+        # or unchanged, observation order is the remaining source-backed
+        # authority; raw_id is only a deterministic last resort.
+        return (candidate, incumbent) if candidate.observed_at_ms > incumbent.observed_at_ms else (incumbent, candidate)
     # No distinguishing provenance or provider timestamp -- these two are
     # already proven identical content, so which raw_id represents them does
     # not matter for correctness; pick deterministically rather than

diff --git a/polylogue/core/message_owner.py b/polylogue/core/message_owner.py
@@ -0,0 +1,42 @@
+"""Private parser-to-writer coordinates for message ownership."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class MessageOwnerCoordinate:
+    """Private linkage between a parsed attachment and its message.
+
+    ``stable_key`` carries reorder-stable provider evidence when the parser
+    has it. ``position`` and ``variant_index`` are the complete transport
+    coordinate used as the fail-closed fallback. The coordinate is excluded
+    from public parser serialization and is never a provider message id.
+    """
+
+    stable_key: str | None = None
+    position: int | None = None
+    variant_index: int = 0
+
+    def __post_init__(self) -> None:
+        if self.position is not None and self.position < 0:
+            raise ValueError("message owner position cannot be negative")
+        if self.variant_index < 0:
+            raise ValueError("message owner variant_index cannot be negative")
+        if self.stable_key == "":
+            raise ValueError("message owner stable_key cannot be empty")
+
+    @property
+    def physical_key(self) -> tuple[int, int] | None:
+        """Return the full position/variant coordinate when it is present."""
+        if self.position is None:
+            return None
+        return self.position, self.variant_index
+
+
+class MessageOwnerAmbiguityError(ValueError):
+    """Raised when an attachment owner cannot be resolved without guessing."""
+
+
+__all__ = ["MessageOwnerAmbiguityError", "MessageOwnerCoordinate"]