Record a short value history per message

The panel is going to sparkline how a value moved, and nothing kept more
than the latest one. Emissions now append to a capped list beside the
value — 120 points, enough to fill a sparkline without making Redis a
time-series store — and `/flows/{flow}/history/{message}` hands it back
oldest first.

Only numbers are recorded, bools included as neither, so a string
message costs nothing at all. The response carries `numeric` so an empty
series reads as "not plottable" rather than "nothing yet". The append is
one pipelined round-trip per emission batch and sits outside the lock,
since the list is append-only.

MemoryState keeps the same window in a deque, so the endpoint answers
without Redis too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KkmeRiyeYmVZqJVwuyHq9o

backend/app/api/routes/flows.py

@@ -26,11 +26,14 @@ from app.flow.schemas import (
     FlowsPublic,
     FlowStatePublic,
     FlowSummary,
+    HistoryPoint,
+    MessageHistory,
     MessageValue,
     NodeSource,
     NodeStatusPublic,
     NodeTypeInfo,
 )
+from app.flow.state import as_number
 from app.flow.store import FlowExists, FlowNotFound
 from app.models import Message
 
@@ -275,6 +278,29 @@ def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
     return _flow_state(controller, name)
 
 
+@router.get("/{name}/history/{message}", response_model=MessageHistory)
+def read_message_history(
+    name: str,
+    message: str,
+    controller: FlowControllerDep,
+) -> Any:
+    """The recent values of one message, for plotting.
+
+    ``message`` may be given bare or qualified; a message that never carried a
+    number comes back with an empty series.
+    """
+    _read_flow(controller, name)
+    key = qualify(name, message)
+    return MessageHistory(
+        message=key,
+        numeric=as_number(controller.state.get(key)) is not None,
+        points=[
+            HistoryPoint(ts=ts, value=value)
+            for ts, value in controller.state.history(key)
+        ],
+    )
+
+
 # -----------------------------------------------------------------------------
 # Live updates
 # -----------------------------------------------------------------------------

backend/app/flow/pipeline.py

@@ -322,6 +322,8 @@ class Pipeline:
                     state.update(
                         {self._timestamp_key(name): ts for name in result},
                     )
+                # Append-only, so it needs no lock of its own.
+                state.append_history(result, ts)
                 self._increment_message_versions(result)
                 for name, value in result.items():
                     self._publish(
@@ -436,6 +438,7 @@ class Pipeline:
             with state.lock():
                 state.update(outputs)
                 state.update({self._timestamp_key(name): ts for name in outputs})
+            state.append_history(outputs, ts)
             self._increment_message_versions(outputs)
             for name, value in outputs.items():
                 self._publish(

backend/app/flow/schemas.py

@@ -91,6 +91,25 @@ class MessageValue(BaseModel):
     ts: float | None = None
 
 
+class HistoryPoint(BaseModel):
+    """One numeric value a message carried, and when."""
+
+    ts: float
+    value: float
+
+
+class MessageHistory(BaseModel):
+    """A message's recent numeric values, oldest first.
+
+    Only numbers are recorded, so ``numeric`` tells the panel whether an empty
+    series means "nothing plottable here" or "nothing has arrived yet".
+    """
+
+    message: str
+    numeric: bool = False
+    points: list[HistoryPoint] = Field(default_factory=list)
+
+
 class FlowSummary(BaseModel):
     name: str
     title: str = ""

backend/app/flow/state.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import json
 from abc import ABC, abstractmethod
+from collections import deque
 from collections.abc import Iterator
 from contextlib import contextmanager
 from threading import RLock
@@ -16,6 +17,19 @@ from typing import Any, cast
 
 import redis
 
+# A sparkline only means something for numbers, so the history keeps the values
+# it can plot and nothing else. 120 points fill a panel-wide chart while leaving
+# Redis a cache rather than a time-series database.
+HISTORY_LIMIT = 120
+
+
+def as_number(value: Any) -> float | None:
+    """The plottable form of a value, or None if it is not a number."""
+    # bool is an int subclass; a flag is not a measurement.
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return None
+    return float(value)
+
 
 class StateBackend(ABC):
     """
@@ -149,6 +163,36 @@ class StateBackend(ABC):
         """
         ...
 
+    # -------------------------------------------------------------------------
+    # Message history
+    # -------------------------------------------------------------------------
+
+    @abstractmethod
+    def append_history(self, values: dict[str, Any], ts: float) -> None:
+        """
+        Record freshly published values in each message's capped history.
+
+        Non-numeric payloads are skipped — the history exists to be plotted.
+
+        :param values: Message names mapped to the value just published.
+        :type values: dict[str, Any]
+        :param ts: When they were published.
+        :type ts: float
+        """
+        ...
+
+    @abstractmethod
+    def history(self, key: str) -> list[tuple[float, float]]:
+        """
+        The recorded ``(timestamp, value)`` pairs of one message, oldest first.
+
+        :param key: The message name.
+        :type key: str
+        :returns: At most ``HISTORY_LIMIT`` points; empty if nothing was recorded.
+        :rtype: list[tuple[float, float]]
+        """
+        ...
+
     def __contains__(self, key: str) -> bool:
         return self.exists(key)
 
@@ -180,10 +224,11 @@ class MemoryState(StateBackend):
         'value'
     """
 
-    __slots__ = ("_data", "_lock")
+    __slots__ = ("_data", "_history", "_lock")
 
     def __init__(self) -> None:
         self._data: dict[str, Any] = {}
+        self._history: dict[str, deque[tuple[float, float]]] = {}
         self._lock = RLock()  # Reentrant lock for nested access
 
     def get(self, key: str, default: Any = None) -> Any:
@@ -201,6 +246,7 @@ class MemoryState(StateBackend):
     def clear(self) -> None:
         with self._lock:
             self._data.clear()
+            self._history.clear()
 
     def keys(self) -> list[str]:
         with self._lock:
@@ -251,6 +297,21 @@ class MemoryState(StateBackend):
             self._data.update(updates)
             return True
 
+    def append_history(self, values: dict[str, Any], ts: float) -> None:
+        """Append the numeric values to their message's bounded series."""
+        with self._lock:
+            for key, value in values.items():
+                number = as_number(value)
+                if number is None:
+                    continue
+                series = self._history.setdefault(key, deque(maxlen=HISTORY_LIMIT))
+                series.append((ts, number))
+
+    def history(self, key: str) -> list[tuple[float, float]]:
+        """The recorded points of one message, oldest first."""
+        with self._lock:
+            return list(self._history.get(key, ()))
+
 
 class RedisState(StateBackend):
     """
@@ -485,3 +546,33 @@ class RedisState(StateBackend):
         except redis.WatchError:
             # Another client modified one of the watched keys
             return False
+
+    def _history_key(self, key: str) -> str:
+        return self._key(f"__history__:{key}")
+
+    def append_history(self, values: dict[str, Any], ts: float) -> None:
+        """Push the numeric values onto their capped list, in one round-trip."""
+        pipe = self._client.pipeline()
+        queued = False
+        for key, value in values.items():
+            number = as_number(value)
+            if number is None:
+                continue
+            history_key = self._history_key(key)
+            pipe.lpush(history_key, json.dumps([ts, number]))
+            pipe.ltrim(history_key, 0, HISTORY_LIMIT - 1)
+            if self._ttl:
+                pipe.expire(history_key, self._ttl)
+            queued = True
+        if queued:
+            pipe.execute()
+
+    def history(self, key: str) -> list[tuple[float, float]]:
+        """The recorded points of one message, oldest first."""
+        entries = cast(list[bytes], self._client.lrange(self._history_key(key), 0, -1))
+        # LPUSH puts the newest first, a chart reads the other way round.
+        points: list[tuple[float, float]] = []
+        for entry in reversed(entries):
+            ts, value = json.loads(entry)
+            points.append((float(ts), float(value)))
+        return points

backend/tests/flow/test_history.py

@@ -0,0 +1,58 @@
+"""Message history: the series a node panel draws as a sparkline."""
+
+from app.flow.messages import MessageSpec
+from app.flow.nodes import Node
+from app.flow.pipeline import Pipeline
+from app.flow.state import HISTORY_LIMIT, MemoryState
+
+
+def emitter(values) -> Node:
+    """A node publishing the given values, one per execution."""
+    remaining = list(values)
+
+    def process(params):
+        return {"out": remaining.pop(0)}
+
+    node = Node(f=process, requires=[], provides=[MessageSpec(name="out")], name="src")
+    node.assign_flow("f", "src")
+    return node
+
+
+def test_the_series_follows_the_order_values_were_published():
+    node = emitter([1.0, 2.0, 3.0])
+    pipeline = Pipeline(nodes=[node])
+
+    # An injecting node publishes directly, an executed one goes through the
+    # executor; both paths land in the same series.
+    node.inject()
+    pipeline.run()
+    pipeline.run()
+
+    points = pipeline.state.history("f.out")
+    assert [value for _, value in points] == [1.0, 2.0, 3.0]
+    assert [ts for ts, _ in points] == sorted(ts for ts, _ in points)
+
+
+def test_the_series_is_capped():
+    state = MemoryState()
+    for i in range(HISTORY_LIMIT + 20):
+        state.append_history({"f.out": float(i)}, ts=float(i))
+
+    points = state.history("f.out")
+    assert len(points) == HISTORY_LIMIT
+    # The oldest twenty fell off the end.
+    assert points[0] == (20.0, 20.0)
+    assert points[-1] == (139.0, 139.0)
+
+
+def test_only_numbers_are_recorded():
+    state = MemoryState()
+    state.append_history({"f.text": "warm", "f.flag": True, "f.temp": 21}, ts=1.0)
+
+    assert state.history("f.text") == []
+    assert state.history("f.flag") == []
+    assert state.history("f.temp") == [(1.0, 21.0)]
+
+
+def test_a_message_without_history_comes_back_empty():
+    assert MemoryState().history("f.never") == []