Give a rate-limit flush its own run, linked to the run that held the value

A flush ran its cascade under no id, so its executions and failures belonged
to no run. It now records a run with source "flush", and the holding run's id
travels on the queue item into a new `flow_run.parent` column, also on the
`/observability/runs` rows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/fluksio/alembic/versions/930ffa094aad_flow_run_parent.py

@@ -0,0 +1,37 @@
+"""flow_run.parent
+
+A rate-limit flush is a run of its own now, and this names the run whose value
+it released. Rows written before this carry an empty string, the same as every
+run that continues no other.
+
+Revision ID: 930ffa094aad
+Revises: b7d2e9c4a1f3
+Create Date: 2026-10-15
+
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "930ffa094aad"
+down_revision = "b7d2e9c4a1f3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "flow_run",
+        sa.Column(
+            "parent",
+            sqlmodel.sql.sqltypes.AutoString(length=64),
+            nullable=False,
+            server_default="",
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("flow_run", "parent")

backend/fluksio/api/routes/observability.py

@@ -85,6 +85,8 @@ class RunRow(BaseModel):
     id: str
     flow: str
     source: str
+    #: The run this one continues: for a flush, the run whose value it released.
+    parent: str = ""
     status: str
     started_at: datetime
     finished_at: datetime | None = None

backend/fluksio/flow/executor.py

@@ -440,13 +440,6 @@ class ExecutionService:
             self.queue.park(node.flow, item)
             return True
 
-        if item.kind == "flush":
-            # A rate-limit window ended; nothing to replay, only to let out.
-            # ponytail: no run record for a flush — it is the tail of the run
-            # that scheduled it, not a run of its own.
-            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)
@@ -461,6 +454,7 @@ class ExecutionService:
                 "flow": item.flow,
                 "node": item.node,
                 "cause": item.cause,
+                "parent": item.parent,
                 "deliveries": item.deliveries,
                 "ts": now,
             }
@@ -483,6 +477,11 @@ class ExecutionService:
         replay = item.deliveries > 1
         emission = item.kind == "emission"
         try:
+            if item.kind == "flush":
+                # A rate-limit window ended; nothing to replay, only to let out.
+                # A run of its own, so what it sets off is recorded somewhere.
+                pipeline.flush(node, item.entry_id)
+                return True
             # An emission's values went into state when the node produced them;
             # this item carries them so its readers get the chunk that caused
             # the wave rather than whichever is newest by the time they run.
@@ -491,7 +490,7 @@ class ExecutionService:
             published = (
                 set(item.outputs)
                 if emission
-                else pipeline.apply_outputs(node, item.outputs or None)
+                else pipeline.apply_outputs(node, item.outputs or None, item.entry_id)
             )
             pipeline.run_downstream(
                 node,

backend/fluksio/flow/metrics.py

@@ -291,6 +291,7 @@ class MetricsCollector:
             "id": run_id,
             "flow": str(event.get("flow") or "")[:NAME_MAX],
             "source": str(event.get("cause") or ""),
+            "parent": str(event.get("parent") or "")[:64],
             "started_ts": ts,
             "started_at": datetime.fromtimestamp(ts, UTC),
             "finished_at": None,

backend/fluksio/flow/pipeline.py

@@ -670,7 +670,7 @@ class Pipeline:
         return passed, held, due_at, stamps.get(flush_key)
 
     def _throttled(
-        self, node: Node, result: dict[str, Any]
+        self, node: Node, result: dict[str, Any], entry_id: str = ""
     ) -> tuple[dict[str, Any], list[str]]:
         """Hold back the outputs whose port is not due to publish yet.
 
@@ -679,6 +679,9 @@ class Pipeline:
         than leaving the consumer on the one before it. Nothing declaring an
         interval means nothing to look up.
 
+        ``entry_id`` is the run doing the holding, which the flush names as
+        its parent.
+
         Returns what passed and the held keys the publish makes stale — the
         caller drops them in the same round trip it writes the values with,
         rather than one DEL per port here.
@@ -699,28 +702,35 @@ class Pipeline:
                 self._state.update(
                     {self._held_key(name): value for name, value in held.items()}
                 )
-                self._schedule_flush(node, due_at, now, pending)
+                self._schedule_flush(node, due_at, now, pending, entry_id)
         return passed, stale
 
     def _schedule_flush(
-        self, node: Node, at: float, now: float, pending: float | None
+        self,
+        node: Node,
+        at: float,
+        now: float,
+        pending: float | None,
+        entry_id: str = "",
     ) -> 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.
+        to do when it fires. So the parent a flush names is the run that booked
+        it, which held the first value of the window, not necessarily the last.
         """
         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")
+        self.defer(node, {}, at - now, kind="flush", parent=entry_id)
 
-    def flush(self, node: Node) -> None:
+    def flush(self, node: Node, entry_id: str = "") -> 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.
+        ``entry_id`` is the flush's own run.
         """
         limited_out = [name for name, s in node.provides.items() if s.interval > 0]
         stored = (
@@ -737,14 +747,17 @@ class Pipeline:
             # Publishing runs the limit again, which is what clears the hold —
             # or puts the timer back, if this fired a hair early. In that case
             # nothing came out and there is nothing downstream to wake.
-            published = self.apply_outputs(node, held)
+            published = self.apply_outputs(node, held, entry_id)
             if published:
-                self.run_downstream(node, changed=published)
+                self.run_downstream(node, entry_id=entry_id, changed=published)
         if any(spec.interval > 0 for spec in node.requires.values()):
             # A wake-up that was held back, not a value that just changed —
             # so this one deliberately walks everything reachable.
             self._execute_parallel(
-                {node, *self._get_downstream(node)}, self._state, check_ready=True
+                {node, *self._get_downstream(node)},
+                self._state,
+                check_ready=True,
+                entry_id=entry_id,
             )
 
     def _increment_message_versions(self, outputs: dict[str, Any]) -> None:
@@ -791,7 +804,7 @@ class Pipeline:
 
         return self._state.compare_and_swap_multi(expected, updates)
 
-    def _input_is_due(self, node: Node) -> bool:
+    def _input_is_due(self, node: Node, entry_id: str = "") -> bool:
         """Has any rate-limited input waited out its interval?
 
         A node runs when *any* of its inputs is due, so a slow port next to a
@@ -827,6 +840,7 @@ class Pipeline:
                 ),
                 now,
                 stamps.get(flush_key),
+                entry_id,
             )
             return False
 
@@ -834,7 +848,7 @@ class Pipeline:
         return True
 
     def _is_node_ready(
-        self, node: Node, state: StateBackend
+        self, node: Node, state: StateBackend, entry_id: str = ""
     ) -> tuple[bool, dict[str, Any]]:
         """Whether the node may run, and the inputs the answer was read from.
 
@@ -851,7 +865,7 @@ class Pipeline:
         if any(msg not in values for msg in waited_on):
             return False, values
 
-        if not self._input_is_due(node):
+        if not self._input_is_due(node, entry_id):
             return False, values
 
         if not node.synchronous:
@@ -1062,7 +1076,7 @@ class Pipeline:
 
             stale: list[str] = []
             if result:
-                result, stale = self._throttled(node, result)
+                result, stale = self._throttled(node, result, entry_id)
 
             if result:
                 self._record_outputs(node, result, state, stale)
@@ -1358,7 +1372,7 @@ class Pipeline:
                         # The values are kept whether or not the answer is yes:
                         # a node that may run reads them below instead of
                         # asking state for the same keys again.
-                        ready, ready_inputs = self._is_node_ready(n, state)
+                        ready, ready_inputs = self._is_node_ready(n, state, entry_id)
                         if not ready:
                             # Completed, not merely passed over: a later trigger
                             # may still make it ready, but not running is not
@@ -1440,11 +1454,14 @@ class Pipeline:
 
         return self._execute_parallel(nodes, self._state, check_ready=False)
 
-    def apply_outputs(self, node: Node, outputs: dict[str, Any] | None) -> set[str]:
+    def apply_outputs(
+        self, node: Node, outputs: dict[str, Any] | None, entry_id: str = ""
+    ) -> set[str]:
         """Record what a node emitted: state, history, versions and events.
 
         Shared by the direct path and by the execution service replaying a
         journaled item, so a value looks the same on the canvas either way.
+        ``entry_id`` is the run this happens in.
 
         Returns the message names that actually reached state — post rate
         limiting — which is what the cascade behind it has to walk from.
@@ -1453,7 +1470,7 @@ class Pipeline:
         if outputs:
             # This is where a chatty subscriber gets thinned out, so a port set
             # to publish every 60s does so whatever the broker sends.
-            outputs, stale = self._throttled(node, outputs)
+            outputs, stale = self._throttled(node, outputs, entry_id)
 
         if not outputs:
             return set()
@@ -1569,6 +1586,7 @@ class Pipeline:
             self._state.update(
                 {self._held_key(name): value for name, value in held.items()}
             )
+            # Held before any run began, so its flush names no parent.
             self._schedule_flush(node, due_at, now, pending)
             return True
         except Exception:
@@ -1607,7 +1625,9 @@ class Pipeline:
         )
         try:
             published = (
-                set(outputs or {}) if delivered else self.apply_outputs(node, outputs)
+                set(outputs or {})
+                if delivered
+                else self.apply_outputs(node, outputs, run_id)
             )
             state = self.run_downstream(
                 node,
@@ -1684,6 +1704,7 @@ class Pipeline:
         seconds: float,
         guard: tuple[str, Any] | None = None,
         kind: str = "cascade",
+        parent: str = "",
     ) -> bool:
         """Publish a node's outputs later, without holding a worker thread.
 
@@ -1691,6 +1712,8 @@ class Pipeline:
         over; if it has moved on, the item is dropped. That is how a wait which
         gets restarted cancels the one it replaced.
 
+        ``parent`` is the run the item continues, recorded beside its own.
+
         Returns False when there is no queue to hold the item, in which case
         the caller has to wait however it waited before.
         """
@@ -1703,9 +1726,10 @@ class Pipeline:
             node=node.id,
             flow=node.flow,
             outputs=outputs,
-            cause="delay",
+            cause="flush" if kind == "flush" else "delay",
             guard_key=guard[0] if guard else "",
             guard_value=str(guard[1]) if guard else "",
+            parent=parent,
         )
         try:
             self._queue.add_delayed(item, time.time() + seconds)

backend/fluksio/flow/queue.py

@@ -94,6 +94,9 @@ class WorkItem:
     #: The run this item is, on a ``run`` item. Everything else about a run is
     #: in its database row, so the journal only has to name it.
     run_id: str = ""
+    #: The run that scheduled this item, on a ``flush``: the one whose value
+    #: the rate limit held back.
+    parent: str = ""
 
     def to_fields(self) -> dict[str, str]:
         return {
@@ -107,6 +110,7 @@ class WorkItem:
             "guard_value": self.guard_value,
             "enqueued_at": str(self.enqueued_at),
             "run_id": self.run_id,
+            "parent": self.parent,
         }
 
     @classmethod
@@ -126,6 +130,7 @@ class WorkItem:
             deliveries=deliveries,
             enqueued_at=float(fields.get("enqueued_at") or time.time()),
             run_id=fields.get("run_id", ""),
+            parent=fields.get("parent", ""),
         )
 
 

backend/fluksio/models.py

@@ -298,6 +298,9 @@ class FlowRun(SQLModel, table=True):
     flow: str = Field(index=True, max_length=255)
     #: What caused it: the work item's cause, "manual", or "run" for a batch.
     source: str = ""
+    #: The run this one continues: on a rate-limit flush, the run whose value
+    #: the window held back. Empty for everything else.
+    parent: str = Field(default="", max_length=64)
     started_at: datetime = Field(
         index=True,
         sa_type=UTCDateTime,

backend/tests/test_metrics.py

@@ -11,7 +11,12 @@ from datetime import UTC, datetime
 from sqlmodel import Session, select
 
 from fluksio.flow.events import EventBus
+from fluksio.flow.executor import ExecutionService
+from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.metrics import DETAIL_CAP, RUN_STALE_S, MetricsCollector
+from fluksio.flow.nodes import Node
+from fluksio.flow.pipeline import Pipeline
+from fluksio.flow.queue import MemoryWorkQueue
 from fluksio.models import EngineEvent, FlowRun, MetricBucket
 
 FLOW = "metrics-test"
@@ -292,3 +297,52 @@ def test_a_failure_keeps_the_run_it_happened_in(db: Session) -> None:
 
     failure = db.exec(select(EngineEvent).where(EngineEvent.run == "run-abc123")).one()
     assert failure.detail.startswith("ValueError: in a run")
+
+
+def test_a_rate_limit_flush_is_a_run_that_names_its_parent(db: Session) -> None:
+    """A flush used to run under no id, so what it set off was in no run."""
+    collector = MetricsCollector(EventBus())
+
+    class Bus:
+        """Hands every event straight to the collector."""
+
+        def publish(self, event: dict) -> None:
+            collector.handle(event)
+
+    window = 0.05
+    source = Node(
+        f=lambda params: None,
+        provides=[MessageSpec(name="raw", dtype=DType.FLOAT)],
+        name="in",
+    )
+    relay = Node(
+        f=lambda raw, params: {"level": raw},
+        requires=[MessageSpec(name="raw", dtype=DType.FLOAT)],
+        provides=[MessageSpec(name="level", dtype=DType.FLOAT, interval=window)],
+        name="relay",
+    )
+    source.assign_flow("metrics-flush", "in")
+    relay.assign_flow("metrics-flush", "relay")
+    queue = MemoryWorkQueue()
+    pipeline = Pipeline(nodes=[source, relay], work_queue=queue, events=Bus())
+    service = ExecutionService(queue, events=Bus())
+    service.bind(pipeline)
+
+    # The second value lands inside the relay's window and is held.
+    held_in = ""
+    for value in (1.0, 2.0):
+        source.inject({"raw": value})
+        for item in queue.claim(10, 10):
+            service._run_item(item)
+            held_in = item.entry_id
+    time.sleep(window * 2)
+    queue.move_due(time.time())
+    for item in queue.claim(10, 10):
+        service._run_item(item)
+    asyncio.run(collector.flush())
+
+    flush = db.exec(select(FlowRun).where(FlowRun.source == "flush")).one()
+    assert (flush.status, flush.parent) == ("ok", held_in)
+    parent = db.get(FlowRun, held_in)
+    assert parent is not None
+    assert (parent.flow, parent.source) == ("metrics-flush", "external")

frontend/src/client/schemas.gen.ts

@@ -4777,6 +4777,11 @@ export const fluksio__api__routes__observability__RunRowSchema = {
             type: 'string',
             title: 'Source'
         },
+        parent: {
+            type: 'string',
+            title: 'Parent',
+            default: ''
+        },
         status: {
             type: 'string',
             title: 'Status'

frontend/src/client/types.gen.ts

@@ -506,6 +506,7 @@ export type fluksio__api__routes__observability__RunRow = {
     id: string;
     flow: string;
     source: string;
+    parent?: string;
     status: string;
     started_at: string;
     finished_at?: (string | null);