orisys-app/apps/studio/tests/test_replay_wiring.py

"""Part B's wiring: the cached analysis views over a recording, in the running app.

The result contract ([replay-analysis.md](../docs/replay-analysis.md)), the playback
contract ([replay-analysis-part-b.md](../docs/replay-analysis-part-b.md)) and the panel
contract ([replay-panel.md](../docs/replay-panel.md)) all describe one state: a recording
is open, the cursor is on the recording's own signal time, and the arrows and the depth
map are served from the stored analysis result while the operator scrubs and plays — with
the recording's own picture one step away in the mode selector, and the recompute switch
left as the explicit "run the chain over this footage again" action.

What was missing was the wiring that reaches it. ``ReplayEngine.use_recording()`` had no
caller, so analysis-on-open moved the cursor onto the *result's* timeline instead: the
derived views were drawn, but the recording's own picture was not offered at all, and the
cursor could not leave the frames the result covers. These tests drive the window's own
「打开录像」 action over a recording whose result is already stored beside it — the result
is generated here into the recording's own cache, so the app *reuses* it and opens no SDK
chain — and assert on the picture the panel actually shows, not on a state a harness
built.

The recording is copied before use: a pass writes its result into the recording's own
directory, and nothing may write into ``dist/``.
"""

from __future__ import annotations

import shutil
import time
from pathlib import Path

import numpy as np
import pytest
from orisys_app_shared.acquisition.arrows import depth_map_to_bgr, flow_colormap
from orisys_app_shared.acquisition.image_backend import (
    ARROW_PURE,
    ARROW_STITCHED,
    DEPTH,
    DERIVED_MODES,
    RECORDED,
    plan_image,
)
from orisys_app_shared.i18n import t
from orisys_app_shared.scene3d.contracts import DisplaySettings
from orisys_app_shared.scene3d.visual_policy import (
    select_flow_samples,
)
from orisys_studio.replay import MODE_RECORDING
from PySide6.QtWidgets import QApplication

REPO = Path(__file__).resolve().parents[3]
STUDIO = REPO / "dist" / "OrisysStudio"
SOURCE_RECORDING = (
    STUDIO
    / "outputs"
    / "recordings"
    / "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
)
#: What makes a directory a recording for the application, and all a replay reads.
RECORDING_FILES = (
    "frames.csv",
    "forces.csv",
    "events.csv",
    "session.json",
    "raw_0001.mp4",
    "preroll_0001.mp4",
)


def _require_inputs() -> None:
    missing = [
        str(SOURCE_RECORDING / name)
        for name in ("frames.csv", "raw_0001.mp4", "session.json")
        if not (SOURCE_RECORDING / name).exists()
    ]
    if missing:
        pytest.skip(
            "the replay wiring tests need the evidence recording: " + ", ".join(missing)
        )
    pytest.importorskip("orisys", reason="the analysis pass drives the SDK chain")


def _copy_recording(destination: Path) -> Path:
    """The evidence recording, copied: nothing may write into ``dist/``."""
    destination.mkdir(parents=True, exist_ok=True)
    for name in RECORDING_FILES:
        source = SOURCE_RECORDING / name
        if source.exists():
            shutil.copy2(source, destination / name)
    return destination


def _run_the_pass(recording: Path, root: Path | None = None):
    """One real pass, with the chain the *application* derives from the recording.

    ``root`` names the cache explicitly because the app's own root is the user's
    derived-data directory: a test must never write there, and the window it then opens
    must find the result, so the root is patched to a temporary one for the whole test.
    """
    from orisys_app_shared.analysis.cache import ResultCache
    from orisys_app_shared.analysis.service import AnalysisService, chain_spec_for
    from orisys_app_shared.record import RecordingReader

    reader = RecordingReader(recording)
    try:
        chain = chain_spec_for(reader.session())
    finally:
        reader.close()
    cache = None if root is None else ResultCache(root)
    return AnalysisService(recording, chain=chain, cache=cache).run()


@pytest.fixture(scope="module")
def cache_root(tmp_path_factory):
    """The analysis cache for this module's tests, in place of the user's own.

    The app resolves its cache root at construction from
    ``paths.analysis_cache_dir()``, so pointing that at a temporary directory is enough
    for the *window* to write and read there — which is what keeps a test from sweeping
    or filling the real ``%LOCALAPPDATA%`` cache, and what makes "the window found the
    stored result" a real statement rather than a coincidence of two roots.
    """
    from orisys_app_shared import paths as paths_module

    root = tmp_path_factory.mktemp("replay-wiring-cache") / "analysis-cache"
    root.mkdir(parents=True)
    patch = pytest.MonkeyPatch()
    patch.setattr(paths_module, "analysis_cache_dir", lambda: root)
    yield root
    patch.undo()


@pytest.fixture(scope="module")
def stored_recording(tmp_path_factory, cache_root) -> Path:
    """A copy of the evidence recording whose cache already holds a result."""
    _require_inputs()
    recording = _copy_recording(tmp_path_factory.mktemp("replay-wiring") / "recording")
    outcome = _run_the_pass(recording, cache_root)
    assert outcome.complete, outcome.reason
    assert outcome.frames_written > 0
    return recording


@pytest.fixture
def window(make_window, qt_app, cache_root):
    """The application, in the replay view, as the operator meets it.

    It shares the module's cache root, so a test may pre-place a result (or reuse the one
    ``stored_recording`` wrote) and have the window find it — which is the whole point of
    a root that is not inside the recording: the same recording is the same result.
    """
    return _replay_window(make_window)


@pytest.fixture
def fresh_cache(tmp_path, monkeypatch):
    """A cache root of this test's own.

    A copy of the recording *is* the same recording to the cache (the key carries the
    identity, not the path), so a test that wants a pass to run needs a root nothing has
    written to — otherwise it would be served another test's result, which is correct
    behaviour and the wrong fixture.
    """
    from orisys_app_shared import paths as paths_module

    root = tmp_path / "analysis-cache"
    root.mkdir(exist_ok=True)
    monkeypatch.setattr(paths_module, "analysis_cache_dir", lambda: root)
    return root


@pytest.fixture
def own_window(make_window, qt_app, fresh_cache):
    """A window whose analysis cache belongs to this test alone."""
    return _replay_window(make_window)


def _replay_window(make_window):
    window = make_window(sensor_type="box", source_mode="camera")
    window.resize(1440, 880)
    window.show()
    QApplication.processEvents()
    window.brand_bar.btn_playback.click()
    QApplication.processEvents()
    assert window.layout_ctl.in_replay_view(), "the window did not switch to the replay view"
    return window


def _open(window, recording: Path, monkeypatch) -> None:
    """Press the window's own 「打开录像」 entry, with the file dialog answered for it."""
    import orisys_studio.replay_controller as replay_controller

    video = Path(recording) / "raw_0001.mp4"
    monkeypatch.setattr(
        replay_controller.QFileDialog,
        "getOpenFileName",
        lambda *args, **kwargs: (str(video), ""),
    )
    action = window.replay_ctl._actions.playback
    assert action.isEnabled(), "the replay entries are not offered in the replay view"
    action.trigger()
    QApplication.processEvents()
    assert window.replay.has_video(), "the recording did not open"


def _sink_frames(window, monkeypatch) -> list:
    """Every frame handed to the window's single 2D sink, in order."""
    captured: list = []
    original = window._show_image_frame

    def spy(frame):
        captured.append(frame)
        return original(frame)

    monkeypatch.setattr(window, "_show_image_frame", spy)
    return captured


def _panel_bgr(window) -> np.ndarray:
    """The 2D panel's picture, in BGR, from the end of the real display path."""
    image = window.arrow_panel._source_image
    assert image is not None and not image.isNull(), "the 2D panel shows no picture"
    width, height = image.width(), image.height()
    rows = np.frombuffer(bytes(image.constBits()), dtype=np.uint8)
    rows = rows.reshape(height, image.bytesPerLine())
    return rows[:, : width * 3].reshape(height, width, 3)[:, :, ::-1].copy()


def _expected_depth(window, frame) -> np.ndarray:
    """A stored frame's depth view, drawn the way the window's own sink draws it."""
    return depth_map_to_bgr(
        frame.depth,
        flow_colormap(window._theme),
        value_range=window._depth_range(),
        invalid_mask=window._domain_mask.mask,
    )


def _set_mode(window, mode: str) -> None:
    index = window.arrow_panel.view_combo.findData(mode)
    assert index >= 0, f"the mode {mode} is not offered: {window.arrow_panel.view_modes()}"
    window.arrow_panel.view_combo.setCurrentIndex(index)
    QApplication.processEvents()
    assert window.arrow_panel.view_mode() == mode


def _pump(seconds: float) -> None:
    """Let real time pass so the playback timer fires, as it does in the app."""
    deadline = time.perf_counter() + float(seconds)
    while time.perf_counter() < deadline:
        QApplication.processEvents()
        time.sleep(0.01)


def _cells(rows, cols) -> set:
    return {(int(row), int(col)) for row, col in zip(np.asarray(rows), np.asarray(cols))}


def _entry_set(directory: Path) -> set:
    """Every entry under ``directory``, as (relative path, size).

    The owner's constraint is about the *recording*, so it is asserted on the recording's
    own file set: a result written beside the footage changes it, and a size is included
    so an in-place rewrite cannot pass as "the same entries".
    """
    found = set()
    for path in Path(directory).rglob("*"):
        relative = str(path.relative_to(directory))
        found.add((relative, -1 if path.is_dir() else path.stat().st_size))
    return found


def _active_cells(samples) -> set:
    return {
        (int(row), int(col))
        for row, col, active in zip(
            np.asarray(samples.rows), np.asarray(samples.cols), np.asarray(samples.active)
        )
        if bool(active)
    }


def test_the_operator_sees_the_cached_views_over_the_recording(
    stored_recording, window, monkeypatch
):
    """The state the contracts describe, reached the way an operator reaches it.

    A recording with a result beside it opens on the **recording's** timeline (the result
    is a cache behind the cursor, not a second timeline), the panel offers the recording's
    own picture *and* the three derived views, and what is on screen is the stored frame —
    not the camera, and not a placeholder.
    """
    captured = _sink_frames(window, monkeypatch)
    _open(window, stored_recording, monkeypatch)

    assert window.replay.mode() == MODE_RECORDING, (
        "the cursor left the recording's own timeline; the result is a cache behind it"
    )
    session = window.replay_ctl.available_result()
    assert session is not None, "the stored result beside the recording was not adopted"
    assert not window.replay_ctl.recompute_active(), (
        "the cache was only reachable by turning the recompute switch on"
    )
    assert window.analysis_progress.state() == "reused", (
        "a pass ran over a recording that already had a result"
    )

    assert window.arrow_panel.view_modes() == (
        RECORDED,
        ARROW_PURE,
        ARROW_STITCHED,
        DEPTH,
    ), "the recording's own picture must stay one step away from the derived views"
    assert window.arrow_panel.view_mode() in DERIVED_MODES, (
        "the operator was left on the camera picture although a result is served"
    )
    assert window.arrow_panel.note_label.text() == "", (
        "the panel says there is nothing to draw while it is drawing a result"
    )

    # The arrow view is the stored measurement, at the cursor's own frame.
    frames = list(session.frames())
    active_frame = next(frame for frame in frames if frame.active_count > 0)
    window.replay.set_position(session.time_of(active_frame.frame_id))
    QApplication.processEvents()
    assert captured, "nothing reached the 2D sink"
    drawn = captured[-1]
    assert drawn.frame_id == active_frame.frame_id
    assert drawn.result_available is True
    stored_cells = _cells(active_frame.samples.rows, active_frame.samples.cols)
    field = np.asarray(drawn.flow)
    finite = np.argwhere(np.isfinite(field[..., 0]))
    assert _cells(finite[:, 0], finite[:, 1]) == stored_cells, (
        "the recovered flow field carries values outside the stored candidate locations"
    )
    # The height last: the shared selector re-reads the same cells, in its own order, and
    # comes to the same activity verdict as the pass stored.
    reselected = select_flow_samples(field, drawn.valid_mask, session.display_policy)
    assert _cells(reselected.rows, reselected.cols) == stored_cells
    assert _active_cells(reselected) == _active_cells(active_frame.samples)
    arrows_bgr = _panel_bgr(window)
    raw_here = window.replay.video_frame_bgr()
    assert raw_here is not None
    assert not np.array_equal(arrows_bgr, np.asarray(raw_here)), (
        "the arrow view is the recording's raw picture"
    )

    # The depth view is the stored map, drawn through the window's own backend. This
    # result's map is exactly zero on every frame it holds — the box chain's depth is
    # flat on this footage — and it is still a stored array of the result's own shape,
    # so the pixels are checked against it rather than against a flat placeholder.
    # The colour-scale caption names the range only.
    _set_mode(window, DEPTH)
    target = active_frame
    assert target.depth is not None and target.depth_nonzero == 0
    QApplication.processEvents()
    drawn = captured[-1]
    assert drawn.frame_id == target.frame_id
    assert np.array_equal(
        np.asarray(drawn.depth), np.asarray(target.depth)
    ), "the depth view is not the stored map"
    shown = _panel_bgr(window)
    expected = _expected_depth(window, target)
    assert expected is not None
    assert shown.shape == expected.shape
    assert np.array_equal(shown, expected), (
        "the depth view's pixels are not depth_map_to_bgr of the stored frame"
    )
    assert not np.array_equal(shown, arrows_bgr), "the depth view drew the arrow view"
    assert window.arrow_panel.depth_scale_label.text() == (
        window.frame_view.depth_scale_text(target)
    ), "the depth caption is not the colour-scale range"

    # …and the recording's own picture is still reachable, at the same cursor.
    _set_mode(window, RECORDED)
    raw = window.replay.video_frame_bgr()
    assert raw is not None, "the recording's own frame could not be decoded"
    assert np.array_equal(_panel_bgr(window), np.asarray(raw)), (
        "the recorded view is not the recording's own picture"
    )
    assert window.arrow_panel.note_label.text() == ""
    assert window.replay.mode() == MODE_RECORDING


def test_playing_forward_and_seeking_draw_the_same_cached_frame(
    stored_recording, window, monkeypatch
):
    """A seek is the frame playing forward passes through — on the operator's path."""
    captured = _sink_frames(window, monkeypatch)
    _open(window, stored_recording, monkeypatch)
    session = window.replay_ctl.available_result()
    assert session is not None
    _set_mode(window, DEPTH)

    ids = [int(value) for value in session.frame_ids]
    target_id = ids[len(ids) // 4]
    moment = session.time_of(target_id)
    assert moment is not None

    # Play forward from the top, a tick's worth of cursor at a time, ending on the
    # target instant itself.
    positions = [step / 240.0 for step in range(int(moment * 240.0) + 1)]
    positions.append(moment)
    for position in positions:
        window.replay.set_position(position)
    played = _panel_bgr(window)
    played_id = captured[-1].frame_id

    # …then seek straight to the same instant.
    window.replay.set_position(moment)
    QApplication.processEvents()
    assert captured[-1].frame_id == played_id == target_id
    assert np.array_equal(played, _panel_bgr(window)), (
        "a seek drew a different frame from playing forward to the same position"
    )

    # Stepping one stored frame on moves to that frame's own picture.
    later = ids[ids.index(target_id) + 1]
    window.replay.set_position(session.time_of(later))
    QApplication.processEvents()
    assert captured[-1].frame_id == later
    assert np.array_equal(
        np.asarray(captured[-1].depth), np.asarray(session.frame(later).depth)
    )


def test_playback_draws_the_cached_frame_not_the_video(
    stored_recording, window, monkeypatch
):
    """Advance the real playback timer and check every drawn frame against the cache."""
    captured = _sink_frames(window, monkeypatch)
    _open(window, stored_recording, monkeypatch)
    session = window.replay_ctl.available_result()
    assert session is not None
    _set_mode(window, DEPTH)

    window.replay.set_position(0.0)
    QApplication.processEvents()
    captured.clear()
    window.replay_ctl.start_playback()
    assert window.replay_ctl.is_playing(), "playback did not start"
    _pump(0.4)
    position = window.replay.position()
    assert position > 0.0, "the playhead did not advance"
    assert captured, "playback drew nothing"
    assert position <= session.duration_s(), "the playhead left the recording's timeline"

    expected_id = session.frame_id_at(position)
    assert expected_id is not None
    for frame in captured:
        assert frame.depth is not None, "a frame drawn during playback carries no depth"
        assert frame.result_available is True
        assert session.has(int(frame.frame_id)), (
            f"playback drew frame {frame.frame_id}, which the result does not hold"
        )
    # Pausing leaves the frame the cursor is on, and it is the cached one.
    window.replay_ctl.pause_playback()
    assert not window.replay_ctl.is_playing()
    assert captured[-1].frame_id == expected_id
    expected = _expected_depth(window, session.frame(expected_id))
    assert expected is not None
    assert np.array_equal(_panel_bgr(window), expected), (
        "the paused frame is not the stored frame for the cursor"
    )


def test_a_finished_pass_puts_the_cache_on_screen(tmp_path, own_window, monkeypatch, fresh_cache):
    """The other trigger: the pass completes and the panel serves what it stored.

    Nothing is pre-placed here, so the window runs the pass over the copy (a box chain:
    a couple of seconds for this recording) and the *completion* is what has to leave the
    derived view on screen — with the cursor still on the recording.
    """
    window = own_window
    recording = _copy_recording(tmp_path / "recording")
    before = _entry_set(recording)
    _sink_frames(window, monkeypatch)
    _open(window, recording, monkeypatch)

    assert window.replay_ctl.wait_for_analysis(180.0), "the pass did not finish"
    QApplication.processEvents()
    assert window.replay_ctl.analysis_error() == "", (
        f"the pass left no usable result: {window.replay_ctl.analysis_error()}"
    )
    assert window.analysis_progress.state() == "complete"
    session = window.replay_ctl.available_result()
    assert session is not None, "the completed result was not adopted"
    assert window.replay.mode() == MODE_RECORDING, (
        "a finished pass moved the cursor off the recording's own timeline"
    )
    assert window.arrow_panel.view_mode() in DERIVED_MODES, (
        "the operator was left on the camera picture after the pass finished"
    )
    assert not window.replay_ctl.recompute_active()

    frame_id = session.frame_id_at(window.replay.position())
    assert frame_id is not None
    _set_mode(window, DEPTH)
    expected = _expected_depth(window, session.frame(frame_id))
    assert expected is not None
    assert np.array_equal(_panel_bgr(window), expected), (
        "the finished result is not what the panel draws"
    )

    # The owner's constraint, asserted where it can actually be broken: the recording
    # directory came back exactly as it was, and the result went to the app's own cache.
    assert _entry_set(recording) == before, (
        "opening and analysing a recording wrote inside the recording folder"
    )
    assert not (recording / "analysis").exists()
    assert Path(session.directory).parent == Path(fresh_cache), (
        f"the result was not written to the app's cache root: {session.directory}"
    )


def test_a_running_session_does_not_hide_a_stored_result(
    stored_recording, window, monkeypatch
):
    """A live session must not cost the operator the cached views.

    The operator's own sequence: record something, and open the recording while the
    acquisition session is still live (paused sessions keep the device, so they count).
    Reading a stored result opens no SDK chain, so the session is no reason to refuse it —
    and refusing it left the panel saying "this recording has no analysis result" over a
    recording whose arrows were in the cache, which is the report this pins. Before the
    fix: the note was shown and the mode selector offered ``recorded`` alone.
    """
    import ui_check

    window._ensure_workers = lambda: None
    stuck = ui_check._StuckWorker()
    window._box_worker = stuck
    window._finger_worker = stuck
    window._start_run()
    window._run_timer.stop()
    window._set_phase("running")
    QApplication.processEvents()
    assert window._running and window._session_active, "no session was started"

    captured = _sink_frames(window, monkeypatch)
    _open(window, stored_recording, monkeypatch)

    session = window.replay_ctl.available_result()
    assert session is not None, (
        "a running session hid a stored result the app did not have to compute"
    )
    assert window.analysis_progress.state() == "reused", "a pass ran beside a session"
    assert window.replay.mode() == MODE_RECORDING
    assert window.arrow_panel.note_label.text() == "", (
        "the panel says it has no result while it is drawing one"
    )
    _set_mode(window, DEPTH)
    frame_id = session.frame_id_at(window.replay.position())
    expected = _expected_depth(window, session.frame(frame_id))
    assert expected is not None
    assert np.array_equal(_panel_bgr(window), expected), (
        "the cache was not what the panel drew"
    )
    assert captured[-1].result_available is True

    window.replay_ctl.stop_activity()
    stuck.release()


def test_the_mode_selector_is_the_trigger_for_a_derived_view(
    stored_recording, window, monkeypatch
):
    """On the recording's timeline, choosing a derived view serves it from the cache."""
    captured = _sink_frames(window, monkeypatch)
    _open(window, stored_recording, monkeypatch)
    session = window.replay_ctl.available_result()
    assert session is not None
    _set_mode(window, RECORDED)

    position = session.time_of(int(session.frame_ids[len(session.frame_ids) // 2]))
    assert position is not None
    window.replay.set_position(position)
    QApplication.processEvents()
    assert window.replay.mode() == MODE_RECORDING
    assert captured[-1].depth is None, "the recorded view drew a stored frame"

    captured.clear()
    _set_mode(window, DEPTH)
    assert captured, "choosing a derived view drew nothing"
    assert captured[-1].depth is not None, (
        "a view that needs derived data drew the recording's raw picture"
    )
    assert captured[-1].frame_id == session.frame_id_at(position)
    assert window.replay.mode() == MODE_RECORDING, (
        "choosing a view moved the cursor's timeline"
    )


def test_the_display_settings_still_redraw_from_the_cache(
    stored_recording, window, monkeypatch, cache_root
):
    """A threshold change is a read: it redraws the stored frame and writes nothing."""
    from orisys_app_shared.analysis.cache import scan_results

    _open(window, stored_recording, monkeypatch)
    session = window.replay_ctl.available_result()
    assert session is not None
    _set_mode(window, ARROW_PURE)

    before = {entry.directory for entry in scan_results(cache_root)}
    captured = _sink_frames(window, monkeypatch)
    window.replay_ctl.on_display_settings(DisplaySettings(flow_on_px=0.9))
    QApplication.processEvents()

    assert session.threshold() == pytest.approx(0.9), (
        "the display threshold never reached the session"
    )
    assert captured, "a display change did not redraw the frame"
    assert captured[-1].frame_id == session.frame_id_at(window.replay.position())
    assert {entry.directory for entry in scan_results(cache_root)} == before, (
        "a display change wrote an analysis result"
    )


def test_a_setting_that_moves_the_grid_is_still_refused(
    stored_recording, window, monkeypatch, cache_root
):
    """The stored candidate locations cannot be re-gridded; the frame is drawn anyway."""
    from orisys_app_shared.analysis.cache import scan_results

    _open(window, stored_recording, monkeypatch)
    session = window.replay_ctl.available_result()
    assert session is not None
    _set_mode(window, ARROW_PURE)
    before_threshold = session.threshold()
    before = {entry.directory for entry in scan_results(cache_root)}

    captured = _sink_frames(window, monkeypatch)
    window.replay_ctl.on_display_settings(DisplaySettings(max_arrows=50))
    QApplication.processEvents()

    assert session.threshold() == pytest.approx(before_threshold), (
        "a refused policy changed the session"
    )
    assert captured, "the frame was not redrawn from the result's own grid"
    assert captured[-1].frame_id == session.frame_id_at(window.replay.position())
    assert {entry.directory for entry in scan_results(cache_root)} == before


# --------------------------------------------------- what the panel says without one
def test_the_panel_says_the_pass_is_running(tmp_path, own_window, monkeypatch):
    """While the pass runs, the line must not tell the operator to start one.

    The old line promised "opening a recording runs a pass" *during* the pass, which is
    worse than useless on a ten-minute recording: it reads as "nothing is happening".
    """
    from orisys_app_shared.acquisition.image_backend import RESULT_RUNNING
    from orisys_app_shared.i18n import t

    window = own_window
    recording = _copy_recording(tmp_path / "recording")
    captured = _sink_frames(window, monkeypatch)
    _open(window, recording, monkeypatch)

    assert window.replay_ctl.analysis_running(), "the pass finished before it was observed"
    assert window.replay_ctl.result_state() == (RESULT_RUNNING, "")
    assert captured[-1].result_state == RESULT_RUNNING
    assert window.arrow_panel.note_label.text() == t("panels.replay_analysis_running"), (
        "the panel did not say the pass is running"
    )
    assert window.replay_ctl.wait_for_analysis(180.0)
    QApplication.processEvents()
    assert window.arrow_panel.note_label.text() == "", (
        "the completed pass left a note behind"
    )


def test_the_panel_states_a_failed_pass_with_its_reason(
    tmp_path, own_window, monkeypatch, fresh_cache
):
    """A pass that ends without a result says so, with the pass's own reason.

    Reached through the window's own open action: only the chain is made unopenable (a
    configuration file that is not there — the shape of a moved install), so the worker
    thread, the completion slot and the panel are all the real ones. The check that
    mattered in the report is the last one: the panel and the progress surface agree.
    """
    from orisys_app_shared.acquisition.image_backend import (
        RAW_NOTE_KEY,
        RESULT_FAILED,
    )
    from orisys_app_shared.analysis.cache import ResultCache
    from orisys_app_shared.analysis.service import AnalysisService, ChainSpec
    from orisys_app_shared.i18n import t

    window = own_window

    def broken_service(recording, cancelled=None, *, on_progress=None):
        return AnalysisService(
            recording,
            chain=ChainSpec(sensor_type="box", config_ref=tmp_path / "absent-sensor.json"),
            cache=ResultCache(fresh_cache),
            cancelled=cancelled,
            on_progress=on_progress,
        )

    monkeypatch.setattr(window, "_analysis_service", broken_service)
    recording = _copy_recording(tmp_path / "recording")
    _open(window, recording, monkeypatch)
    assert window.replay_ctl.wait_for_analysis(180.0), "the failing pass never finished"
    QApplication.processEvents()

    state, reason = window.replay_ctl.result_state()
    assert reason, "a failed pass left no reason to show"
    assert state == RESULT_FAILED, f"the failed pass is reported as {state!r}"
    assert window.analysis_progress.state() == "failed"
    assert window.arrow_panel.note_label.text() == t(
        "panels.replay_analysis_failed", reason=reason
    ), "the panel's line did not carry the pass's reason"
    assert reason in window.arrow_panel.note_label.text()
    assert t(RAW_NOTE_KEY) not in window.arrow_panel.note_label.text(), (
        "the panel promised the automatic pass that had just failed"
    )


def test_a_bare_video_with_a_declined_prompt_promises_no_pass(
    tmp_path, own_window, monkeypatch, decline_video_prompt
):
    """A bare video whose setup prompt was dismissed: playback, and one honest line.

    A bare video *can* be analysed now — that is what ``test_bare_video_analysis.py`` is
    about, and it asks the prompt and answers it. This test is the other answer: the operator
    opened the video and dismissed the question, so no pass is coming and the panel must not
    say one is. The line names the action instead (the setup prompt), which is what the
    "failed or refused, with its reason" state is for.
    """
    from orisys_app_shared.acquisition.image_backend import (
        RAW_NOTE_KEY,
        RESULT_FAILED,
    )
    from orisys_app_shared.i18n import t

    window = own_window
    recording = _copy_recording(tmp_path / "recording")
    clip = tmp_path / "clip.mp4"
    shutil.copy2(recording / "raw_0001.mp4", clip)

    captured = _sink_frames(window, monkeypatch)
    assert window.replay_ctl.load_video(clip) is True
    QApplication.processEvents()

    assert window.replay.is_video_only(), "the clip did not open as a bare video"
    assert not window.replay_ctl.analysis_running(), "a pass ran after a dismissed prompt"
    assert window.replay_ctl.result_state() == (
        RESULT_FAILED,
        t("playback.analysis_setup_declined"),
    )
    assert window.arrow_panel.note_label.text() == t(
        "panels.replay_analysis_failed", reason=t("playback.analysis_setup_declined")
    )
    assert t(RAW_NOTE_KEY) not in window.arrow_panel.note_label.text()
    assert plan_image(captured[-1]).note_key == "panels.replay_analysis_failed"