diff --git a/src/k1link/sessions/store.py b/src/k1link/sessions/store.py index 8c1367f..e027d33 100644 --- a/src/k1link/sessions/store.py +++ b/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 " - "FROM observation_sessions WHERE session_id = ?", + "SELECT * 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: diff --git a/tests/test_observatory_portable_result_publisher.py b/tests/test_observatory_portable_result_publisher.py index 688d0d5..0691ff5 100644 --- a/tests/test_observatory_portable_result_publisher.py +++ b/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]: - repository = tmp_path / "repository" +) -> ObservationArchiveSource: 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) diff --git a/tests/test_session_store.py b/tests/test_session_store.py index 74c278e..7660408 100644 --- a/tests/test_session_store.py +++ b/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"