Record batch runs beside cascades so Home lists them

A batch run opens no cascade, and FlowRun was written only from
cascade_started — so `fluksio run` showed on /runs and in `fluksio status` and
was simply absent from Home. The collector now folds the run_started and
run_finished events RunService already published. Such a record is exempt from
the staleness sweep in both places: a training step of an hour is a normal one,
and only run_finished ends it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/fluksio/flow/metrics.py

@@ -35,6 +35,10 @@ FLUSH_INTERVAL_S = 15.0
 DETAIL_CAP = 8000
 #: A cascade still open this long after it started is never finishing.
 RUN_STALE_S = 600.0
+#: What ``flow_run.source`` says for a batch run, as against the cause a
+#: cascade carries. A training step can legitimately run for hours, so this is
+#: also what exempts one from the staleness sweep above.
+RUN_SOURCE = "run"
 #: ``metric_minute.flow``/``node`` and ``flow_run.flow`` are this wide, and a
 #: node id has no length of its own. One row over it aborts the transaction,
 #: which would take every other row in the flush with it.
@@ -214,9 +218,18 @@ class MetricsCollector:
 
         if kind == "cascade_finished":
             if run is not None:
-                run["finished_at"] = datetime.fromtimestamp(ts, UTC)
-                run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2)
-                run["status"] = "error" if run["errors"] else "ok"
+                self._finish_run(run, ts)
+            return
+
+        # A batch run opens no cascade, so without these it ran, recorded its
+        # nodes and its failures, and appeared in no history at all.
+        if kind == "run_started":
+            self._start_run({**event, "cause": RUN_SOURCE}, ts)
+            return
+
+        if kind == "run_finished":
+            if run is not None:
+                self._finish_run(run, ts, str(event.get("status") or ""))
             return
 
         if kind == "node_health":
@@ -257,6 +270,14 @@ class MetricsCollector:
                 )
             )
 
+    def _finish_run(
+        self, run: dict[str, Any], ts: float, status: str = ""
+    ) -> None:
+        """Close an open record. A run says how it ended; a cascade is told."""
+        run["finished_at"] = datetime.fromtimestamp(ts, UTC)
+        run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2)
+        run["status"] = status or ("error" if run["errors"] else "ok")
+
     def _start_run(self, event: dict[str, Any], ts: float) -> None:
         run_id = str(event.get("run") or "")
         if not run_id:
@@ -305,7 +326,11 @@ class MetricsCollector:
             self._last_prune = time.monotonic()
         cutoff = time.time() - RUN_STALE_S
         for run_id, run in list(self._runs.items()):
-            if run["status"] != "running" or run["started_ts"] < cutoff:
+            if run["status"] != "running":
+                del self._runs[run_id]
+            # A batch run is held however long it takes: a training step of an
+            # hour is a normal one, and `run_finished` is what ends it.
+            elif run["source"] != RUN_SOURCE and run["started_ts"] < cutoff:
                 del self._runs[run_id]
         # Held tracebacks survive the flush: the log and the failure it belongs
         # to are two events, and a flush can fall between them. One per node,
@@ -405,12 +430,15 @@ class MetricsCollector:
         session.execute(delete(MetricBucket).where(col(MetricBucket.bucket) < cutoff))
         session.execute(delete(EngineEvent).where(col(EngineEvent.ts) < cutoff))
         session.execute(delete(FlowRun).where(col(FlowRun.started_at) < cutoff))
-        # A run still open long after it started did not finish; saying so is
-        # more honest than leaving it running forever.
+        # A cascade still open long after it started did not finish; saying so
+        # is more honest than leaving it running forever. A batch run is not
+        # swept: it is allowed to take hours, and `Run` has its own lease sweep
+        # for the case where the engine died holding one.
         session.execute(
             update(FlowRun)
             .where(
                 col(FlowRun.status) == "running",
+                col(FlowRun.source) != RUN_SOURCE,
                 col(FlowRun.started_at) < now - timedelta(seconds=RUN_STALE_S),
             )
             .values(status="abandoned", finished_at=now)