Read the analysis cache before refusing a pass, and own the cache root

Opening a recording while the session that recorded it was still live never reached
``stored_result()``, so the panel reported no result over a recording whose result was on
disk. The guard now follows the cache read, results move to the app's derived-data root
(swept at startup, the old in-recording root ignored), and the panel states which of the
four no-result states it is in, with the pass's own reason.
Verified: apps/studio 462/0/18, shared 431/0/1, autocal 358, ui_check 628/0, both gates
clean, real-GL snapshots per state.

shared/tests/test_replay_analysis.py

@@ -32,11 +32,14 @@ import numpy as np
 import pytest
 from orisys_app_shared.analysis import model as analysis_model
 from orisys_app_shared.analysis.cache import (
+    DEFAULT_INCOMPLETE_GRACE_S,
     INCOMPLETE_MARKER,
+    LEGACY_RESULT_DIRNAME,
     META_NAME,
     ResultCache,
     evict,
     scan_results,
+    sweep,
 )
 from orisys_app_shared.analysis.model import (
     FRAME_HEADER,
@@ -624,6 +627,144 @@ def test_eviction_removes_whole_results_against_the_cap(tmp_path):
     assert kept.is_dir()
 
 
+# ------------------------------------------------- cache root and lifetime policy
+def _write_result(root: Path, name: str, *, complete: bool, modified: float, blob: int = 4096):
+    """One result directory as the writer leaves it, with a chosen age."""
+    directory = root / name
+    directory.mkdir(parents=True)
+    (directory / META_NAME).write_text(
+        json.dumps({"complete": complete, "key": name, "format": "x"}), encoding="utf-8"
+    )
+    (directory / "blob.bin").write_bytes(b"\0" * blob)
+    if not complete:
+        (directory / INCOMPLETE_MARKER).write_text("{}", encoding="utf-8")
+    for path in directory.rglob("*"):
+        os.utime(path, (modified, modified))
+    os.utime(directory, (modified, modified))
+    return directory
+
+
+def test_the_cache_root_is_the_app_s_own_and_never_the_recording_s(tmp_path):
+    """Results go to the app's derived-data root; a recording folder is never written.
+
+    This is the owner's constraint made checkable: the recording a result describes is
+    an input, and an input directory must come back byte-identical. It is also what makes
+    the cache findable at all — the *key* is path-independent, so one root for every
+    recording serves a copy exactly as it serves the original, while a root inside the
+    recording only ever served the one folder it was written beside.
+    """
+    from orisys_app_shared import paths as paths_module
+
+    assert paths_module.analysis_cache_dir() == (
+        paths_module.derived_data_dir() / paths_module.ANALYSIS_CACHE_DIRNAME
+    )
+    assert paths_module.ANALYSIS_CACHE_DIRNAME in str(paths_module.analysis_cache_dir())
+    # The recording's own folder is not the root, and ``for_recording`` does not put it
+    # back: the recording is named for the key, not for the layout.
+    recording = tmp_path / "recording"
+    recording.mkdir()
+    cache = ResultCache.for_recording(recording)
+    assert cache.root != recording / LEGACY_RESULT_DIRNAME
+    assert not str(cache.root).startswith(str(recording))
+    assert cache.directory(result_key(
+        recording="r", sensor_config="s", calibration="c"
+    )).parent == cache.root
+    # …and the root it defaults to is the OS one, resolved at construction so a caller
+    # can point it elsewhere without patching this class.
+    assert ResultCache().root == paths_module.analysis_cache_dir()
+
+
+def test_a_result_under_the_old_root_is_ignored_not_migrated(tmp_path):
+    """A leftover ``<recording>/analysis`` neither breaks an open nor gets touched.
+
+    The old root is a *recording the operator owns*, so the app may neither read it
+    (the path is not the identity) nor delete it (it is not the app's data). A result
+    for the same input under the app's own root is what is served, and the old directory
+    is found exactly as it was left.
+    """
+    from orisys_app_shared.analysis.model import result_key
+
+    recording = tmp_path / "recording"
+    legacy = recording / LEGACY_RESULT_DIRNAME
+    legacy.mkdir(parents=True)
+    key = result_key(recording="r", sensor_config="s", calibration="c")
+    (legacy / key.digest).mkdir()
+    (legacy / key.digest / META_NAME).write_text(
+        json.dumps({"complete": True, "key": key.digest}), encoding="utf-8"
+    )
+    before = sorted(str(path.relative_to(recording)) for path in recording.rglob("*"))
+
+    app_root = tmp_path / "app-cache"
+    app_root.mkdir()
+    cache = ResultCache(app_root)
+    assert cache.exists(key) is False, "a result beside the recording answered for it"
+    _write_result(app_root, key.digest, complete=True, modified=1_600_000_000)
+    assert cache.exists(key) is True
+
+    # A sweep of the app root does not reach into the recording: the old result and the
+    # recording's own files are all still there, untouched.
+    assert sweep(app_root, max_bytes=0) is not None
+    assert legacy.is_dir()
+    assert sorted(
+        str(path.relative_to(recording)) for path in recording.rglob("*")
+    ) == before
+
+
+def test_the_sweep_removes_unfinished_results_but_only_after_the_grace(tmp_path):
+    """Garbage first: an unreadable result goes once it is old, and not before.
+
+    An incomplete result can never be served (Part A §7), so it is the cheapest thing to
+    lose — but it is also the only evidence of how far a cancelled pass got, which is why
+    the age test exists rather than "delete on sight".
+    """
+    root = tmp_path / "cache"
+    root.mkdir()
+    now = 1_700_000_000.0
+    stale = _write_result(root, "stale", complete=False, modified=now - 7200)
+    fresh = _write_result(root, "fresh", complete=False, modified=now - 30)
+    finished = _write_result(root, "finished", complete=True, modified=now - 7200)
+
+    report = sweep(root, incomplete_grace_s=DEFAULT_INCOMPLETE_GRACE_S, now=now)
+    assert report.removed == [stale], "the sweep took something it should not have"
+    assert report.incomplete == [stale]
+    assert stale.exists() is False
+    assert fresh.is_dir(), "a pass that may still be running lost its directory"
+    assert finished.is_dir(), "a readable result was removed while it fits the cap"
+    assert report.freed_bytes > 0
+
+
+def test_the_sweep_protects_what_is_in_use_and_honours_the_cap(tmp_path):
+    """Then the cap: whole results, oldest first, and never the one being served."""
+    root = tmp_path / "cache"
+    root.mkdir()
+    now = 1_700_000_000.0
+    made = [
+        _write_result(root, f"result-{index}", complete=True, modified=now - 100 + index)
+        for index in range(3)
+    ]
+    kept = made[2]
+    size = sum(path.stat().st_size for path in kept.rglob("*") if path.is_file())
+    # A cap that fits only the protected result: both others must go, and the protected
+    # one must survive even when the arithmetic would prefer it (it is the newest).
+    report = sweep(root, max_bytes=size, keep=[kept], now=now)
+    assert kept.is_dir()
+    assert made[0].exists() is False and made[1].exists() is False
+    assert set(report.removed) == {made[0], made[1]}
+    assert report.incomplete == []
+    assert report.remaining_bytes <= size + 64
+
+    # A dry run removes nothing and reports what would go.
+    again = _write_result(root, "result-3", complete=True, modified=now)
+    plan = sweep(root, max_bytes=1, keep=[kept], now=now, dry_run=True)
+    assert again.is_dir() and kept.is_dir()
+    assert set(plan.removed) == {again}
+
+    # A root that does not exist is not an error: the first run of an app has no cache.
+    missing = sweep(tmp_path / "absent", now=now)
+    assert missing.removed == [] and missing.remaining_bytes == 0
+
+
+
 # ------------------------------------------------------------------ incompleteness gate
 def test_a_result_without_metadata_is_never_served(tmp_path):
     """The only thing that opens is a directory whose metadata says it finished."""