Do time-based shm-index lookup for annots on server

Fix annotation misalignment during backfill by switching from
client-computed indices to server-side timestamp lookups against
current shm state. Store absolute coords on annotations and
reposition on viz redraws.

Lowlevel impl deats,
- add `time` param to `.add_arrow()`, `.add_text()`, `.add_rect()`
- lookup indices from shm via timestamp matching in IPC handlers
- force chart redraw before `markup_gaps()` annotation creation
- wrap IPC send/receive in `trio.fail_after(3)` for timeout when
  server fails to respond, likely hangs on no-case-match/error.
- cache `_meth`/`_kwargs` on rects, `_abs_x`/`_abs_y` on arrows
- auto-reposition all annotations after viz reset in redraw cmd

Also,
- handle `KeyError` for missing timeframes in chart lookup
- return `-1` aid on annotation creation failures (lol oh `claude`..)
- reconstruct rect positions from timestamps + BGM offset logic
- log repositioned annotation counts on viz redraw

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

piker/tsp/_annotate.py

@@ -94,6 +94,15 @@ async def markup_gaps(
     with rectangles.
 
     '''
+    # XXX: force chart redraw FIRST to ensure PlotItem coordinate
+    # system is properly initialized before we position annotations!
+    # Without this, annotations may be misaligned on first creation
+    # due to Qt/pyqtgraph initialization race conditions.
+    await actl.redraw(
+        fqme=fqme,
+        timeframe=timeframe,
+    )
+
     aids: dict[int] = {}
     for i in range(gaps.height):
         row: pl.DataFrame = gaps[i]
@@ -101,6 +110,7 @@ async def markup_gaps(
         # the gap's RIGHT-most bar's OPEN value
         # at that time (sample) step.
         iend: int = row['index'][0]
+
         # dt: datetime = row['dt'][0]
         # dt_prev: datetime = row['dt_prev'][0]
         # dt_end_t: float = dt.timestamp()
@@ -174,6 +184,10 @@ async def markup_gaps(
         gap_dur_s: float = row['s_diff'][0]
         gap_label: str = humanize_duration(gap_dur_s)
 
+        # XXX: get timestamps for server-side index lookup
+        start_time: float = prev_r['time'][0]
+        end_time: float = row['time'][0]
+
         # BGM=0.16 is the normal diff from overlap between bars, SO
         # just go slightly "in" from that "between them".
         from_idx: int = BGM - .06  # = .10
@@ -201,6 +215,8 @@ async def markup_gaps(
             start_pos=lc,
             end_pos=ro,
             color=color,
+            start_time=start_time,
+            end_time=end_time,
         )
 
         # add up/down rects
@@ -213,11 +229,15 @@ async def markup_gaps(
         )
         # TODO! mk this a `msgspec.Struct` which we deserialize
         # on the server side!
+        # XXX: send timestamp for server-side index lookup
+        # to ensure alignment with current shm state
+        gap_time: float = row['time'][0]
         arrow_kwargs: dict[str, Any] = dict(
             fqme=fqme,
             timeframe=timeframe,
-            x=iend,
+            x=iend,  # fallback if timestamp lookup fails
             y=cls,
+            time=gap_time,  # for server-side index lookup
             color=color,
             alpha=169,
             pointing=direction,
@@ -249,8 +269,9 @@ async def markup_gaps(
             fqme=fqme,
             timeframe=timeframe,
             text=gap_label,
-            x=iend + 1,
+            x=iend + 1,  # fallback if timestamp lookup fails
             y=cls,
+            time=gap_time,  # server-side index lookup
             color=color,
             anchor=anchor,
             font_size=font_size,

piker/ui/_remote_ctl.py

@@ -149,12 +149,72 @@ async def serve_rc_annots(
                 'kwargs': dict(kwargs),
             }:
                 ds: DisplayState = _dss[fqme]
-                chart: ChartPlotWidget = {
+                try:
-                    60: ds.hist_chart,
+                    chart: ChartPlotWidget = {
-                    1: ds.chart,
+                        60: ds.hist_chart,
-                }[timeframe]
+                        1: ds.chart,
+                    }[timeframe]
+                except KeyError:
+                    log.warning(
+                        f'No chart for timeframe={timeframe}s, '
+                        f'skipping rect annotation'
+                    )
+                    await annot_req_stream.send(-1)
+                    continue
                 cv: ChartView = chart.cv
 
+                # NEW: if timestamps provided, lookup current indices
+                # from shm to ensure alignment with current buffer
+                # state
+                start_time = kwargs.pop('start_time', None)
+                end_time = kwargs.pop('end_time', None)
+                if (
+                    start_time is not None
+                    and end_time is not None
+                ):
+                    viz: Viz = chart.get_viz(fqme)
+                    shm = viz.shm
+                    arr = shm.array
+
+                    # lookup start index
+                    start_matches = arr[arr['time'] == start_time]
+                    if len(start_matches) == 0:
+                        log.error(
+                            f'No shm entry for start_time='
+                            f'{start_time}, skipping rect'
+                        )
+                        await annot_req_stream.send(-1)
+                        continue
+
+                    # lookup end index
+                    end_matches = arr[arr['time'] == end_time]
+                    if len(end_matches) == 0:
+                        log.error(
+                            f'No shm entry for end_time={end_time}, '
+                            f'skipping rect'
+                        )
+                        await annot_req_stream.send(-1)
+                        continue
+
+                    # get close price from start bar, open from end
+                    # bar
+                    start_idx = float(start_matches[0]['index'])
+                    end_idx = float(end_matches[0]['index'])
+                    start_close = float(start_matches[0]['close'])
+                    end_open = float(end_matches[0]['open'])
+
+                    # reconstruct start_pos and end_pos with
+                    # looked-up indices
+                    from_idx: float = 0.16 - 0.06  # BGM offset
+                    kwargs['start_pos'] = (
+                        start_idx + 1 - from_idx,
+                        start_close,
+                    )
+                    kwargs['end_pos'] = (
+                        end_idx + from_idx,
+                        end_open,
+                    )
+
                 # annot type lookup from cmd
                 rect = SelectRect(
                     viewbox=cv,
@@ -173,6 +233,12 @@ async def serve_rc_annots(
                 # delegate generically to the requested method
                 getattr(rect, meth)(**kwargs)
                 rect.show()
+
+                # XXX: store absolute coords for repositioning
+                # during viz redraws (eg backfill updates)
+                rect._meth = meth
+                rect._kwargs = kwargs
+
                 aid: int = id(rect)
                 annots[aid] = rect
                 aids: set[int] = ctxs[ipc_key][1]
@@ -196,18 +262,47 @@ async def serve_rc_annots(
                     'tailLen': int()|float()|None as tailLen,
                     'tailWidth': int()|float()|None as tailWidth,
                     'pxMode': bool(pxMode),
+                    'time': int()|float()|None as timestamp,
                 },
                 # ?TODO? split based on method fn-sigs?
                 # 'pointing',
             }:
                 ds: DisplayState = _dss[fqme]
-                chart: ChartPlotWidget = {
+                try:
-                    60: ds.hist_chart,
+                    chart: ChartPlotWidget = {
-                    1: ds.chart,
+                        60: ds.hist_chart,
-                }[timeframe]
+                        1: ds.chart,
+                    }[timeframe]
+                except KeyError:
+                    log.warning(
+                        f'No chart for timeframe={timeframe}s, '
+                        f'skipping arrow annotation'
+                    )
+                    # return -1 to indicate failure
+                    await annot_req_stream.send(-1)
+                    continue
                 cv: ChartView = chart.cv
                 godw = chart.linked.godwidget
 
+                # NEW: if timestamp provided, lookup current index
+                # from shm to ensure alignment with current buffer
+                # state
+                if timestamp is not None:
+                    viz: Viz = chart.get_viz(fqme)
+                    shm = viz.shm
+                    arr = shm.array
+                    # find index where time matches timestamp
+                    matches = arr[arr['time'] == timestamp]
+                    if len(matches) == 0:
+                        log.error(
+                            f'No shm entry for timestamp={timestamp}, '
+                            f'skipping arrow annotation'
+                        )
+                        await annot_req_stream.send(-1)
+                        continue
+                    # use the matched row's index as x
+                    x = float(matches[0]['index'])
+
                 arrows = ArrowEditor(godw=godw)
                 # `.add/.remove()` API
                 if meth != 'add':
@@ -232,6 +327,11 @@ async def serve_rc_annots(
                     tailWidth=tailWidth,
                     pxMode=pxMode,
                 )
+                # XXX: store absolute coords for repositioning
+                # during viz redraws (eg backfill updates)
+                arrow._abs_x = x
+                arrow._abs_y = y
+
                 annots[aid] = arrow
                 _editors[aid] = arrows
                 aids: set[int] = ctxs[ipc_key][1]
@@ -249,13 +349,42 @@ async def serve_rc_annots(
                     'color': color,
                     'anchor': list(anchor),
                     'font_size': int()|None as font_size,
+                    'time': int()|float()|None as timestamp,
                 },
             }:
                 ds: DisplayState = _dss[fqme]
-                chart: ChartPlotWidget = {
+                try:
-                    60: ds.hist_chart,
+                    chart: ChartPlotWidget = {
-                    1: ds.chart,
+                        60: ds.hist_chart,
-                }[timeframe]
+                        1: ds.chart,
+                    }[timeframe]
+                except KeyError:
+                    log.warning(
+                        f'No chart for timeframe={timeframe}s, '
+                        f'skipping text annotation'
+                    )
+                    await annot_req_stream.send(-1)
+                    continue
+
+                # NEW: if timestamp provided, lookup current index
+                # from shm to ensure alignment with current buffer
+                # state
+                if timestamp is not None:
+                    viz: Viz = chart.get_viz(fqme)
+                    shm = viz.shm
+                    arr = shm.array
+                    # find index where time matches timestamp
+                    matches = arr[arr['time'] == timestamp]
+                    if len(matches) == 0:
+                        log.error(
+                            f'No shm entry for timestamp={timestamp}, '
+                            f'skipping text annotation'
+                        )
+                        await annot_req_stream.send(-1)
+                        continue
+                    # use the matched row's index as x, +1 for text
+                    # offset
+                    x = float(matches[0]['index']) + 1
 
                 # convert named color to hex
                 color_hex: str = hcolor(color)
@@ -284,6 +413,11 @@ async def serve_rc_annots(
                 text_item.setPos(x, y)
                 chart.plotItem.addItem(text_item)
 
+                # XXX: store absolute coords for repositioning
+                # during viz redraws (eg backfill updates)
+                text_item._abs_x = x
+                text_item._abs_y = y
+
                 aid: str = str(uuid4())
                 annots[aid] = text_item
                 aids: set[int] = ctxs[ipc_key][1]
@@ -329,6 +463,38 @@ async def serve_rc_annots(
                 )
                 viz.reset_graphics()
 
+                # XXX: reposition all annotations to ensure they
+                # stay aligned with viz data after reset (eg during
+                # backfill when abs-index range changes)
+                n_repositioned: int = 0
+                for aid, annot in annots.items():
+                    # arrows and text items use abs x,y coords
+                    if (
+                        hasattr(annot, '_abs_x')
+                        and
+                        hasattr(annot, '_abs_y')
+                    ):
+                        annot.setPos(
+                            annot._abs_x,
+                            annot._abs_y,
+                        )
+                        n_repositioned += 1
+
+                    # rects use method + kwargs
+                    elif (
+                        hasattr(annot, '_meth')
+                        and
+                        hasattr(annot, '_kwargs')
+                    ):
+                        getattr(annot, annot._meth)(**annot._kwargs)
+                        n_repositioned += 1
+
+                if n_repositioned:
+                    log.info(
+                        f'Repositioned {n_repositioned} annotation(s) '
+                        f'after viz redraw'
+                    )
+
             case _:
                 log.error(
                     'Unknown remote annotation cmd:\n'
@@ -417,6 +583,10 @@ class AnnotCtl(Struct):
 
         from_acm: bool = False,
 
+        # NEW: optional timestamps for server-side index lookup
+        start_time: float|None = None,
+        end_time: float|None = None,
+
     ) -> int:
         '''
         Add a `SelectRect` annotation to the target view, return
@@ -424,29 +594,32 @@ class AnnotCtl(Struct):
 
         '''
         ipc: MsgStream = self._get_ipc(fqme)
-        await ipc.send({
+        with trio.fail_after(3):
-            'fqme': fqme,
+            await ipc.send({
-            'cmd': 'SelectRect',
+                'fqme': fqme,
-            'timeframe': timeframe,
+                'cmd': 'SelectRect',
-            # 'meth': str(meth),
+                'timeframe': timeframe,
-            'meth': 'set_view_pos' if domain == 'view' else 'set_scene_pos',
+                # 'meth': str(meth),
-            'kwargs': {
+                'meth': 'set_view_pos' if domain == 'view' else 'set_scene_pos',
-                'start_pos': tuple(start_pos),
+                'kwargs': {
-                'end_pos': tuple(end_pos),
+                    'start_pos': tuple(start_pos),
-                'color': color,
+                    'end_pos': tuple(end_pos),
-                'update_label': False,
+                    'color': color,
-            },
+                    'update_label': False,
-        })
+                    'start_time': start_time,
-        aid: int = await ipc.receive()
+                    'end_time': end_time,
-        self._ipcs[aid] = ipc
+                },
-        if not from_acm:
+            })
-            self._annot_stack.push_async_callback(
+            aid: int = await ipc.receive()
-                partial(
+            self._ipcs[aid] = ipc
-                    self.remove,
+            if not from_acm:
-                    aid,
+                self._annot_stack.push_async_callback(
+                    partial(
+                        self.remove,
+                        aid,
+                    )
                 )
-            )
+            return aid
-        return aid
 
     async def remove(
         self,
@@ -516,6 +689,9 @@ class AnnotCtl(Struct):
 
         from_acm: bool = False,
 
+        # NEW: optional timestamp for server-side index lookup
+        time: float|None = None,
+
     ) -> int:
         '''
         Add a `SelectRect` annotation to the target view, return
@@ -523,36 +699,38 @@ class AnnotCtl(Struct):
 
         '''
         ipc: MsgStream = self._get_ipc(fqme)
-        await ipc.send({
+        with trio.fail_after(3):
-            'fqme': fqme,
+            await ipc.send({
-            'cmd': 'ArrowEditor',
+                'fqme': fqme,
-            'timeframe': timeframe,
+                'cmd': 'ArrowEditor',
-            # 'meth': str(meth),
+                'timeframe': timeframe,
-            'meth': 'add',
+                # 'meth': str(meth),
-            'kwargs': {
+                'meth': 'add',
-                'x': float(x),
+                'kwargs': {
-                'y': float(y),
+                    'x': float(x),
-                'color': color,
+                    'y': float(y),
-                'pointing': pointing,  # up|down
+                    'color': color,
-                'alpha': alpha,
+                    'pointing': pointing,  # up|down
-                'aid': None,
+                    'alpha': alpha,
-                'headLen': headLen,
+                    'aid': None,
-                'headWidth': headWidth,
+                    'headLen': headLen,
-                'tailLen': tailLen,
+                    'headWidth': headWidth,
-                'tailWidth': tailWidth,
+                    'tailLen': tailLen,
-                'pxMode': pxMode,
+                    'tailWidth': tailWidth,
-            },
+                    'pxMode': pxMode,
-        })
+                    'time': time,  # for server-side index lookup
-        aid: int = await ipc.receive()
+                },
-        self._ipcs[aid] = ipc
+            })
-        if not from_acm:
+            aid: int = await ipc.receive()
-            self._annot_stack.push_async_callback(
+            self._ipcs[aid] = ipc
-                partial(
+            if not from_acm:
-                    self.remove,
+                self._annot_stack.push_async_callback(
-                    aid,
+                    partial(
+                        self.remove,
+                        aid,
+                    )
                 )
-            )
+            return aid
-        return aid
 
     async def add_text(
         self,
@@ -567,6 +745,9 @@ class AnnotCtl(Struct):
 
         from_acm: bool = False,
 
+        # NEW: optional timestamp for server-side index lookup
+        time: float|None = None,
+
     ) -> int:
         '''
         Add a `pg.TextItem` annotation to the target view.
@@ -576,29 +757,31 @@ class AnnotCtl(Struct):
 
         '''
         ipc: MsgStream = self._get_ipc(fqme)
-        await ipc.send({
+        with trio.fail_after(3):
-            'fqme': fqme,
+            await ipc.send({
-            'cmd': 'TextItem',
+                'fqme': fqme,
-            'timeframe': timeframe,
+                'cmd': 'TextItem',
-            'kwargs': {
+                'timeframe': timeframe,
-                'text': text,
+                'kwargs': {
-                'x': float(x),
+                    'text': text,
-                'y': float(y),
+                    'x': float(x),
-                'color': color,
+                    'y': float(y),
-                'anchor': tuple(anchor),
+                    'color': color,
-                'font_size': font_size,
+                    'anchor': tuple(anchor),
-            },
+                    'font_size': font_size,
-        })
+                    'time': time,  # for server-side index lookup
-        aid: int = await ipc.receive()
+                },
-        self._ipcs[aid] = ipc
+            })
-        if not from_acm:
+            aid: int = await ipc.receive()
-            self._annot_stack.push_async_callback(
+            self._ipcs[aid] = ipc
-                partial(
+            if not from_acm:
-                    self.remove,
+                self._annot_stack.push_async_callback(
-                    aid,
+                    partial(
+                        self.remove,
+                        aid,
+                    )
                 )
-            )
+            return aid
-        return aid
 
 
 @acm