Let a sweep run more than four at a time, and name the run a failure was in

Concurrent runs sat at 4 whatever FLOW_MAX_CASCADES said: that setting bounds
cascades, and the run drivers read a hardcoded MAX_PARALLEL nobody could reach.
FLOW_MAX_RUNS is the knob they read now, --max-runs/--max-cascades/--max-workers
are the same three as flags on serve, and the engine says which numbers it
started with — which is the only way to tell that a settings file was read.

Events keep the run they happened in. The payload always carried it and the
persist path dropped it, so reading one run's failures meant filtering the
engine-wide list; a batch run's id reaches those events now too, since a run
has no journaled item to name itself by.

Also: a provisioner's 0 means "no deadline" rather than "cancel on the next
reconcile", and a command that reaches no engine says how to start one.

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

backend/tests/api/routes/test_observability.py

@@ -312,3 +312,30 @@ def test_events_narrow_to_one_minute(
 
     # The upper bound is exclusive, so the failure a minute later is not in it.
     assert [event["detail"] for event in events] == ["minute-in"]
+
+
+def test_events_narrow_to_one_run(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    """Reading one run's failures without filtering the engine-wide list."""
+    ts = datetime.now(UTC) - timedelta(hours=6)
+    db.add(
+        EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="mine", run="run-mine")
+    )
+    db.add(
+        EngineEvent(
+            ts=ts, type="node_error", flow=FLOW, detail="theirs", run="run-theirs"
+        )
+    )
+    # What a live cascade unrelated to any run leaves, and what every row
+    # written before the column existed looks like.
+    db.add(EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="neither"))
+    db.commit()
+
+    events = client.get(
+        f"{PREFIX}/events",
+        headers=superuser_token_headers,
+        params={"flow": FLOW, "run": "run-mine"},
+    ).json()
+
+    assert [event["detail"] for event in events] == ["mine"]

backend/tests/api/routes/test_runs.py

@@ -598,3 +598,47 @@ def test_a_step_the_x_metric_never_reached_is_left_out(
     ).json()
 
     assert [point[0] for point in answer["lines"][0]["points"]] == [10.0, 20.0, 30.0]
+
+
+def test_a_run_records_the_code_it_started_with_not_the_code_it_was_queued_with():
+    """A sweep queues every run at once and the tree moves while it waits.
+
+    `_restamp` is what runs at claim time, so the digest the record keeps is
+    the one the run actually executed.
+    """
+    service = RunService(controller=_Unusable(), queue=_Unusable())
+    with Session(db_engine) as session:
+        session.add(
+            Run(
+                id="stamp-1",
+                flow="study",
+                status="queued",
+                code_digest="at-submit",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+        run = session.get(Run, "stamp-1")
+
+    assert service._restamp(run, "at-claim") == "at-claim"
+
+    with Session(db_engine) as session:
+        assert session.get(Run, "stamp-1").code_digest == "at-claim"
+
+
+def test_a_tree_that_did_not_move_is_not_written_again():
+    service = RunService(controller=_Unusable(), queue=_Unusable())
+    with Session(db_engine) as session:
+        session.add(
+            Run(
+                id="stamp-2",
+                flow="study",
+                status="queued",
+                code_digest="same",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+        run = session.get(Run, "stamp-2")
+
+    assert service._restamp(run, "same") == "same"

backend/tests/flow/test_provision.py

@@ -8,6 +8,7 @@ it should, and that a job nobody ever attached does not sit in the queue.
 
 import subprocess
 import threading
+import time
 
 from fluksio.flow.placement import Placer
 from fluksio.flow.provision import SlurmProfile, SlurmProvisioner, load_provisioners
@@ -116,6 +117,22 @@ def test_a_machine_that_arrives_clears_the_way_for_the_next_ask(monkeypatch):
 
 
 def test_a_job_that_never_attaches_is_cancelled(monkeypatch):
+    ssh = FakeSsh()
+    monkeypatch.setattr(subprocess, "run", ssh)
+    hpc = cluster(provision_timeout_s=0.01)
+
+    hpc.provision(cpus=2, gpus=0, ram_mb=0)
+    assert ssh.ran.wait(5)
+    # Slept rather than set to zero: zero is what says "no deadline" now.
+    time.sleep(0.02)
+    hpc.reconcile(set())
+
+    assert hpc.status()["outstanding"] == []
+    assert ["scancel", "4711"] == ssh.calls[-1][0][-2:]
+
+
+def test_no_deadline_waits_for_as_long_as_the_queue_does(monkeypatch):
+    """0 is "never give up", not "give up now": a cluster can queue for days."""
     ssh = FakeSsh()
     monkeypatch.setattr(subprocess, "run", ssh)
     hpc = cluster(provision_timeout_s=0)
@@ -124,8 +141,8 @@ def test_a_job_that_never_attaches_is_cancelled(monkeypatch):
     assert ssh.ran.wait(5)
     hpc.reconcile(set())
 
-    assert hpc.status()["outstanding"] == []
-    assert ["scancel", "4711"] == ssh.calls[-1][0][-2:]
+    assert hpc.status()["outstanding"][0]["job"] == "4711"
+    assert [argv for argv, _ in ssh.calls if "scancel" in argv] == []
 
 
 def test_a_refused_submission_says_what_the_cluster_said(monkeypatch):

backend/tests/test_metrics.py

@@ -224,3 +224,27 @@ def test_a_traceback_no_failure_ever_claims_is_dropped() -> None:
     asyncio.run(collector.flush())
 
     assert collector._tracebacks == {}
+
+
+def test_a_failure_keeps_the_run_it_happened_in(db: Session) -> None:
+    """The payload always carried it; the row used to drop it.
+
+    A batch run publishes no `cascade_started`, so there is no `FlowRun` beside
+    this — which is exactly the case that had no way of being asked about.
+    """
+    collector = MetricsCollector(EventBus())
+    ts = datetime.now(UTC).replace(second=0, microsecond=0).timestamp()
+    collector.handle(
+        {
+            "type": "node_error",
+            "flow": FLOW,
+            "node": NODE,
+            "error": "ValueError: in a run",
+            "run": "run-abc123",
+            "ts": ts,
+        }
+    )
+    asyncio.run(collector.flush())
+
+    failure = db.exec(select(EngineEvent).where(EngineEvent.run == "run-abc123")).one()
+    assert failure.detail.startswith("ValueError: in a run")