Version 0.2.1

Version 0.2.1 now has derive unit context for derivative calculus

docs/API.md

@@ -127,10 +127,13 @@ Recalculates derived signals in the current processing buffer using the bound
 derive units. Call it in `"offline"` mode after `load_processing_buffer()`, with
 `CoreConfig(playback_mode="read")`. Units with source inputs run first; units
 that depend on their derived outputs run in dependency order. Each unit runs
-in sample order using per-signal binary searches instead of full-buffer scans.
-For other inputs it selects the nearest recorded value, preferring the earlier
-sample when distances tie. This retains the previous offline alignment policy,
-including selection of future samples. The operation
+in sample order. Legacy alignment uses per-signal binary searches; contexts
+advance bounded histories.
+For functions without a processing context, other inputs select the nearest
+recorded value, preferring the earlier sample when distances tie; this can
+select future samples. Functions accepting `DeriveContext` use causal history
+and preserve each input's timestamp instead. See [processing and derivatives](Processing.md).
+The operation
 works on a deep copy, removes its previously recorded derived samples and
 descriptors, and merges the new outputs. It replaces the processing buffer
 only after all units finish successfully. Calling it again replaces those
@@ -141,8 +144,10 @@ then `play()`.
 The mode, configuration, and buffer checks raise `CoreError` with codes
 `"invalid_mode"`, `"invalid_config"`, and `"empty_buffer"`, respectively.
 Missing inputs raise `DeriveError(code="missing_inputs")`, cycles raise
-`DeriveError(code="dependency_cycle")`, and a unit that produces no values raises
-`DeriveError(code="no_values")`. Errors from a derive function propagate to the
+`DeriveError(code="dependency_cycle")`. A unit that produces no values raises
+`DeriveError(code="no_values")` unless its return annotation allows `None`.
+An optional-output unit may produce an empty series; units requiring that output
+are skipped with a diagnostic. Errors from a derive function propagate to the
 synchronous caller or are stored in the background operation status. Failure
 leaves the original processing buffer, including its prior derived results
 and metadata, unchanged.
@@ -366,6 +371,14 @@ Derive source operates on internal `ValueDescriptor` and `SignalDescriptor`
 dataclasses, not their `Json*` equivalents. Source is executed with `exec()`;
 only load units from trusted sources.
 
+Functions may optionally accept a `DeriveContext` between their input values and
+the final output descriptor. `ctx.derivative(value, order=1, max_gap_s=None)`
+estimates time derivatives of orders 1–3 using bounded history in realtime,
+offline processing, and recalculated playback. Annotate the return type as
+`ValueDescriptor | None` to skip outputs during warm-up. Existing functions keep
+their original signatures and behavior. See [DLUnit processing and derivatives](Processing.md)
+for examples, numerical behavior, input alignment, and reset rules.
+
 ### `validate_derive_units() -> None`
 
 Checks the configured graph against signals available in the current mode.
@@ -375,6 +388,8 @@ raise `DeriveError(code="missing_inputs")`; dependency cycles raise
 `"duplicate_output"`. Outputs must have `origin="derived"`, otherwise validation
 raises `"invalid_origin"`. A derive function must return a `ValueDescriptor`
 whose `signal_id` is its configured output, or it raises `"invalid_result"`.
+Functions annotated `-> ValueDescriptor | None` may also return `None` to emit
+no sample for that invocation.
 
 Structural validation runs before a unit is registered, so cycles and duplicate
 outputs cannot begin realtime execution. Units can be bound before their sources

docs/Processing.md

@@ -0,0 +1,152 @@
+# DLUnit processing and derivatives
+
+DLUnits can optionally request a `DeriveContext` to calculate time derivatives
+of their declared inputs. The same interface works during realtime acquisition,
+offline `Core.reprocess()`, and playback with `playback_mode="reprocess"`.
+Playback with `playback_mode="read"` reads previously saved outputs instead.
+
+## Example: speed to acceleration
+
+Use this as the DLUnit's `unit.py` source:
+
+```python
+from dynalab_core.processing import DeriveContext
+from dynalab_core.protocols.packets.data import ValueDescriptor
+from dynalab_core.protocols.packets.handshake import SignalDescriptor
+
+
+def process(
+    speed: ValueDescriptor,
+    ctx: DeriveContext,
+    output: SignalDescriptor,
+) -> ValueDescriptor | None:
+    acceleration = ctx.derivative(speed, order=1, max_gap_s=0.5)
+    if acceleration is None:
+        return None
+    return ValueDescriptor(output.id, acceleration, speed.timestamp)
+```
+
+Bind it using the existing API:
+
+```python
+from uuid import uuid4
+from dynalab_core.dlunit import DLUnit
+from dynalab_core.protocols.packets.handshake import SignalDescriptor
+
+acceleration_signal = SignalDescriptor(
+    id=uuid4(), name="Acceleration (m/s²)", type="number", origin="derived"
+)
+unit = DLUnit()
+unit.set_manifest(uuid4(), [speed_signal], acceleration_signal)
+unit.set_data(source_text)  # The unit.py source above.
+core.bind_derive_unit(dlunit=unit)
+# Optionally persist it with unit.write(directory, "acceleration.dlunit").
+```
+
+`speed_signal` must describe the existing speed input. If its values are in m/s,
+the output values are in m/s². Core does not infer physical units or rename
+signals automatically. Derivative outputs are ordinary derived signals: they
+can feed another DLUnit, be recorded, and be saved in a DLPak.
+
+## Function contract
+
+A function takes its `ValueDescriptor` inputs in manifest order, optionally one
+`DeriveContext`, then one final `SignalDescriptor` describing the output. Use
+the actual imported classes in annotations, without postponed/string
+annotations. The source must expose exactly one Python function; importing the
+context class is supported.
+
+Existing functions such as `process(a, b, output) -> ValueDescriptor` continue
+to work. To skip output, explicitly annotate `-> ValueDescriptor | None` and
+return `None`. This also works without a context. Returning `None` with the
+original `-> ValueDescriptor` annotation remains an invalid result.
+
+No placeholder is emitted when a function returns `None`: live values are not
+updated, recording gets no sample, and downstream units receive no event for
+that output. A previously published live output remains subject to its normal
+timeout; skipping an output does not explicitly clear it.
+
+An offline unit annotated to allow `None` may produce an empty series. Core logs
+`derive.offline_no_values` and skips units requiring that empty output, logging
+`derive.offline_upstream_empty`. Reprocessing still succeeds, and descriptors
+are retained for the empty outputs. Units that do not allow `None` retain the
+`DeriveError(code="no_values")` behavior if they execute but emit nothing.
+
+## `ctx.derivative(value, order=1, *, max_gap_s=None) -> float | None`
+
+- `value` is one of the function's supplied input descriptors.
+- `order` is an integer: 1, 2, or 3. Other values raise `ValueError`.
+- `max_gap_s` is an optional finite positive maximum interval between
+  consecutive samples in the derivative window. Invalid limits raise
+  `ValueError`. The default places no gap limit.
+- The result uses seconds: input units/s, input units/s², or input units/s³.
+- Warm-up requires 2, 3, or 4 distinct timestamps, respectively. Until then,
+  the result is `None`. A window containing a gap above `max_gap_s` also returns
+  `None`; estimates resume after enough samples arrive following that gap.
+- Repeated calls are read-only and return the same estimate. You can request
+  different orders for the same input in one invocation.
+
+The estimator differentiates the degree-`order` polynomial interpolating the
+most recent `order + 1` samples. For order 1 this is the usual backward secant:
+`(new_value - previous_value) / elapsed_seconds`. Orders 2 and 3 use the actual
+unequal sample intervals rather than repeatedly dividing by a nominal period.
+Timestamp differences are taken as integers before converting to seconds.
+
+These are numerical estimates without smoothing. They use only present and
+past samples, and higher orders can magnify measurement noise. Very uneven
+spacing and floating-point precision also limit accuracy. A non-finite
+computed estimate returns `None`.
+
+## History and alignment
+
+Core retains at most four samples per declared input, per context-enabled unit.
+The router advances history only when that input itself updates and queues a
+snapshot for each invocation. Later arrivals cannot alter an already queued
+context, and calls to `derivative()` do not advance history.
+
+Context-enabled units wait until all declared inputs have a sample at or before
+the triggering event's timestamp. Each input retains its own original timestamp.
+Only the most recent four accepted samples are available, so severely late
+events may no longer have a companion sample in history and are skipped.
+Equal or older timestamps for a signal are ignored by that unit's history and
+do not trigger its processing function. A newer non-finite value clears that
+signal's history and skips the invocation; finite samples restart warm-up.
+These rules do not remove or alter raw recording samples.
+
+When another input changes, a derivative of an unchanged input reuses that
+input's existing estimate. Its history does not gain another sample. If the
+output combines multiple inputs, choose its timestamp explicitly, for example
+`max(a.timestamp, b.timestamp)`. Companion values are retained in the bounded
+history; they do not expire through the realtime live-value timeout. Apply any
+additional age policy in the function using the supplied timestamps.
+
+Offline processing uses the same bounded history and causal alignment, with a
+fresh history on every pass. Functions without a context retain the original
+offline nearest-neighbor alignment, including possible future samples and
+timestamps assigned from the triggering event.
+
+Online and offline estimates match for the same chronological accepted input
+sequence. Offline processing sorts stored samples; online out-of-order arrivals
+can therefore differ. Queue overload can also change the observed sequence:
+router drops omit samples from history, while a unit queue drop omits that
+invocation but its sample remains in later history snapshots. Asynchronous
+multi-input derived chains can observe a different event ordering online.
+Offline processing does not drop samples due to queue capacity.
+
+Starting or stopping recording does not reset ongoing online history. If a
+recording begins midstream, its first online derivative outputs may depend on
+samples preceding the recording. Offline recalculation warms up from the
+recorded samples and can omit those early estimates.
+
+## Lifecycle
+
+Online histories are invalidated when Core changes mode, a source connector
+registers/disconnects, processing or playback data is replaced, or a playback
+control changes its generation (play/resume, pause, seek, or restart). Connector
+changes conservatively invalidate all units' histories. Queued requests and
+results from an older revision are discarded, including results completing
+after the reset. New data warms up the estimator again.
+
+Offline histories are local to the processing pass and never share online
+state. DLUnit archives save source and manifest, not history. DLPak and connector
+packet formats are unchanged. No numerical-library dependency is required.

docs/Protocol.md

@@ -83,7 +83,7 @@ packet classes.
 {
   "type": "dynalab_hello",
   "instance_id": "2f3c7f62-0a45-4d66-9f32-7fbeb92f9d36",
-  "core_version": {"type": "alpha", "major": 0, "minor": 1, "patch": 0},
+  "core_version": {"type": "alpha", "major": 0, "minor": 2, "patch": 1},
   "protocol_version": {"type": "alpha", "major": 0, "minor": 0, "patch": 1},
   "heartbeat_interval_ms": 1000,
   "heartbeat_timeout_ms": 5000

docs/README.md

@@ -12,4 +12,6 @@ This documentation is ordered in the following sections
 
 [**Protocol**](Protocol.md) - Documents the data ingress protocol, this is useful for developing connectors to pipe data into the DynaLab Core
 
+[**Processing and derivatives**](Processing.md) - DLUnit context API, first through third time derivatives, warm-up, and online/offline behavior
+
 [**Internal**](Internal/README.md) - Documents the inner working of DynaLab Core including data flow, internal API's, custom file types and much more

src/dynalab_core/__init__.py

@@ -22,7 +22,11 @@ import psutil
 from dynalab_core.buffer import ValueBuffer
 from dynalab_core.config import CoreConfig
 from dynalab_core.constants import CORE_VERSION
-from dynalab_core.derive import DeriveRegistry
+from dynalab_core.derive import (
+    DeriveRegistry,
+    DerivedValueDescriptor,
+    DerivedPlaybackValueDescriptor,
+)
 from dynalab_core.dlpak import DLPak
 from dynalab_core.dlunit import DLUnit
 from dynalab_core.errors import CoreError
@@ -67,6 +71,7 @@ class Core:
         self._input_error: str | None = None
         self._recording_signals: dict[UUID, SignalDescriptor] = {}
         self._mode: CoreModeState = CoreModeState()
+        self._derive_epoch = 0
         self._recording = threading.Event()
         self._recording_buffer = ValueBuffer()
         self._recording_timestamp: datetime = datetime.now(timezone.utc)
@@ -282,14 +287,17 @@ class Core:
                 if self._mode.value == "playback" and mode != "playback":
                     self._playback_engine.pause()
                 self._mode.value = mode
+                self._derive_epoch += 1
                 self._live_values.clear()
 
     def _get_source_revision(self) -> object:
         return (
+            self._derive_epoch,
             self._mode.value,
             id(self._processing_buffer),
             id(self._playback_engine._data),
             self._connector_registry.get_revision(),
+            self._playback_engine.get_generation(),
         )
 
     def get_connectors(self) -> list[ConnectorStatus]:
@@ -771,6 +779,12 @@ class Core:
                     continue
                 if isinstance(message, ValueDescriptor):
                     with self._live_values_lock:
+                        if isinstance(
+                            message,
+                            (DerivedValueDescriptor, DerivedPlaybackValueDescriptor),
+                        ):
+                            if not self._derive_registry.is_current_result(message):
+                                continue
                         if isinstance(message, PlaybackValueDescriptor):
                             if (
                                 self._mode.value != "playback"

src/dynalab_core/constants.py

@@ -12,7 +12,7 @@ from dynalab_core.protocols.packets.handshake import (
 )
 
 
-CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1)
+CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=2, patch=1)
 
 HELLO_PACKET = JsonDynaLabHello(
     instance_id=uuid4(),

src/dynalab_core/derive.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Callable
 from datetime import datetime, timezone
+from dataclasses import dataclass
 from enum import IntEnum
 import inspect
 import logging
@@ -23,10 +24,28 @@ from dynalab_core.protocols.packets.handshake import SignalDescriptor
 from dynalab_core.types import CoreModeState, DeriveUnitStatus, WorkerStatus
 from dynalab_core.values import SignalValueDescriptor
 from dynalab_core.operations import ProgressCallback
+from dynalab_core.processing import DeriveContext, _SampleHistory
 
 log = logging.getLogger(__name__)
 
 
+@dataclass(slots=True)
+class DerivedValueDescriptor(SignalValueDescriptor):
+    revision: object
+
+
+@dataclass(slots=True)
+class DerivedPlaybackValueDescriptor(PlaybackValueDescriptor):
+    revision: object
+
+
+@dataclass
+class _DeriveRequest:
+    inputs: list[ValueDescriptor]
+    context: DeriveContext | None
+    revision: object
+
+
 class ParserState(IntEnum):
     INIT = 0
     VALUES_DONE = 1
@@ -37,13 +56,14 @@ class DeriveUnit:
     def __init__(
         self,
         unit_uuid: UUID,
-        processing_function: Callable[..., ValueDescriptor],
+        processing_function: Callable[..., ValueDescriptor | None],
         input_signals: list[SignalDescriptor],
         return_signal: SignalDescriptor,
         return_queue: Queue[ValueDescriptor],
         is_current_playback_value: Callable[[PlaybackValueDescriptor], bool]
         | None = None,
         queue_capacity: int = 16384,
+        get_revision: Callable[[], object] | None = None,
     ) -> None:
         self._id: UUID = unit_uuid
         self._process_function: Callable = processing_function
@@ -51,7 +71,10 @@ class DeriveUnit:
         self._input_signals: list[SignalDescriptor] = input_signals
         self._stop_event: threading.Event = threading.Event()
         self._stopped_event: threading.Event = threading.Event()
-        self._input_queue: Queue[list[ValueDescriptor]] = Queue(queue_capacity)
+        self._input_queue: Queue[_DeriveRequest] = Queue(queue_capacity)
+        self._get_revision = get_revision or (lambda: None)
+        self._history = _SampleHistory()
+        self._history_revision: object = object()
         self._error: str | None = None
         self._dropped_samples = 0
         self._status_lock = threading.Lock()
@@ -60,6 +83,7 @@ class DeriveUnit:
 
         self._num_input_args: int = 0
         self._parser_state: int = ParserState.INIT
+        self._uses_context = False
         sig = inspect.signature(self._process_function)
 
         for name, param in sig.parameters.items():
@@ -70,6 +94,9 @@ class DeriveUnit:
                     self._parser_state = ParserState.VALUES_DONE
 
             if self._parser_state == ParserState.VALUES_DONE:
+                if param.annotation is DeriveContext and not self._uses_context:
+                    self._uses_context = True
+                    continue
                 if param.annotation is SignalDescriptor:
                     self._parser_state = ParserState.SIGNAL_DONE
                     continue
@@ -85,9 +112,10 @@ class DeriveUnit:
                     code="invalid_signature",
                 )
 
-        if sig.return_annotation is not ValueDescriptor:
+        self._allows_none = sig.return_annotation == (ValueDescriptor | None)
+        if sig.return_annotation is not ValueDescriptor and not self._allows_none:
             raise DeriveError(
-                "The derive function must return ValueDescriptor",
+                "The derive function must return ValueDescriptor or ValueDescriptor | None",
                 code="invalid_signature",
             )
 
@@ -134,9 +162,12 @@ class DeriveUnit:
         try:
             while not self._stop_event.is_set():
                 try:
-                    input_args = self._input_queue.get(timeout=0.1)
+                    request = self._input_queue.get(timeout=0.1)
                 except Empty:
                     continue
+                input_args = request.inputs
+                if request.revision != self._get_revision():
+                    continue
 
                 if len(input_args) != self._num_input_args:
                     log.warning(
@@ -156,25 +187,31 @@ class DeriveUnit:
                 if self._has_stale_playback_input(input_args):
                     continue
                 try:
-                    value = self._process_function(*input_args, self._return_signal)
-                    self._validate_result(value)
+                    value = self._process(input_args, request.context)
                 except Exception:
-                    if self._has_stale_playback_input(input_args):
+                    if (
+                        request.revision != self._get_revision()
+                        or self._has_stale_playback_input(input_args)
+                    ):
                         continue
                     raise
+                if value is None or request.revision != self._get_revision():
+                    continue
                 if input_args and isinstance(input_args[0], PlaybackValueDescriptor):
-                    value = PlaybackValueDescriptor(
+                    value = DerivedPlaybackValueDescriptor(
                         signal_id=value.signal_id,
                         value=value.value,
                         timestamp=value.timestamp,
                         generation=input_args[0].generation,
+                        revision=request.revision,
                     )
                 else:
-                    value = SignalValueDescriptor(
+                    value = DerivedValueDescriptor(
                         value.signal_id,
                         value.value,
                         value.timestamp,
                         self._return_signal,
+                        request.revision,
                     )
                 while not self._stop_event.is_set():
                     try:
@@ -207,11 +244,17 @@ class DeriveUnit:
                 },
             )
 
-    def put_data(self, input: list[ValueDescriptor]) -> bool:
+    def put_data(
+        self,
+        input: list[ValueDescriptor],
+        *,
+        context: DeriveContext | None = None,
+        revision: object = None,
+    ) -> bool:
         if self._stop_event.is_set() or self._stopped_event.is_set():
             return False
         try:
-            self._input_queue.put_nowait(input)
+            self._input_queue.put_nowait(_DeriveRequest(input, context, revision))
             return True
         except Full:
             with self._status_lock:
@@ -243,7 +286,19 @@ class DeriveUnit:
             worker,
         )
 
-    def _validate_result(self, value: ValueDescriptor) -> None:
+    def _process(
+        self,
+        inputs: list[ValueDescriptor],
+        context: DeriveContext | None,
+    ) -> ValueDescriptor | None:
+        args = [*inputs, context] if self._uses_context else inputs
+        value = self._process_function(*args, self._return_signal)
+        self._validate_result(value)
+        return value
+
+    def _validate_result(self, value: ValueDescriptor | None) -> None:
+        if value is None and self._allows_none:
+            return
         if (
             not isinstance(value, ValueDescriptor)
             or value.signal_id != self._return_signal.id
@@ -294,10 +349,26 @@ class DeriveUnit:
         buffer.sort()
         input_ids = {signal.id for signal in self._input_signals}
         output_buffer = ValueBuffer()
+        history = _SampleHistory()
         for timestamp, signal_id, value in buffer.get_samples():
             if signal_id not in input_ids:
                 continue
 
+            if self._uses_context:
+                if not history.update(ValueDescriptor(signal_id, value, timestamp)):
+                    continue
+                args = history.inputs(
+                    [signal.id for signal in self._input_signals], timestamp
+                )
+                if args is None:
+                    continue
+                result = self._process(args, history.snapshot(timestamp))
+                if result is not None:
+                    output_buffer.append(
+                        result.timestamp, self._return_signal.id, result.value
+                    )
+                continue
+
             input_args: list[ValueDescriptor] = []
             for signal in self._input_signals:
                 if signal.id == signal_id:
@@ -308,9 +379,11 @@ class DeriveUnit:
                     )
                 input_args.append(ValueDescriptor(signal.id, input_value, timestamp))
 
-            result = self._process_function(*input_args, self._return_signal)
-            self._validate_result(result)
-            output_buffer.append(result.timestamp, self._return_signal.id, result.value)
+            result = self._process(input_args, None)
+            if result is not None:
+                output_buffer.append(
+                    result.timestamp, self._return_signal.id, result.value
+                )
         return output_buffer
 
 
@@ -332,7 +405,9 @@ class DeriveRegistry:
         self._get_live_value = get_live_value
         self._get_all_core_signal_descriptors = get_all_core_signal_descriptors
         self._is_current_playback_value = is_current_playback_value
-        self._routing_queue: Queue[ValueDescriptor] = Queue(queue_capacity)
+        self._routing_queue: Queue[tuple[object, ValueDescriptor]] = Queue(
+            queue_capacity
+        )
         self._queue_capacity = queue_capacity
         self._get_source_revision = get_source_revision
         self._units_revision = 0
@@ -420,6 +495,7 @@ class DeriveRegistry:
                 self._core_input_queue,
                 self._is_current_playback_value,
                 self._queue_capacity,
+                self._current_revision,
             )
             self._units[unit_id] = unit
             self._units_revision += 1
@@ -520,7 +596,7 @@ class DeriveRegistry:
         if self._routing_stop_event.is_set() or self._routing_stopped_event.is_set():
             return False
         try:
-            self._routing_queue.put_nowait(message)
+            self._routing_queue.put_nowait((self._current_revision(), message))
             return True
         except Full:
             with self._lock:
@@ -529,6 +605,15 @@ class DeriveRegistry:
                     log.warning("Derive routing queue is full; dropping new requests")
             return False
 
+    def _current_revision(self) -> object:
+        return self._get_source_revision() if self._get_source_revision else None
+
+    def is_current_result(
+        self,
+        message: DerivedValueDescriptor | DerivedPlaybackValueDescriptor,
+    ) -> bool:
+        return message.revision == self._current_revision()
+
     def stop(self) -> bool:
         log.debug(
             "Stopping derive registry",
@@ -661,11 +746,12 @@ class DeriveRegistry:
         try:
             while not self._routing_stop_event.is_set():
                 try:
-                    message = self._routing_queue.get(timeout=0.1)
+                    revision, message = self._routing_queue.get(timeout=0.1)
                 except Empty:
                     continue
 
-                self._route_data(message)
+                if revision == self._current_revision():
+                    self._route_data(message, revision)
         except Exception as error:
             self._error = str(error)
             log.exception(
@@ -686,7 +772,7 @@ class DeriveRegistry:
                 },
             )
 
-    def _route_data(self, message: ValueDescriptor) -> None:
+    def _route_data(self, message: ValueDescriptor, revision: object = None) -> None:
         if (
             isinstance(message, PlaybackValueDescriptor)
             and self._is_current_playback_value is not None
@@ -705,6 +791,32 @@ class DeriveRegistry:
                 continue
 
             args: list[ValueDescriptor] = []
+            context = None
+            if unit._uses_context:
+                if unit._history_revision != revision:
+                    unit._history = _SampleHistory()
+                    unit._history_revision = revision
+                if not unit._history.update(message):
+                    continue
+                inputs = unit._history.inputs(
+                    [signal.id for signal in input_signals], message.timestamp
+                )
+                if inputs is None:
+                    continue
+                context = unit._history.snapshot(message.timestamp)
+                if isinstance(message, PlaybackValueDescriptor):
+                    inputs = [
+                        PlaybackValueDescriptor(
+                            value.signal_id,
+                            value.value,
+                            value.timestamp,
+                            message.generation,
+                        )
+                        for value in inputs
+                    ]
+                unit.put_data(inputs, context=context, revision=revision)
+                continue
+
             for signal in input_signals:
                 if signal.id == message.signal_id:
                     args.append(message)
@@ -726,7 +838,7 @@ class DeriveRegistry:
                         )
                         for value in args
                     ]
-                unit.put_data(args)
+                unit.put_data(args, revision=revision)
 
     def process_offline(
         self, buffer: DLPak, progress: ProgressCallback | None = None
@@ -740,6 +852,7 @@ class DeriveRegistry:
         for timestamp, signal_id, value in buffer._data._samples:
             working_buffer.append(timestamp, signal_id, value)
         processed_buffer = ValueBuffer()
+        empty_outputs: set[UUID] = set()
 
         for layer_index, layer in enumerate(layers):
             layer_units = [self._units[uuid] for uuid in layer]
@@ -750,6 +863,23 @@ class DeriveRegistry:
                         total_units,
                         f"Processing derive unit {unit.uuid()}",
                     )
+                if any(
+                    signal.id in empty_outputs for signal in unit.get_input_signals()
+                ):
+                    log.info(
+                        "Skipping derive unit %s: upstream unit produced no values",
+                        unit.uuid(),
+                        extra={"event": "derive.offline_upstream_empty"},
+                    )
+                    empty_outputs.add(unit.return_signal().id)
+                    completed_units += 1
+                    if progress:
+                        progress(
+                            completed_units,
+                            total_units,
+                            "Upstream signal has no values",
+                        )
+                    continue
                 returned_buffer = unit.process_offline(working_buffer)
                 completed_units += 1
                 if progress:
@@ -764,9 +894,11 @@ class DeriveRegistry:
                             "layer_index": layer_index,
                         },
                     )
-                    raise DeriveError(
-                        "Offline processing returned no values", code="no_values"
-                    )
+                    if not unit._allows_none:
+                        raise DeriveError(
+                            "Offline processing returned no values", code="no_values"
+                        )
+                    empty_outputs.add(unit.return_signal().id)
                 log.debug(
                     "Offline derive unit %s produced %d sample(s)",
                     unit.uuid(),

src/dynalab_core/playback.py

@@ -113,6 +113,11 @@ class PlaybackEngine:
                 not self._stop_event.is_set() and message.generation == self._generation
             )
 
+    def get_generation(self) -> int:
+        """Revision used to invalidate processing history across playback controls."""
+        with self._condition:
+            return self._generation
+
     def get_status(self) -> WorkerStatus:
         return WorkerStatus(
             self._worker_thread.name,

src/dynalab_core/processing.py

@@ -0,0 +1,117 @@
+# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
+# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""Causal sample history and numerical derivatives for DLUnit functions."""
+
+from collections import deque
+from math import factorial, fsum, isfinite, prod
+from types import MappingProxyType
+from uuid import UUID
+
+from dynalab_core.protocols.packets.data import ValueDescriptor
+
+
+class DeriveContext:
+    """Read-only history snapshot supplied by Core to one processing call."""
+
+    def __init__(self, samples: dict[UUID, tuple[tuple[int, float], ...]]) -> None:
+        self._samples = MappingProxyType(dict(samples))
+
+    def derivative(
+        self,
+        value: ValueDescriptor,
+        order: int = 1,
+        *,
+        max_gap_s: float | None = None,
+    ) -> float | None:
+        """Estimate d**order(value)/dt**order in seconds, or return None.
+
+        Uses the last order + 1 distinct samples, including ``value``. Repeated
+        queries do not advance history. No smoothing or future samples are used.
+        ``max_gap_s`` optionally restarts warm-up across excessive sample gaps.
+        """
+        if (
+            not isinstance(order, int)
+            or isinstance(order, bool)
+            or order not in (1, 2, 3)
+        ):
+            raise ValueError("Derivative order must be 1, 2, or 3")
+        if max_gap_s is not None and (not isfinite(max_gap_s) or max_gap_s <= 0):
+            raise ValueError("max_gap_s must be finite and positive")
+        samples = self._samples.get(value.signal_id, ())
+        if len(samples) < order + 1 or samples[-1] != (value.timestamp, value.value):
+            return None
+        samples = samples[-(order + 1) :]
+        if max_gap_s is not None and any(
+            (right[0] - left[0]) / 1_000_000_000 > max_gap_s
+            for left, right in zip(samples, samples[1:])
+        ):
+            return None
+
+        # The nth derivative of the degree-n interpolant is constant. Normalize
+        # time to the window span and subtract the latest value to limit loss of
+        # precision from large absolute timestamps and constant value offsets.
+        span_ns = samples[-1][0] - samples[0][0]
+        span_s = span_ns / 1_000_000_000
+        try:
+            result = (
+                factorial(order)
+                * fsum(
+                    (sample - samples[-1][1])
+                    / prod(
+                        (timestamp - other_timestamp) / span_ns
+                        for j, (other_timestamp, _) in enumerate(samples)
+                        if i != j
+                    )
+                    for i, (timestamp, sample) in enumerate(samples)
+                )
+                / span_s**order
+            )
+        except (ArithmeticError, ValueError):
+            # Extreme value ranges can overflow even when each input is finite.
+            return None
+        return result if isfinite(result) else None
+
+
+class _SampleHistory:
+    """Mutable history owned by a unit's router, or by one offline pass."""
+
+    def __init__(self) -> None:
+        self._samples: dict[UUID, deque[tuple[int, float]]] = {}
+
+    def update(self, value: ValueDescriptor) -> bool:
+        samples = self._samples.setdefault(value.signal_id, deque(maxlen=4))
+        if samples and value.timestamp <= samples[-1][0]:
+            return False
+        if not isfinite(value.value):
+            samples.clear()
+            return False
+        samples.append((value.timestamp, value.value))
+        return True
+
+    def snapshot(self, timestamp: int) -> DeriveContext:
+        return DeriveContext(
+            {
+                key: tuple(sample for sample in samples if sample[0] <= timestamp)
+                for key, samples in self._samples.items()
+            }
+        )
+
+    def inputs(
+        self, signal_ids: list[UUID], timestamp: int
+    ) -> list[ValueDescriptor] | None:
+        values = []
+        for key in signal_ids:
+            sample = next(
+                (
+                    sample
+                    for sample in reversed(self._samples.get(key, ()))
+                    if sample[0] <= timestamp
+                ),
+                None,
+            )
+            if sample is None:
+                return None
+            values.append(ValueDescriptor(key, sample[1], sample[0]))
+        return values

uv.lock

@@ -1,5 +1,5 @@
 version = 1
-revision = 3
+revision = 5
 requires-python = ">=3.13"
 
 [[package]]
@@ -22,7 +22,7 @@ wheels = [
 
 [[package]]
 name = "dynalab-core"
-version = "0.1.0"
+version = "0.2.1"
 source = { editable = "." }
 dependencies = [
     { name = "psutil" },