fix(observatory): preserve source identity on unchanged reindex

src/k1link/sessions/store.py

@@ -9,7 +9,7 @@ import shutil
 import sqlite3
 import stat
 import threading
-from collections.abc import Iterator
+from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any, Literal, cast
@@ -1118,10 +1118,16 @@ class SessionStore:
         with self._lock, self._connect() as connection:
             connection.execute("BEGIN IMMEDIATE")
             existing = connection.execute(
-                "SELECT plugin_id, archive_id, allowed_root, session_root, created_at_utc "
+                "SELECT * FROM observation_sessions WHERE session_id = ?",
-                "FROM observation_sessions WHERE session_id = ?",
                 (candidate.session_id,),
             ).fetchone()
+            previous_snapshot = (
+                None
+                if existing is None
+                else _catalog_snapshot_sha256_for_session(
+                    connection, candidate.session_id, session_row=existing
+                )
+            )
             unclaimed_pre_plugin_row = existing is not None and (
                 existing["plugin_id"] == "" and existing["archive_id"] == ""
             )
@@ -1271,6 +1277,20 @@ class SessionStore:
                     for source in sources
                 ],
             )
+            if existing is not None:
+                reconciled = dict(connection.execute(
+                    "SELECT * FROM observation_sessions WHERE session_id = ?",
+                    (candidate.session_id,),
+                ).fetchone())
+                reconciled["updated_at_utc"] = existing["updated_at_utc"]
+                if _catalog_snapshot_sha256_for_session(
+                    connection, candidate.session_id, session_row=reconciled
+                ) == previous_snapshot:
+                    # Discovery is not a source mutation. Preserve the exact
+                    # admitted snapshot when every session/source/artifact field
+                    # is identical; real changes still commit with the new clock.
+                    connection.rollback()
+                    return
             connection.commit()
 
     @contextmanager
@@ -1571,7 +1591,7 @@ def _catalog_snapshot_sha256_for_session(
     connection: sqlite3.Connection,
     session_id: str,
     *,
-    session_row: sqlite3.Row | None = None,
+    session_row: sqlite3.Row | Mapping[str, object] | None = None,
 ) -> str:
     row = (
         session_row
@@ -1597,7 +1617,7 @@ def _catalog_snapshot_sha256_for_session(
 
 
 def _catalog_snapshot_sha256(
-    session_row: sqlite3.Row,
+    session_row: sqlite3.Row | Mapping[str, object],
     source_rows: list[sqlite3.Row],
     artifact_rows: list[sqlite3.Row],
 ) -> str:

tests/test_observatory_portable_result_publisher.py

@@ -126,14 +126,13 @@ def _ready_registry(tmp_path: Path) -> PortableRunDefinitionRegistry:
     return PortableRunDefinitionRegistry.from_file(path)
 
 
-def _source_store(
+def _source_archive(
     tmp_path: Path,
     definition: PortableRunDefinition,
-) -> tuple[SessionStore, str, str, str]:
+) -> ObservationArchiveSource:
-    repository = tmp_path / "repository"
     archive_root = tmp_path / "source-archive"
     session_root = archive_root / SOURCE_SESSION_ID
-    session_root.mkdir(parents=True)
+    session_root.mkdir(parents=True, exist_ok=True)
     candidate = ObservationSessionCandidate(
         session_id=SOURCE_SESSION_ID,
         display_name="Portable result source",
@@ -152,12 +151,20 @@ def _source_store(
         sources=(),
         artifacts=(),
     )
-    archive = ObservationArchiveSource(
+    return ObservationArchiveSource(
         plugin_id=definition.source_requirements.plugin_id,
         archive_id=definition.source_requirements.archive_id,
         root=archive_root,
         discover=lambda _root: (candidate,),
     )
+
+
+def _source_store(
+    tmp_path: Path,
+    definition: PortableRunDefinition,
+) -> tuple[SessionStore, str, str, str]:
+    repository = tmp_path / "repository"
+    archive = _source_archive(tmp_path, definition)
     store = SessionStore(repository, data_dir=tmp_path / "mission-core-data")
     assert store.reconcile_archive(archive) == (SOURCE_SESSION_ID,)
     _detail, catalog_sha256 = store.get_session_with_catalog_snapshot(SOURCE_SESSION_ID)
@@ -738,6 +745,26 @@ def test_cache_does_not_hide_another_source_snapshot_or_profile_version(tmp_path
     assert sessions.get_lab_instance(RESULT_ID) is not None
 
 
+def test_cache_survives_new_store_and_archive_reconciliation(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T08:00:00.000Z")
+    cache, queue, sessions, artifacts, registry, definition, job, _, _ = _cache_fixture(tmp_path)
+    before = cache.find(SOURCE_SESSION_ID, definition)
+    assert before
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T09:00:00.000Z")
+    reopened = SessionStore(sessions.repository_root, data_dir=sessions.data_dir)
+    reopened.reconcile_archive(_source_archive(tmp_path, definition))
+    cold_cache = PortableResultCache(
+        sessions=reopened, artifacts=artifacts, queue=queue, definitions=registry,
+        calculation_profiles=PortableCalculationProfileRegistry((_profile(definition),)),
+    )
+    assert cold_cache.find(SOURCE_SESSION_ID, definition) == before
+    with pytest.raises(ObservatoryRecordedQueueDuplicateError):
+        queue.submit(_retry_intent(job), published_result_available=cold_cache.available)
+    assert len(queue.list_jobs()) == 1
+
+
 @pytest.mark.parametrize("field", ["executor_image_sha256", "source_bundle_sha256"])
 def test_queue_rejects_mutated_job_identity_before_cache_lookup(tmp_path: Path, field: str) -> None:
     cache, _, _, _, _, definition, job, _, _ = _cache_fixture(tmp_path)

tests/test_session_store.py

@@ -186,6 +186,62 @@ def test_evidence_root_is_private_and_configurable(
     assert resolve_missioncore_evidence_dir(repository) == configured.resolve()
 
 
+def test_unchanged_archive_preserves_exact_snapshot_across_reopen(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repository = tmp_path / "repo"
+    sessions = repository / "sessions"
+    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T08:00:00.000Z")
+    store = SessionStore(repository, data_dir=tmp_path / "data")
+    archive = xgrids_k1_archive_source(sessions)
+    store.reconcile_archive(archive)
+    before = store.get_session_with_catalog_snapshot(session.name)
+    with sqlite3.connect(store.database_path) as connection:
+        rows = tuple(connection.iterdump())
+
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T09:00:00.000Z")
+    reopened = SessionStore(repository, data_dir=store.data_dir)
+    for _ in range(2):
+        assert reopened.reconcile_archive(archive) == (session.name,)
+        assert reopened.get_session_with_catalog_snapshot(session.name) == before
+    with sqlite3.connect(store.database_path) as connection:
+        assert tuple(connection.iterdump()) == rows
+
+
+@pytest.mark.parametrize("table,assignment", [
+    ("observation_sessions", "total_bytes = total_bytes + 1"),
+    ("observation_sessions", "duration_seconds = duration_seconds + 1"),
+    ("observation_session_sources", "seekable = 0"),
+    ("observation_session_artifacts", "byte_length = byte_length + 1"),
+    ("observation_session_artifacts", "sha256 = NULL"),
+])
+def test_reconcile_retains_clock_change_for_real_catalog_differences(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, table: str, assignment: str,
+) -> None:
+    repository = tmp_path / "repo"
+    sessions = repository / "sessions"
+    session = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T08:00:00.000Z")
+    store = SessionStore(repository, data_dir=tmp_path / "data")
+    archive = xgrids_k1_archive_source(sessions)
+    store.reconcile_archive(archive)
+    original = store.get_session_with_catalog_snapshot(session.name)
+    with sqlite3.connect(store.database_path) as connection:
+        connection.execute(f"UPDATE {table} SET {assignment}")  # noqa: S608 - fixed test cases
+    changed = store.get_session_with_catalog_snapshot(session.name)
+    assert changed[1] != original[1]
+    monkeypatch.setattr("k1link.sessions.store.utc_now_iso", lambda: "2026-09-03T09:00:00.000Z")
+    store.reconcile_archive(archive)
+    after = store.get_session_with_catalog_snapshot(session.name)
+    assert after[1] not in (original[1], changed[1])
+    with sqlite3.connect(store.database_path) as connection:
+        assert connection.execute(
+            "SELECT updated_at_utc FROM observation_sessions WHERE session_id = ?",
+            (session.name,),
+        ).fetchone()[0] == "2026-09-03T09:00:00.000Z"
+
+
 def test_catalog_reconciles_sessions_removed_from_one_evidence_root(tmp_path: Path) -> None:
     repository = tmp_path / "repo"
     sessions = repository / "sessions"