Formalize `FspStream` and fix builtin call contracts

Export `FeedQuote` and `Tick`, describe historical and realtime
FSP yields, and preserve callable parameters through `@fsp`.
Align builtin history arrays and emit named realtime updates.

Keep the staged Tuicr review fixes. Let Numba specialize `ema()`
lazily so omitted defaults work alongside explicit arguments,
retaining nopython/nogil compilation and single-sample seeding.
Cover builtin yields and EMA argument forms with regressions.

Prompt-IO: ai/prompt-io/opencode/20260908T020619Z_fadab3d2_prompt_io.md
Prompt-IO: ai/prompt-io/codex/20260915T155229Z_fadab3d2_prompt_io.md

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

ai/prompt-io/codex/20260915T155229Z_fadab3d2_prompt_io.md

@@ -0,0 +1,36 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+session: fsp-session-recovery-takeover-20260915
+timestamp: 2026-09-15T15:52:29Z
+git_ref: fadab3d2
+scope: code
+substantive: true
+raw_file: 20260915T155229Z_fadab3d2_prompt_io.raw.md
+---
+
+## Prompt
+
+Recover the interrupted OpenCode FSP session and take over its ongoing
+work. The recovered child instruction records human authorization for
+the minimal EMA omitted-default fix followed by a complete commit plan.
+
+## Response summary
+
+Removed the eager Numba signature that rejected Python's omitted
+default arguments. Preserved nopython/nogil compilation and numerical
+behavior. Added nine numerical regressions and verified all 38 focused
+FSP cases. Rebuilt planning artifacts separately from production code.
+
+## Files changed
+
+- `piker/fsp/_momo.py`: drop the eager signature and unused type imports.
+- `tests/test_fsp_momo.py`: omitted/explicit argument and seed coverage.
+
+## Human edits
+
+The human identified the omitted-default limitation in the earlier
+session, authorized the minimal lazy-specialization approach, and
+requested this takeover after OpenCode failed. Existing source changes
+and older OpenCode provenance were retained. The new fix remains
+unstaged and awaits human review; no acceptance is inferred.

ai/prompt-io/codex/20260915T155229Z_fadab3d2_prompt_io.raw.md

@@ -0,0 +1,50 @@
+---
+model: gpt-6 (exact variant not exposed)
+service: codex
+timestamp: 2026-09-15T15:52:29Z
+git_ref: fadab3d2
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt
+
+> i'm thinking maybe you can just take over that work and we can
+> abandon that oc dialog for now?
+
+Recovered child-session instructions describe the earlier human
+authorization to remove EMA's eager Numba signature while retaining
+`jit(nopython=True, nogil=True)`, verify defaults, and finish the commit
+plan without staging or committing. This is recovered context, not a
+verbatim quote of the earlier human authorization.
+
+## Response
+
+> `git diff HEAD~1..HEAD -- piker/fsp/_momo.py`
+
+Removed EMA's eager Numba signature and now-unused Numba type imports.
+Retained nopython/nogil compilation, the numerical kernel, historical
+comments, and pre-existing staged and unstaged edits.
+
+> `git diff HEAD~1..HEAD -- tests/test_fsp_momo.py`
+
+Added production-dispatcher regressions for omitted positional and
+keyword arguments, explicit None and numeric parameters, and the
+previously corrected one-sample continuation. Expected outputs are
+hand-computed recurrences, rather than the Python implementation.
+
+Before the fix, the first case failed with:
+
+```text
+TypeError: No matching definition for argument type(s)
+array(float64, 1d, C), omitted(default=None), omitted(default=None)
+```
+
+After the fix, the focused run outside the socket-restricted sandbox
+reported:
+
+```text
+38 passed in 0.66s
+```
+
+Ruff passed for the modified momentum module and new regression file.
+Combined-worktree results do not attest to isolated commit boundaries.

ai/prompt-io/codex/README.md

@@ -0,0 +1,10 @@
+# AI Prompt I/O Log — codex
+
+This directory tracks prompts and outputs for AI-assisted development
+using Codex. Substantive contributions include model identification,
+prompt context, output records, and human contribution accounting.
+
+Code output uses Git diff references; non-code output is retained in
+the paired `.raw.md` file. Human contributors remain responsible for
+review and acceptance. Entries follow the
+[NLNet generative AI policy](https://nlnet.nl/foundation/policies/generativeAI/).

ai/prompt-io/opencode/20260908T020619Z_fadab3d2_prompt_io.md

@@ -0,0 +1,40 @@
+---
+model: gpt-5.6-sol
+service: opencode
+session: tuicr-fsp-api-review
+timestamp: 2026-09-08T02:06:19Z
+git_ref: fadab3d2
+scope: code
+substantive: true
+raw_file: 20260908T020619Z_fadab3d2_prompt_io.raw.md
+---
+
+## Prompt
+
+Address all 14 comments from the local staged FSP Tuicr review, account
+for relevant unstaged behavior changes, verify the fixes, and add local
+responses without staging or committing repository changes.
+
+## Response summary
+
+Clarified typed wire-schema and generic-decorator intent, adopted more
+specific FSP and feed type names, made momentum array typing explicit,
+documented the WAP tuple fields, and verified both FSP regressions and
+the production Numba call shape.
+
+## Files changed
+
+- `piker/data/__init__.py` - export the renamed feed quote type.
+- `piker/data/ticktools.py` - clarify wire dicts and future structs.
+- `piker/fsp/_api.py` - clarify history and decorator types.
+- `piker/fsp/_engine.py` - consume the renamed public types.
+- `piker/fsp/_momo.py` - make array and stream types explicit.
+- `piker/fsp/_volume.py` - document WAP return fields.
+- `tests/test_fsp_sync.py` - consume the renamed feed quote type.
+
+## Human edits
+
+The human reviewed the staged API boundary in Tuicr, requested the type
+renames and documentation, rejected an opaque NumPy alias, identified
+missing local annotations, required per-field tuple comments, and asked
+that existing unstaged fixes be considered when responding.

ai/prompt-io/opencode/20260908T020619Z_fadab3d2_prompt_io.raw.md

@@ -0,0 +1,47 @@
+---
+model: gpt-5.6-sol
+service: opencode
+timestamp: 2026-09-08T02:06:19Z
+git_ref: fadab3d2
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt
+
+The human supplied 14 comments from the local Tuicr review session
+`fsp_backfill_sync@wkt-fsp_backfill_sync/staged/fadab3d` and invoked
+`/code-review-changes -r`. They asked that comments already addressed
+by unstaged behavioral work be recognized when choosing each response.
+
+## Response
+
+The review was correlated one-to-one with its persisted Tuicr records.
+The resulting adjustments clarify the existing dict wire schema and a
+future paired `msgspec.Struct` migration, and rename the broad `Quote`
+and `HistoryOutput` types to `FeedQuote` and `FspHistory`.
+
+> `git diff HEAD~1..HEAD -- piker/data/__init__.py`
+
+> `git diff HEAD~1..HEAD -- piker/data/ticktools.py`
+
+> `git diff HEAD~1..HEAD -- piker/fsp/_api.py`
+
+> `git diff HEAD~1..HEAD -- piker/fsp/_engine.py`
+
+The FSP decorator now documents its PEP 695 parameter specification and
+its bare and configured return forms. Momentum types use explicit
+`NDArray[np.float64]` annotations, and WMA locals and the ignored tick
+target are named and typed explicitly. The WAP tuple return is split
+across documented fields.
+
+> `git diff HEAD~1..HEAD -- piker/fsp/_momo.py`
+
+> `git diff HEAD~1..HEAD -- piker/fsp/_volume.py`
+
+> `git diff HEAD~1..HEAD -- tests/test_fsp_sync.py`
+
+The focused FSP regression file passed with 29 tests. Ruff, Ruff's E501
+line-length selection, compileall, and diff whitespace checks passed.
+The production Numba call shape was also executed successfully with all
+three arguments; its pre-existing explicit signature does not accept
+omitted defaults.

piker/data/__init__.py

@@ -22,7 +22,11 @@ and storing data from your brokers as well as
 sharing live streams over a network.
 
 """
-from .ticktools import iterticks
+from .ticktools import (
+    FeedQuote,
+    iterticks,
+    Tick,
+)
 from tractor.ipc._shm import (
     ShmArray,
     get_shm_token,
@@ -54,7 +58,9 @@ __all__: list[str] = [
     'Feed',
     'open_feed',
     'ShmArray',
+    'FeedQuote',
     'iterticks',
+    'Tick',
     'maybe_open_shm_array',
     'match_from_pairs',
     'attach_shm_array',

piker/data/ticktools.py

@@ -20,10 +20,31 @@ Tick event stream processing, filter-by-types, format-normalization.
 '''
 from itertools import chain
 from typing import (
-    Any,
+    Iterable,
-    AsyncIterator,
+    Iterator,
+    Sequence,
+    TypedDict,
 )
 
+
+# TODO: migrate feed emitters and consumers to ``msgspec.Struct``
+# together. These ``TypedDict`` types only describe the builtin dicts
+# currently transported over IPC; changing only the receiver-side
+# annotations would not change serialization.
+class Tick(TypedDict, total=False):
+    type: str
+    price: float
+    size: float
+    time: float
+
+
+class FeedQuote(TypedDict, total=False):
+    ticks: list[Tick]
+    tradeRate: float
+    volumeRate: float
+    broker_ts: float
+    brokerd_ts: float
+
 # tick-type-classes template for all possible "lowest level" events
 # that can can be emitted by the "top of book" L1 queues and
 # price-matching (with eventual clearing) in a double auction
@@ -41,15 +62,12 @@ _auction_ticks: set[str] = set.union(*_tick_groups.values())
 
 
 def frame_ticks(
-    quote: dict[str, Any],
+    quote: FeedQuote,
 
-    ticks_by_type: dict | None = None,
+    ticks_by_type: dict[str, list[Tick]]|None = None,
-    ticks_in_order: list[dict[str, Any]] | None = None
+    ticks_in_order: list[Tick]|None = None,
 
-) -> dict[
+) -> dict[str, list[Tick]]:
-    str,
-    list[dict[str, Any]]
-]:
     '''
     XXX: build a tick-by-type table of lists
     of tick messages. This allows for less
@@ -82,7 +100,7 @@ def frame_ticks(
 
     # append in reverse FIFO order for in-order iteration on
     # receiver side.
-    tick: dict[str, Any]
+    tick: Tick
     for tick in ticks:
         tbt.setdefault(
             tick['type'],
@@ -104,8 +122,8 @@ def frame_ticks(
 
 
 def iterticks(
-    quote: dict,
+    quote: FeedQuote,
-    types: tuple[str] = (
+    types: Sequence[str] = (
         'trade',
         'dark_trade',
     ),
@@ -116,7 +134,7 @@ def iterticks(
     # with this?
     frame_by_type: bool = False,
 
-) -> AsyncIterator:
+) -> Iterator[Tick]:
     '''
     Iterate through ticks delivered per quote cycle, filter and
     yield any declared in `types`.
@@ -163,10 +181,13 @@ def iterticks(
             ticks.extend(list(chain(trades.values(), darks.values())))
 
         # most-recent-first
-        if reverse:
+        ticks_iter: Iterable[Tick] = (
-            ticks = reversed(ticks)
+            reversed(ticks)
+            if reverse
+            else ticks
+        )
 
-        for tick in ticks:
+        for tick in ticks_iter:
             # print(f"{quote['symbol']}: {tick}")
             ttype = tick.get('type')
             if ttype in types:

piker/fsp/_api.py

@@ -25,12 +25,11 @@ FSP (financial signal processing) apis.
 # - composition of fsps / implicit chaining syntax (we need an issue)
 
 from __future__ import annotations
-from functools import partial
 from typing import (
-    Any,
+    AsyncGenerator,
     Callable,
-    Awaitable,
+    overload,
-    Optional,
+    Protocol,
 )
 
 import numpy as np
@@ -47,11 +46,32 @@ from ..log import get_logger
 
 log = get_logger(__name__)
 
+type FspHistory = dict[str, np.ndarray|None]|np.ndarray
+type RealtimeValue = np.ndarray|np.number|int|float
+type FspYield = FspHistory|tuple[str, RealtimeValue]
+type FspStream = AsyncGenerator[FspYield, None]
+type FspConfigValue = str|bool|int|float
+
+
+# ``**P`` declares a PEP 695 ``ParamSpec`` so wrappers retain each
+# decorated FSP's positional and keyword parameter types:
+# https://docs.python.org/3/reference/compound_stmts.html#type-params
+class FspFunc[**P](Protocol):
+    @property
+    def __name__(self) -> str: ...
+
+    def __call__(
+        self,
+        *args: P.args,
+        **kwargs: P.kwargs,
+
+    ) -> FspStream: ...
+
 # global fsp registry filled out by @fsp decorator below
-_fsp_registry = {}
+_fsp_registry: dict[NamespacePath, Fsp] = {}
 
 
-def _load_builtins() -> dict[tuple, Callable]:
+def _load_builtins() -> dict[NamespacePath, Fsp]:
 
     # import to implicity trigger registration via ``@fsp``
     from . import _momo  # noqa
@@ -60,7 +80,7 @@ def _load_builtins() -> dict[tuple, Callable]:
     return _fsp_registry
 
 
-class Fsp:
+class Fsp[**P]:
     '''
     "Financial signal processor" decorator wrapped async function.
 
@@ -79,28 +99,28 @@ class Fsp:
     # shm flow.
     _flow_registry: dict[
         tuple[NDToken, str],
-        tuple[NDToken, Optional[ShmArray]],
+        tuple[NDToken, ShmArray|None],
     ] = {}
 
     def __init__(
         self,
-        func: Callable[..., Awaitable],
+        func: FspFunc[P],
         *,
-        outputs: tuple[str] = (),
+        outputs: tuple[str, ...] = (),
-        display_name: Optional[str] = None,
+        display_name: str|None = None,
-        **config,
+        **config: FspConfigValue,
 
     ) -> None:
 
         # TODO (maybe):
         # - type introspection?
         # - should we make this a wrapt object proxy?
-        self.func = func
+        self.func: FspFunc[P] = func
         self.__name__ = func.__name__  # XXX: must have func-object name
 
-        self.ns_path: tuple[str, str] = NamespacePath.from_ref(func)
+        self.ns_path: NamespacePath = NamespacePath.from_ref(func)
         self.outputs = outputs
-        self.config: dict[str, Any] = config
+        self.config: dict[str, FspConfigValue] = config
 
         # register with declared set.
         _fsp_registry[self.ns_path] = self
@@ -116,9 +136,10 @@ class Fsp:
         # type annots from pep 612:
         # https://www.python.org/dev/peps/pep-0612/
         # instance,
-        *args,
+        *args: P.args,
-        **kwargs
+        **kwargs: P.kwargs,
-    ):
+
+    ) -> FspStream:
         return self.func(*args, **kwargs)
 
     def get_shm(
@@ -154,22 +175,57 @@ class Fsp:
         return maybe_array
 
 
-def fsp(
+@overload
-    wrapped=None,
+def fsp[**P](
+    wrapped: FspFunc[P],
     *,
-    outputs: tuple[str] = (),
+    outputs: tuple[str, ...] = (),
-    display_name: Optional[str] = None,
+    display_name: str|None = None,
-    **config,
+    **config: FspConfigValue,
 
-) -> Fsp:
+) -> Fsp[P]:
+    ...
+
+
+@overload
+def fsp[**P](
+    wrapped: None = None,
+    *,
+    outputs: tuple[str, ...] = (),
+    display_name: str|None = None,
+    **config: FspConfigValue,
+
+) -> Callable[[FspFunc[P]], Fsp[P]]:
+    ...
+
+
+def fsp[**P](
+    wrapped: FspFunc[P]|None = None,
+    *,
+    outputs: tuple[str, ...] = (),
+    display_name: str|None = None,
+    **config: FspConfigValue,
+
+) -> Fsp[P]|Callable[[FspFunc[P]], Fsp[P]]:
+    '''
+    Wrap an FSP function directly or return its configured decorator.
+
+    Bare ``@fsp`` calls this function with ``wrapped`` and returns an
+    ``Fsp``. Parameterized ``@fsp(...)`` returns ``decorate`` first;
+    Python then passes the decorated function to that closure.
+
+    '''
 
     if wrapped is None:
-        return partial(
+        def decorate(func: FspFunc[P]) -> Fsp[P]:
-            Fsp,
+            return Fsp(
-            outputs=outputs,
+                func,
-            display_name=display_name,
+                outputs=outputs,
-            **config,
+                display_name=display_name,
-        )
+                **config,
+            )
+
+        return decorate
 
     return Fsp(wrapped, outputs=(wrapped.__name__,))
 
@@ -180,7 +236,7 @@ def maybe_mk_fsp_shm(
     size: int,
     readonly: bool = True,
 
-) -> (str, ShmArray, bool):
+) -> tuple[str, ShmArray, bool]:
     '''
     Allocate a single row shm array for an symbol-fsp pair if none
     exists, otherwise load the shm already existing for that token.

piker/fsp/_momo.py

@@ -18,32 +18,37 @@
 Momentum bby.
 
 """
-from typing import AsyncIterator, Optional
+from typing import (
+    AsyncIterator,
+)
 
 import numpy as np
-from numba import jit, float64, optional, int64
+from numpy.typing import NDArray
+from numba import jit
 
-from ._api import fsp
+from ._api import (
-from ..data import iterticks
+    fsp,
+    FspStream,
+)
+from ..data import (
+    FeedQuote,
+    iterticks,
+    Tick,
+)
 from tractor.ipc._shm import ShmArray
 
 
 @jit(
-    float64[:](
-        float64[:],
-        optional(float64),
-        optional(float64)
-    ),
     nopython=True,
     nogil=True
 )
 def ema(
 
-    y: 'np.ndarray[float64]',
+    y: NDArray[np.float64],
-    alpha: optional(float64) = None,
+    alpha: float|None = None,
-    ylast: optional(float64) = None,
+    ylast: float|None = None,
 
-) -> 'np.ndarray[float64]':
+) -> NDArray[np.float64]:
     r'''
     Exponential weighted moving average owka 'Exponential smoothing'.
 
@@ -80,10 +85,14 @@ def ema(
         # directly to the com of a SMA or WMA:
         alpha = 2 / float(n + 1)
 
-    s = np.empty(n, dtype=float64)
+    s = np.empty(n, dtype=np.float64)
 
     if n == 1:
-        s[0] = y[0] * alpha + ylast * (1 - alpha)
+        s[0] = (
+            y[0]
+            if ylast is None
+            else y[0] * alpha + ylast * (1 - alpha)
+        )
 
     else:
         if ylast is None:
@@ -109,13 +118,12 @@ def ema(
 # )
 def _rsi(
 
-    # TODO: use https://github.com/ramonhagenaars/nptyping
+    signal: NDArray[np.float64],
-    signal: 'np.ndarray[float64]',
+    period: int = 14,
-    period: int64 = 14,
+    up_ema_last: float|None = None,
-    up_ema_last: float64 = None,
+    down_ema_last: float|None = None,
-    down_ema_last: float64 = None,
 
-) -> 'np.ndarray[float64]':
+) -> tuple[NDArray[np.float64], float, float]:
     '''
     relative strengggth.
 
@@ -125,10 +133,18 @@ def _rsi(
     df = np.diff(signal, prepend=0)
 
     up = np.where(df > 0, df, 0)
-    up_ema = ema(up, alpha, up_ema_last)
+    up_ema = ema(
+        up,
+        alpha,
+        up_ema_last,
+    )
 
     down = np.where(df < 0, -df, 0)
-    down_ema = ema(down, alpha, down_ema_last)
+    down_ema = ema(
+        down,
+        alpha,
+        down_ema_last,
+    )
 
     # avoid dbz errors, this leaves the first
     # index == 0 right?
@@ -151,7 +167,7 @@ def _wma(
 
     signal: np.ndarray,
     length: int,
-    weights: Optional[np.ndarray] = None,
+    weights: np.ndarray|None = None,
 
 ) -> np.ndarray:
     '''
@@ -174,11 +190,11 @@ def _wma(
 @fsp
 async def wma(
 
-    source,  #: AsyncStream[np.ndarray],
+    source: AsyncIterator[FeedQuote],
-    length: int,
+    ohlcv: ShmArray,
-    ohlcv: np.ndarray,  # price time-frame "aware"
+    length: int = 14,
 
-) -> AsyncIterator[np.ndarray]:  # maybe something like like FspStream?
+) -> FspStream:
     '''
     Streaming weighted moving average.
 
@@ -188,23 +204,33 @@ async def wma(
 
     '''
     # deliver historical output as "first yield"
-    yield _wma(ohlcv.array['close'], length)
+    close: NDArray[np.float64] = ohlcv.array['close']
+    history: NDArray[np.float64] = np.full(len(close), np.nan)
+    if len(close) >= length:
+        history[length - 1:] = _wma(close, length)
+    yield history
 
     # begin real-time section
 
     async for quote in source:
-        for tick in iterticks(quote, type='trade'):
+        _tick: Tick
-            yield _wma(ohlcv.last(length))
+        for _tick in iterticks(
+            quote,
+            types=['trade'],
+        ):
+            closes: np.ndarray = ohlcv.last(length)['close']
+            if len(closes) == length:
+                yield 'wma', _wma(closes, length)[-1]
 
 
 @fsp
 async def rsi(
 
-    source: 'QuoteStream[Dict[str, Any]]',  # noqa
+    source: AsyncIterator[FeedQuote],
     ohlcv: ShmArray,
     period: int = 14,
 
-) -> AsyncIterator[np.ndarray]:
+) -> FspStream:
     '''
     Multi-timeframe streaming RSI.
 
@@ -250,4 +276,4 @@ async def rsi(
                 up_ema_last=last_up_ema_close,
                 down_ema_last=last_down_ema_close,
             )
-            yield rsi_out[-1:]
+            yield 'rsi', rsi_out[-1]

piker/fsp/_volume.py

@@ -14,13 +14,17 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
-from typing import AsyncIterator, Optional, Union
-
 import numpy as np
 from tractor.trionics._broadcast import AsyncReceiver
 
-from ._api import fsp
+from ._api import (
-from ..data import iterticks
+    fsp,
+    FspStream,
+)
+from ..data import (
+    FeedQuote,
+    iterticks,
+)
 from tractor.ipc._shm import ShmArray
 from ._momo import _wma
 from ..log import get_logger
@@ -36,7 +40,11 @@ def wap(
     signal: np.ndarray,
     weights: np.ndarray,
 
-) -> np.ndarray:
+) -> tuple[
+    np.ndarray,  # weighted average price
+    np.ndarray,  # cumulative weighted input
+    np.ndarray,  # cumulative weights
+]:
     '''
     Weighted average price from signal and weights.
 
@@ -62,16 +70,13 @@ def wap(
 @fsp
 async def tina_vwap(
 
-    source: AsyncReceiver[dict],
+    source: AsyncReceiver[FeedQuote],
     ohlcv: ShmArray,  # OHLC sampled history
 
     # TODO: anchor logic (eg. to session start)
-    anchors: Optional[np.ndarray] = None,
+    anchors: np.ndarray|None = None,
 
-) -> Union[
+) -> FspStream:
-    AsyncIterator[np.ndarray],
-    float
-]:
     '''
     Streaming volume weighted moving average.
 
@@ -127,12 +132,10 @@ async def tina_vwap(
     curve_style='step',
 )
 async def dolla_vlm(
-    source: AsyncReceiver[dict],
+    source: AsyncReceiver[FeedQuote],
     ohlcv: ShmArray,  # OHLC sampled history
 
-) -> AsyncIterator[
+) -> FspStream:
-    tuple[str, Union[np.ndarray, float]],
-]:
     '''
     "Dollar Volume", aka the volume in asset-currency-units (usually
     a fiat) computed from some price function for the sample step
@@ -227,14 +230,14 @@ async def dolla_vlm(
     curve_style='line',
 )
 async def flow_rates(
-    source: AsyncReceiver[dict],
+    source: AsyncReceiver[FeedQuote],
     ohlcv: ShmArray,  # OHLC sampled history
 
     # TODO (idea): a dynamic generic / boxing type that can be updated by other
     # FSPs, user input, and possibly any general event stream in
     # real-time. Hint: ideally implemented with caching until mutated
     # ;)
-    period: 'Param[int]' = 1,  # noqa
+    period: int = 1,
 
     # TODO: support other means by providing a map
     # to weights `partial()`-ed with `wma()`?
@@ -252,9 +255,7 @@ async def flow_rates(
     # lazy copy in that case?
     # dvlm: 'Fsp[dolla_vlm]'
 
-) -> AsyncIterator[
+) -> FspStream:
-    tuple[str, Union[np.ndarray, float]],
-]:
     # generally no history available prior to real-time calcs
     yield {
         # from ib

tests/test_fsp_momo.py

@@ -0,0 +1,64 @@
+'''
+Momentum FSP numerical regressions.
+
+'''
+import numpy as np
+import pytest
+
+from piker.fsp._momo import ema
+
+
+@pytest.mark.parametrize(
+    'args,kwargs,expected',
+    [
+        ((), {}, [1., 1.5, 2.25]),
+        ((None,), {}, [1., 1.5, 2.25]),
+        ((None, None), {}, [1., 1.5, 2.25]),
+        ((0.25,), {}, [1., 1.25, 1.6875]),
+        ((), {'alpha': 0.25}, [1., 1.25, 1.6875]),
+        ((), {'ylast': 4.}, [4., 3., 3.]),
+        ((None, 4.), {}, [4., 3., 3.]),
+        ((0.5, 4.), {}, [4., 3., 3.]),
+    ],
+)
+def test_ema_optional_arguments(
+    args: tuple[float|None, ...],
+    kwargs: dict[str, float],
+    expected: list[float],
+) -> None:
+    '''
+    Accept omitted EMA defaults through the real Numba dispatcher.
+
+    `ema()` exposed optional smoothing and seed arguments, but its
+    eager signature accepted only explicit values or `None`.
+    Numba's omitted-argument types therefore raised `TypeError`
+    before the numerical kernel ran. Exercise positional and keyword
+    omissions as well as explicit arguments, checking the resulting
+    recurrence against hand-computed values. Import the production
+    dispatcher so this catches signature regressions that a call to
+    `ema.py_func` would miss.
+
+    '''
+    signal: np.ndarray = np.array([1., 2., 3.])
+    result: np.ndarray = ema(signal, *args, **kwargs)
+
+    np.testing.assert_allclose(result, expected)
+    assert result.dtype == np.float64
+
+
+def test_ema_single_sample_continuation() -> None:
+    '''
+    Preserve the previous EMA when advancing one realtime sample.
+
+    The old one-sample path multiplied an absent seed by a float,
+    failing instead of initializing from the sample. Removing the
+    eager Numba signature must retain the existing initialization
+    fix and the previous-value update used by realtime RSI. Use a
+    distinct seed and smoothing factor so copying either the seed
+    or sample fails; verify an omitted seed uses the sole sample.
+
+    '''
+    signal: np.ndarray = np.array([3.])
+
+    np.testing.assert_allclose(ema(signal), [3.])
+    np.testing.assert_allclose(ema(signal, 0.25, 7.), [6.])

tests/test_fsp_sync.py

@@ -0,0 +1,133 @@
+'''
+FSP history synchronization regressions.
+
+'''
+from collections.abc import AsyncIterator
+from typing import cast
+
+import numpy as np
+import pytest
+import trio
+
+from piker.fsp._momo import (
+    rsi,
+    wma,
+)
+from piker.fsp._volume import (
+    tina_vwap,
+)
+from piker.data._sharedmem import NDTokenMsg
+from piker.data.ticktools import FeedQuote
+from piker.fsp._api import Fsp
+from tractor.ipc._shm import (
+    NDToken,
+    ShmArray,
+)
+
+class Value:
+    def __init__(self, value: int) -> None:
+        self.value: int = value
+
+
+class Shm:
+    def __init__(
+        self,
+        first: int,
+        last: int,
+        token: str = 'fsp',
+
+    ) -> None:
+        self._first: Value = Value(first)
+        self._last: Value = Value(last)
+        self._array: np.ndarray = np.ones(4096)
+        self._len: int = len(self._array)
+        self._token: NDToken = NDToken(
+            shm_name=token,
+            shm_first_index_name=f'{token}_first',
+            shm_last_index_name=f'{token}_last',
+            dtype_descr=(('value', '<f8'),),
+            size=len(self._array),
+        )
+
+    @property
+    def array(self) -> np.ndarray:
+        return self._array[
+            self._first.value:self._last.value
+        ]
+
+    @property
+    def token(self) -> NDTokenMsg:
+        return cast(NDTokenMsg, self._token.as_msg())
+
+    @property
+    def index(self) -> int:
+        return self._last.value % len(self._array)
+
+    def last(self, length: int = 1) -> np.ndarray:
+        return self.array[-length:]
+
+
+class OhlcvShm(Shm):
+    def __init__(self, length: int = 32) -> None:
+        dtype = np.dtype([
+            ('index', '<i8'),
+            ('time', '<i8'),
+            ('open', '<f8'),
+            ('high', '<f8'),
+            ('low', '<f8'),
+            ('close', '<f8'),
+            ('volume', '<f8'),
+        ])
+        self._first = Value(0)
+        self._last = Value(length)
+        self._array = np.ones(length + 8, dtype=dtype)
+        self._array['index'] = np.arange(length + 8)
+        self._array['time'] = np.arange(length + 8)
+        self._array['close'] = np.arange(length + 8) + 1
+        self._array['high'] = self._array['close'] + 1
+        self._array['low'] = self._array['close'] - 1
+        self._len = len(self._array)
+        self._token = NDToken(
+            shm_name='ohlcv',
+            shm_first_index_name='ohlcv_first',
+            shm_last_index_name='ohlcv_last',
+            dtype_descr=tuple(dtype.descr),
+            size=self._len,
+        )
+
+
+@pytest.mark.parametrize('target', [wma, rsi, tina_vwap])
+def test_builtin_fsp_stream_contract(target: Fsp) -> None:
+    '''
+    Keep every registered scalar FSP on the engine's yield protocol.
+
+    The momentum operators previously had incompatible call signatures,
+    short historical arrays, and bare realtime yields, all hidden by an
+    engine-side cast. Run each against one OHLCV snapshot and one trade,
+    proving the first yield is a source-aligned array and the next yield
+    is a named realtime field/value pair.
+
+    '''
+    shm = cast(ShmArray, OhlcvShm())
+
+    async def source() -> AsyncIterator[FeedQuote]:
+        yield {
+            'ticks': [{
+                'type': 'trade',
+                'price': 42.0,
+                'size': 1.0,
+            }],
+        }
+
+    async def main() -> None:
+        stream = target.func(source(), shm)
+        history = await anext(stream)
+        assert isinstance(history, np.ndarray)
+        assert len(history) == len(shm.array)
+
+        realtime = await anext(stream)
+        assert isinstance(realtime, tuple)
+        assert realtime[0] == target.name
+        await stream.aclose()
+
+    trio.run(main)