piker/tests/test_gap_overlays.py

'''
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
import tractor
from tractor._testing import tractor_test
import trio
from trio.testing import wait_all_tasks_blocked

from piker.ui._annotate import GapAnnotations
from piker.ui._display import _register_gap_overlays
from piker.ui._gaps import (
    GapOverlay,
    GapOverlayMngr,
    GapOverlayPayload,
    GapSpec,
    SetGapOverlay,
    gap_specs_from_ohlcv,
)
from piker.ui._interaction import (
    ChartView,
    _toggle_gap_overlays,
)
from piker.ui.qt import (
    QApplication,
    QEvent,
    QPointF,
    QRectF,
    Qt,
    QWidget,
)


FQME: str = 'gap.test'


class _ChartStub:
    '''
    Real Qt plot with the small chart API required by `gapman`.

    '''
    def __init__(
        self,
        fqme: str,
        array: np.ndarray,

    ) -> None:
        self.fqme: str = fqme
        self.widget: pg.PlotWidget = pg.PlotWidget()
        self.viz: SimpleNamespace = SimpleNamespace(
            plot=self.widget.plotItem,
            shm=SimpleNamespace(array=array),
        )

    def get_viz(
        self,
        fqme: str,

    ) -> SimpleNamespace:
        '''
        Return this chart's only viz.

        '''
        assert fqme == self.fqme
        return self.viz

    def close(self) -> None:
        '''
        Release this test-owned plot widget.

        '''
        self.widget.close()
        self.widget.deleteLater()


def _ohlcv_array(
    times: tuple[float, ...],

) -> np.ndarray:
    '''
    Build the minimal structured OHLCV used by gap rendering.

    '''
    dtype: np.dtype = np.dtype([
        ('index', 'i8'),
        ('time', 'f8'),
        ('open', 'f8'),
        ('close', 'f8'),
    ])
    rows: list[tuple[int, float, float, float]] = []
    i: int
    time_s: float
    for i, time_s in enumerate(times):
        price: float = 100 + i
        rows.append((i, time_s, price, price + 0.5))

    return np.array(rows, dtype=dtype)


def _display_state(
    fqme: str,

) -> tuple[
    SimpleNamespace,
    tuple[_ChartStub, _ChartStub],
]:
    '''
    Build independent realtime and history chart stubs.

    '''
    rt_chart: _ChartStub = _ChartStub(
        fqme,
        _ohlcv_array((1, 2, 4, 5)),
    )
    hist_chart: _ChartStub = _ChartStub(
        fqme,
        _ohlcv_array((60, 120, 300, 360)),
    )
    ds: SimpleNamespace = SimpleNamespace(
        fqme=fqme,
        chart=rt_chart,
        viz=rt_chart.viz,
        hist_chart=hist_chart,
        hist_viz=hist_chart.viz,
    )
    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.

    Build a small sorted OHLCV array with one positive timestamp gap,
    derive its geometry without a per-row dataframe scan, and
    roundtrip the request/reply through msgpack. This proves callers
    invoke one chart renderer with one concrete schema.

    '''
    dtype: np.dtype = np.dtype([
        ('index', 'i8'),
        ('time', 'f8'),
        ('open', 'f8'),
        ('close', 'f8'),
    ])
    array: np.ndarray = np.array([
        (10, 60, 100, 101),
        (11, 120, 101, 102),
        (12, 300, 95, 96),
        (13, 360, 96, 97),
    ], dtype=dtype)

    specs: list[GapSpec] = gap_specs_from_ohlcv(
        array,
        period_s=60,
    )
    assert len(specs) == 1
    spec: GapSpec = specs[0]
    assert spec.start_time == 120
    assert spec.end_time == 300
    assert spec.start_pos == (11.9, 102)
    assert spec.end_pos == (12.1, 95)
    assert spec.pointing == 'down'

    req: SetGapOverlay = SetGapOverlay(
        fqme='mnq.cme.ib',
        timeframe=60,
        specs=specs,
        request_id='req-616',
    )
    req_wire: bytes = msgspec.msgpack.encode(req)
    assert msgspec.msgpack.decode(
        req_wire,
        type=SetGapOverlay,
    ) == req

    resp: GapOverlay = GapOverlay(
        fqme=req.fqme,
        timeframe=req.timeframe,
        visible=True,
        gap_count=1,
        aid=616,
        request_id=req.request_id,
    )
    resp_wire: bytes = msgspec.msgpack.encode(resp)
    assert msgspec.msgpack.decode(
        resp_wire,
        type=GapOverlay,
    ) == resp


def test_gap_overlay_payload_rejects_other_msgs() -> None:
    '''
    Reject payloads outside the strict gap-overlay dialog protocol.

    The sibling Tractor context installs `GapOverlayPayload` so
    legacy annotation dicts cannot enter its stream as partially
    validated commands. Encode one such dict and prove the same
    msgspec decoder rejects it before endpoint dispatch.

    '''
    invalid_wire: bytes = msgspec.msgpack.encode({
        'cmd': 'SelectRect',
    })

    with pytest.raises(msgspec.ValidationError):
        msgspec.msgpack.decode(
            invalid_wire,
            type=GapOverlayPayload,
        )

    req: SetGapOverlay = SetGapOverlay(
        fqme='mnq.cme.ib',
        timeframe=60,
        specs=[],
    )
    resp: GapOverlay = GapOverlay(
        fqme=req.fqme,
        timeframe=req.timeframe,
        visible=False,
        gap_count=0,
    )
    with pytest.raises(msgspec.ValidationError):
        msgspec.msgpack.decode(
            msgspec.msgpack.encode(resp),
            type=SetGapOverlay,
        )
    with pytest.raises(msgspec.ValidationError):
        msgspec.msgpack.decode(
            msgspec.msgpack.encode(req),
            type=GapOverlay,
        )


def test_gap_overlay_unknown_fqme_returns_typed_error() -> None:
    '''
    Keep a stale remote FQME request inside the typed dialog.

    A chart can unload or replace display state after a client learns
    its FQME set. Submit a valid request to an empty controller and
    prove it returns `GapOverlay.error` instead of leaking a
    `KeyError` which would terminate the whole Tractor context.

    '''
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={},
        annots={},
    )
    req: SetGapOverlay = SetGapOverlay(
        fqme='gone.test',
        timeframe=60,
        specs=[],
        request_id='req-stale',
    )

    resp: GapOverlay = gapman.apply(req)

    assert resp.request_id == req.request_id
    assert resp.error == 'No display state for fqme=gone.test'


def test_gap_manager_real_qt_lifecycle(
    qapp: QApplication,

) -> None:
    '''
    Remove overlays from every PyQtGraph registry.

    Gap layers are inserted through `PlotItem.addItem()`. Removing
    the item directly from its scene leaves stale `PlotItem.items`
    and `ViewBox.addedItems` refs which can leak memory and affect
    range calculations. This test uses real offscreen Qt plots,
    replaces one remotely owned layer, hides it, and proves each
    registry releases the old graphics object.

    '''
    ds: SimpleNamespace
    charts: tuple[_ChartStub, _ChartStub]
    ds, charts = _display_state(FQME)
    annots: dict[int, GapAnnotations] = {}
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={FQME: ds},
        annots=annots,
    )
    specs: list[GapSpec] = gap_specs_from_ohlcv(
        ds.hist_viz.shm.array,
        period_s=60,
    )
    aids: set[int] = set()
    req: SetGapOverlay = SetGapOverlay(
        fqme=FQME,
        timeframe=60,
        specs=specs,
        request_id='first',
    )

    try:
        first: GapOverlay = gapman.apply(
            req=req,
            owner='remote-test',
            aids=aids,
        )
        assert first.aid is not None
        first_item: GapAnnotations = annots[first.aid]
        assert first_item.scene() is charts[1].widget.scene()
        assert first_item in ds.hist_viz.plot.items
        assert first_item not in ds.hist_viz.plot.vb.addedItems
        assert aids == {first.aid}

        second: GapOverlay = gapman.apply(
            req=SetGapOverlay(
                fqme=FQME,
                timeframe=60,
                specs=specs,
                request_id='second',
            ),
            owner='remote-test',
            aids=aids,
        )
        assert second.aid is not None
        assert second.aid != first.aid
        assert first.aid not in annots
        assert first_item.scene() is None
        assert first_item not in ds.hist_viz.plot.items
        assert first_item not in ds.hist_viz.plot.vb.addedItems
        assert aids == {second.aid}

        hidden: GapOverlay = gapman.apply(
            req=SetGapOverlay(
                fqme=FQME,
                timeframe=60,
                specs=specs,
                visible=False,
                request_id='hidden',
            ),
            owner='remote-test',
            aids=aids,
        )
        assert hidden.visible is False
        assert hidden.aid is None
        assert aids == set()
        assert annots == {}
        assert gapman._layers == {}
        assert gapman._plots == {}
        item: object
        for item in ds.hist_viz.plot.items:
            assert not isinstance(item, GapAnnotations)

        local: GapOverlay = gapman.apply(
            req=SetGapOverlay(
                fqme=FQME,
                timeframe=60,
                specs=specs,
                request_id='local',
            ),
        )
        remote: GapOverlay = gapman.apply(
            req=SetGapOverlay(
                fqme=FQME,
                timeframe=60,
                specs=specs,
                request_id='remote',
            ),
            owner='remote-test',
        )
        assert local.aid is not None
        assert remote.aid is not None
        gapman.remove_owner('remote-test')
        assert local.aid in annots
        assert remote.aid not in annots
        assert annots[local.aid].scene() is not None
        gapman.remove_owner('chart-local')
        assert annots == {}
    finally:
        chart: _ChartStub
        for chart in charts:
            chart.close()
        qapp.processEvents()


def test_gap_annotations_reposition_after_prepend(
    qapp: QApplication,

) -> None:
    '''
    Reposition live gap geometry after history index offsets change.

    A history prepend preserves timestamps but shifts every absolute
    chart index. Gap rectangles and arrows otherwise remain attached
    to their pre-prepend x coordinates. Render one real gap item,
    shift its source indexes by ten, call `reposition()`, and prove
    its rectangle bounds and arrow spec move together while the item
    stays attached to its real Qt scene.

    '''
    ds: SimpleNamespace
    charts: tuple[_ChartStub, _ChartStub]
    ds, charts = _display_state(FQME)
    annots: dict[int, GapAnnotations] = {}
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={FQME: ds},
        annots=annots,
    )

    try:
        gap: GapOverlay = gapman.refresh(
            ds=ds,
            timeframe=60,
        )
        assert gap.aid is not None
        item: GapAnnotations = annots[gap.aid]
        old_br: QRectF = item.boundingRect()
        scene: QGraphicsScene = charts[1].widget.scene()
        old_scene_point: QPointF = item.mapToScene(
            old_br.center()
        )
        qapp.processEvents()
        assert item in scene.items(old_scene_point)
        old_rects: np.ndarray = (
            item._rectarray.ndarray().copy()
        )
        shifted: np.ndarray = ds.hist_viz.shm.array.copy()
        shifted['index'] += 10

        item.reposition(
            array=shifted,
            fqme=FQME,
            timeframe=60,
        )
        qapp.processEvents()

        new_rects: np.ndarray = item._rectarray.ndarray()
        np.testing.assert_allclose(
            new_rects[:, 0],
            old_rects[:, 0] + 10,
        )
        assert item._gap_specs[0]['arrow_x'] == 12
        new_br: QRectF = item.boundingRect()
        new_scene_point: QPointF = item.mapToScene(
            new_br.center()
        )
        assert new_br.left() == old_br.left() + 10
        assert item in scene.items(new_scene_point)
        assert item not in scene.items(old_scene_point)
        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,

) -> None:
    '''
    Keep cached displays for one FQME in separate local layers.

    The actor-global display lookup is keyed by FQME and can
    represent only its currently registered display. Local cached
    charts still pass their exact `DisplayState` into
    `GapOverlayMngr.refresh()`. Build two independent chart pairs
    with the same FQME, refresh both, and prove their layers survive
    independently.

    '''
    first_ds: SimpleNamespace
    first_charts: tuple[_ChartStub, _ChartStub]
    first_ds, first_charts = _display_state(FQME)
    second_ds: SimpleNamespace
    second_charts: tuple[_ChartStub, _ChartStub]
    second_ds, second_charts = _display_state(FQME)
    annots: dict[int, GapAnnotations] = {}
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={FQME: second_ds},
        annots=annots,
    )

    try:
        first: GapOverlay = gapman.refresh(
            ds=first_ds,
            timeframe=60,
        )
        second: GapOverlay = gapman.refresh(
            ds=second_ds,
            timeframe=60,
        )
        assert first.aid is not None
        assert second.aid is not None
        assert first.aid != second.aid
        assert len(gapman._layers) == 2
        chart_ids: set[int] = set()
        key: tuple[str, int, str, float]
        for key in gapman._layers:
            chart_ids.add(key[1])
        assert chart_ids == {
            id(first_ds.hist_chart),
            id(second_ds.hist_chart),
        }
        assert annots[first.aid].scene() is not None
        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,
    monkeypatch: pytest.MonkeyPatch,

) -> None:
    '''
    Register history gaps and toggle the focused chart only.

    Startup previously lived inside the quote loop and keyboard
    routing treated every state unrelated to the focused history
    chart as a realtime target. Two independent chart pairs reproduce
    that skew. The startup helper must create only 60-second layers;
    then a focused history toggle must hide only its own layer while
    leaving the other cached chart's overlay intact.

    '''
    from piker.ui import _remote_ctl

    first_ds: SimpleNamespace
    first_charts: tuple[_ChartStub, _ChartStub]
    first_ds, first_charts = _display_state('first.test')
    second_ds: SimpleNamespace
    second_charts: tuple[_ChartStub, _ChartStub]
    second_ds, second_charts = _display_state('second.test')
    dss: dict[str, SimpleNamespace] = {
        first_ds.fqme: first_ds,
        second_ds.fqme: second_ds,
    }
    annots: dict[int, GapAnnotations] = {}
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={},
        annots=annots,
    )
    remote_dss: dict[str, SimpleNamespace] = {}
    monkeypatch.setattr(_remote_ctl, '_dss', remote_dss)
    monkeypatch.setattr(_remote_ctl, '_annots', annots)
    monkeypatch.setattr(_remote_ctl, '_gapman', gapman)
    godwidget: SimpleNamespace = SimpleNamespace(gapman=None)

    try:
        started: GapOverlayMngr = _register_gap_overlays(
            dss=dss,
            godwidget=godwidget,
        )
        assert started is gapman
        assert godwidget.gapman is gapman
        assert remote_dss == dss
        assert len(gapman._layers) == 2
        key: tuple[str, int, str, float]
        for key in gapman._layers:
            assert key[3] == 60

        view: SimpleNamespace = SimpleNamespace(
            _chart=first_ds.hist_chart,
        )
        toggled: list[GapOverlay] = _toggle_gap_overlays(
            view=view,
            dss=dss,
            gapman=gapman,
        )
        assert len(toggled) == 1
        assert toggled[0].fqme == first_ds.fqme
        assert toggled[0].timeframe == 60
        assert toggled[0].visible is False
        assert (
            'chart-local',
            id(first_ds.hist_chart),
            first_ds.fqme,
            60,
        ) not in gapman._layers
        assert (
            'chart-local',
            id(second_ds.hist_chart),
            second_ds.fqme,
            60,
        ) in gapman._layers

        view._chart = first_ds.chart
        rt_toggled: list[GapOverlay] = _toggle_gap_overlays(
            view=view,
            dss=dss,
            gapman=gapman,
        )
        assert len(rt_toggled) == 1
        assert rt_toggled[0].fqme == first_ds.fqme
        assert rt_toggled[0].timeframe == 1
        assert rt_toggled[0].visible is True
        assert (
            'chart-local',
            id(first_ds.chart),
            first_ds.fqme,
            1,
        ) in gapman._layers

        unrelated: list[GapOverlay] = _toggle_gap_overlays(
            view=SimpleNamespace(_chart=object()),
            dss=dss,
            gapman=gapman,
        )
        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,

) -> None:
    '''
    Route a real Qt Ctrl-G event 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`,
    and wait until the handler blocks again. The resulting manager
    and scene state prove the complete local keyboard path ran.

    '''
    ds: SimpleNamespace
    charts: tuple[_ChartStub, _ChartStub]
    ds, charts = _display_state(FQME)
    annots: dict[int, GapAnnotations] = {}
    gapman: GapOverlayMngr = GapOverlayMngr(
        dss={FQME: ds},
        annots=annots,
    )
    nav: SimpleNamespace = SimpleNamespace(
        hide_info=lambda: None,
    )
    lines: SimpleNamespace = SimpleNamespace(
        unstage_line=lambda: None,
    )
    godwidget: SimpleNamespace = SimpleNamespace(gapman=gapman)
    order_mode: SimpleNamespace = SimpleNamespace(
        godw=godwidget,
        cancel_all_orders=lambda: None,
        current_pp=SimpleNamespace(nav=nav),
        lines=lines,
        active=False,
    )
    source: QWidget = QWidget()
    source.order_mode = order_mode
    source._chart = ds.hist_chart
    source.linked = SimpleNamespace(
        cursor=SimpleNamespace(in_query_mode=False),
    )
    source.setMouseMode = lambda mode: None

    async def drive_key_event() -> None:
        '''
        Run the production input context around one key event.

        '''
        async with ChartView.open_async_input_handler(
            source,
            dss={FQME: ds},
        ):
            key_event: QKeyEvent = QKeyEvent(
                QEvent.Type.KeyPress,
                Qt.Key.Key_G,
                Qt.KeyboardModifier.ControlModifier,
                'g',
            )
            assert QApplication.sendEvent(source, key_event)
            await wait_all_tasks_blocked()

    try:
        trio.run(drive_key_event)
        assert len(gapman._layers) == 1
        layer_key: tuple[str, int, str, float] = next(
            iter(gapman._layers)
        )
        assert layer_key == (
            'chart-local',
            id(ds.hist_chart),
            FQME,
            60,
        )
        aid: int = gapman._layers[layer_key]
        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] = []
_actor_removed: list[str] = []
_delayed_request_ids: list[str] = []


class _ActorGapMngr:
    '''
    Minimal child-actor manager used around the real endpoint.

    '''

    def apply(
        self,
        req: SetGapOverlay,
        owner: str,
        aids: set[int],
        ds: object|None = None,

    ) -> GapOverlay:
        '''
        Record a typed request and return its correlated state.

        '''
        aid: int = 616
        aids.add(aid)
        _actor_requests.append({
            'fqme': req.fqme,
            'timeframe': req.timeframe,
            'request_id': req.request_id,
            'owner': owner,
        })
        return GapOverlay(
            fqme=req.fqme,
            timeframe=req.timeframe,
            visible=req.visible,
            gap_count=len(req.specs),
            aid=aid,
            request_id=req.request_id,
        )

    def remove_owner(
        self,
        owner: str,

    ) -> None:
        '''
        Record endpoint-finally cleanup in the child actor.

        '''
        _actor_removed.append(owner)


async def _init_gap_actor() -> None:
    '''
    Install isolated endpoint globals inside the child actor.

    '''
    from piker.ui import _remote_ctl

    _actor_requests.clear()
    _actor_removed.clear()
    _delayed_request_ids.clear()
    _remote_ctl._dss.clear()
    _remote_ctl._dss[FQME] = object()
    _remote_ctl._gapman = _ActorGapMngr()


async def _gap_actor_snapshot() -> dict:
    '''
    Return serializable endpoint evidence from the child actor.

    '''
    return {
        'requests': list(_actor_requests),
        'removed': list(_actor_removed),
        'delayed_request_ids': list(_delayed_request_ids),
    }


@tractor.context(pld_spec=GapOverlayPayload)
async def _delayed_gap_dialog(
    ctx: tractor.Context,

) -> None:
    '''
    Delay the first reply so client cancellation leaves it queued.

    '''
    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
            req: SetGapOverlay
            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,
                        timeframe=req.timeframe,
                        visible=req.visible,
                        gap_count=0,
                        request_id='actor-received',
                    ))
                    await trio.sleep(0.05)

                await stream.send(GapOverlay(
                    fqme=req.fqme,
                    timeframe=req.timeframe,
                    visible=req.visible,
                    gap_count=len(req.specs),
                    request_id=req.request_id,
                ))


@tractor_test(timeout=20)
async def test_remote_gap_dialog_real_actor(
    monkeypatch: pytest.MonkeyPatch,

) -> None:
    '''
    Exchange typed gap state through the production Tractor endpoint.

    Schema-only tests can pass while endpoint discovery, directional
    decoding, stream setup, request correlation or context-finally
    cleanup is broken. Boot a real child actor, initialize only its
    actor-local display and manager globals, call the production
    `remote_gap_overlays()` context through `AnnotClient`, and verify
    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
    reply. Cancel that exact client task after the receipt, submit a
    second request on the shared stream, 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.

    '''
    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(
            'gap-overlay-test-server',
            enable_modules=[
                __name__,
                'piker.ui._remote_ctl',
            ],
        )
        try:
            await portal.run(_init_gap_actor)
            ctx: tractor.Context
            fqmes: list[str]
            async with portal.open_context(
                remote_gap_overlays,
            ) as (ctx, fqmes):
                assert fqmes == [FQME]

                stream: tractor.MsgStream
                async with ctx.open_stream() as stream:
                    client: AnnotClient = AnnotClient(
                        ctx2fqmes={ctx.cid: {FQME}},
                        fqme2ipc={},
                        fqme2gap_ipc={FQME: stream},
                        _gap_locks={FQME: trio.Lock()},
                        _annot_stack=AsyncExitStack(),
                        _ipcs={},
                    )
                    supplied_id: str = 'client-id-is-replaced'
                    gap: GapOverlay = await client.set_gap_overlay(
                        SetGapOverlay(
                            fqme=FQME,
                            timeframe=60,
                            specs=[],
                            request_id=supplied_id,
                        )
                    )
                    assert gap.aid == 616
                    assert gap.request_id != supplied_id

            snapshot: dict = await portal.run(
                _gap_actor_snapshot,
            )
            requests: list[dict] = snapshot['requests']
            assert len(requests) == 1
            assert requests[0]['request_id'] == gap.request_id
            assert snapshot['removed'] == [requests[0]['owner']]

            delayed_ctx: tractor.Context
            delayed_fqmes: list[str]
            async with portal.open_context(
                _delayed_gap_dialog,
            ) as (delayed_ctx, delayed_fqmes):
                assert delayed_fqmes == [FQME]

                delayed_stream: tractor.MsgStream
                async with (
                    delayed_ctx.open_stream() as delayed_stream,
                ):
                    delayed_client: AnnotClient = AnnotClient(
                        ctx2fqmes={delayed_ctx.cid: {FQME}},
                        fqme2ipc={},
                        fqme2gap_ipc={FQME: delayed_stream},
                        _gap_locks={FQME: trio.Lock()},
                        _annot_stack=AsyncExitStack(),
                        _ipcs={},
                    )
                    first_scope: trio.CancelScope = (
                        trio.CancelScope()
                    )

                    async def cancel_first_request() -> None:
                        '''
                        Wait for cancellation inside the first
                        dialog.

                        '''
                        with first_scope:
                            await delayed_client.set_gap_overlay(
                                SetGapOverlay(
                                    fqme=FQME,
                                    timeframe=60,
                                    specs=[],
                                )
                            )

                    nursery: trio.Nursery
                    async with trio.open_nursery() as nursery:
                        nursery.start_soon(cancel_first_request)
                        await receipt_seen.wait()
                        first_scope.cancel()
                        await wait_all_tasks_blocked()

                        recovered: GapOverlay = (
                            await delayed_client.set_gap_overlay(
                                SetGapOverlay(
                                    fqme=FQME,
                                    timeframe=60,
                                    specs=[],
                                )
                            )
                        )
                        assert recovered.fqme == FQME
                        assert recovered.request_id
                        nursery.cancel_scope.cancel()

            delayed_snapshot: dict = await portal.run(
                _gap_actor_snapshot,
            )
            delayed_ids: list[str] = (
                delayed_snapshot['delayed_request_ids']
            )
            assert len(delayed_ids) == 2
            assert delayed_ids[0] != delayed_ids[1]
            assert delayed_ids[1] == recovered.request_id
        finally:
            await portal.cancel_actor()