Step a paused flow, and deliver what a rate limit held back

Three things a pause and an interval were quietly losing:

- A rebuild builds a fresh pipeline, so nothing is paused any more and no
  resume ever comes for what the old one parked. Release it on rebuild.
- POST /flows/{name}/step takes the oldest parked item and runs that one wave
  while the flow stays paused, so a held-back cascade can be walked through.
  Nothing parked answers plainly rather than failing.
- A per-port interval was leading-edge only: a producer going quiet inside the
  window left the consumer on the value before it. The held value is kept and
  a flush item scheduled on the queue's existing timer, so the window ends with
  a delivery. One timer in flight per node, and none without a queue to run it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H7LwYgJfpkbLCTeiAf8U4A

backend/app/api/routes/flows.py

@@ -541,6 +541,16 @@ async def resume_flow(name: str, controller: FlowControllerDep) -> Any:
     return Message(message=f"Resumed flow '{name}'")
 
 
+@router.post("/{name}/step", response_model=Message)
+async def step_flow(name: str, controller: FlowControllerDep) -> Any:
+    """Run one message a pause is holding back, leaving the flow paused."""
+    _read_flow(controller, name)
+    node = await run_in_threadpool(controller.step_flow, name)
+    if node is None:
+        return Message(message=f"Nothing held back in flow '{name}'")
+    return Message(message=f"Stepped '{node}'")
+
+
 # -----------------------------------------------------------------------------
 # Validation and execution
 # -----------------------------------------------------------------------------

backend/app/flow/controller.py

@@ -352,6 +352,10 @@ class FlowController:
 
             if self.execution is not None:
                 self.execution.resume_intake()
+                # A rebuild clears the pause, so no resume will ever come for
+                # what the old pipeline parked. Release it here or it is lost.
+                for flow in published:
+                    self._release_parked(flow.name)
 
         self._publish(
             {
@@ -597,11 +601,21 @@ class FlowController:
         if self.pipeline is None:
             return
         self.pipeline.resume(flow)
-        if self.execution is not None:
-            # Whatever arrived while the flow was held is queued again, oldest
-            # first, so a pause loses nothing.
-            for item in self.execution.queue.unpark(flow):
-                self.execution.queue.add(item)
+        self._release_parked(flow)
+
+    def _release_parked(self, flow: str) -> None:
+        """Queue what a pause held back again, oldest first, so it is not lost."""
+        if self.execution is None:
+            return
+        for item in self.execution.queue.unpark(flow):
+            self.execution.queue.add(item)
+
+    def step_flow(self, flow: str) -> str | None:
+        """Run one held-back item, leaving the flow paused. Blocking.
+
+        Returns the node it came from, or None when nothing is held back.
+        """
+        return self.execution.step(flow) if self.execution is not None else None
 
     def set_history_limits(self, limits: dict[str, int]) -> None:
         """How deep to keep each charted message's series. Applies at once."""

backend/app/flow/executor.py

@@ -187,6 +187,25 @@ class ExecutionService:
     # Handling one item
     # -------------------------------------------------------------------------
 
+    def step(self, flow: str) -> str | None:
+        """Run one item a pause is holding, and hold everything else still.
+
+        Blocking, so the caller sees the wave finish. Returns the node the item
+        came from, or None when nothing is parked for this flow.
+        """
+        pipeline = self._pipeline
+        if pipeline is None:
+            return None
+        item = self.queue.unpark_one(flow)
+        if item is None:
+            return None
+        with pipeline.stepping(flow):
+            try:
+                self._run_item(item)
+            except Exception:
+                logger.exception("Step of '%s' failed", item.node)
+        return item.node
+
     def _handle(self, item: WorkItem) -> None:
         handled = True
         try:
@@ -235,10 +254,15 @@ class ExecutionService:
         if pipeline.is_disabled(node.flow):
             return True
 
-        if pipeline.is_paused(node.flow):
+        if pipeline.is_paused(node.flow) and not pipeline.is_stepping(node.flow):
             self.queue.park(node.flow, item)
             return True
 
+        if item.kind == "flush":
+            # A rate-limit window ended; nothing to replay, only to let out.
+            pipeline.flush(node)
+            return True
+
         if item.guard_key and str(node.recall(item.guard_key, "")) != item.guard_value:
             # The node moved on while this waited — a restarted timer, say.
             logger.debug("Guard no longer holds for '%s', dropped", item.node)

backend/app/flow/pipeline.py

@@ -12,7 +12,9 @@ import logging
 import threading
 import time
 from collections import deque
+from collections.abc import Iterator
 from concurrent.futures import Future, ThreadPoolExecutor, wait
+from contextlib import contextmanager
 from typing import Any, Literal
 
 from pydantic import BaseModel
@@ -84,6 +86,7 @@ class Pipeline:
         "_downstream_cache",
         "_disabled",
         "_paused",
+        "_stepping",
         "_gate_lock",
         "_queue",
         "_node_pool",
@@ -106,6 +109,7 @@ class Pipeline:
         # debugging state that a rebuild is meant to clear.
         self._disabled = frozenset(disabled_flows or ())
         self._paused: set[str] = set()
+        self._stepping: set[str] = set()
         self._gate_lock = threading.Lock()
         # An empty state backend is falsy, so this cannot be ``state or ...``:
         # that would quietly hand the pipeline a second, private state and
@@ -327,11 +331,21 @@ class Pipeline:
         """When a rate-limited input last woke this node."""
         return f"__in_ts__:{node_name}:{msg_name}"
 
-    def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]:
-        """Drop the outputs whose port is not due to publish yet.
+    def _held_key(self, msg_name: str) -> str:
+        """The value a rate-limited output kept back, waiting for its window."""
+        return f"__held__:{msg_name}"
 
-        The value is not lost — the port publishes the current one next time it
-        is due. Nothing declaring an interval means nothing to look up.
+    def _flush_key(self, node_name: str) -> str:
+        """When a rate-limit window of this node is due to be let through."""
+        return f"__flush__:{node_name}"
+
+    def _throttled(self, node: Node, result: dict[str, Any]) -> dict[str, Any]:
+        """Hold back the outputs whose port is not due to publish yet.
+
+        The value is not lost: it is kept and published when the window ends,
+        so a producer that goes quiet still delivers its last reading rather
+        than leaving the consumer on the one before it. Nothing declaring an
+        interval means nothing to look up.
         """
         limited = {
             name: spec.interval
@@ -342,13 +356,79 @@ class Pipeline:
             return result
 
         now = time.time()
-        stamps = self._state.get_multi([self._timestamp_key(name) for name in limited])
-        return {
-            name: value
-            for name, value in result.items()
-            if name not in limited
-            or now - (stamps.get(self._timestamp_key(name)) or 0) >= limited[name]
+        flush_key = self._flush_key(node.id)
+        stamps = self._state.get_multi(
+            [self._timestamp_key(name) for name in limited] + [flush_key]
+        )
+
+        passed: dict[str, Any] = {}
+        held: dict[str, Any] = {}
+        due_at = 0.0
+        for name, value in result.items():
+            if name not in limited:
+                passed[name] = value
+                continue
+            window_ends = (stamps.get(self._timestamp_key(name)) or 0) + limited[name]
+            if now >= window_ends:
+                passed[name] = value
+            else:
+                held[name] = value
+                due_at = min(due_at or window_ends, window_ends)
+
+        if self._queue is not None:
+            # Without a queue there is no timer to let the value out later, so
+            # holding it would only mean losing it more slowly.
+            for name in limited:
+                if name in passed:
+                    # A fresh publish makes anything held for that port stale.
+                    self._state.delete(self._held_key(name))
+            if held:
+                self._state.update(
+                    {self._held_key(name): value for name, value in held.items()}
+                )
+                self._schedule_flush(node, due_at, now, stamps.get(flush_key))
+        return passed
+
+    def _schedule_flush(
+        self, node: Node, at: float, now: float, pending: float | None
+    ) -> None:
+        """Come back to this node when its rate-limit window ends.
+
+        One timer in flight per node: a pending one that is early enough is
+        left alone, and one that turns out to be too late simply finds nothing
+        to do when it fires.
+        """
+        if self._queue is None or (pending and now < pending <= at):
+            return
+        self._state[self._flush_key(node.id)] = at
+        self.defer(node, {}, at - now, kind="flush")
+
+    def flush(self, node: Node) -> None:
+        """Let through what this node's rate limits held back.
+
+        Runs when a window ends: the last value a limited output kept back is
+        published, and a node whose inputs were all held back gets its run.
+        """
+        limited_out = [name for name, s in node.provides.items() if s.interval > 0]
+        stored = (
+            self._state.get_multi([self._held_key(name) for name in limited_out])
+            if limited_out
+            else {}
+        )
+        held = {
+            name: stored[self._held_key(name)]
+            for name in limited_out
+            if stored.get(self._held_key(name)) is not None
         }
+        if held:
+            # Publishing runs the limit again, which is what clears the hold —
+            # or puts the timer back, if this fired a hair early.
+            self.apply_outputs(node, held)
+            self.run_downstream(node)
+        if any(spec.interval > 0 for spec in node.requires.values()):
+            self._execute_parallel(
+                {node, *self._get_downstream(node)}, self._state, check_ready=True
+            )
 
     def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
         for msg_name in outputs:
@@ -411,8 +491,9 @@ class Pipeline:
             return True
 
         now = time.time()
+        flush_key = self._flush_key(node.id)
         keys = [self._delivered_key(node.id, name) for name in limited]
-        stamps = self._state.get_multi(keys)
+        stamps = self._state.get_multi(keys + [flush_key])
         due = [
             name
             for name, interval in limited.items()
@@ -420,6 +501,17 @@ class Pipeline:
         ]
         # An unlimited input alongside a held-back one still wakes the node.
         if not due and len(limited) == len(node.requires):
+            # The value is in state, only the wake-up is held back. Come back
+            # for it, so a producer going quiet does not strand it there.
+            self._schedule_flush(
+                node,
+                min(
+                    (stamps.get(self._delivered_key(node.id, name)) or 0) + interval
+                    for name, interval in limited.items()
+                ),
+                now,
+                stamps.get(flush_key),
+            )
             return False
 
         with self._state.lock():
@@ -585,17 +677,37 @@ class Pipeline:
         # Whatever was held back is free to run now.
         self._execute_parallel(self.flow_nodes(flow), self._state, check_ready=True)
 
+    @contextmanager
+    def stepping(self, flow: str) -> Iterator[None]:
+        """Let one held-back wave through without ending the pause.
+
+        The flow stays paused throughout, so everything else arriving for it
+        keeps piling up where the next step can find it.
+        """
+        with self._gate_lock:
+            self._stepping.add(flow)
+        try:
+            yield
+        finally:
+            with self._gate_lock:
+                self._stepping.discard(flow)
+
     def _gate_blocks(self, node: Node) -> bool:
         """Is this node's flow held back from executing?"""
         if node.flow in self._disabled:
             return True
         with self._gate_lock:
-            return node.flow in self._paused
+            return node.flow in self._paused and node.flow not in self._stepping
 
     def is_paused(self, flow: str) -> bool:
         with self._gate_lock:
             return flow in self._paused
 
+    def is_stepping(self, flow: str) -> bool:
+        """Is a single step of this paused flow running right now?"""
+        with self._gate_lock:
+            return flow in self._stepping
+
     def _execute_parallel(
         self,
         nodes_subset: set[Node] | None,
@@ -838,6 +950,7 @@ class Pipeline:
         outputs: dict[str, Any],
         seconds: float,
         guard: tuple[str, Any] | None = None,
+        kind: str = "cascade",
     ) -> bool:
         """Publish a node's outputs later, without holding a worker thread.
 
@@ -853,7 +966,7 @@ class Pipeline:
         if self._queue is None or seconds <= 0:
             return False
         item = WorkItem(
-            kind="cascade",
+            kind=kind,
             node=node.id,
             flow=node.flow,
             outputs=outputs,

backend/app/flow/queue.py

@@ -40,7 +40,7 @@ class WorkItem:
     """One unit of journaled work.
 
     :param kind: ``cascade`` replays a node's outputs and runs what is
-        downstream; ``node`` executes exactly one node.
+        downstream; ``flush`` lets out what a node's rate limits held back.
     :param node: The node the item is about — the source for a cascade, the
         target for a node item.
     :param flow: The flow that node belongs to, so gating needs no lookup.
@@ -135,6 +135,10 @@ class WorkQueue(ABC):
     def unpark(self, flow: str) -> list[WorkItem]:
         """Return a paused flow's held items to the queue, oldest first."""
 
+    @abstractmethod
+    def unpark_one(self, flow: str) -> WorkItem | None:
+        """Take the oldest held item, leaving the rest parked. That is a step."""
+
     @abstractmethod
     def clear_flow(self, flow: str) -> None:
         """Forget anything held for a flow that no longer exists."""
@@ -227,6 +231,11 @@ class MemoryWorkQueue(WorkQueue):
         with self._lock:
             return self._parked.pop(flow, [])
 
+    def unpark_one(self, flow: str) -> WorkItem | None:
+        with self._lock:
+            held = self._parked.get(flow)
+            return held.pop(0) if held else None
+
     def clear_flow(self, flow: str) -> None:
         with self._lock:
             self._parked.pop(flow, None)
@@ -387,6 +396,10 @@ class RedisWorkQueue(WorkQueue):
         self._redis.delete(key)
         return [WorkItem.from_fields(json.loads(r), "") for r in raw]
 
+    def unpark_one(self, flow: str) -> WorkItem | None:
+        raw = cast("str | None", self._redis.lpop(self._parked_key(flow)))
+        return WorkItem.from_fields(json.loads(raw), "") if raw else None
+
     def clear_flow(self, flow: str) -> None:
         self._redis.delete(self._parked_key(flow))