Reorient E2E plan toward live charts

Deats,
- park replay until a concrete regression needs controlled data
- prioritize nonblocking `QtractorSession` lifecycle ownership
- define isolated real Kraken chart and QtBot qualification
- record entry-point plugin migration for replay resumption

Prompt-IO: ai/prompt-io/opencode/20260903T192744Z_aadfd1cf_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

.claude/skills/run-tests/test-harness-reference.md

@@ -10,30 +10,34 @@ here.
 - Project/import: `piker`
 - Test root: `tests/`
 - Supported Python: `>=3.12,<3.14`
-- Preferred complete environment: worktree-local `py313` inside the current
+- Preferred verified environment: worktree-local `.venv`.
-  `nix develop` shell
+- The testing group includes pytest and the locked `pytest-qt==4.5.0`.
-- The flake shell pins CPython 3.13 and sets
+- Verify the interpreter, local package resolution, and dependency versions
-  `UV_PROJECT_ENVIRONMENT=py313`.
+  before running tests.
-- Verify the interpreter, package resolution, and dependency import before
-  running tests. A bare `py313` may lack the Qt binding supplied by Nix.
 
-Use an already-provisioned `py313` only when `import piker` succeeds:
+This worktree's environment was provisioned with:
 
 ```text
-py313/bin/python -m pytest -p no:xonsh
+env UV_PROJECT_ENVIRONMENT=.venv uv sync --group testing --frozen
 ```
 
-If direct environment paths are unavailable, an existing uv environment can
+Do not rerun provisioning without approval. Use the already-provisioned
-be used without changing it, subject to the same import check:
+environment directly:
 
 ```text
-UV_PROJECT_ENVIRONMENT=py313 uv run --frozen --no-sync python -m pytest -p no:xonsh
+.venv/bin/python -m pytest
 ```
 
-Ask before running provisioning commands such as:
+If direct environment paths are unavailable, use uv without changing the
+environment, subject to the same import check:
+
+```text
+env UV_PROJECT_ENVIRONMENT=.venv uv run --frozen --no-sync python -m pytest
+```
+
+Ask before entering or provisioning alternate environments such as:
 
 ```text
-UV_PROJECT_ENVIRONMENT=py313 uv sync --dev --all-extras --no-group lint
 nix develop
 nix-shell default.nix
 ```
@@ -42,43 +46,41 @@ nix-shell default.nix
 X11 shell. Do not use `develop.nix` for current testing; it retains the old
 Python 3.11, Poetry, and Qt 5 stack.
 
-The current root-checkout `py313` resolves `piker` locally but fails
+Do not enter `nix develop` without approval: its shell hook may provision its
-`import piker` outside Nix because PyQtGraph cannot import PyQt or PySide. Do
+own configured environment. Plain `uv sync` does not include the testing
-not treat that environment as test-ready and do not enter `nix develop`
+group; use `--group testing` when provisioning is explicitly authorized. The
-without approval: its shell hook may recreate and sync `py313`.
+`dbs` dependency group is also absent from normal dev-shell provisioning.
-
-Plain `uv sync` does not include the testing group. The `dbs` dependency group
-is also absent from normal dev-shell provisioning.
 
 ## Commands
 
 Base command in the preferred environment:
 
 ```text
-py313/bin/python -m pytest -p no:xonsh
+.venv/bin/python -m pytest
 ```
 
-The explicit `-p no:xonsh` is required. The tracked comments-only
+The comments-only `pytest.ini` has been deleted. The authoritative
-`pytest.ini` takes precedence over `pyproject.toml`, so the intended
+`[tool.pytest.ini_options]` in `pyproject.toml` sets `testpaths = ["tests"]`,
-`addopts = "-p no:xonsh"` and `testpaths = ["tests"]` are inactive.
+`addopts = "-p no:xonsh"`, and `qt_api = "pyqt6"`; do not repeat
-Always pass a test path or node ID explicitly.
+`-p no:xonsh` in ordinary commands. Still pass a test path or node ID when a
+deterministic scope is required.
 
 Package-resolution check that does not import Piker's dependencies:
 
 ```text
-py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+.venv/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
 ```
 
 Dependency import check, required before collection or execution:
 
 ```text
-py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+.venv/bin/python -c 'from importlib.metadata import version; import pathlib, piker, pytestqt, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); qt_ver = version("pytest-qt"); print(sys.executable); print(mod); print(qt_ver); assert mod.is_relative_to(root); assert qt_ver == "4.5.0"'
 ```
 
 Safe core collection check:
 
 ```text
-py313/bin/python -m pytest -p no:xonsh -q --collect-only tests/test_watchlists.py tests/test_accounting.py tests/test_services.py tests/test_ems.py tests/test_feeds.py tests/test_cli.py
+.venv/bin/python -m pytest -q --collect-only tests/test_watchlists.py tests/test_accounting.py tests/test_services.py tests/test_ems.py tests/test_feeds.py tests/test_cli.py
 ```
 
 Default first-pass flags are `-q -x --tb=short --no-header` unless the user
@@ -86,7 +88,7 @@ requests otherwise. For actor-heavy tests, use one file or node per process
 with an outer timeout:
 
 ```text
-timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q <one-file-or-node>
+timeout -k 5 300 .venv/bin/python -m pytest -q <one-file-or-node>
 ```
 
 If an actor-heavy command exits `124` or `143`, retry that exact command once
@@ -111,6 +113,8 @@ Deterministic or local first-pass targets:
 - `tests/test_ib_method_proxy.py`
 - `tests/test_history_backfill.py`
 - `tests/test_ldshm.py`
+- `tests/test_dpi_font.py`
+- `tests/test_gap_overlays.py`
 - `tests/test_accounting.py::test_account_file_default_empty`
 - `tests/test_services.py::test_runtime_boot`
 - `tests/test_services.py::test_datad_spawn`
@@ -128,7 +132,6 @@ Require explicit authorization before running:
   fixtures plus possible live symcache generation;
 - `tests/test_accounting.py::test_ib_account_with_duplicated_mktids` - active
   broker/account configuration and state writes;
-- `tests/test_dpi_font.py` - Qt/UI import and user-config side effects;
 - `tests/test_docker_services.py` - optional dependencies and containers;
 - `tests/test_questrade.py` - obsolete credentialed imports.
 
@@ -169,9 +172,12 @@ whole suite supports UDS merely because the Tractor plugin exposes it.
   `tests/_inputs/account.binance.paper.toml` are used in place. Accounting
   contexts can write them on exit. Inspect `git diff -- tests/_inputs` after
   any selected accounting case.
-- Importing `tests/test_dpi_font.py` constructs module-global font objects
+- `tests/conftest.py` selects PyQt6 defaults and owns temporary
-  before fixtures can isolate config. If explicitly requested, isolate
+  `XDG_CONFIG_HOME` and `XDG_CONFIG_DIRS` trees before importing Piker or Qt.
-  `XDG_CONFIG_HOME` before Python starts and use the proper Qt/Nix shell.
+  `--headless` force-selects Qt's `offscreen` platform at that same early
+  point, overriding compositor values inherited from a development shell.
+  A pytest config cleanup restores the caller environment even when only
+  collection runs. The `qapp_args` fixture only supplies `piker-tests`.
 - Piker and the installed Tractor pytest plugin do not provide a
   repository-local process or socket reaper. Never apply historical broad
   `pkill -f tractor._child` guidance automatically.
@@ -181,6 +187,60 @@ whole suite supports UDS merely because the Tractor plugin exposes it.
   unlinks exact surviving names before failing the leaking test. It never
   scans `/dev/shm` or unlinks attachments created by another process.
 
+## Qt/UI Tier
+
+The pytest process defaults to `QT_QPA_PLATFORM=offscreen` only when the
+caller has not selected a platform. Development shells can select a real
+compositor such as Wayland. Pass `--headless` on every automated Qt run to
+force `offscreen` before imports and avoid opening desktop windows. Without
+`--headless`, an explicit caller platform wins. `PYTEST_QT_API` defaults to
+PyQt6, the authoritative pytest config selects `qt_api = "pyqt6"`, and
+`pytest-qt==4.5.0` is locked.
+
+Root conftest setup assigns temporary `XDG_CONFIG_HOME` and `XDG_CONFIG_DIRS`
+trees before importing Piker or Qt. This keeps import-cached Piker paths and
+Qt user/system settings below test-owned storage. The session `qapp_args`
+fixture uses `piker-tests` as the application argument. Deterministic
+offscreen Qt tests are normal local targets; they are not opt-in merely
+because they import Qt or might otherwise read user config.
+
+`tests/ui/conftest.py` reuses pytest-qt's session `QApplication` and applies
+an autouse function guard. After pytest-qt closes registered widgets, the
+guard checks and restores top-level widgets, `ViewBox.AllViews`,
+`ViewBox.NamedViews`, `pyqtgraph.CONFIG_OPTIONS`, Piker config paths,
+`QSettings`, and `quitOnLastWindowClosed()`. Register every test-owned widget
+with `qtbot.addWidget()` so a surviving object is reported as a leak.
+
+Exact verified gap-overlay commands:
+
+```text
+.venv/bin/python -m pytest --headless -q --collect-only tests/ui/test_harness.py tests/test_gap_overlays.py
+.venv/bin/python -m pytest --headless -q -x --tb=short --no-header tests/ui/test_harness.py tests/test_gap_overlays.py
+.venv/bin/python -m pytest --headless -q -x --tb=short --no-header tests/ui/test_harness.py
+.venv/bin/python -m pytest --headless -q --collect-only tests/test_gap_overlays.py
+.venv/bin/python -m pytest --headless -q -x --tb=short --no-header tests/test_gap_overlays.py
+```
+
+The combined collection command collected 11 tests. The combined headless
+command passed all 11 in 2.00s, including the UI platform assertion, with one
+upstream Tractor `trio.Event` boolean-use deprecation warning.
+
+- Let `qtbot` own every widget registered with `qtbot.addWidget()` through
+  teardown. Do not manually close or delete the same widget a second time.
+- Keep production `MainWindow` out of this tier: its `closeEvent()` sends
+  `SIGINT` to the pytest process.
+- A real compositor is an explicit process override, for example:
+
+```text
+QT_QPA_PLATFORM=wayland .venv/bin/python -m pytest -q -x --tb=short --no-header tests/test_gap_overlays.py
+QT_QPA_PLATFORM=xcb .venv/bin/python -m pytest -q -x --tb=short --no-header tests/test_gap_overlays.py
+```
+
+Real-compositor, visual, manual, and live UI tests remain opt-in. Announce
+them before execution and obtain explicit authorization. Never omit
+`--headless` from an automated Qt run merely because a compositor is
+available.
+
 ## Test Layout
 
 ```text
@@ -193,6 +253,7 @@ tests/
   test_dpi_font.py          Qt DPI/font behavior
   test_ems.py               actor, EMS, and paper-position behavior
   test_feeds.py             live Binance/Kraken feeds and shared memory
+  test_gap_overlays.py      typed gap logic, offscreen Qt, and Tractor IPC
   test_ib_history.py        deterministic IB history request formatting
   test_ib_method_proxy.py   deterministic IB asyncio proxy routing
   test_history_backfill.py  deterministic history/SHM orchestration
@@ -203,6 +264,7 @@ tests/
   test_storage_audit.py     read-only NativeDB audit and JSON CLI
   test_backfill_audit_snippet.py
                             disposable xonsh qualification helpers
+  ui/                       repeated-session Qt isolation and leak proofs
   test_watchlists.py        deterministic watchlist JSON operations
 ```
 
@@ -220,7 +282,9 @@ tests/
 | `piker/storage/cli.py` SHM null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation |
 | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch |
 | `piker/accounting/` | targeted accounting node | some cases use live/configured state |
-| `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | GUI/config-isolated opt-in |
+| root/UI test fixtures | `tests/ui/test_harness.py` | same-process Qt and config restoration |
+| `piker/ui/_gaps.py`, `_annotate.py`, `_display.py`, `_interaction.py`, `_remote_ctl.py` | `tests/test_gap_overlays.py` | offscreen PyQt6 plus one local Tractor actor |
+| `piker/ui/_style.py`, `piker/ui/qt.py` | `tests/test_dpi_font.py` | deterministic offscreen Qt |
 | `piker/service/_actor_runtime.py`, `_registry.py`, `_mngr.py` | `test_runtime_boot` | then `test_datad_spawn` |
 | `piker/service/`, `piker/data/_daemon.py` | `test_datad_spawn` | feed lifecycle cases are live |
 | `piker/data/feed.py`, `flows.py`, `_sharedmem.py`, `_sampling.py` | collect first | feed execution needs live permission |
@@ -230,26 +294,28 @@ tests/
 | Docker/service adapters | `tests/test_docker_services.py` | optional deps and containers |
 | project, lock, or Nix files | import check and safe collection | full collection is not safe by default |
 
-Prefer deterministic filesystem/config tests, then local actor-runtime nodes,
+Prefer deterministic filesystem/config and offscreen Qt tests, then local
-then explicitly approved live broker, GUI, or container coverage.
+actor-runtime nodes, then explicitly approved live broker, visual UI, or
+container coverage.
 
 ## Quick Checks
 
 ```text
-py313/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+.venv/bin/python -c 'import importlib.util, pathlib, sys; root = pathlib.Path.cwd().resolve(); spec = importlib.util.find_spec("piker"); mod = pathlib.Path(spec.origin).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
-py313/bin/python -c 'import pathlib, piker, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); print(sys.executable); print(mod); assert mod.is_relative_to(root)'
+.venv/bin/python -c 'from importlib.metadata import version; import pathlib, piker, pytestqt, sys; root = pathlib.Path.cwd().resolve(); mod = pathlib.Path(piker.__file__).resolve(); qt_ver = version("pytest-qt"); print(sys.executable); print(mod); print(qt_ver); assert mod.is_relative_to(root); assert qt_ver == "4.5.0"'
-py313/bin/python -m pytest -p no:xonsh -q tests/test_watchlists.py
+.venv/bin/python -m pytest --headless -q --collect-only tests/test_gap_overlays.py
-py313/bin/python -m pytest -p no:xonsh -q tests/test_accounting.py::test_account_file_default_empty
+.venv/bin/python -m pytest --headless -q -x --tb=short --no-header tests/test_gap_overlays.py
-timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_runtime_boot
+.venv/bin/python -m pytest -q tests/test_watchlists.py
-timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_services.py::test_datad_spawn
+.venv/bin/python -m pytest -q tests/test_accounting.py::test_account_file_default_empty
-timeout -k 5 300 py313/bin/python -m pytest -p no:xonsh -q tests/test_ems.py::test_ems_err_on_bad_broker
+timeout -k 5 300 .venv/bin/python -m pytest -q tests/test_services.py::test_runtime_boot
+timeout -k 5 300 .venv/bin/python -m pytest -q tests/test_services.py::test_datad_spawn
+timeout -k 5 300 .venv/bin/python -m pytest -q tests/test_ems.py::test_ems_err_on_bad_broker
 ```
 
 ## Known Outcomes
 
-- The current root-checkout `py313` fails `import piker` outside the Nix shell
+- The verified `.venv` import check resolves `piker` beneath this worktree and
-  with `ImportError: PyQtGraph requires one of PyQt5, PyQt6, PySide2 or
+  imports the locked `pytest-qt` 4.5.0.
-  PySide6`. This is an incomplete environment, not an application regression.
 - `tests/test_accounting.py::test_root_conf_networking_section` currently
   expects `network.tsdb`, which is absent from the tracked config template.
   Match the current `KeyError: 'tsdb'` before classifying it as the known

ai/prompt-io/opencode/20260831T005712Z_09ddcf50_prompt_io.md

@@ -0,0 +1,44 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: unavailable
+timestamp: 2026-08-31T00:57:12Z
+git_ref: 09ddcf50
+scope: tests
+substantive: true
+raw_file: 20260831T005712Z_09ddcf50_prompt_io.raw.md
+---
+
+## Prompt
+
+Deeply evaluate Piker's lacking suites and build a comprehensive plan for
+non-mocked, human-driven E2E coverage, beginning with the pytest-qt plan on
+`chart_local_gapper`. Work in `/open-wkt`, build the first feasible GUI test
+slice, add minimal adjacent CLI and Qt guide docs, keep the `/run-tests`
+reference current, and stop before the stated Eastern deadline.
+
+## Response summary
+
+Audits the existing suite and human-facing surfaces, creates a risk-ranked
+cross-interface E2E plan, activates pytest-qt, migrates the real-Qt gap suite
+to pytest-qt widget ownership, and records verified commands and minimal user
+guides. Full `MainWindow` E2E remains deferred behind safe shutdown, guest
+outcome, symbol-session ownership, and deterministic provider seams.
+
+## Files changed
+
+- `pyproject.toml`, `uv.lock`, `pytest.ini` - activate pytest-qt config.
+- `tests/conftest.py` - establish the Qt process and settings contract.
+- `tests/test_gap_overlays.py` - migrate real widgets to `qtbot` ownership.
+- `piker/cli/README.rst` - add the minimal command-line guide.
+- `piker/ui/README.rst` - add the minimal chart UI guide.
+- `plans/opencode/human-facing-e2e-coverage.md` - persist the suite plan.
+- `.claude/skills/run-tests/test-harness-reference.md` - record verified
+  test commands and boundaries.
+
+## Human edits
+
+The human requires isolated worktree execution, adjacent guide-style docs,
+continuous `/run-tests` reference maintenance, non-mocked Qt interactions,
+and a hard time cutoff. These constraints materially determine the patch
+scope and defer unsafe full-window work.

ai/prompt-io/opencode/20260831T005712Z_09ddcf50_prompt_io.raw.md

@@ -0,0 +1,45 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-31T00:57:12Z
+git_ref: 09ddcf50
+diff_cmd: git diff HEAD
+---
+
+## Generated changes
+
+> `git diff HEAD -- pyproject.toml pytest.ini uv.lock`
+
+Activates pytest-qt with PyQt6 through the repository's authoritative
+pytest configuration and locks pytest-qt 4.5.0.
+
+> `git diff HEAD -- tests/conftest.py tests/test_gap_overlays.py`
+
+Selects the offscreen PyQt6 test process before Qt imports, isolates Qt
+configuration, migrates real chart-gap widgets to pytest-qt ownership,
+and removes the conflicting manual widget teardown path.
+
+> `git diff HEAD -- piker/cli/README.rst piker/ui/README.rst`
+
+Adds minimal adjacent guides for the installed command and Qt chart
+interfaces without freezing volatile subsystem details.
+
+> `git diff HEAD -- plans/opencode/human-facing-e2e-coverage.md`
+
+Records the evidence-based test-suite audit and phased human-facing E2E
+strategy, distinguishing component, application, system, and live
+qualification boundaries.
+
+> `git diff HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Records the verified environment, commands, Qt ownership rules, current
+tier boundaries, and test outcomes.
+
+## Verification
+
+- Local package and pytest-qt 4.5.0 import check passes.
+- Nine gap-overlay tests collect.
+- The targeted file passes: 9 tests, one upstream Tractor warning.
+- A duplicated same-process run passes: 18 tests, two equivalent
+  upstream Tractor warnings.
+- `git diff --check` passes.

ai/prompt-io/opencode/20260831T215640Z_233fa590_prompt_io.md

@@ -0,0 +1,64 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: d3e773b1-63ab-44da-a30f-33bc8a05e778
+timestamp: 2026-08-31T21:56:40Z
+git_ref: 233fa590
+scope: code
+substantive: true
+raw_file: 20260831T215640Z_233fa590_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue Phase 1 of the human-facing E2E foundation from the existing
+worktree state: establish robust pytest-qt isolation and leak proofs,
+use real QtBot Ctrl-G input, replace arbitrary actor sleeps with typed
+deterministic synchronization, keep the test-harness reference current,
+and verify the work end to end. Preserve the user's staged snapshot,
+with permission to unstage if useful. After implementation, unstage
+the snapshot and prepare a complete multi-commit plan.
+
+## Response summary
+
+Establishes import-time XDG isolation with exact pytest cleanup, adds
+same-process QApplication and mutable-state restoration proofs, and
+migrates Ctrl-G to real QtBot input. An explicit `--headless` option
+forces offscreen execution before imports even inside the Wayland
+development shell. The real input sequence exposed and fixed
+modifier-only event handling in the production keyboard path. The real
+Tractor regression now uses a typed receipt and explicit cancellation
+barrier instead of timing and logger instrumentation.
+
+## Files changed
+
+- `tests/conftest.py` - own and restore process XDG config roots before
+  importing Piker or Qt.
+- `tests/ui/conftest.py` - guard and restore mutable Qt, PyQtGraph,
+  QSettings, filesystem, and Piker config state after each UI test.
+- `tests/ui/test_harness.py` - prove shared QApplication reuse,
+  isolation, sentinel preservation, and leak-free teardown.
+- `piker/ui/_interaction.py` - ignore modifier-only chord events before
+  action dispatch.
+- `tests/test_gap_overlays.py` - use real QtBot Ctrl-G and a typed,
+  deterministic real-actor cancellation barrier.
+- `.claude/skills/run-tests/test-harness-reference.md` - document the
+  verified fixture contract, commands, and results.
+
+## Human edits
+
+The human directed continuation of the existing Phase 1 scope,
+authorized `nix develop` for runtime verification, requested Prompt-IO
+capture, and stated that the current staged snapshot may be unstaged if
+needed. After an unannounced test run opened real Wayland windows, the
+human identified the problem and required an explicit headless mode,
+while preserving separately authorized real-window coverage. This
+material correction added `--headless`, offscreen platform proof, and
+the rule that real-compositor runs must be announced and explicitly
+authorized. The staged snapshot was preserved; subsequent fixes and
+these provenance files initially remained unstaged. The human later
+explicitly requested a mixed reset before commit-plan generation.
+During review of the first staged boundary, the human required a TODO
+for a public config-path API and detailed rationale around each UI
+teardown check. The agent applied both changes and added authorized
+responses to the persisted local Tuicr session.

ai/prompt-io/opencode/20260831T215640Z_233fa590_prompt_io.raw.md

@@ -0,0 +1,52 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-31T21:56:40Z
+git_ref: 233fa590
+diff_cmd: git diff HEAD
+---
+
+## Generated changes
+
+> `git diff HEAD -- tests/conftest.py tests/ui/conftest.py tests/ui/test_harness.py`
+
+Moves XDG user and system config isolation ahead of Piker and Qt
+imports, guarantees process-environment restoration through pytest
+cleanup, and adds repeated-session proofs for one shared
+`QApplication`. The function-scoped UI guard restores QSettings,
+Piker config paths, PyQtGraph options and view registries, Qt window
+state, and exact test-owned files while preserving process sentinels.
+An early `--headless` option force-selects `offscreen` even when the
+development shell exports Wayland, and the test asserts the effective
+Qt platform.
+
+> `git diff HEAD -- piker/ui/_interaction.py tests/test_gap_overlays.py`
+
+Routes Ctrl-G through a shown and focused real widget with
+`QtBot.keyPress()`. The production input handler now ignores the
+modifier-only events emitted by real key chords. Replaces a timed
+Tractor cancellation window and patched logger with a typed stream
+receipt, explicit task-completion event, and correlated stale-response
+proof across a real child actor.
+
+> `git diff HEAD -- .claude/skills/run-tests/test-harness-reference.md`
+
+Records the verified early-XDG lifecycle, same-process UI isolation
+contract, leak checks, targeted commands, layout, and current results.
+
+## Verification
+
+- Local package and pytest-qt 4.5.0 resolve in the approved
+  `nix develop` environment.
+- The combined UI and gap-overlay scope collects 11 tests.
+- Both UI isolation tests pass; a duplicated same-process run reports
+  six passing cases.
+- The real QtBot Ctrl-G test passes after exercising Qt's standalone
+  Control event before G.
+- The real Tractor actor test passes with no arbitrary sleep or
+  monkeypatched synchronization.
+- The combined target passes: 11 tests and one upstream Tractor
+  `trio.Event.__bool__` deprecation warning.
+- The final combined run passes with `--headless`, and the harness
+  proves `QApplication.platformName()` is `offscreen`.
+- Python static compilation and `git diff --check` pass.

ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.md

@@ -0,0 +1,51 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 8737d0a9-98da-4fa1-ba35-ca8a3a42bbc9
+timestamp: 2026-09-03T19:27:42Z
+git_ref: wkt/replay_provider_e2e
+scope: code
+substantive: true
+raw_file: 20260903T192742Z_aadfd1cf_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue Phase 2 of the human-facing E2E plan by implementing a
+protocol-faithful offline replay provider. Use normal backend discovery,
+datad, history, quote, search, and typed Tractor control contracts with
+versioned fixtures, deterministic ordering, no network, no positive
+sleeps, and no log parsing.
+
+During review, explain the provider's purpose and placement, evaluate a
+general package-entry-point backend mechanism, then park replay when the
+human prioritizes real-backend chart E2E instead.
+
+## Response summary
+
+Adds a versioned offline market-data provider implementing normal Piker
+backend endpoint shapes. Typed control commands advance, pause, resume,
+snapshot, and inject declared failures while preserving deterministic
+transcripts. Focused tests validate fixture symbology and finite,
+repeatable OHLCV history behavior.
+
+## Files changed
+
+- `piker/brokers/replay.py` - implement replay scenario, provider, and
+  control protocol machinery.
+- `tests/_inputs/replay/basic-v1.json` - provide deterministic history
+  and quote input.
+- `tests/_inputs/replay/failure-v1.json` - provide an acknowledged fault
+  point.
+- `tests/replay/test_contract.py` - validate scenario and history
+  contracts.
+
+## Human edits
+
+The human challenged the original replay-first premise, questioned why
+test machinery lived beside production backends, proposed dynamic
+registration, and selected standard package entry points as the future
+direction for external backend support. The human directly added a
+source note questioning replay placement, then explicitly placed replay
+on hold and requested that the completed experiment be committed only
+as parked work with plugin migration recorded for possible resumption.

ai/prompt-io/opencode/20260903T192742Z_aadfd1cf_prompt_io.raw.md

@@ -0,0 +1,41 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-03T19:27:42Z
+git_ref: wkt/replay_provider_e2e
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Generated changes
+
+> `git diff HEAD~1..HEAD -- piker/brokers/replay.py`
+
+Implements a fixture-driven market-data backend with typed scenario,
+quote, failure, command, acknowledgement, snapshot, and transcript
+structures. It provides the normal market-info, bounded history,
+quote-stream, search, and Tractor control endpoints without external
+network access.
+
+> `git diff HEAD~1..HEAD -- tests/_inputs/replay/ tests/replay/test_contract.py`
+
+Adds versioned basic and failure scenarios plus focused validation and
+history-boundary tests. The scenarios carry normalized market metadata,
+one-second and one-minute OHLCV arrays, sequenced quotes, and a named
+failure point.
+
+## Verification
+
+- Python compilation, Ruff, JSON parsing, and staged whitespace checks
+  pass.
+- Scenario normalization and bounded repeatable history tests pass.
+- The production history contract exposed a need for Pendulum datetime
+  values; the implementation was corrected before final verification.
+
+## Design discussion
+
+The user questioned placing test infrastructure beside production
+backends and proposed dynamic test registration. The resulting review
+identified standard Python package entry points as the preferred future
+mechanism for external and testing backends. The user then deprioritized
+replay entirely in favor of real-backend chart E2E and requested that
+the completed experiment be parked with that follow-up recorded.

ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.md

@@ -0,0 +1,42 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 8737d0a9-98da-4fa1-ba35-ca8a3a42bbc9
+timestamp: 2026-09-03T19:27:43Z
+git_ref: wkt/replay_provider_e2e
+scope: code
+substantive: true
+raw_file: 20260903T192743Z_aadfd1cf_prompt_io.raw.md
+---
+
+## Prompt
+
+Prove the replay provider through the real Piker feed runtime rather
+than monkeypatching `open_feed()`: exercise datad, samplerd, history and
+quote SHM, search, pause/resume, typed failures, repeatability, and clean
+actor/resource teardown without sleeps or log assertions.
+
+## Response summary
+
+Connects feed subscription mutations to an optional provider observer
+and adds real Tractor integration tests. The suite compares complete
+replay generations, verifies production pause/resume barriers, exercises
+typed producer controls, and proves acknowledged failure ordering and
+named-SHM cleanup.
+
+## Files changed
+
+- `piker/data/feed.py` - notify an optional backend subscriber-state
+  observer after add/remove operations.
+- `tests/replay/conftest.py` - select tracked replay scenarios for child
+  actors.
+- `tests/replay/test_provider.py` - exercise real feed services, SHM,
+  controls, repeatability, failures, and teardown.
+
+## Human edits
+
+The human reviewed the completed integration, questioned whether fake
+provider traffic offered enough real bug-detection value, and directed
+the work to stop before additional replay features. The human requires
+future E2E effort to use live broker machinery wherever practical and
+to derive focused data tests from observed real-world defects.

ai/prompt-io/opencode/20260903T192743Z_aadfd1cf_prompt_io.raw.md

@@ -0,0 +1,35 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-03T19:27:43Z
+git_ref: wkt/replay_provider_e2e
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Generated changes
+
+> `git diff HEAD~1..HEAD -- piker/data/feed.py tests/replay/`
+
+Adds an optional backend notification at real feed-bus subscriber
+mutations so replay can acknowledge the production `Feed.pause()` and
+`Feed.resume()` state. Integration fixtures select tracked scenarios
+through inherited configuration, while real actor tests exercise
+pikerd, datad, samplerd, history and quote SHM, search, typed replay
+control, failure acknowledgement, transcript equality, and exact
+resource teardown.
+
+## Verification
+
+- The basic scenario passes twice through fresh actor-tree generations
+  with identical control, quote, and history transcripts.
+- The failure scenario acknowledges its declared fault before any
+  failed quote reaches SHM.
+- Exact datad-created SHM names disappear after each generation.
+- The combined replay suite passes four tests; remaining warnings are
+  existing Tractor and Piker deprecations.
+
+## Design discussion
+
+The user concluded this middleware-focused work should not delay real
+Qt/Trio application testing. The integration remains a completed parked
+experiment rather than the foundation for the immediate chart roadmap.

ai/prompt-io/opencode/20260903T192744Z_aadfd1cf_prompt_io.md

@@ -0,0 +1,46 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 8737d0a9-98da-4fa1-ba35-ca8a3a42bbc9
+timestamp: 2026-09-03T19:27:44Z
+git_ref: wkt/replay_provider_e2e
+scope: docs
+substantive: true
+raw_file: 20260903T192744Z_aadfd1cf_prompt_io.raw.md
+---
+
+## Prompt
+
+Perform a deep review of the human-facing E2E plan against real broker
+and application machinery. Prioritize booting production charts against
+mostly continuous live backends, isolate tests from any production Piker
+instance on the host, put replay entirely on hold, and provide immediate
+practical next steps for Qt plus Trio application testing.
+
+Record the package-entry-point replay move as a follow-up if replay is
+ever resumed.
+
+## Response summary
+
+Reorients the roadmap around the actual application blockers and a live
+Kraken vertical slice. The review separates test boundary from data
+source, defines exact runtime isolation, specifies a nonblocking
+`QtractorSession`, and defers replay until a real regression supplies a
+concrete reason for controlled input.
+
+## Files changed
+
+- `plans/opencode/live-backend-chart-e2e-review.md` - record the revised
+  live-backend-first roadmap, lifecycle seam, isolation contract, and
+  replay/plugin resumption design.
+
+## Human edits
+
+The human rejected deterministic replay as the organizing prerequisite
+for E2E, prioritized failures observable through real broker and chart
+machinery, selected the Qt/Trio lifecycle as immediate work, approved a
+live Kraken chart journey followed by real interactions and installed
+process coverage, and required replay to remain parked unless motivated
+by a concrete regression. The human also required the future replay
+move to use general package-entry-point backend support and explicitly
+approved provenance capture for this design document.

ai/prompt-io/opencode/20260903T192744Z_aadfd1cf_prompt_io.raw.md

@@ -0,0 +1,30 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-03T19:27:44Z
+git_ref: wkt/replay_provider_e2e
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Generated document
+
+> `git diff HEAD~1..HEAD -- plans/opencode/live-backend-chart-e2e-review.md`
+
+Reviews the original replay-centered phase order against the actual
+Qt/Trio host, chart startup, real backend, runtime isolation, and test
+harness machinery. It parks replay, promotes the nonblocking production
+application seam and live Kraken chart journey, defines exact isolation
+and teardown expectations, and records package-entry-point migration for
+any future replay resumption.
+
+## Review conclusions
+
+- Replay is not a technical prerequisite for Qt/Trio lifecycle work.
+- `run_qtractor()` blocking, discarded guest outcomes, pytest-qt app
+  shutdown, process-wide close SIGINT, and missing readiness are the
+  immediate blockers.
+- Existing XDG, config, Qt, and registry isolation is reusable.
+- A public Kraken chart can exercise real REST, WebSocket, datad,
+  samplerd, SHM, paper EMS, rendering, input, and teardown machinery.
+- Replay should return only for a concrete captured payload, race,
+  network-free CI requirement, or fault-control regression.

piker/brokers/replay.py

@@ -0,0 +1,715 @@
+'''
+Deterministic offline market-data replay backend.
+
+'''
+from __future__ import annotations
+
+from contextlib import (
+    asynccontextmanager as acm,
+)
+from dataclasses import dataclass
+from datetime import datetime
+import os
+from pathlib import Path
+from typing import (
+    Any,
+    Literal,
+)
+
+import msgspec
+import numpy as np
+import pendulum
+import tractor
+import trio
+from trio_typing import TaskStatus
+
+from piker.accounting import MktPair
+from piker.brokers import (
+    DataUnavailable,
+    SymbolNotFound,
+)
+from piker.data._source import def_iohlcv_fields
+from piker.data.validate import FeedInit
+from piker.types import Struct
+
+
+name: str = 'replay'
+_scenario_env: str = 'PIKER_REPLAY_SCENARIO'
+
+
+class ReplayBar(Struct, frozen=True):
+    '''
+    One normalized OHLCV fixture row.
+
+    '''
+    index: int
+    time: int
+    open: float
+    high: float
+    low: float
+    close: float
+    volume: float
+
+
+class ReplayTick(Struct, frozen=True):
+    '''
+    One normalized tick in a replay quote.
+
+    '''
+    type: str
+    price: float
+    size: float
+
+
+class ReplayQuote(Struct, frozen=True):
+    '''
+    One sequenced provider event from a replay scenario.
+
+    '''
+    sequence: int
+    fqme: str
+    broker_ts: int
+    last: float
+    ticks: list[ReplayTick]
+
+
+class ReplayFailure(Struct, frozen=True):
+    '''
+    A failure which can be armed at one event sequence.
+
+    '''
+    sequence: int
+    code: str
+    message: str
+
+
+class ReplayScenario(Struct, frozen=True):
+    '''
+    Versioned input for one deterministic replay generation.
+
+    '''
+    version: int
+    scenario_id: str
+    markets: list[MktPair]
+    history_1s: list[ReplayBar]
+    history_1m: list[ReplayBar]
+    quotes: list[ReplayQuote]
+    failures: list[ReplayFailure] = []
+
+
+ReplayState = Literal[
+    'ready',
+    'paused',
+    'failed',
+    'exhausted',
+]
+
+
+class Advance(Struct, frozen=True, tag=True):
+    '''
+    Publish one exact replay event.
+
+    '''
+    command_id: int
+    event_sequence: int
+
+
+class Pause(Struct, frozen=True, tag=True):
+    '''
+    Close the replay producer gate.
+
+    '''
+    command_id: int
+
+
+class Resume(Struct, frozen=True, tag=True):
+    '''
+    Open the replay producer gate.
+
+    '''
+    command_id: int
+
+
+class FailAt(Struct, frozen=True, tag=True):
+    '''
+    Arm one scenario-declared failure injection.
+
+    '''
+    command_id: int
+    event_sequence: int
+
+
+class AwaitState(Struct, frozen=True, tag=True):
+    '''
+    Wait for one provider state without scheduler polling.
+
+    '''
+    command_id: int
+    state: ReplayState
+
+
+class Snapshot(Struct, frozen=True, tag=True):
+    '''
+    Request the current deterministic replay state.
+
+    '''
+    command_id: int
+
+
+ReplayCommand = (
+    Advance
+    | Pause
+    | Resume
+    | FailAt
+    | AwaitState
+    | Snapshot
+)
+
+
+class ReplayRecord(Struct, frozen=True):
+    '''
+    One deterministic control transcript entry.
+
+    '''
+    command_id: int
+    command: str
+    state: ReplayState
+    event_sequence: int
+    ok: bool
+    error: str = ''
+
+
+class ReplaySnapshot(Struct, frozen=True, tag=True):
+    '''
+    Provider state returned through the control protocol.
+
+    '''
+    scenario_id: str
+    state: ReplayState
+    command_id: int
+    event_sequence: int
+    subscriber_active: bool
+    armed_failure: int | None
+    transcript: list[ReplayRecord]
+
+
+class ReplayAck(Struct, frozen=True, tag=True):
+    '''
+    Correlated result for one replay control command.
+
+    '''
+    command_id: int
+    ok: bool
+    snapshot: ReplaySnapshot
+    error: str = ''
+
+
+ReplayPayload = (
+    ReplayCommand
+    | ReplayAck
+    | ReplaySnapshot
+)
+
+
+# hmm maybe put this in a new piker._testing.brokers ?
+# or is there some reason to put it alongside the production
+# backends?
+def load_scenario(
+    path: Path | str,
+) -> ReplayScenario:
+    '''
+    Decode and validate one versioned replay scenario.
+
+    '''
+    scenario_path: Path = Path(path)
+    scenario: ReplayScenario = msgspec.json.decode(
+        scenario_path.read_bytes(),
+        type=ReplayScenario,
+    )
+    if scenario.version != 1:
+        raise ValueError(
+            f'Unsupported replay scenario version: '
+            f'{scenario.version}'
+        )
+    if not scenario.markets:
+        raise ValueError('Replay scenario has no markets')
+    if not scenario.quotes:
+        raise ValueError('Replay scenario has no quotes')
+
+    quote_sequences: list[int] = [
+        quote.sequence
+        for quote in scenario.quotes
+    ]
+    expected: list[int] = list(range(
+        quote_sequences[0],
+        quote_sequences[-1] + 1,
+    ))
+    if quote_sequences != expected:
+        raise ValueError(
+            'Replay quote sequences must be contiguous'
+        )
+
+    market_fqmes: set[str] = {
+        mkt.fqme
+        for mkt in scenario.markets
+    }
+    unknown_fqmes: set[str] = {
+        quote.fqme
+        for quote in scenario.quotes
+    } - market_fqmes
+    if unknown_fqmes:
+        raise ValueError(
+            f'Replay quotes reference unknown markets: '
+            f'{sorted(unknown_fqmes)!r}'
+        )
+    return scenario
+
+
+def _scenario_path() -> Path:
+    path: str | None = os.environ.get(_scenario_env)
+    if not path:
+        raise RuntimeError(
+            f'Set `{_scenario_env}` to a replay scenario path'
+        )
+    return Path(path)
+
+
+def _quote_to_msg(
+    quote: ReplayQuote,
+) -> dict[str, Any]:
+    return {
+        'symbol': quote.fqme,
+        'last': quote.last,
+        'broker_ts': quote.broker_ts,
+        'brokerd_ts': quote.broker_ts,
+        'replay_seq': quote.sequence,
+        'ticks': [
+            {
+                'type': tick.type,
+                'price': tick.price,
+                'size': tick.size,
+            }
+            for tick in quote.ticks
+        ],
+    }
+
+
+def _bars_to_array(
+    bars: list[ReplayBar],
+) -> np.ndarray:
+    return np.asarray(
+        [
+            (
+                bar.index,
+                bar.time,
+                bar.open,
+                bar.high,
+                bar.low,
+                bar.close,
+                bar.volume,
+            )
+            for bar in bars
+        ],
+        dtype=def_iohlcv_fields,
+    )
+
+
+def _find_market(
+    scenario: ReplayScenario,
+    fqme: str,
+) -> MktPair:
+    normalized: str = fqme.lower()
+    for mkt in scenario.markets:
+        aliases: set[str] = {
+            mkt.fqme.lower(),
+            mkt.bs_fqme.lower(),
+            mkt.bs_mktid.lower(),
+        }
+        if normalized in aliases:
+            return mkt
+    raise SymbolNotFound(fqme)
+
+
+async def get_mkt_info(
+    fqme: str,
+) -> tuple[MktPair, MktPair]:
+    '''
+    Return normalized and backend market records from the fixture.
+
+    '''
+    scenario: ReplayScenario = load_scenario(_scenario_path())
+    mkt: MktPair = _find_market(scenario, fqme)
+    return mkt, mkt
+
+
+@acm
+async def open_history_client(
+    mkt: MktPair,
+):
+    '''
+    Open a bounded in-memory OHLCV history client.
+
+    '''
+    scenario: ReplayScenario = load_scenario(_scenario_path())
+    _find_market(scenario, mkt.fqme)
+    frames: dict[int, np.ndarray] = {
+        1: _bars_to_array(scenario.history_1s),
+        60: _bars_to_array(scenario.history_1m),
+    }
+
+    async def get_ohlc(
+        timeframe: float,
+        end_dt: datetime | None = None,
+        start_dt: datetime | None = None,
+
+    ) -> tuple[np.ndarray, datetime, datetime]:
+        period: int = int(timeframe)
+        try:
+            frame: np.ndarray = frames[period]
+        except KeyError:
+            raise DataUnavailable(
+                f'Unsupported replay timeframe: {timeframe}'
+            ) from None
+
+        selected: np.ndarray = frame
+        if start_dt is not None:
+            start_ts: float = start_dt.timestamp()
+            selected = selected[selected['time'] >= start_ts]
+        if end_dt is not None:
+            end_ts: float = end_dt.timestamp()
+            selected = selected[selected['time'] < end_ts]
+        if not selected.size:
+            raise DataUnavailable(
+                f'No replay history before {end_dt!r}'
+            )
+
+        result: np.ndarray = selected.copy()
+        start: datetime = pendulum.from_timestamp(
+            int(result['time'][0]),
+        )
+        end: datetime = pendulum.from_timestamp(
+            int(result['time'][-1]),
+        )
+        return result, start, end
+
+    yield get_ohlc, {
+        'erlangs': 1,
+        'rate': 1,
+    }
+
+
+@dataclass
+class _RuntimeRequest:
+    command: ReplayCommand
+    reply: trio.MemorySendChannel[ReplayAck]
+
+
+class _ReplayRuntime:
+    '''
+    Actor-local owner of replay progression and transcript state.
+
+    '''
+    def __init__(
+        self,
+        scenario: ReplayScenario,
+    ) -> None:
+        self.scenario = scenario
+        self.command_tx: (
+            trio.MemorySendChannel[_RuntimeRequest]
+            | None
+        ) = None
+        self.command_id: int = 0
+        self.event_sequence: int = scenario.quotes[0].sequence
+        self.producer_paused: bool = False
+        self.subscriber_active: bool = True
+        self.armed_failure: int | None = None
+        self.failed: bool = False
+        self.transcript: list[ReplayRecord] = []
+        self._state_changed: trio.Event = trio.Event()
+
+    @property
+    def state(self) -> ReplayState:
+        if self.failed:
+            return 'failed'
+        if (
+            self.event_sequence
+            == self.scenario.quotes[-1].sequence
+        ):
+            return 'exhausted'
+        if (
+            self.producer_paused
+            or not self.subscriber_active
+        ):
+            return 'paused'
+        return 'ready'
+
+    def snapshot(self) -> ReplaySnapshot:
+        return ReplaySnapshot(
+            scenario_id=self.scenario.scenario_id,
+            state=self.state,
+            command_id=self.command_id,
+            event_sequence=self.event_sequence,
+            subscriber_active=self.subscriber_active,
+            armed_failure=self.armed_failure,
+            transcript=list(self.transcript),
+        )
+
+    def signal_state_change(self) -> None:
+        changed: trio.Event = self._state_changed
+        self._state_changed = trio.Event()
+        changed.set()
+
+    async def await_state(
+        self,
+        state: ReplayState,
+    ) -> None:
+        while self.state != state:
+            changed: trio.Event = self._state_changed
+            await changed.wait()
+
+    def record(
+        self,
+        command: ReplayCommand,
+        ok: bool,
+        error: str = '',
+    ) -> ReplayAck:
+        self.transcript.append(ReplayRecord(
+            command_id=command.command_id,
+            command=type(command).__name__,
+            state=self.state,
+            event_sequence=self.event_sequence,
+            ok=ok,
+            error=error,
+        ))
+        return ReplayAck(
+            command_id=command.command_id,
+            ok=ok,
+            error=error,
+            snapshot=self.snapshot(),
+        )
+
+
+_runtime: _ReplayRuntime | None = None
+_runtime_path: Path | None = None
+
+
+def _get_runtime() -> _ReplayRuntime:
+    global _runtime, _runtime_path
+    path: Path = _scenario_path().resolve()
+    if (
+        _runtime is None
+        or _runtime_path != path
+    ):
+        _runtime = _ReplayRuntime(load_scenario(path))
+        _runtime_path = path
+    return _runtime
+
+
+def on_feed_subscription_change(
+    bs_fqme: str,
+    active: bool,
+) -> None:
+    '''
+    Synchronize the provider gate with feed-bus subscriptions.
+
+    '''
+    runtime: _ReplayRuntime = _get_runtime()
+    _find_market(runtime.scenario, bs_fqme)
+    if runtime.subscriber_active != active:
+        runtime.subscriber_active = active
+        runtime.signal_state_change()
+
+
+async def _execute_command(
+    runtime: _ReplayRuntime,
+    command: ReplayCommand,
+    send_chan: trio.abc.SendChannel,
+) -> ReplayAck:
+    expected_id: int = runtime.command_id + 1
+    if command.command_id != expected_id:
+        error: str = (
+            f'Expected command_id={expected_id}, got '
+            f'{command.command_id}'
+        )
+        return runtime.record(command, False, error)
+
+    runtime.command_id = command.command_id
+    match command:
+        case Pause():
+            runtime.producer_paused = True
+            runtime.signal_state_change()
+
+        case Resume():
+            runtime.producer_paused = False
+            runtime.signal_state_change()
+
+        case AwaitState(state=state):
+            await runtime.await_state(state)
+
+        case FailAt(event_sequence=sequence):
+            failures: dict[int, ReplayFailure] = {
+                failure.sequence: failure
+                for failure in runtime.scenario.failures
+            }
+            if sequence not in failures:
+                error = (
+                    f'No scenario failure at event_sequence='
+                    f'{sequence}'
+                )
+                return runtime.record(command, False, error)
+            runtime.armed_failure = sequence
+
+        case Advance(event_sequence=sequence):
+            if runtime.state != 'ready':
+                error = (
+                    f'Cannot advance replay while state='
+                    f'{runtime.state!r}'
+                )
+                return runtime.record(command, False, error)
+
+            expected_sequence: int = runtime.event_sequence + 1
+            if sequence != expected_sequence:
+                error = (
+                    f'Expected event_sequence={expected_sequence}, '
+                    f'got {sequence}'
+                )
+                return runtime.record(command, False, error)
+
+            if runtime.armed_failure == sequence:
+                failure: ReplayFailure = next(
+                    failure
+                    for failure in runtime.scenario.failures
+                    if failure.sequence == sequence
+                )
+                runtime.failed = True
+                runtime.signal_state_change()
+                error = f'{failure.code}: {failure.message}'
+                return runtime.record(command, False, error)
+
+            quote: ReplayQuote = runtime.scenario.quotes[
+                sequence
+                - runtime.scenario.quotes[0].sequence
+            ]
+            mkt: MktPair = _find_market(
+                runtime.scenario,
+                quote.fqme,
+            )
+            await send_chan.send({
+                mkt.bs_fqme: _quote_to_msg(quote),
+            })
+            runtime.event_sequence = sequence
+            runtime.signal_state_change()
+
+        case Snapshot():
+            pass
+
+    return runtime.record(command, True)
+
+
+async def stream_quotes(
+    send_chan: trio.abc.SendChannel,
+    symbols: list[str],
+    feed_is_live: trio.Event,
+    loglevel: str | None = None,
+    task_status: TaskStatus[
+        tuple[list[FeedInit], dict[str, Any]]
+    ] = trio.TASK_STATUS_IGNORED,
+) -> None:
+    '''
+    Publish fixture quotes only after acknowledged control commands.
+
+    '''
+    if len(symbols) != 1:
+        raise ValueError(
+            'The replay backend currently supports one symbol'
+        )
+
+    runtime: _ReplayRuntime = _get_runtime()
+    mkt: MktPair = _find_market(runtime.scenario, symbols[0])
+    first: ReplayQuote = runtime.scenario.quotes[0]
+    if first.fqme != mkt.fqme:
+        raise ValueError(
+            f'First replay quote does not target {mkt.fqme!r}'
+        )
+
+    command_tx: trio.MemorySendChannel[_RuntimeRequest]
+    command_rx: trio.MemoryReceiveChannel[_RuntimeRequest]
+    command_tx, command_rx = trio.open_memory_channel(0)
+    runtime.command_tx = command_tx
+    try:
+        async with (
+            send_chan,
+            command_rx,
+        ):
+            task_status.started((
+                [FeedInit(mkt_info=mkt)],
+                _quote_to_msg(first),
+            ))
+            feed_is_live.set()
+
+            async for request in command_rx:
+                ack: ReplayAck = await _execute_command(
+                    runtime,
+                    request.command,
+                    send_chan,
+                )
+                async with request.reply:
+                    await request.reply.send(ack)
+    finally:
+        runtime.command_tx = None
+
+
+@tractor.context(pld_spec=ReplayPayload)
+async def open_replay_control(
+    ctx: tractor.Context,
+) -> None:
+    '''
+    Open the typed replay command and acknowledgement stream.
+
+    '''
+    runtime: _ReplayRuntime = _get_runtime()
+    if runtime.command_tx is None:
+        raise RuntimeError('Replay quote stream is not running')
+
+    await ctx.started(runtime.snapshot())
+    async with ctx.open_stream() as stream:
+        with ctx.pld_rx.limit_plds(spec=ReplayCommand):
+            command: ReplayCommand
+            async for command in stream:
+                reply_tx: trio.MemorySendChannel[ReplayAck]
+                reply_rx: trio.MemoryReceiveChannel[ReplayAck]
+                reply_tx, reply_rx = trio.open_memory_channel(1)
+                request = _RuntimeRequest(
+                    command=command,
+                    reply=reply_tx,
+                )
+                await runtime.command_tx.send(request)
+                async with reply_rx:
+                    ack: ReplayAck = await reply_rx.receive()
+                await stream.send(ack)
+
+
+@tractor.context
+async def open_symbol_search(
+    ctx: tractor.Context,
+) -> None:
+    '''
+    Search replay fixture markets by normalized FQME substring.
+
+    '''
+    scenario: ReplayScenario = load_scenario(_scenario_path())
+    await ctx.started()
+    async with ctx.open_stream() as stream:
+        pattern: str
+        async for pattern in stream:
+            lowered: str = pattern.lower()
+            matches: dict[str, dict[str, Any]] = {
+                mkt.fqme: mkt.to_dict()
+                for mkt in scenario.markets
+                if lowered in mkt.fqme.lower()
+            }
+            await stream.send(matches)
+
+
+_datad_mods: list[str] = []
+__enable_modules__: list[str] = []

piker/cli/README.rst

@@ -0,0 +1,65 @@
+piker command-line guide
+========================
+
+Piker installs three human-facing entry points with different jobs:
+
+* ``piker`` is the main command group for broker, data, storage and UI tasks.
+* ``pikerd`` is the long-running root service supervisor.  Clients such as
+  ``chart`` connect to it, or start a service tree when none is running.
+* ``ledger`` is the separate trade-ledger and position-accounting tool.  Its
+  commands can contact brokers or update account data; it is not read-only.
+
+See the `project README <../../README.rst>`_ for installation and runtime context.
+
+Safe discovery and first run
+----------------------------
+
+From a checkout, these enter no broker, UI or account command body::
+
+    uv run piker --help
+    uv run piker chart --help
+    uv run pikerd --help
+    uv run ledger --help
+
+Explicit ``--help`` prints usage and exits 0.  Repeat it at each command level.
+
+A low-risk first runtime starts only the supervisor; stop it with ``Ctrl-C``::
+
+    uv run pikerd -l info
+
+It waits for client requests before starting broker or feed work.  Run this
+intentionally long-lived process in its own terminal.
+
+Root options come first
+-----------------------
+
+Click options precede the subcommand that consumes them.  ``--brokers``,
+``--loglevel``, ``--configdir`` and ``--pdb`` are ``piker`` root options::
+
+    uv run piker -l info -c /tmp/piker-profile chart <fqme>
+
+Create the configuration directory first.  Do not move ``-l`` or ``-c`` after
+``chart``; use ``piker chart --help`` for options owned by that subcommand.
+
+For isolation, set ``XDG_CONFIG_HOME`` before any Piker process.  ``piker -c``
+can also select an existing directory for that ``piker`` invocation::
+
+    export XDG_CONFIG_HOME=/tmp/piker-xdg
+    mkdir -p "$XDG_CONFIG_HOME/piker"
+    uv run piker -c "$XDG_CONFIG_HOME/piker" --help
+
+Output contracts
+----------------
+
+Parser errors are non-zero, but operational callbacks do not yet share one
+failure-code contract.  Tables, colors, help prose and logs are human output.
+Consume JSON only when that command advertises ``--json`` and pin automation to
+its tested schema.  No CLI-wide JSON or stdout/stderr stability is promised.
+
+Intentionally not guaranteed
+----------------------------
+
+Registry/service-list details, multiaddr syntax/routing, and storage commands,
+SHM identities and disk layouts are volatile and intentionally not guaranteed.
+Consult the running checkout; focused storage UX expectations live in
+`tests/test_store_cli.py <../../tests/test_store_cli.py>`_.

piker/data/feed.py

@@ -38,6 +38,7 @@ from typing import (
     Any,
     AsyncContextManager,
     Awaitable,
+    Callable,
     Sequence,
     TYPE_CHECKING,
 )
@@ -506,6 +507,28 @@ async def open_feed_bus(
     assert brokername in servicename
 
     bus: _FeedsBus = get_feed_bus(brokername)
+    try:
+        mod: ModuleType = get_brokermod(brokername)
+    except ImportError:
+        mod = get_ingestormod(brokername)
+
+    on_sub_change: Callable[[str, bool], None] | None = getattr(
+        mod,
+        'on_feed_subscription_change',
+        None,
+    )
+
+    def notify_sub_change(bs_fqme: str) -> None:
+        '''
+        Notify an optional backend subscriber-state observer.
+
+        '''
+        if on_sub_change is not None:
+            on_sub_change(
+                bs_fqme,
+                bool(bus.get_subs(bs_fqme)),
+            )
+
     sub_registered = trio.Event()
 
     flumes: dict[str, Flume] = {}
@@ -638,6 +661,7 @@ async def open_feed_bus(
                 bs_fqme,
                 {sub}
             )
+            notify_sub_change(bs_fqme)
 
         # sync caller with all subs registered state
         sub_registered.set()
@@ -654,12 +678,14 @@ async def open_feed_bus(
                         log.info(
                             f'Pausing {bs_fqme} feed for {uid}')
                         bus.remove_subs(bs_fqme, subs)
+                        notify_sub_change(bs_fqme)
 
                 elif msg == 'resume':
                     for bs_fqme, subs in local_subs.items():
                         log.info(
                             f'Resuming {bs_fqme} feed for {uid}')
                         bus.add_subs(bs_fqme, subs)
+                        notify_sub_change(bs_fqme)
 
                 else:
                     raise ValueError(msg)
@@ -675,6 +701,7 @@ async def open_feed_bus(
             # drop all subs for this task from the bus
             for bs_fqme, subs in local_subs.items():
                 bus.remove_subs(bs_fqme, subs)
+                notify_sub_change(bs_fqme)
 
 
 class Feed(Struct):

piker/ui/README.rst

@@ -0,0 +1,68 @@
+Qt chart guide
+==============
+
+The Qt chart is Piker's keyboard-first realtime market view.  Follow the
+`project README <../../README.rst>`_, include the UI group, then launch an FQME::
+
+    uv sync --group uis
+    uv run piker -l info chart btcusdt.spot.binance
+
+Use a market supported by your provider; a name without its provider suffix is
+rejected.  This is not an offline demo: launch may contact provider services
+and needs a Qt display.  Put root options before ``chart``; see
+`piker/cli/README.rst <../cli/README.rst>`_.
+
+Daemon lifetime
+---------------
+
+The chart looks for ``pikerd`` and starts a supervisor for its session if none
+is available.  To retain service state across chart restarts, run
+``uv run pikerd`` separately first.  That daemon has its own lifetime; stop it
+in its terminal.
+
+Keyboard journeys
+-----------------
+
+Keep focus on the chart or search pane whose action you want:
+
+* Search: press ``Ctrl-L`` (``L`` for "list" symbols), type, move with
+  ``Ctrl-J``/``Ctrl-K`` (or ``Ctrl-Down``/``Ctrl-Up``), then ``Enter``.
+  ``Ctrl-C`` or ``Ctrl-Space`` returns focus to the chart.
+* Chart: use the wheel to zoom and press ``R`` to restore the default view.
+  ``Ctrl-I`` and ``Ctrl-O`` provide keyboard zoom in and out.
+* Gaps: press ``Ctrl-G`` on the focused realtime or history chart to toggle
+  chart-local OHLC gap markers for that pane and timeframe.
+* Orders: hold ``F`` ("fill") for buy, ``D`` ("dump") for sell or ``A`` for
+  an alert to stage at the cursor.  Buy/sell default to dark; add ``S`` or
+  ``Ctrl`` for live.  A left-click submits.  ``C`` or ``Delete`` cancels under
+  the cursor; quick ``cc`` ("complete clear") asks to cancel all orders.
+
+These are real controls.  Confirm the mode label, account and paper/live setup;
+a live account can send a real order.
+
+Closing safely
+--------------
+
+Closing the main window saves geometry and sends ``SIGINT`` to the chart process
+so its async runtime can unwind.  Do not use ``MainWindow.close()`` as generic
+cleanup inside another app or test runner: the signal targets the whole process.
+The full in-process test tier therefore needs a dedicated shutdown seam.
+
+Testing contract
+----------------
+
+The intended default automated gate is layered and deterministic:
+
+* Drive real PyQt6/PyQtGraph objects with real Qt key and mouse events through
+  production event filters.
+* Use synthetic market data and deterministic feed, search, EMS and service
+  boundaries; assert visible state, scene ownership and clean teardown.
+* Keep screenshots as failure artifacts, not pixel or visual goldens.
+* Keep live brokers, credentials, network feeds and compositor qualification
+  out of the default gate.
+
+Current `gap tests <../../tests/test_gap_overlays.py>`_ send a real ``QKeyEvent``
+through ``Ctrl-G`` and use real graphics scenes.  The actor case stubs rendering,
+so this is integration evidence, not full chart E2E.  Intended tiers and close
+work live in the `pytest-qt chart plan
+<../../plans/opencode/pytest-qt-chart-ui-e2e.md>`_.

piker/ui/_interaction.py

@@ -226,6 +226,16 @@ async def handle_viewmode_kb_inputs(
             if mods == Qt.KeyboardModifier.ControlModifier:
                 ctrl = True
 
+            # Real key chords deliver modifier-only events before the
+            # bound key. Nothing below handles those as actions.
+            if key in {
+                Qt.Key.Key_Alt,
+                Qt.Key.Key_Control,
+                Qt.Key.Key_Meta,
+                Qt.Key.Key_Shift,
+            }:
+                continue
+
             # UI REPL-shell, with ctrl-p (for "pause")
             if (
                 ctrl

plans/opencode/live-backend-chart-e2e-review.md

@@ -0,0 +1,348 @@
+# Live-backend chart E2E plan review
+
+## Decision
+
+Reorient the human-facing test effort around the production Qt chart
+running against real public backends. Put replay implementation,
+scenario design, provider failure injection, and replay-backed CI on
+hold until a real application or backend failure demonstrates a need
+for controlled data.
+
+The first valuable target is an in-process application journey using
+pytest-qt, Trio guest mode, an isolated Piker actor tree, and public
+Kraken data. Installed-process and paper-order journeys follow after
+that path can start, report readiness, and stop without process-wide
+signals.
+
+Entry-point backend discovery remains independently useful, but it is
+not a prerequisite for chart E2E. Treat it as a short standalone patch
+or defer it with replay.
+
+## Review of the original plan
+
+The original plan correctly identifies the end goal: user-facing QtBot
+journeys paired with installed-process tests. It also correctly requires
+real Tractor, IPC, SHM, feed, search, and EMS boundaries.
+
+Its central ordering assumption is now rejected. It makes a deterministic
+offline provider the organizing dependency for chart boot, search, paper
+orders, public APIs, process tests, and failure handling. It schedules
+that provider and a general subprocess harness before the production
+Qt/Trio lifecycle seam.
+
+That ordering optimizes for deterministic default CI. It does not optimize
+for finding failures users currently encounter with real venues, real
+history payloads, real quote timing, real symbology, and the composed chart.
+
+The replay phase is not technically required by the later lifecycle work:
+
+1. Qt/Trio startup and shutdown are provider-independent.
+2. A chart can use Kraken's public history and quote APIs without credentials.
+3. Search can use the real Kraken symbol service.
+4. Paper EMS can consume a real public feed without spawning live brokerd.
+5. Installed chart startup can use an isolated registry and real provider.
+6. Guest errors, child failures, SIGINT behavior, and repeated-session leaks
+   do not require synthetic provider failures.
+
+Replay remains potentially useful for forcing exact races, reproducing a
+captured venue payload, denying network in required CI, or proving a fix for
+a specific data-dependent regression. Those are follow-up uses, not current
+prerequisites.
+
+## What replay would and would not audit
+
+A replay-backed chart journey would prove that the composed application can
+consume one protocol-conforming source, render known bars, process known
+quotes, and shut down. It would be useful interface and lifecycle coverage.
+
+It would not prove the behavior most likely to drift outside the repository:
+
+- exchange REST and WebSocket availability;
+- TLS, DNS, geo-routing, and endpoint changes;
+- authentication behavior accidentally applied to public requests;
+- real symbology and market metadata changes;
+- incomplete, duplicated, delayed, or out-of-order history;
+- backend normalization against current venue payloads;
+- reconnect and rate-limit behavior;
+- the timing and volume of actual chart updates.
+
+The current priority is the second list. Therefore real-provider application
+qualification should precede further replay work.
+
+## Current practical blockers
+
+### Blocking application host
+
+`run_qtractor()` currently creates or reuses `QApplication`, constructs the
+production `MainWindow` and `GodWidget`, starts Trio guest mode, shows the
+window, and then blocks in `app.exec_()`.
+
+This prevents a synchronous pytest-qt test from receiving a session handle
+while Qt owns the host loop.
+
+### Unobservable Trio completion
+
+The Trio guest `done_callback` prints unexpected errors and unconditionally
+calls `app.quit()`. It does not retain the `outcome.Outcome`, expose a Qt
+completion signal, or let pytest fail on the original exception.
+
+Calling `app.quit()` also violates pytest-qt ownership of its session-scoped
+application.
+
+### Process-wide chart shutdown
+
+`MainWindow.closeEvent()` saves geometry and sends SIGINT to the entire test
+process. A QtBot-owned window cannot exercise normal close behavior without
+interrupting pytest.
+
+### No application readiness contract
+
+`_async_main()` has a meaningful readiness point after the initial chart,
+paper order mode, search handlers, and status cleanup are active, but it does
+not publish that state outside its Trio task.
+
+### Incomplete child-resource accounting
+
+The test harness isolates XDG paths, config paths, registry addresses, and
+current-process SHM. A chart application test must additionally record the
+specific child actor IDs and SHM names created by datad so teardown can prove
+those exact resources disappeared without broad cleanup.
+
+## Existing foundation to keep
+
+The completed harness foundation already provides:
+
+- pytest-qt ownership of one `QApplication`;
+- import-time PyQt6 and offscreen selection with `--headless`;
+- process-level XDG isolation before Piker and Qt imports;
+- function-scoped restoration of Piker, Qt, QSettings, and PyQtGraph state;
+- real QtBot key delivery and widget registration;
+- unique Tractor registry addresses from the Tractor pytest plugin;
+- test-owned config directories propagated to child actors through
+  `piker_test_dir` runtime variables;
+- exact current-process SHM ownership checks.
+
+This is sufficient foundation for the lifecycle extraction. Replay is not
+needed to begin it.
+
+## Revised test model
+
+Keep independent dimensions instead of defining application coverage by its
+data source:
+
+| Dimension | Initial selection | Later selections |
+|---|---|---|
+| Boundary | In-process production application | Installed chart process |
+| Data source | Live Kraken public data | Binance, other qualified backends |
+| Display | Qt offscreen through pytest-qt | Real Wayland/X11 compositor |
+| Clearing | Real paper EMS | Credentialed live broker qualification |
+| Cadence | Explicit local/CI qualification | Repeated soak and release gate |
+
+The first test is therefore a headless, live-provider, in-process application
+journey. It is not a deterministic default unit test, and it should not be
+presented as one.
+
+## Isolation contract
+
+Real provider access does not require attaching to a developer's production
+Piker runtime. Each application test must:
+
+1. Use the Tractor pytest plugin's unique registry address.
+2. Pass that address explicitly through `run_qtractor()` into
+   `maybe_open_pikerd()`; never allow fallback to the default registry.
+3. Propagate the test-owned config directory through
+   `tractor_runtime_overrides['piker_test_dir']`.
+4. Start a fresh local `pikerd`, `datad.kraken`, and `samplerd` under the
+   test's registry.
+5. Use public Kraken data only; do not load user credentials.
+6. Record every actor identity and each chart feed's exact SHM names.
+7. Request structured shutdown through the application session.
+8. Assert those actors, streams, registry connections, and SHM names are gone.
+9. Never use process-name matching, broad SHM scans, `pkill`, or global ports.
+10. Treat the external network as intentionally shared qualification input.
+
+An explicitly selected live test must fail on connection or provider errors,
+not silently skip after startup. Tests skip only when live qualification was
+not requested.
+
+## Immediate vertical slice
+
+### Production lifecycle seam
+
+Extract a nonblocking `start_qtractor()` from `run_qtractor()`.
+
+It should return a `QtractorSession` containing:
+
+- the production `MainWindow` and `GodWidget`;
+- a Trio token and owned cancellation scope;
+- an application-ready Qt signal;
+- a completion Qt signal;
+- the final `outcome.Outcome`;
+- a structured `request_shutdown()` method;
+- whether the adapter or caller owns `QApplication.exec()`.
+
+`start_qtractor()` must configure and show the production objects but must not
+call `app.exec()` or `app.quit()`. `run_qtractor()` remains the blocking CLI
+adapter: it calls `start_qtractor()`, owns `app.exec()`, and unwraps the final
+outcome into a process result.
+
+Change `MainWindow.closeEvent()` to request session shutdown through an
+injected callback. Keep geometry persistence. Remove process-wide SIGINT from
+normal window closure; terminal SIGINT remains an outer CLI concern.
+
+Publish application readiness from `_async_main()` only after the initial
+feed, charts, paper order mode, search handlers, and startup status have all
+entered their live scopes.
+
+### First live chart journey
+
+Add an explicitly selected test such as:
+
+`tests/app/test_live_chart.py::test_kraken_chart_boots_and_closes`
+
+Use `xbtusd.spot.kraken`, because Kraken public REST/WebSocket data is
+credential-free and the pair is continuously active.
+
+The test should:
+
+1. Start `QtractorSession` with pytest-qt's `QApplication`.
+2. Pass the unique test registry and child config root.
+3. Register `session.window` with `qtbot` immediately.
+4. Wait for the application-ready signal with a bounded timeout.
+5. Assert the displayed FQME and production window title.
+6. Assert both historical and real-time chart visualizations have nonempty
+   data from their real SHM arrays.
+7. Observe at least one post-start quote/display update from Kraken.
+8. Drive Ctrl-L with QtBot and assert the real search bar receives focus.
+9. Drive the existing search-dismiss interaction and assert chart focus.
+10. Request structured close and wait for the completion signal.
+11. Unwrap a successful Trio outcome.
+12. Prove exact actor and SHM teardown.
+
+Do not assert pixels, exact prices, exact bar counts, result ordering, actor
+PIDs, generated SHM names, or quote arrival within an unrealistically short
+interval. Those values vary while the semantic behavior remains correct.
+
+## Live-provider selection
+
+Use Kraken as the initial baseline:
+
+- public market metadata, OHLC, and WebSocket data need no credentials;
+- `xbtusd.spot.kraken` and `ethusdt.spot.kraken` already appear in live feed
+  tests;
+- the existing suite allows Kraken feed coverage in CI;
+- crypto trading is continuous outside venue maintenance and outages.
+
+Use Binance spot as a second qualification only where network and geographic
+access are known. Keep Kucoin informational until its public configuration and
+startup behavior receive direct repair. Exclude IB, Questrade, and current
+Deribit from credential-free chart qualification.
+
+Add one explicit selection mechanism, preferably `--live-provider=kraken`.
+Without it, live application tests skip. With it, backend startup failure is a
+test failure and retains the original Trio/Tractor exception and diagnostics.
+
+Do not add a blanket retry. Record cold-start duration and failure signatures.
+A separate repeated-run or soak command can measure operational reliability
+after the first journey is stable.
+
+## Reordered implementation phases
+
+### Phase A: Qt/Trio lifecycle
+
+Implement `QtractorSession`, nonblocking startup, structured close, observable
+outcome, and repeated-session cleanup. Verify first with a provider-independent
+guest coroutine so failures are localized to lifecycle ownership.
+
+### Phase B: real Kraken chart
+
+Boot the production `_async_main()` against `xbtusd.spot.kraken` using the
+isolated runtime contract. Prove chart readiness, live data, one QtBot focus
+journey, and complete teardown.
+
+### Phase C: real chart behavior
+
+Add A-to-B-to-A Kraken symbol switching, chart navigation, timeframe changes,
+and paper order open/cancel behavior. Prefer stable semantic assertions and
+record real failures before extracting narrow regression tests.
+
+### Phase D: installed-process coverage
+
+Run the installed `piker chart` command with the same isolated registry/config
+contract. Prove cold startup, readiness, graceful close, exit status, and exact
+descendant cleanup. Add standalone `pikerd` attachment only after cold start
+works.
+
+### Phase E: qualification breadth
+
+Add Binance where available, real compositor runs, repeated cold starts, and
+bounded soak sessions. Keep provider outages visible as qualification results
+rather than rewriting them into synthetic success.
+
+### Deferred work
+
+Defer replay scenarios, provider-conformance tests, deterministic quote
+sequencing, synthetic provider failures, and replay-backed required CI.
+
+Reconsider replay only when one of these concrete needs appears:
+
+- a real payload must be preserved as a regression fixture;
+- an ordering race cannot be reproduced reliably against a live venue;
+- required network-free CI needs a minimal known-good chart source;
+- a reconnect or malformed-data defect needs deterministic fault control.
+
+Backend entry-point discovery may land separately if its implementation stays
+small and useful to external providers. Do not make it block Phase A or B, and
+do not use plugin support as a reason to continue replay work.
+
+## Replay resumption record
+
+If replay work resumes, first move its implementation out of the production
+backend namespace and use it to introduce general external backend discovery:
+
+1. Move the implementation to `piker.testing.brokers.replay`.
+2. Add a `piker.brokers` Python entry-point group which resolves installed
+   external backend modules after conventional built-in imports.
+3. Register replay through that entry point in Piker's package metadata.
+4. Preserve a backend's actual module path when constructing datad and brokerd
+   `enable_modules` lists so Tractor authorizes the external endpoint module.
+5. Reject duplicate names, built-in shadowing, non-module entry points, name
+   mismatches, and unsupported backend API versions.
+6. Preserve dependency import failures from built-in backends instead of
+   misclassifying them as absent modules.
+7. Cover resolver behavior with focused unit tests and prove one installed
+   entry-point backend through a spawned datad actor.
+
+Do not use pytest-only `sys.modules` or package-attribute monkeypatching: those
+registrations do not reliably cross spawn or forkserver actor boundaries.
+External backends under active development can use editable installation in
+the same Python environment as Piker.
+
+## Practical sequence from the current worktree
+
+The replay worktree contains the completed replay experiment. The unfinished
+entry-point resolver experiment was deliberately excluded. Preserve this
+branch as parked work rather than mixing it into the live chart patch.
+
+Create the live-chart work on a fresh branch/worktree from the completed
+Phase 1 head. The first implementation boundary should touch only the Qt/Trio
+lifecycle and its provider-independent lifecycle tests. The second boundary
+should add the opt-in real Kraken chart journey and any defects that journey
+demonstrates.
+
+If entry-point discovery is finished first, cap it at one independent resolver
+commit with focused unit tests. Do not move or expand replay as part of that
+commit.
+
+## Effort estimate
+
+- Entry-point resolver and unit tests: roughly half a day if kept independent.
+- Qt/Trio lifecycle extraction and tests: one to two focused days.
+- First isolated real Kraken chart journey: one to two days, depending on
+  defects revealed during startup and teardown.
+- First QtBot search/focus behavior journey: roughly one additional day after
+  lifecycle stability.
+- Installed-process chart coverage: one to two days after the in-process path.
+
+The first real chart should therefore be reachable in roughly two to four
+focused days without further replay work.

pyproject.toml

@@ -149,6 +149,7 @@ repl = [
 ]
 testing = [
   "pytest",
+  "pytest-qt",
 ]
 de = [ # (linux) specific DEs
   "i3ipc>=2.2.1",
@@ -177,6 +178,7 @@ console_output_style = 'progress'
 # https://docs.pytest.org/en/stable/how-to/plugins.html#disabling-plugins-from-autoloading
 # https://docs.pytest.org/en/stable/how-to/plugins.html#deactivating-unregistering-a-plugin-by-name
 addopts = '-p no:xonsh'
+qt_api = 'pyqt6'
 # ------ tool.pytest ------
 
 

pytest.ini

@@ -1,3 +0,0 @@
-#[pytest]
-#trio_mode=True
-#log_cli=1

tests/_inputs/replay/basic-v1.json

@@ -0,0 +1,45 @@
+{
+  "version": 1,
+  "scenario_id": "basic-v1",
+  "markets": [
+    {
+      "dst": {
+        "name": "btc",
+        "atype": "crypto",
+        "tx_tick": "0.00000001"
+      },
+      "src": {
+        "name": "usd",
+        "atype": "fiat",
+        "tx_tick": "0.01"
+      },
+      "price_tick": "0.01",
+      "size_tick": "0.0001",
+      "bs_mktid": "BTCUSD",
+      "broker": "replay",
+      "venue": "test"
+    }
+  ],
+  "history_1s": [
+    {"index": 0, "time": 1700000035, "open": 100.0, "high": 100.2, "low": 99.9, "close": 100.1, "volume": 1.0},
+    {"index": 1, "time": 1700000036, "open": 100.1, "high": 100.3, "low": 100.0, "close": 100.2, "volume": 2.0},
+    {"index": 2, "time": 1700000037, "open": 100.2, "high": 100.4, "low": 100.1, "close": 100.3, "volume": 3.0},
+    {"index": 3, "time": 1700000038, "open": 100.3, "high": 100.5, "low": 100.2, "close": 100.4, "volume": 4.0},
+    {"index": 4, "time": 1700000039, "open": 100.4, "high": 100.6, "low": 100.3, "close": 100.5, "volume": 5.0},
+    {"index": 5, "time": 1700000040, "open": 100.5, "high": 100.7, "low": 100.4, "close": 100.6, "volume": 6.0}
+  ],
+  "history_1m": [
+    {"index": 0, "time": 1699999740, "open": 99.0, "high": 99.4, "low": 98.8, "close": 99.2, "volume": 10.0},
+    {"index": 1, "time": 1699999800, "open": 99.2, "high": 99.6, "low": 99.0, "close": 99.4, "volume": 11.0},
+    {"index": 2, "time": 1699999860, "open": 99.4, "high": 99.8, "low": 99.2, "close": 99.6, "volume": 12.0},
+    {"index": 3, "time": 1699999920, "open": 99.6, "high": 100.0, "low": 99.4, "close": 99.8, "volume": 13.0},
+    {"index": 4, "time": 1699999980, "open": 99.8, "high": 100.4, "low": 99.6, "close": 100.2, "volume": 14.0},
+    {"index": 5, "time": 1700000040, "open": 100.2, "high": 100.8, "low": 100.0, "close": 100.6, "volume": 15.0}
+  ],
+  "quotes": [
+    {"sequence": 1, "fqme": "btcusd.test.replay", "broker_ts": 1700000041, "last": 100.7, "ticks": [{"type": "trade", "price": 100.7, "size": 0.1}]},
+    {"sequence": 2, "fqme": "btcusd.test.replay", "broker_ts": 1700000042, "last": 100.8, "ticks": [{"type": "trade", "price": 100.8, "size": 0.2}]},
+    {"sequence": 3, "fqme": "btcusd.test.replay", "broker_ts": 1700000043, "last": 100.9, "ticks": [{"type": "trade", "price": 100.9, "size": 0.3}]},
+    {"sequence": 4, "fqme": "btcusd.test.replay", "broker_ts": 1700000044, "last": 101.0, "ticks": [{"type": "trade", "price": 101.0, "size": 0.4}]}
+  ]
+}

tests/_inputs/replay/failure-v1.json

@@ -0,0 +1,45 @@
+{
+  "version": 1,
+  "scenario_id": "failure-v1",
+  "markets": [
+    {
+      "dst": {
+        "name": "btc",
+        "atype": "crypto",
+        "tx_tick": "0.00000001"
+      },
+      "src": {
+        "name": "usd",
+        "atype": "fiat",
+        "tx_tick": "0.01"
+      },
+      "price_tick": "0.01",
+      "size_tick": "0.0001",
+      "bs_mktid": "BTCUSD",
+      "broker": "replay",
+      "venue": "test"
+    }
+  ],
+  "history_1s": [
+    {"index": 0, "time": 1700000038, "open": 100.0, "high": 100.2, "low": 99.9, "close": 100.1, "volume": 1.0},
+    {"index": 1, "time": 1700000039, "open": 100.1, "high": 100.3, "low": 100.0, "close": 100.2, "volume": 2.0},
+    {"index": 2, "time": 1700000040, "open": 100.2, "high": 100.4, "low": 100.1, "close": 100.3, "volume": 3.0}
+  ],
+  "history_1m": [
+    {"index": 0, "time": 1699999920, "open": 99.0, "high": 99.5, "low": 98.8, "close": 99.3, "volume": 10.0},
+    {"index": 1, "time": 1699999980, "open": 99.3, "high": 99.9, "low": 99.1, "close": 99.7, "volume": 11.0},
+    {"index": 2, "time": 1700000040, "open": 99.7, "high": 100.5, "low": 99.5, "close": 100.3, "volume": 12.0}
+  ],
+  "quotes": [
+    {"sequence": 1, "fqme": "btcusd.test.replay", "broker_ts": 1700000041, "last": 100.4, "ticks": [{"type": "trade", "price": 100.4, "size": 0.1}]},
+    {"sequence": 2, "fqme": "btcusd.test.replay", "broker_ts": 1700000042, "last": 100.5, "ticks": [{"type": "trade", "price": 100.5, "size": 0.2}]},
+    {"sequence": 3, "fqme": "btcusd.test.replay", "broker_ts": 1700000043, "last": 100.6, "ticks": [{"type": "trade", "price": 100.6, "size": 0.3}]}
+  ],
+  "failures": [
+    {
+      "sequence": 3,
+      "code": "fixture_disconnect",
+      "message": "offline provider disconnected"
+    }
+  ]
+}

tests/conftest.py

@@ -1,14 +1,48 @@
 from contextlib import asynccontextmanager as acm
-from collections.abc import Callable
+from collections.abc import (
+    Callable,
+    Iterator,
+)
 from functools import partial
 import logging
 import os
 from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
 from weakref import (
     ReferenceType,
     ref,
 )
 
+# These must be selected before test modules import Qt or Piker.
+_original_qt_qpa_platform: str|None = os.environ.get(
+    'QT_QPA_PLATFORM',
+)
+_original_pytest_qt_api: str|None = os.environ.get(
+    'PYTEST_QT_API',
+)
+_headless_qt: bool = '--headless' in sys.argv
+if _headless_qt:
+    os.environ['QT_QPA_PLATFORM'] = 'offscreen'
+else:
+    os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
+os.environ.setdefault('PYTEST_QT_API', 'pyqt6')
+_original_xdg_config_home: str|None = os.environ.get(
+    'XDG_CONFIG_HOME',
+)
+_original_xdg_config_dirs: str|None = os.environ.get(
+    'XDG_CONFIG_DIRS',
+)
+_xdg_config_home_owner: TemporaryDirectory[str] = (
+    TemporaryDirectory(prefix='piker-pytest-xdg-')
+)
+_xdg_config_home: Path = Path(_xdg_config_home_owner.name)
+_xdg_config_dirs: Path = _xdg_config_home / 'system-config'
+_xdg_config_dirs.mkdir()
+os.environ['XDG_CONFIG_HOME'] = str(_xdg_config_home)
+os.environ['XDG_CONFIG_DIRS'] = str(_xdg_config_dirs)
+_xdg_config_restored: bool = False
+
 import pytest
 import tractor
 from piker import (
@@ -26,7 +60,74 @@ pytest_plugins: tuple[str] = (
 )
 
 
+@pytest.fixture(scope='session')
+def isolated_xdg_config_home() -> Iterator[Path]:
+    '''
+    Own process-wide config isolation through session teardown.
+
+    '''
+    yield _xdg_config_home
+
+
+def _restore_test_process_environment() -> None:
+    '''
+    Release process-owned XDG isolation exactly once.
+
+    '''
+    global _xdg_config_restored
+
+    if _xdg_config_restored:
+        return
+    _xdg_config_restored = True
+
+    if _original_xdg_config_home is None:
+        os.environ.pop('XDG_CONFIG_HOME', None)
+    else:
+        os.environ['XDG_CONFIG_HOME'] = _original_xdg_config_home
+    if _original_xdg_config_dirs is None:
+        os.environ.pop('XDG_CONFIG_DIRS', None)
+    else:
+        os.environ['XDG_CONFIG_DIRS'] = _original_xdg_config_dirs
+    if _original_qt_qpa_platform is None:
+        os.environ.pop('QT_QPA_PLATFORM', None)
+    else:
+        os.environ['QT_QPA_PLATFORM'] = (
+            _original_qt_qpa_platform
+        )
+    if _original_pytest_qt_api is None:
+        os.environ.pop('PYTEST_QT_API', None)
+    else:
+        os.environ['PYTEST_QT_API'] = _original_pytest_qt_api
+    _xdg_config_home_owner.cleanup()
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    '''
+    Guarantee import-time XDG ownership is released by pytest.
+
+    '''
+    config.add_cleanup(_restore_test_process_environment)
+
+
+@pytest.fixture(scope='session')
+def qapp_args(
+    isolated_xdg_config_home: Path,
+
+) -> list[str]:
+    '''
+    Configure pytest-qt after process-wide config isolation.
+
+    '''
+    assert isolated_xdg_config_home == _xdg_config_home
+    return ['piker-tests']
+
+
 def pytest_addoption(parser):
+    parser.addoption(
+        '--headless',
+        action='store_true',
+        help='Force Qt onto the offscreen platform before imports',
+    )
     parser.addoption("--ll", action="store", dest='loglevel',
                      default=None, help="logging level to set when testing")
     parser.addoption("--confdir", default=None,

tests/replay/conftest.py

@@ -0,0 +1,31 @@
+'''
+Offline replay integration fixtures.
+
+'''
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def replay_scenario(
+    request: pytest.FixtureRequest,
+    monkeypatch: pytest.MonkeyPatch,
+) -> Path:
+    '''
+    Select one tracked scenario for parent and child actors.
+
+    '''
+    inputs: Path = (
+        Path(__file__).parents[1]
+        / '_inputs'
+        / 'replay'
+    )
+    path: Path = inputs / f'{request.param}.json'
+    if not path.is_file():
+        raise FileNotFoundError(path)
+    monkeypatch.setenv(
+        'PIKER_REPLAY_SCENARIO',
+        str(path),
+    )
+    return path

tests/replay/test_contract.py

@@ -0,0 +1,100 @@
+'''
+Offline replay provider contract tests.
+
+'''
+from datetime import (
+    UTC,
+    datetime,
+)
+from pathlib import Path
+
+import numpy as np
+import pytest
+import trio
+
+from piker.brokers import DataUnavailable
+from piker.brokers import replay
+
+
+INPUTS: Path = (
+    Path(__file__).parents[1]
+    / '_inputs'
+    / 'replay'
+)
+
+
+def test_versioned_scenario_normalizes_market_data() -> None:
+    '''
+    Reject fixture drift before a datad actor obscures its cause.
+
+    Replay scenarios are durable test inputs rather than loose mock
+    dictionaries. Decode the versioned basic scenario through the
+    production loader and prove its market identity, Decimal fields,
+    contiguous event IDs, and normalized tick records survive typed
+    decoding. These assertions catch schema or symbology changes at
+    the provider boundary without starting services or using network
+    resources.
+
+    '''
+    scenario: replay.ReplayScenario = replay.load_scenario(
+        INPUTS / 'basic-v1.json'
+    )
+
+    assert scenario.version == 1
+    assert scenario.scenario_id == 'basic-v1'
+    assert scenario.markets[0].fqme == 'btcusd.test.replay'
+    assert str(scenario.markets[0].price_tick) == '0.01'
+    assert [quote.sequence for quote in scenario.quotes] == [
+        1,
+        2,
+        3,
+        4,
+    ]
+    assert scenario.quotes[1].ticks[0].type == 'trade'
+
+
+def test_history_queries_are_bounded_and_repeatable(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    Keep history replay finite, offline, and repeatable.
+
+    A history fixture which mutates a cursor per request can make the
+    two concurrent 1-second and 1-minute backfill tasks race,
+    while an unbounded latest frame can make datad backfill
+    forever. Select the same 1-minute frame twice, then request
+    data ending at its first timestamp. Equal arrays prove calls
+    are immutable and the explicit `DataUnavailable` proves
+    reverse backfill terminates at the fixture boundary without
+    a clock delay or external request.
+
+    '''
+    scenario_path: Path = INPUTS / 'basic-v1.json'
+    monkeypatch.setenv(
+        'PIKER_REPLAY_SCENARIO',
+        str(scenario_path),
+    )
+    scenario: replay.ReplayScenario = replay.load_scenario(
+        scenario_path
+    )
+    mkt = scenario.markets[0]
+
+    async def main() -> None:
+        async with replay.open_history_client(mkt) as (
+            get_hist,
+            config,
+        ):
+            first, start, end = await get_hist(60)
+            second, second_start, second_end = await get_hist(60)
+            np.testing.assert_array_equal(first, second)
+            assert (start, end) == (second_start, second_end)
+            assert config == {'erlangs': 1, 'rate': 1}
+
+            boundary: datetime = datetime.fromtimestamp(
+                int(first['time'][0]),
+                tz=UTC,
+            )
+            with pytest.raises(DataUnavailable):
+                await get_hist(60, end_dt=boundary)
+
+    trio.run(main)

tests/replay/test_provider.py

@@ -0,0 +1,370 @@
+'''
+Real service-tree coverage for the offline replay provider.
+
+'''
+from contextlib import AbstractAsyncContextManager
+from multiprocessing.shared_memory import SharedMemory
+from pathlib import Path
+from typing import (
+    Any,
+    Callable,
+)
+
+import msgspec
+import numpy as np
+import pytest
+import tractor
+import trio
+
+from piker.brokers import replay
+from piker.data import open_feed
+from piker.service import Services
+
+
+FQME: str = 'btcusd.test.replay'
+
+
+async def _request(
+    stream: tractor.MsgStream,
+    command: replay.ReplayCommand,
+) -> replay.ReplayAck:
+    '''
+    Send one replay command and receive its correlated reply.
+
+    '''
+    await stream.send(command)
+    ack: replay.ReplayAck = await stream.receive()
+    assert ack.command_id == command.command_id
+    return ack
+
+
+async def _search_replay_symbols(
+    portal: tractor.Portal,
+) -> dict[str, Any]:
+    '''
+    Exercise the backend's normal symbol-search context.
+
+    '''
+    ctx: tractor.Context
+    async with portal.open_context(
+        replay.open_symbol_search,
+    ) as (ctx, _):
+        async with ctx.open_stream() as stream:
+            await stream.send('btc')
+            return await stream.receive()
+
+
+async def _run_basic_generation(
+    open_test_pikerd: Callable[
+        ...,
+        AbstractAsyncContextManager,
+    ],
+    loglevel: str,
+) -> tuple[bytes, list[dict[str, Any]], list[float]]:
+    '''
+    Run one complete replay through real data services.
+
+    '''
+    quote_transcript: list[dict[str, Any]] = []
+    shm_names: list[str] = []
+    services: Services
+    async with (
+        open_test_pikerd() as (_, _, _, services),
+        open_feed(
+            [FQME],
+            loglevel=loglevel,
+        ) as feed,
+    ):
+        assert 'datad.replay' in services.service_tasks
+        assert 'samplerd' in services.service_tasks
+
+        stream: tractor.MsgStream = feed.streams['replay']
+        first_msg: dict[str, Any] = await stream.receive()
+        quote_transcript.append(first_msg)
+        assert first_msg[FQME]['replay_seq'] == 1
+
+        flume = feed.flumes[FQME]
+        history_closes: list[float] = (
+            flume.hist_shm.array['close'].tolist()
+        )
+        shm_names.extend([
+            flume.hist_shm._shm.name,
+            flume.rt_shm._shm.name,
+        ])
+        assert history_closes == [
+            99.2,
+            99.4,
+            99.6,
+            99.8,
+            100.2,
+            100.6,
+        ]
+
+        portal: tractor.Portal = feed.portals[replay]
+        matches: dict[str, Any] = await _search_replay_symbols(
+            portal
+        )
+        assert list(matches) == [FQME]
+
+        control_ctx: tractor.Context
+        started: replay.ReplaySnapshot
+        async with portal.open_context(
+            replay.open_replay_control,
+        ) as (control_ctx, started):
+            assert started.state == 'ready'
+            assert started.event_sequence == 1
+
+            async with control_ctx.open_stream() as control:
+                with control_ctx.pld_rx.limit_plds(
+                    spec=replay.ReplayAck,
+                ):
+                    await feed.pause()
+                    paused: replay.ReplayAck = await _request(
+                        control,
+                        replay.AwaitState(
+                            command_id=1,
+                            state='paused',
+                        ),
+                    )
+                    assert paused.ok
+                    assert not paused.snapshot.subscriber_active
+
+                    rejected: replay.ReplayAck = await _request(
+                        control,
+                        replay.Advance(
+                            command_id=2,
+                            event_sequence=2,
+                        ),
+                    )
+                    assert not rejected.ok
+                    assert rejected.snapshot.event_sequence == 1
+
+                    await feed.resume()
+                    ready: replay.ReplayAck = await _request(
+                        control,
+                        replay.AwaitState(
+                            command_id=3,
+                            state='ready',
+                        ),
+                    )
+                    assert ready.ok
+                    assert ready.snapshot.subscriber_active
+
+                    producer_paused = await _request(
+                        control,
+                        replay.Pause(command_id=4),
+                    )
+                    assert producer_paused.snapshot.state == 'paused'
+                    gate_rejected = await _request(
+                        control,
+                        replay.Advance(
+                            command_id=5,
+                            event_sequence=2,
+                        ),
+                    )
+                    assert not gate_rejected.ok
+                    producer_ready = await _request(
+                        control,
+                        replay.Resume(command_id=6),
+                    )
+                    assert producer_ready.snapshot.state == 'ready'
+
+                    for command_id, event_sequence in (
+                        (7, 2),
+                        (8, 3),
+                        (9, 4),
+                    ):
+                        advanced: replay.ReplayAck = (
+                            await _request(
+                                control,
+                                replay.Advance(
+                                    command_id=command_id,
+                                    event_sequence=(
+                                        event_sequence
+                                    ),
+                                ),
+                            )
+                        )
+                        assert advanced.ok
+                        quote: dict[str, Any] = (
+                            await stream.receive()
+                        )
+                        quote_transcript.append(quote)
+                        assert (
+                            quote[FQME]['replay_seq']
+                            == event_sequence
+                        )
+
+                    snap_ack: replay.ReplayAck = await _request(
+                        control,
+                        replay.Snapshot(command_id=10),
+                    )
+                    assert snap_ack.snapshot.state == 'exhausted'
+                    transcript: bytes = msgspec.json.encode(
+                        snap_ack.snapshot
+                    )
+
+        np.testing.assert_allclose(
+            flume.rt_shm.array['close'][-1],
+            101.0,
+        )
+
+    assert not services.service_tasks
+    for shm_name in shm_names:
+        with pytest.raises(FileNotFoundError):
+            SharedMemory(name=shm_name)
+
+    return transcript, quote_transcript, history_closes
+
+
+@pytest.mark.parametrize(
+    'replay_scenario',
+    ['basic-v1'],
+    indirect=True,
+)
+def test_real_feed_replay_is_repeatable(
+    replay_scenario: Path,
+    open_test_pikerd: Callable[
+        ...,
+        AbstractAsyncContextManager,
+    ],
+    loglevel: str,
+) -> None:
+    '''
+    Replay identical typed transcripts through fresh actor trees.
+
+    Mocked feed calls can hide backend discovery, datad and samplerd
+    startup, history publication, SHM writes, stream subscription
+    control, and context teardown failures. Run the same tracked
+    scenario through two fresh `open_test_pikerd()` generations.
+    Each generation receives the initial quote, pauses the production
+    feed, awaits the typed subscriber-state barrier, proves
+    advancement is rejected, resumes, repeats that gate check
+    through typed producer controls, and explicitly advances every
+    remaining event. Exact control bytes, quote dictionaries, and
+    history values must match, while service and named-SHM checks
+    prove each generation fully tears down before the next one
+    starts.
+
+    '''
+    assert replay_scenario.name == 'basic-v1.json'
+
+    async def main() -> None:
+        first = await _run_basic_generation(
+            open_test_pikerd,
+            loglevel,
+        )
+        second = await _run_basic_generation(
+            open_test_pikerd,
+            loglevel,
+        )
+        assert first == second
+
+    trio.run(main)
+
+
+@pytest.mark.parametrize(
+    'replay_scenario',
+    ['failure-v1'],
+    indirect=True,
+)
+def test_failure_injection_is_acknowledged(
+    replay_scenario: Path,
+    open_test_pikerd: Callable[
+        ...,
+        AbstractAsyncContextManager,
+    ],
+    loglevel: str,
+) -> None:
+    '''
+    Stop at a fixture failure without publishing its quote.
+
+    An injected provider disconnect used to require timing a task
+    crash and inferring its position from logs. Start the real feed,
+    arm the fixture's third event through typed control, publish
+    event two, then request event three. Its correlated negative
+    acknowledgement must identify the fixture error, retain event
+    sequence two, and expose `failed` state. Reading SHM after the
+    acknowledgement proves the rejected event was not hidden in the
+    sampler before normal actor and SHM teardown.
+
+    '''
+    assert replay_scenario.name == 'failure-v1.json'
+
+    async def main() -> None:
+        services: Services
+        shm_names: list[str] = []
+        async with (
+            open_test_pikerd() as (_, _, _, services),
+            open_feed(
+                [FQME],
+                loglevel=loglevel,
+            ) as feed,
+        ):
+            quotes: tractor.MsgStream = feed.streams['replay']
+            first: dict[str, Any] = await quotes.receive()
+            assert first[FQME]['replay_seq'] == 1
+
+            flume = feed.flumes[FQME]
+            shm_names.extend([
+                flume.hist_shm._shm.name,
+                flume.rt_shm._shm.name,
+            ])
+            portal: tractor.Portal = feed.portals[replay]
+            ctx: tractor.Context
+            async with portal.open_context(
+                replay.open_replay_control,
+            ) as (ctx, _):
+                async with ctx.open_stream() as control:
+                    with ctx.pld_rx.limit_plds(
+                        spec=replay.ReplayAck,
+                    ):
+                        armed: replay.ReplayAck = await _request(
+                            control,
+                            replay.FailAt(
+                                command_id=1,
+                                event_sequence=3,
+                            ),
+                        )
+                        assert armed.ok
+
+                        advanced: replay.ReplayAck = (
+                            await _request(
+                                control,
+                                replay.Advance(
+                                    command_id=2,
+                                    event_sequence=2,
+                                ),
+                            )
+                        )
+                        assert advanced.ok
+                        second: dict[str, Any] = (
+                            await quotes.receive()
+                        )
+                        assert second[FQME]['replay_seq'] == 2
+
+                        failed: replay.ReplayAck = await _request(
+                            control,
+                            replay.Advance(
+                                command_id=3,
+                                event_sequence=3,
+                            ),
+                        )
+                        assert not failed.ok
+                        assert failed.snapshot.state == 'failed'
+                        assert failed.snapshot.event_sequence == 2
+                        assert failed.error == (
+                            'fixture_disconnect: offline provider '
+                            'disconnected'
+                        )
+                        assert (
+                            flume.rt_shm.array['close'][-1]
+                            == 100.5
+                        )
+
+        assert not services.service_tasks
+        for shm_name in shm_names:
+            with pytest.raises(FileNotFoundError):
+                SharedMemory(name=shm_name)
+
+    trio.run(main)

tests/test_gap_overlays.py

@@ -2,22 +2,15 @@
 Typed chart-local gap-overlay regressions.
 
 '''
-from collections.abc import (
-    Callable,
-    Iterator,
-)
 from contextlib import AsyncExitStack
-import os
 from types import SimpleNamespace
 
-os.environ['QT_QPA_PLATFORM'] = 'offscreen'
-
 import msgspec
 import numpy as np
-from PyQt6.QtGui import QKeyEvent
 from PyQt6.QtWidgets import QGraphicsScene
 import pyqtgraph as pg
 import pytest
+from pytestqt.qtbot import QtBot
 import tractor
 from tractor._testing import tractor_test
 import trio
@@ -39,7 +32,6 @@ from piker.ui._interaction import (
 )
 from piker.ui.qt import (
     QApplication,
-    QEvent,
     QPointF,
     QRectF,
     Qt,
@@ -59,10 +51,12 @@ class _ChartStub:
         self,
         fqme: str,
         array: np.ndarray,
+        qtbot: QtBot,
 
     ) -> None:
         self.fqme: str = fqme
         self.widget: pg.PlotWidget = pg.PlotWidget()
+        qtbot.addWidget(self.widget)
         self.viz: SimpleNamespace = SimpleNamespace(
             plot=self.widget.plotItem,
             shm=SimpleNamespace(array=array),
@@ -115,6 +109,7 @@ def _ohlcv_array(
 
 def _display_state(
     fqme: str,
+    qtbot: QtBot,
 
 ) -> tuple[
     SimpleNamespace,
@@ -127,10 +122,12 @@ def _display_state(
     rt_chart: _ChartStub = _ChartStub(
         fqme,
         _ohlcv_array((1, 2, 4, 5)),
+        qtbot,
     )
     hist_chart: _ChartStub = _ChartStub(
         fqme,
         _ohlcv_array((60, 120, 300, 360)),
+        qtbot,
     )
     ds: SimpleNamespace = SimpleNamespace(
         fqme=fqme,
@@ -142,22 +139,6 @@ def _display_state(
     return ds, (rt_chart, hist_chart)
 
 
-@pytest.fixture(scope='session')
-def qapp() -> Iterator[QApplication]:
-    '''
-    Keep one offscreen Qt application alive for graphics tests.
-
-    '''
-    app: QApplication|None = QApplication.instance()
-    if app is None:
-        app = QApplication(['piker-gap-tests'])
-    app.setQuitOnLastWindowClosed(False)
-
-    yield app
-
-    app.processEvents()
-
-
 def test_gap_specs_and_wire_roundtrip() -> None:
     '''
     Local detection and remote IPC share one typed request model.
@@ -291,7 +272,7 @@ def test_gap_overlay_unknown_fqme_returns_typed_error() -> None:
 
 
 def test_gap_manager_real_qt_lifecycle(
-    qapp: QApplication,
+    qtbot: QtBot,
 
 ) -> None:
     '''
@@ -307,7 +288,7 @@ def test_gap_manager_real_qt_lifecycle(
     '''
     ds: SimpleNamespace
     charts: tuple[_ChartStub, _ChartStub]
-    ds, charts = _display_state(FQME)
+    ds, charts = _display_state(FQME, qtbot)
     annots: dict[int, GapAnnotations] = {}
     gapman: GapOverlayMngr = GapOverlayMngr(
         dss={FQME: ds},
@@ -403,14 +384,11 @@ def test_gap_manager_real_qt_lifecycle(
         gapman.remove_owner('chart-local')
         assert annots == {}
     finally:
-        chart: _ChartStub
+        gapman.remove_owner('chart-local')
-        for chart in charts:
-            chart.close()
-        qapp.processEvents()
 
 
 def test_gap_annotations_reposition_after_prepend(
-    qapp: QApplication,
+    qtbot: QtBot,
 
 ) -> None:
     '''
@@ -426,7 +404,7 @@ def test_gap_annotations_reposition_after_prepend(
     '''
     ds: SimpleNamespace
     charts: tuple[_ChartStub, _ChartStub]
-    ds, charts = _display_state(FQME)
+    ds, charts = _display_state(FQME, qtbot)
     annots: dict[int, GapAnnotations] = {}
     gapman: GapOverlayMngr = GapOverlayMngr(
         dss={FQME: ds},
@@ -445,7 +423,7 @@ def test_gap_annotations_reposition_after_prepend(
         old_scene_point: QPointF = item.mapToScene(
             old_br.center()
         )
-        qapp.processEvents()
+        QApplication.processEvents()
         assert item in scene.items(old_scene_point)
         old_rects: np.ndarray = (
             item._rectarray.ndarray().copy()
@@ -458,7 +436,7 @@ def test_gap_annotations_reposition_after_prepend(
             fqme=FQME,
             timeframe=60,
         )
-        qapp.processEvents()
+        QApplication.processEvents()
 
         new_rects: np.ndarray = item._rectarray.ndarray()
         np.testing.assert_allclose(
@@ -476,14 +454,10 @@ def test_gap_annotations_reposition_after_prepend(
         assert item.scene() is charts[1].widget.scene()
     finally:
         gapman.remove_owner('chart-local')
-        chart: _ChartStub
-        for chart in charts:
-            chart.close()
-        qapp.processEvents()
 
 
 def test_duplicate_fqme_layers_use_local_chart_identity(
-    qapp: QApplication,
+    qtbot: QtBot,
 
 ) -> None:
     '''
@@ -499,10 +473,10 @@ def test_duplicate_fqme_layers_use_local_chart_identity(
     '''
     first_ds: SimpleNamespace
     first_charts: tuple[_ChartStub, _ChartStub]
-    first_ds, first_charts = _display_state(FQME)
+    first_ds, first_charts = _display_state(FQME, qtbot)
     second_ds: SimpleNamespace
     second_charts: tuple[_ChartStub, _ChartStub]
-    second_ds, second_charts = _display_state(FQME)
+    second_ds, second_charts = _display_state(FQME, qtbot)
     annots: dict[int, GapAnnotations] = {}
     gapman: GapOverlayMngr = GapOverlayMngr(
         dss={FQME: second_ds},
@@ -534,14 +508,10 @@ def test_duplicate_fqme_layers_use_local_chart_identity(
         assert annots[second.aid].scene() is not None
     finally:
         gapman.remove_owner('chart-local')
-        chart: _ChartStub
-        for chart in (*first_charts, *second_charts):
-            chart.close()
-        qapp.processEvents()
 
 
 def test_startup_and_focused_toggle_use_real_qt(
-    qapp: QApplication,
+    qtbot: QtBot,
     monkeypatch: pytest.MonkeyPatch,
 
 ) -> None:
@@ -560,10 +530,16 @@ def test_startup_and_focused_toggle_use_real_qt(
 
     first_ds: SimpleNamespace
     first_charts: tuple[_ChartStub, _ChartStub]
-    first_ds, first_charts = _display_state('first.test')
+    first_ds, first_charts = _display_state(
+        'first.test',
+        qtbot,
+    )
     second_ds: SimpleNamespace
     second_charts: tuple[_ChartStub, _ChartStub]
-    second_ds, second_charts = _display_state('second.test')
+    second_ds, second_charts = _display_state(
+        'second.test',
+        qtbot,
+    )
     dss: dict[str, SimpleNamespace] = {
         first_ds.fqme: first_ds,
         second_ds.fqme: second_ds,
@@ -642,30 +618,28 @@ def test_startup_and_focused_toggle_use_real_qt(
         assert unrelated == []
     finally:
         gapman.remove_owner('chart-local')
-        chart: _ChartStub
-        for chart in (*first_charts, *second_charts):
-            chart.close()
-        qapp.processEvents()
 
 
 def test_ctrl_g_event_renders_real_history_overlay(
-    qapp: QApplication,
+    qtbot: QtBot,
 
 ) -> None:
     '''
-    Route a real Qt Ctrl-G event into a rendered history gap layer.
+    Route real QtBot Ctrl-G input into a rendered history gap layer.
 
     Calling the toggle helper directly does not prove that Qt event
     filtering, `KeyboardMsg` conversion or the asynchronous view-mode
     handler recognizes the configured binding. Install the production
-    `EventRelay` on a real widget, send one offscreen `QKeyEvent`,
+    `EventRelay` on a shown and focused widget, press Ctrl-G through
+    `QtBot.keyPress()` so Qt emits its modifier-only event before G,
     and wait until the handler blocks again. The resulting manager
-    and scene state prove the complete local keyboard path ran.
+    and scene state prove the complete local keyboard path ran
+    without crashing or directly dispatching a synthetic event.
 
     '''
     ds: SimpleNamespace
     charts: tuple[_ChartStub, _ChartStub]
-    ds, charts = _display_state(FQME)
+    ds, charts = _display_state(FQME, qtbot)
     annots: dict[int, GapAnnotations] = {}
     gapman: GapOverlayMngr = GapOverlayMngr(
         dss={FQME: ds},
@@ -686,12 +660,20 @@ def test_ctrl_g_event_renders_real_history_overlay(
         active=False,
     )
     source: QWidget = QWidget()
+    qtbot.addWidget(source)
     source.order_mode = order_mode
     source._chart = ds.hist_chart
     source.linked = SimpleNamespace(
         cursor=SimpleNamespace(in_query_mode=False),
     )
     source.setMouseMode = lambda mode: None
+    source.setFocusPolicy(Qt.FocusPolicy.StrongFocus)
+    with qtbot.waitExposed(source):
+        source.show()
+    source.raise_()
+    source.activateWindow()
+    source.setFocus()
+    qtbot.waitUntil(lambda: source.hasFocus())
 
     async def drive_key_event() -> None:
         '''
@@ -702,13 +684,13 @@ def test_ctrl_g_event_renders_real_history_overlay(
             source,
             dss={FQME: ds},
         ):
-            key_event: QKeyEvent = QKeyEvent(
+            qtbot.keyPress(
-                QEvent.Type.KeyPress,
+                source,
                 Qt.Key.Key_G,
-                Qt.KeyboardModifier.ControlModifier,
+                modifier=(
-                'g',
+                    Qt.KeyboardModifier.ControlModifier
+                ),
             )
-            assert QApplication.sendEvent(source, key_event)
             await wait_all_tasks_blocked()
 
     try:
@@ -727,12 +709,6 @@ def test_ctrl_g_event_renders_real_history_overlay(
         assert annots[aid].scene() is charts[1].widget.scene()
     finally:
         gapman.remove_owner('chart-local')
-        source.close()
-        source.deleteLater()
-        chart: _ChartStub
-        for chart in charts:
-            chart.close()
-        qapp.processEvents()
 
 
 _actor_requests: list[dict] = []
@@ -820,28 +796,28 @@ async def _delayed_gap_dialog(
 
 ) -> None:
     '''
-    Delay the first reply so client cancellation leaves it queued.
+    Hold the first reply until a second request arrives.
 
     '''
     await ctx.started([FQME])
     stream: tractor.MsgStream
     async with ctx.open_stream() as stream:
         with ctx.pld_rx.limit_plds(spec=SetGapOverlay):
-            req_count: int = 0
+            first: SetGapOverlay = await stream.receive()
-            req: SetGapOverlay
+            _delayed_request_ids.append(first.request_id)
-            async for req in stream:
-                req_count += 1
-                _delayed_request_ids.append(req.request_id)
-                if req_count == 1:
             await stream.send(GapOverlay(
-                        fqme=req.fqme,
+                fqme=first.fqme,
-                        timeframe=req.timeframe,
+                timeframe=first.timeframe,
-                        visible=req.visible,
+                visible=first.visible,
                 gap_count=0,
-                        request_id='actor-received',
+                request_id=(
+                    f'actor-received:{first.request_id}'
+                ),
             ))
-                    await trio.sleep(0.05)
 
+            second: SetGapOverlay = await stream.receive()
+            _delayed_request_ids.append(second.request_id)
+            for req in (first, second):
                 await stream.send(GapOverlay(
                     fqme=req.fqme,
                     timeframe=req.timeframe,
@@ -850,12 +826,15 @@ async def _delayed_gap_dialog(
                     request_id=req.request_id,
                 ))
 
+            async for unexpected in stream:
+                raise AssertionError(
+                    'unexpected third delayed request: '
+                    f'{unexpected!r}'
+                )
+
 
 @tractor_test(timeout=20)
-async def test_remote_gap_dialog_real_actor(
+async def test_remote_gap_dialog_real_actor() -> None:
-    monkeypatch: pytest.MonkeyPatch,
-
-) -> None:
     '''
     Exchange typed gap state through the production Tractor endpoint.
 
@@ -867,45 +846,22 @@ async def test_remote_gap_dialog_real_actor(
     the child receives the generated request ID and removes its owner
     after stream closure.
 
-    A second child context emits a receipt before delaying the first
+    A second child context emits a typed receipt containing the first
-    reply. Cancel that exact client task after the receipt, submit a
+    generated request ID, then blocks on receipt of a second request
-    second request on the shared stream, and prove the client
+    before publishing either correlated reply. A subscribed real
+    `MsgStream` receiver gives the parent an explicit publication
+    barrier. Cancel that exact client task after the receipt, await
+    its completion, submit the second request, and prove the client
     discards the queued first reply before returning the second. The
     child-side request IDs prove cancellation happened after
-    publication instead of merely preventing the first send.
+    publication instead of merely preventing the first send, without
+    an arbitrary scheduler delay or a patched logger.
 
     '''
     from piker.ui._remote_ctl import (
         AnnotClient,
         remote_gap_overlays,
     )
-    from piker.ui import _remote_ctl
-
-    receipt_seen: trio.Event = trio.Event()
-    original_warning: Callable[..., None] = (
-        _remote_ctl.log.warning
-    )
-
-    def observe_warning(
-        msg: str,
-        *args: object,
-        **kwargs: object,
-
-    ) -> None:
-        '''
-        Observe the stale receipt before cancelling its client task.
-
-        '''
-        if 'actor-received' in msg:
-            receipt_seen.set()
-        original_warning(msg, *args, **kwargs)
-
-    monkeypatch.setattr(
-        _remote_ctl.log,
-        'warning',
-        observe_warning,
-    )
-
     actor_nursery: tractor.ActorNursery
     async with tractor.open_nursery() as actor_nursery:
         portal: tractor.Portal = await actor_nursery.start_actor(
@@ -976,28 +932,59 @@ async def test_remote_gap_dialog_real_actor(
                     first_scope: trio.CancelScope = (
                         trio.CancelScope()
                     )
+                    first_done: trio.Event = trio.Event()
 
                     async def cancel_first_request() -> None:
                         '''
-                        Wait for cancellation inside the first
+                        Publish once and expose exact cancellation.
-                        dialog.
 
                         '''
+                        try:
                             with first_scope:
-                            await delayed_client.set_gap_overlay(
+                                await (
+                                    delayed_client.set_gap_overlay(
                                         SetGapOverlay(
                                             fqme=FQME,
                                             timeframe=60,
                                             specs=[],
                                         )
                                     )
+                                )
+                        finally:
+                            first_done.set()
 
                     nursery: trio.Nursery
                     async with trio.open_nursery() as nursery:
-                        nursery.start_soon(cancel_first_request)
+                        receipt: GapOverlay
-                        await receipt_seen.wait()
+                        async with (
+                            delayed_stream.subscribe()
+                            as receipt_stream,
+                        ):
+                            nursery.start_soon(
+                                cancel_first_request,
+                            )
+                            with (
+                                delayed_ctx.pld_rx.limit_plds(
+                                    spec=GapOverlay,
+                                )
+                            ):
+                                receipt = (
+                                    await receipt_stream.receive()
+                                )
+
+                        receipt_prefix: str = 'actor-received:'
+                        assert receipt.request_id.startswith(
+                            receipt_prefix,
+                        )
+                        first_request_id: str = (
+                            receipt.request_id.removeprefix(
+                                receipt_prefix,
+                            )
+                        )
+                        assert first_request_id
                         first_scope.cancel()
-                        await wait_all_tasks_blocked()
+                        await first_done.wait()
+                        assert first_scope.cancelled_caught
 
                         recovered: GapOverlay = (
                             await delayed_client.set_gap_overlay(
@@ -1010,7 +997,6 @@ async def test_remote_gap_dialog_real_actor(
                         )
                         assert recovered.fqme == FQME
                         assert recovered.request_id
-                        nursery.cancel_scope.cancel()
 
             delayed_snapshot: dict = await portal.run(
                 _gap_actor_snapshot,
@@ -1019,6 +1005,7 @@ async def test_remote_gap_dialog_real_actor(
                 delayed_snapshot['delayed_request_ids']
             )
             assert len(delayed_ids) == 2
+            assert delayed_ids[0] == first_request_id
             assert delayed_ids[0] != delayed_ids[1]
             assert delayed_ids[1] == recovered.request_id
         finally:

tests/ui/conftest.py

@@ -0,0 +1,252 @@
+from collections.abc import Iterator
+from pathlib import Path
+import shutil
+from types import SimpleNamespace
+
+from PyQt6.QtCore import (
+    QCoreApplication,
+    QEvent,
+    QSettings,
+)
+from PyQt6.QtWidgets import (
+    QApplication,
+    QWidget,
+)
+import pyqtgraph as pg
+from pyqtgraph import ViewBox
+import pytest
+from pytestqt.qtbot import QtBot
+
+from piker import config
+
+
+_PREEXISTING_SETTING: str = 'pytest/preexisting'
+_OWNED_SETTING: str = 'pytest/owned'
+
+
+@pytest.fixture(scope='session', autouse=True)
+def ui_process_sentinel(
+    isolated_xdg_config_home: Path,
+
+) -> Iterator[Path]:
+    '''
+    Keep known process-owned state across all UI tests.
+
+    '''
+    # TODO: expose public config-dir get/set APIs like
+    # `modden.config.dirs.get_conf_dir()` and `set_conf_dir()` so
+    # tests do not reach into Piker's import-cached path globals.
+    config_dir: Path = config._click_config_dir
+    config_dir.mkdir(parents=True, exist_ok=True)
+    sentinel_path: Path = config_dir / 'pytest-preexisting'
+    sentinel_existed: bool = sentinel_path.exists()
+    sentinel_bytes: bytes|None = None
+    if sentinel_existed:
+        sentinel_bytes = sentinel_path.read_bytes()
+    sentinel_path.write_text('keep-me', encoding='utf-8')
+
+    settings: QSettings = QSettings('pikers', 'piker')
+    settings.setFallbacksEnabled(False)
+    setting_existed: bool = settings.contains(
+        _PREEXISTING_SETTING,
+    )
+    setting_value: object = settings.value(
+        _PREEXISTING_SETTING,
+    )
+    settings.setValue(_PREEXISTING_SETTING, 'keep-me')
+    settings.sync()
+
+    try:
+        assert config_dir.is_relative_to(isolated_xdg_config_home)
+        yield sentinel_path
+    finally:
+        if setting_existed:
+            settings.setValue(
+                _PREEXISTING_SETTING,
+                setting_value,
+            )
+        else:
+            settings.remove(_PREEXISTING_SETTING)
+        settings.sync()
+
+        if sentinel_existed:
+            assert sentinel_bytes is not None
+            sentinel_path.write_bytes(sentinel_bytes)
+        else:
+            sentinel_path.unlink(missing_ok=True)
+
+
+@pytest.fixture(autouse=True)
+def isolated_ui_state(
+    qapp: QApplication,
+    qtbot: QtBot,
+    tmp_path: Path,
+    isolated_xdg_config_home: Path,
+    ui_process_sentinel: Path,
+
+) -> Iterator[SimpleNamespace]:
+    '''
+    Restore mutable process state after one real Qt test.
+
+    Pytest-qt owns the shared `QApplication` and closes widgets
+    before this fixture unwinds. This guard snapshots the remaining
+    process globals, removes exact test-owned config, and reports any
+    widget or PyQtGraph object that survived normal pytest-qt
+    cleanup.
+
+    '''
+    top_levels: set[QWidget] = set(
+        qapp.topLevelWidgets()
+    )
+    all_views: set[ViewBox] = set(ViewBox.AllViews)
+    named_views: dict[str, ViewBox] = dict(ViewBox.NamedViews)
+    pg_options: dict[str, object] = dict(pg.CONFIG_OPTIONS)
+    quit_on_last_window: bool = (
+        qapp.quitOnLastWindowClosed()
+    )
+    config_paths: dict[str, Path] = {
+        '_click_config_dir': config._click_config_dir,
+        '_config_dir': config._config_dir,
+        '_watchlists_data_path': config._watchlists_data_path,
+    }
+    settings: QSettings = QSettings('pikers', 'piker')
+    settings.setFallbacksEnabled(False)
+    settings_state: dict[str, object] = {
+        key: settings.value(key)
+        for key in settings.allKeys()
+    }
+    test_config_dir: Path = tmp_path / 'piker-config'
+    test_config_dir.mkdir()
+    state: SimpleNamespace = SimpleNamespace(
+        qapp=qapp,
+        process_config_home=isolated_xdg_config_home,
+        preexisting_config=ui_process_sentinel,
+        test_config_dir=test_config_dir,
+        preexisting_setting=_PREEXISTING_SETTING,
+        owned_setting=_OWNED_SETTING,
+    )
+
+    try:
+        yield state
+    finally:
+        # Collect every isolation failure so cleanup of one process
+        # global cannot prevent restoration of the remaining globals.
+        errors: list[Exception] = []
+
+        # Pytest-qt closes registered widgets before fixture
+        # teardown, but Qt completes `QWidget.deleteLater()`
+        # asynchronously.
+        QCoreApplication.sendPostedEvents(
+            None,
+            QEvent.Type.DeferredDelete,
+        )
+        qapp.processEvents()
+
+        # Any addition to `QApplication.topLevelWidgets()` after the
+        # deferred-delete drain is a widget pytest-qt failed to own.
+        leaked_widgets: set[QWidget] = (
+            set(qapp.topLevelWidgets()) - top_levels
+        )
+        for widget in leaked_widgets:
+            widget.close()
+            widget.deleteLater()
+        QCoreApplication.sendPostedEvents(
+            None,
+            QEvent.Type.DeferredDelete,
+        )
+        qapp.processEvents()
+        if leaked_widgets:
+            errors.append(AssertionError(
+                'pytest-qt left test-owned top-level widgets alive: '
+                f'{leaked_widgets!r}'
+            ))
+
+        # `ViewBox.AllViews` and `ViewBox.NamedViews` are
+        # process-wide weak registries used for linked-axis
+        # discovery.
+        # Close leaked views, report either registry delta, then
+        # restore the exact baseline so a failure cannot contaminate
+        # the next test.
+        leaked_views: set[ViewBox] = (
+            set(ViewBox.AllViews) - all_views
+        )
+        for view in leaked_views:
+            view.close()
+        if leaked_views:
+            errors.append(AssertionError(
+                'test-owned `ViewBox.AllViews` entries survived: '
+                f'{leaked_views!r}'
+            ))
+        named_views_changed: bool = (
+            dict(ViewBox.NamedViews) != named_views
+        )
+        if named_views_changed:
+            errors.append(AssertionError(
+                'test mutated `ViewBox.NamedViews` without cleanup'
+            ))
+        ViewBox.AllViews.clear()
+        ViewBox.AllViews.update({
+            view: None
+            for view in all_views
+        })
+        ViewBox.NamedViews.clear()
+        ViewBox.NamedViews.update(named_views)
+        ViewBox.updateAllViewLists()
+
+        # PyQtGraph options and Qt's last-window policy are mutable
+        # process globals, not properties owned by one test widget.
+        pg.CONFIG_OPTIONS.clear()
+        pg.CONFIG_OPTIONS.update(pg_options)
+        qapp.setQuitOnLastWindowClosed(quit_on_last_window)
+
+        # Rebuild this test application's `QSettings` keys exactly:
+        # test-owned keys disappear while pre-existing values
+        # survive.
+        for key in settings.allKeys():
+            settings.remove(key)
+        for key, value in settings_state.items():
+            settings.setValue(key, value)
+        settings.sync()
+
+        # Restore Piker's import-cached path globals before deleting
+        # the function-owned config tree to prevent a dangling path.
+        for name, path in config_paths.items():
+            setattr(config, name, path)
+        shutil.rmtree(test_config_dir, ignore_errors=True)
+
+        # Re-read every restored process value instead of trusting
+        # the cleanup assignments; each mismatch joins one report.
+        if dict(pg.CONFIG_OPTIONS) != pg_options:
+            errors.append(AssertionError(
+                '`pyqtgraph.CONFIG_OPTIONS` was not restored'
+            ))
+        if qapp.quitOnLastWindowClosed() != quit_on_last_window:
+            errors.append(AssertionError(
+                '`QApplication.quitOnLastWindowClosed()` changed'
+            ))
+        if {
+            name: getattr(config, name)
+            for name in config_paths
+        } != config_paths:
+            errors.append(AssertionError(
+                'Piker config path globals were not restored'
+            ))
+        if {
+            key: settings.value(key)
+            for key in settings.allKeys()
+        } != settings_state:
+            errors.append(AssertionError(
+                '`QSettings` keys were not restored exactly'
+            ))
+        if test_config_dir.exists():
+            errors.append(AssertionError(
+                'test-owned config directory survived teardown'
+            ))
+
+        # Raise only after all restoration and verification
+        # completes, preserving each leak signal for diagnosis.
+        if errors:
+            raise ExceptionGroup(
+                'UI isolation teardown failed',
+                errors,
+            )

tests/ui/test_harness.py

@@ -0,0 +1,126 @@
+from pathlib import Path
+from types import SimpleNamespace
+
+from PyQt6.QtCore import QSettings
+from PyQt6.QtWidgets import (
+    QApplication,
+    QWidget,
+)
+import pyqtgraph as pg
+import pytest
+from pytestqt.qtbot import QtBot
+
+from piker import config
+
+
+@pytest.fixture(scope='module')
+def module_qapp(
+    qapp: QApplication,
+
+) -> QApplication:
+    '''
+    Pin pytest-qt's session application for this test module.
+
+    '''
+    return qapp
+
+
+def test_ui_state_is_process_isolated_and_restored(
+    isolated_ui_state: SimpleNamespace,
+    module_qapp: QApplication,
+    pytestconfig: pytest.Config,
+    qtbot: QtBot,
+
+) -> None:
+    '''
+    Prove imports and one UI test cannot reach user config state.
+
+    Previously `qapp_args` assigned `XDG_CONFIG_HOME` only after
+    Piker was imported, so `piker.config` cached real user paths
+    before Qt setup. This test verifies every cached path starts
+    below the process-owned temporary XDG root, then redirects those
+    globals and writes both a file and `QSettings` key. The fixture's
+    exact-state teardown proves those mutations disappear while its
+    process sentinels survive for the next test.
+
+    '''
+    state: SimpleNamespace = isolated_ui_state
+    assert module_qapp is state.qapp
+    assert QApplication.instance() is state.qapp
+    if pytestconfig.getoption('headless'):
+        assert state.qapp.platformName() == 'offscreen'
+    assert config._click_config_dir.is_relative_to(
+        state.process_config_home,
+    )
+    assert config._config_dir.is_relative_to(
+        state.process_config_home,
+    )
+    assert config._watchlists_data_path.is_relative_to(
+        state.process_config_home,
+    )
+
+    widget: QWidget = QWidget()
+    qtbot.addWidget(widget)
+    widget.show()
+
+    owned_dir: Path = state.test_config_dir
+    config._click_config_dir = owned_dir
+    config._config_dir = owned_dir
+    config._watchlists_data_path = owned_dir / 'watchlists.json'
+    (owned_dir / 'owned').write_text('remove-me', encoding='utf-8')
+
+    settings: QSettings = QSettings('pikers', 'piker')
+    settings.setFallbacksEnabled(False)
+    settings_path: Path = Path(settings.fileName())
+    assert settings_path.is_relative_to(state.process_config_home)
+    assert settings.value(state.preexisting_setting) == 'keep-me'
+    settings.setValue(state.owned_setting, 'remove-me')
+    settings.sync()
+
+
+def test_ui_state_reuses_qapp_without_previous_test_leaks(
+    isolated_ui_state: SimpleNamespace,
+    module_qapp: QApplication,
+    qtbot: QtBot,
+
+) -> None:
+    '''
+    Prove repeated tests reuse Qt without inheriting mutable state.
+
+    A session `QApplication` can retain widgets, settings and
+    PyQtGraph's `ViewBox.AllViews` entries from an earlier test. This
+    test runs after the mutation case, verifies the same application
+    and preserved process sentinel but no test-owned key, then
+    creates a real named `PlotWidget` and mutates Qt/PyQtGraph
+    options. Normal pytest-qt widget closure plus the isolation
+    fixture's registry assertions prove the second test also leaves
+    no state behind.
+
+    '''
+    state: SimpleNamespace = isolated_ui_state
+    assert module_qapp is state.qapp
+    assert QApplication.instance() is state.qapp
+
+    settings: QSettings = QSettings('pikers', 'piker')
+    settings.setFallbacksEnabled(False)
+    assert settings.value(state.preexisting_setting) == 'keep-me'
+    assert not settings.contains(state.owned_setting)
+    assert state.preexisting_config.read_text(
+        encoding='utf-8',
+    ) == 'keep-me'
+    assert list(state.test_config_dir.iterdir()) == []
+
+    plot: pg.PlotWidget = pg.PlotWidget(name='pytest-owned-view')
+    qtbot.addWidget(plot)
+    plot.show()
+    assert pg.ViewBox.NamedViews['pytest-owned-view'] is (
+        plot.plotItem.vb
+    )
+
+    pg.setConfigOption(
+        'antialias',
+        not pg.getConfigOption('antialias'),
+    )
+    state.qapp.setQuitOnLastWindowClosed(
+        not state.qapp.quitOnLastWindowClosed(),
+    )

uv.lock

@@ -1132,6 +1132,7 @@ dev = [
     { name = "pyqt6" },
     { name = "pyqtgraph" },
     { name = "pytest" },
+    { name = "pytest-qt" },
     { name = "qdarkstyle" },
     { name = "rapidfuzz" },
     { name = "xonsh" },
@@ -1149,6 +1150,7 @@ repl = [
 ]
 testing = [
     { name = "pytest" },
+    { name = "pytest-qt" },
 ]
 uis = [
     { name = "pyqt6" },
@@ -1206,6 +1208,7 @@ dev = [
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
     { name = "pyqtgraph", specifier = ">=0.14.0" },
     { name = "pytest" },
+    { name = "pytest-qt" },
     { name = "qdarkstyle", specifier = ">=3.0.2,<4.0.0" },
     { name = "rapidfuzz", specifier = ">=3.2.0,<4.0.0" },
     { name = "xonsh", specifier = ">=0.23.0" },
@@ -1219,7 +1222,10 @@ repl = [
     { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "xonsh", specifier = ">=0.23.0" },
 ]
-testing = [{ name = "pytest" }]
+testing = [
+    { name = "pytest" },
+    { name = "pytest-qt" },
+]
 uis = [
     { name = "pyqt6", specifier = ">=6.7.0,<7.0.0" },
     { name = "pyqtgraph", specifier = ">=0.14.0" },
@@ -1629,6 +1635,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
 ]
 
+[[package]]
+name = "pytest-qt"
+version = "4.5.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pluggy" },
+    { name = "pytest" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/d3/61/8bdec02663c18bf5016709b909411dce04a868710477dc9b9844ffcf8dd2/pytest_qt-4.5.0.tar.gz", hash = "sha256:51620e01c488f065d2036425cbc1cbcf8a6972295105fd285321eb47e66a319f", size = 128702, upload-time = "2025-07-01T17:24:39.889Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/cc/d0/8339b888ad64a3d4e508fed8245a402b503846e1972c10ad60955883dcbb/pytest_qt-4.5.0-py3-none-any.whl", hash = "sha256:ed21ea9b861247f7d18090a26bfbda8fb51d7a8a7b6f776157426ff2ccf26eff", size = 37214, upload-time = "2025-07-01T17:24:38.226Z" },
+]
+
 [[package]]
 name = "python-baseconv"
 version = "1.2.2"