Add opt-in Kraken chart lifecycle reproducer

Preserve the production chart journey behind
`--live-provider=kraken`; ordinary runs skip it without network
access. Public Kraken data needs no API key. Supply empty owned
broker config, a unique registry and explicit paper EMS without
submitting orders.

Observe readiness or guest failure, retain original errors, and
check observed actor endpoints and chart SHM after cancellation.
These observations do not prove all descendant PIDs have exited
or account for unpublished child SHM.

The selected live run is known-failing: readiness can precede
`rt_linked.mode`, Qt reports abstract paint calls, and teardown
left a paper actor requiring explicit SIGINT recovery. Preserve
this reproducer for the follow-up; do not treat its default skip
as successful live qualification.

Generation and failure evidence were recorded with the preceding
prerequisite commit in the existing session provenance.

Prompt-IO: ai/prompt-io/opencode/20260908T215045Z_1a2d4893_prompt_io.md

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

tests/conftest.py

@@ -123,6 +123,12 @@ def qapp_args(
 
 
 def pytest_addoption(parser):
+    parser.addoption(
+        '--live-provider',
+        choices=('kraken',),
+        default=None,
+        help='Opt into public-provider chart qualification',
+    )
     parser.addoption(
         '--headless',
         action='store_true',

tests/ui/test_live_chart.py

@@ -0,0 +1,213 @@
+from collections.abc import Callable
+from pathlib import Path
+import socket
+from time import monotonic
+from types import SimpleNamespace
+
+import numpy as np
+from PyQt6.QtWidgets import QApplication
+import pytest
+from pytestqt.qtbot import QtBot
+import tractor
+from tractor.ipc import _shm
+
+from piker import config
+from piker.brokers import get_brokermod
+from piker.ui._app import _async_main
+from piker.ui._exec import start_qtractor
+from piker.ui._widget import GodWidget
+
+
+def test_kraken_chart_boots_and_closes(
+    request: pytest.FixtureRequest,
+    isolated_ui_state: SimpleNamespace,
+    qapp: QApplication,
+    qtbot: QtBot,
+    reg_addr: tuple[str, int],
+    monkeypatch: pytest.MonkeyPatch,
+    record_property: Callable[[str, object], None],
+
+) -> None:
+    '''
+    Qualify production chart boot against public Kraken, not mocks.
+
+    A blocking Qt host and hidden Trio errors previously prevented
+    testing the composed chart. Explicit selection starts GodWidget
+    and `_async_main` with paper EMS, an empty owned broker config,
+    and the test's unique registry. No orders or input are submitted.
+    Readiness races guest completion so a provider/startup exception
+    is unwrapped rather than replaced by a readiness timeout.
+
+    Nonempty historical/realtime visualizations and a later price
+    change prove actual feed consumption. Registry snapshots check
+    paperboi and absence of brokerd; exact chart SHM names and actor
+    endpoints are retained for non-mutating teardown checks. These
+    snapshots cannot rule out short-lived actors between polls or
+    account for SHM never published to a chart before early failure.
+
+    Every exit requests cancellation and pumps Qt to guest completion
+    before pytest-qt may destroy widgets. Cleanup and guest failures
+    are retained together rather than hiding either original error.
+
+    '''
+    if request.config.getoption('--live-provider') != 'kraken':
+        pytest.skip('requires --live-provider=kraken')
+
+    fqme: str = 'xbtusd.spot.kraken'
+    confdir: Path = isolated_ui_state.test_config_dir
+    # A missing brokers.toml loads templates with example accounts.
+    (confdir / 'brokers.toml').write_text('', encoding='utf-8')
+    for name in ('_config_dir', '_click_config_dir'):
+        monkeypatch.setattr(config, name, confdir)
+    monkeypatch.setattr(
+        config,
+        '_watchlists_data_path',
+        confdir / 'watchlists.json',
+    )
+    assert config.load('brokers')[0] == {}
+
+    owned_actors: dict[tuple[str, str], list] = {}
+    shm_names: set[str] = set()
+    errors: list[BaseException] = []
+    started: float = monotonic()
+    session = start_qtractor(
+        func=_async_main,
+        args=(
+            [fqme],
+            {'kraken': get_brokermod('kraken')},
+            'warning',
+            'paper',
+        ),
+        main_widget_type=GodWidget,
+        tractor_kwargs={
+            'registry_addrs': [reg_addr],
+            'tractor_runtime_overrides': {
+                'piker_test_dir': str(confdir),
+            },
+            'loglevel': 'warning',
+        },
+        app=qapp,
+    )
+
+    def observe_runtime() -> None:
+        '''
+        Retain identities from this guest's local Registrar only.
+
+        '''
+        actor = tractor.current_actor(err_on_no_runtime=False)
+        if actor is not None:
+            assert actor.is_registrar
+            assert reg_addr in actor.accept_addrs
+            owned_actors.update(actor._registry)
+            assert not any(
+                uid[0].startswith('brokerd.')
+                for uid in owned_actors
+            )
+        for attr in ('hist_linked', 'rt_linked'):
+            linked = getattr(session.main_widget, attr, None)
+            chart = getattr(linked, 'chart', None)
+            viz = getattr(chart, 'main_viz', None)
+            if viz is not None:
+                token: dict = viz.shm.token
+                shm_names.update(
+                    token[key] for key in (
+                        'shm_name',
+                        'shm_first_index_name',
+                        'shm_last_index_name',
+                    )
+                )
+
+    def boot_finished() -> bool:
+        observe_runtime()
+        return session.is_ready or session.outcome is not None
+
+    try:
+        qtbot.addWidget(session.window)
+        qtbot.waitUntil(boot_finished, timeout=120_000)
+        record_property('boot_seconds', monotonic() - started)
+        if session.outcome is None:
+            godw = session.main_widget
+            assert session.is_ready
+            assert godw.exec_mode == 'paper'
+            assert config.get_conf_dir() == confdir
+            assert tractor.runtime._state._runtime_vars[
+                'piker_vars'
+            ]['piker_test_dir'] == str(confdir)
+            names: set[str] = {uid[0] for uid in owned_actors}
+            assert {
+                'pikerd', 'datad.kraken', 'samplerd',
+                'emsd', 'paperboi.kraken',
+            } <= names
+            assert set(godw.rt_linked.mode.trackers) == {'paper'}
+            assert not godw.rt_linked.mode.client._sent_orders
+            assert session.window.windowTitle().startswith(fqme)
+            for linked in (godw.hist_linked, godw.rt_linked):
+                assert linked.mkt.fqme == fqme
+                viz = linked.chart.main_viz
+                data = viz.shm.array
+                assert len(data) > 1
+                assert np.isfinite(data['close']).all()
+                assert (data['close'] > 0).any()
+
+            rt_shm = godw.rt_linked.chart.main_viz.shm
+            before = rt_shm.array[-1].copy()
+
+            def live_update_or_exit() -> bool:
+                observe_runtime()
+                if session.outcome is not None:
+                    return True
+                # A sampler clock tick alone must not satisfy this.
+                return bool(
+                    rt_shm.array[-1]['close'] != before['close']
+                )
+
+            qtbot.waitUntil(live_update_or_exit, timeout=60_000)
+    except BaseException as exc:
+        errors.append(exc)
+    finally:
+        session.request_shutdown()
+        try:
+            qtbot.waitUntil(
+                lambda: session.outcome is not None,
+                timeout=60_000,
+            )
+            session.outcome.unwrap()
+        except BaseException as exc:
+            errors.append(exc)
+        if session.outcome is not None:
+            # Early failure can leave these outside Qt parent trees.
+            for attr in ('search', 'hist_linked', 'rt_linked'):
+                widget = getattr(session.main_widget, attr, None)
+                if widget is not None and widget.parent() is None:
+                    qtbot.addWidget(widget)
+        record_property('owned_actors', repr(owned_actors))
+        record_property('chart_shm_names', sorted(shm_names))
+
+    # Check exact survivors even when boot assertions already failed.
+    if tractor.current_actor(err_on_no_runtime=False) is not None:
+        errors.append(AssertionError('Guest runtime survived'))
+    for name in shm_names:
+        try:
+            segment = _shm.SharedMemory(name=name, create=False)
+        except FileNotFoundError:
+            continue
+        segment.close()
+        errors.append(AssertionError(
+            f'Chart SHM survived guest completion: {name}'
+        ))
+    endpoints: set[tuple] = {reg_addr}
+    for addrs in owned_actors.values():
+        endpoints.update(map(tuple, addrs))
+    for addr in endpoints:
+        with socket.socket() as sock:
+            sock.settimeout(1)
+            if sock.connect_ex(addr) == 0:
+                errors.append(AssertionError(
+                    f'Owned actor endpoint survived: {addr}'
+                ))
+
+    if errors:
+        if len(errors) == 1:
+            raise errors[0]
+        raise BaseExceptionGroup('Chart and shutdown failed', errors)
+    assert session.is_ready, 'guest returned before readiness'