From ee05f55171e1b24d4f2043f1fd0fde9176d3f2fc Mon Sep 17 00:00:00 2001 From: Sinity Date: Fri, 31 Jul 2026 14:57:43 +0200 Subject: [PATCH] refactor(storage): consolidate the three skip-stale-replace tie-breaks Problem polylogue-t83e investigated 6 live session_id collisions between drive- cache Claude Code transcript re-uploads (a copy the operator uploaded into an AI Studio conversation, re-downloaded via Drive sync) and their genuinely local ~/.claude/projects/ counterparts, where the drive copy (a strict byte-prefix of the local file) was winning the archive row. Investigation traced this to a revision-arbitration bug, not an identity/detection bug: session_revision_membership.py's content-only set relation already resolves this correctly (verified by direct simulation against the real raw bytes with current code: relation computes a_contains_b, local strictly dominates). The live archive's raw_session_memberships rows for these 6 cohorts simply predate the fix that landed the same day (PR #3401/#3405, polylogue-aggz) and will self-heal via the daemon's automagic bulk rebuild path (daemon/bulk_rebuild.py -> rebuild_index_from_source -> backfill_historical_revision_evidence -> classify_membership_revisions) on the operator's already-planned `ops reset --index && polylogued run`. No SEMANTIC_REPARSE declaration applies -- no detection or parsing semantics changed. Two approaches were tried and reverted before reaching this conclusion (see polylogue-t83e's closing comment for the full trail): an OriginSpec artifact rule refusing session admission for drive-cache Claude-Code-shaped paths, and a matching ingest_worker.py path check. Both worked but were wrong: they would have permanently suppressed the 6 non-colliding drive-cache raws (real content with no local counterpart to supersede them) and added unneeded machinery for the 6 colliding ones, which the already-fixed revision-membership relation resolves once the data catches up. Solution What remained a real, independently-valuable defect: the freshness-tie "skip this write as stale" comparison was duplicated three times (archive_tiers/write.py, pipeline/services/ingest_batch/_core.py, archive_tiers/revision_governance.py) with identical logic (`existing_updated_at_ms is not None and incoming_freshness_ms < existing_updated_at_ms`) that could silently drift apart. Consolidated into one `should_skip_stale_replace()` in archive_tiers/ingest_precedence.py (the module that already owns browser-capture write precedence), called from all three sites. 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 the check applies at all, not the comparison itself. The docstring records that this is a narrow per-write timestamp fallback, not where content- subset supersession is decided (that's revision membership, upstream). Verification - devtools test tests/unit/pipeline/test_archive_write.py::test_older_full_replace_does_not_overwrite_newer_session_body tests/unit/pipeline/test_ingest_batch.py::test_write_session_force_write_replaces_older_freshness -- 2 passed - devtools test tests/unit/storage/test_revision_replay.py tests/unit/storage/test_raw_revision_authority.py tests/unit/sources/test_revision_backfill.py -- 103 passed, 1 failed (test_parse_one_still_replays_real_claude_code_sessions_with_no_path_rule -- pre-existing on origin/master, zero diff in the files it exercises, unrelated to this change: an analysis/-path content-gate regression, not touched here) - devtools verify --seed-testmon --skip-slow -- ruff format, ruff check, mypy --strict, render all --check, topology/layering/closure-matrix, schema-versioning policy, schema-promotion audit all green; the full pytest seed pass was still running a broad corpus sweep when the 580s budget for this verification pass was reached (unrelated to this change's narrow surface) Ref polylogue-t83e Co-Authored-By: Claude --- .../pipeline/services/ingest_batch/_core.py | 6 ++- .../sqlite/archive_tiers/ingest_precedence.py | 54 +++++++++++++++++++ .../archive_tiers/revision_governance.py | 6 ++- .../storage/sqlite/archive_tiers/write.py | 6 ++- 4 files changed, 69 insertions(+), 3 deletions(-) diff --git a/polylogue/pipeline/services/ingest_batch/_core.py b/polylogue/pipeline/services/ingest_batch/_core.py index 1f10dd712b..4cf203646f 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 ( @@ -544,7 +545,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)