diff --git a/polylogue/pipeline/services/ingest_batch/_core.py b/polylogue/pipeline/services/ingest_batch/_core.py index de68782f30..c7992d845e 100644 --- a/polylogue/pipeline/services/ingest_batch/_core.py +++ b/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 diff --git a/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py b/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py index 8b50502835..e9c0b5b156 100644 --- a/polylogue/storage/sqlite/archive_tiers/ingest_precedence.py +++ b/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, diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index 0ba602ce65..4b6350a861 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/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, diff --git a/polylogue/storage/sqlite/archive_tiers/write.py b/polylogue/storage/sqlite/archive_tiers/write.py index 9073e688fb..4113090b61 100644 --- a/polylogue/storage/sqlite/archive_tiers/write.py +++ b/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)