Synchronize `replay` with feed subscribers

Deats,
- notify optional backend observers after subscriber mutations
- gate replay advancement on real `Feed.pause()` state changes
- exercise `pikerd`, `datad`, `samplerd`, SHM and typed IPC
- prove repeatable transcripts, acknowledged faults and teardown

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

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

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.

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):

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_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)