refactor(storage): consolidate the three skip-stale-replace tie-breaks

polylogue/pipeline/services/ingest_batch/_core.py

@@ -60,6 +60,7 @@
     record_capture_gap_event,
     record_source_outage_events,
     session_has_parser_ingest_flag,
+    should_skip_stale_replace,
     stored_message_count,
 )
 from polylogue.storage.sqlite.archive_tiers.write import (
@@ -580,7 +581,10 @@ def _write_session(
     ):
         existing_updated_at_ms = existing_row["updated_at_ms"]
         existing_updated_at_int = int(existing_updated_at_ms) if existing_updated_at_ms is not None else None
-        if existing_updated_at_int is not None and incoming_freshness_ms < existing_updated_at_int:
+        if should_skip_stale_replace(
+            incoming_freshness_ms=incoming_freshness_ms,
+            existing_updated_at_ms=existing_updated_at_int,
+        ):
             counts["skipped_sessions"] = 1
             counts["skipped_messages"] = payload.message_count
             counts["skipped_attachments"] = payload.attachment_count

polylogue/storage/sqlite/archive_tiers/ingest_precedence.py

@@ -16,6 +16,60 @@
 BrowserCapturePrecedence = Literal["default", "replace", "skip"]
 
 
+def should_skip_stale_replace(
+    *,
+    incoming_freshness_ms: int | None,
+    existing_updated_at_ms: int | None,
+) -> bool:
+    """Return whether an incoming full-replace write is strictly staler than what is stored.
+
+    This is the ONE freshness-tie policy for whether an incoming session
+    write should be skipped as stale, consolidated from three previously
+    independent copies (polylogue-t83e): ``write_parsed_session_to_archive``
+    in ``archive_tiers/write.py``, the daemon batch-write path in
+    ``pipeline/services/ingest_batch/_core.py``, and
+    ``revision_governance.py``'s raw-parsed write path. Each call site keeps
+    its own surrounding guard conditions (``force_write``/``force_replace``,
+    browser-capture precedence, append-only, revision-authority membership,
+    ``source_index`` gating) — those decide *whether this check applies at
+    all*, not the comparison itself.
+
+    Deliberately a strict ``<``, not ``<=``: a genuine tie (same
+    ``updated_at_ms``) still replaces, so a replay that legitimately carries
+    identical freshness but different/corrected content is not silently
+    dropped.
+
+    This is a *timestamp* tie-break only, deliberately narrow: it is not
+    where "which raw is the real content" gets decided when two raws
+    resolve to the same ``session_id`` and one is a strict content subset of
+    the other (e.g. an appended-to Claude Code transcript re-acquired twice
+    at different completeness). That is decided upstream, before this
+    function's timestamp comparison is ever reached, by
+    ``archive/session_revision_membership.py``'s content-only set relation
+    (``equal``/``a_contains_b``/``b_contains_a``/``conflict`` — polylogue-aggz,
+    landed in #3401/#3405) via ``raw_session_memberships``/
+    ``raw_revision_heads``: a cohort with an accepted head is written from
+    that head, never from this per-write timestamp comparison. This function
+    only governs the fallback for raws revision governance never classified
+    into a cohort (single-raw sessions, or a governance table not yet
+    populated for older data) — see polylogue-t83e for a live example where
+    stale pre-#3401 ``raw_session_memberships`` rows briefly left two raws
+    of one session_id -- a local Claude Code transcript and an older byte-
+    prefix copy of the same file the operator had uploaded into an AI Studio
+    conversation -- without an accepted head, and this function's ordinary
+    timestamp comparison (not content-subset awareness) decided the
+    outcome. Recomputing revision membership under current code (any
+    ``rebuild_index_from_source`` replay, which calls
+    ``backfill_historical_revision_evidence``) resolves that case correctly
+    upstream of this function.
+    """
+    return (
+        existing_updated_at_ms is not None
+        and incoming_freshness_ms is not None
+        and incoming_freshness_ms < existing_updated_at_ms
+    )
+
+
 def browser_capture_precedence(
     *,
     existing_is_dom_fallback: bool,

polylogue/storage/sqlite/archive_tiers/revision_governance.py

@@ -130,6 +130,7 @@
     record_capture_gap_event,
     record_source_outage_events,
     session_has_parser_ingest_flag,
+    should_skip_stale_replace,
     stored_message_count,
 )
 from polylogue.storage.sqlite.archive_tiers.revision_application import (
@@ -370,7 +371,10 @@ def _write_parsed_precedence_result(
     ):
         existing_updated_at_ms = existing_row["updated_at_ms"]
         existing_updated_at_int = int(existing_updated_at_ms) if existing_updated_at_ms is not None else None
-        if existing_updated_at_int is not None and incoming_freshness_ms < existing_updated_at_int:
+        if should_skip_stale_replace(
+            incoming_freshness_ms=incoming_freshness_ms,
+            existing_updated_at_ms=existing_updated_at_int,
+        ):
             return ArchiveRawParsedWriteResult(
                 raw_id=raw_id,
                 session_id=session_id,

polylogue/storage/sqlite/archive_tiers/write.py

@@ -61,6 +61,7 @@
 from polylogue.storage.search.query_support import normalize_fts5_query
 from polylogue.storage.sqlite.action_pairs import refresh_action_pairs
 from polylogue.storage.sqlite.archive_tiers import archive_tiers_specs
+from polylogue.storage.sqlite.archive_tiers.ingest_precedence import should_skip_stale_replace
 from polylogue.storage.sqlite.archive_tiers.session_annotations_write import (
     ArchiveSessionPhase,
     ArchiveSessionTag,
@@ -417,7 +418,10 @@ def add_timing(name: str, started_at: float) -> None:
             (session_id,),
         ).fetchone()
         existing_updated_at_ms = int(row[0]) if row is not None and row[0] is not None else None
-        if existing_updated_at_ms is not None and incoming_freshness_ms < existing_updated_at_ms:
+        if should_skip_stale_replace(
+            incoming_freshness_ms=incoming_freshness_ms,
+            existing_updated_at_ms=existing_updated_at_ms,
+        ):
             add_timing("index.skip_stale_replace", t0)
             return session_id
     event_duplicate_message_native_ids = _duplicate_message_native_ids(messages)