Close the nine open SDK tasks: one engine per directory, a tabbed dashboard, re-pairing, run recovery

serve: refuse a second engine for one data directory whatever port it was
asked for, using the pidfile and a token this directory signed. The check
runs before the database is touched and before the credential is written,
which is what left every later CLI call pointing at a dead port.

The terminal dashboard is three tabs (Overview, Runs, Logs) with the toolbar
following the focused pane, the engine's output goes to serve.log rather than
down a pipe, and closing the screen stops both reader threads so the prompt
comes back. It adopts a running engine on every start, so stop/start and
restart work on one it did not start, and a stop waits for the process to be
gone before the next start. Enrolment reports itself in the modal.

enroll: a new claim code replaces the pairing instead of being refused. The
code is redeemed before anything is written, mappings to a portal being left
are cleared, and a running engine redials when the stored enrolment changes.

runs: an engine re-queues the runs left `queued` by the one before it, and
`fluksio retry <id>` / `retry --group <sweep>` submits an interrupted run
again with the same inputs and group, recorded through Run.parent_id.

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

backend/tests/api/routes/test_runs.py

@@ -492,6 +492,107 @@ def test_the_seed_is_recorded_the_same_way_however_it_arrived():
             session.commit()
 
 
+def test_a_retry_is_a_new_run_that_names_the_one_it_repeats():
+    """The way back from a run an engine restart interrupted.
+
+    Re-issuing the whole sweep is the blunt version; this keeps the group, so
+    what was missing is filled in rather than run twice.
+    """
+    flow = FlowDef(
+        name="study",
+        mode="batch",
+        inputs=[
+            FlowInput(spec=MessageSpec(name="lr", dtype=DType.FLOAT), initial=0.01)
+        ],
+    )
+    queue = _Collect()
+    service = RunService(controller=_OneFlow(flow), queue=queue)
+    made = []
+    with Session(db_engine) as session:
+        session.add(
+            Run(
+                id="abandoned-1",
+                flow="study",
+                status="abandoned",
+                params={"lr": 0.3},
+                group_id="sweep-9",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.add(
+            Run(
+                id="going-1",
+                flow="study",
+                status="running",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+    try:
+        again = service.retry("abandoned-1", actor="someone@example.com")
+        made.append(again.id)
+        assert again.parent_id == "abandoned-1"
+        assert (again.params, again.group_id, again.cause) == (
+            {"lr": 0.3},
+            "sweep-9",
+            "retry",
+        )
+        assert queue.items[-1].run_id == again.id
+
+        # A run that has not finished is cancelled, not retried.
+        with pytest.raises(RunRejected):
+            service.retry("going-1")
+        with pytest.raises(RunRejected):
+            service.retry("never-existed")
+    finally:
+        with Session(db_engine) as session:
+            for run in session.exec(
+                select(Run).where(col(Run.id).in_([*made, "abandoned-1", "going-1"]))
+            ).all():
+                session.delete(run)
+            session.commit()
+
+
+def test_runs_left_queued_are_woken_by_the_next_engine():
+    """The row is the record, and the work item is not part of its transaction.
+
+    An in-memory queue loses the item with the process and a stream item
+    nobody claimed is nobody's, so without this a run submitted just before a
+    restart waits for an engine that will never be told about it.
+    """
+    queue = _Collect()
+    service = RunService(controller=_Unusable(), queue=queue)
+    with Session(db_engine) as session:
+        session.add(
+            Run(
+                id="orphan-1",
+                flow="study",
+                status="queued",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.add(
+            Run(
+                id="finished-1",
+                flow="study",
+                status="ok",
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+    try:
+        service._requeue_queued()
+        assert [item.run_id for item in queue.items] == ["orphan-1"]
+        assert queue.items[0].flow == "study"
+    finally:
+        with Session(db_engine) as session:
+            for run in session.exec(
+                select(Run).where(col(Run.id).in_(["orphan-1", "finished-1"]))
+            ).all():
+                session.delete(run)
+            session.commit()
+
+
 def test_a_key_nobody_used_submits_normally(
     client, superuser_token_headers, monkeypatch
 ):

backend/tests/test_cli.py

@@ -4,6 +4,8 @@ import subprocess
 import sys
 from pathlib import Path
 
+import pytest
+
 from fluksio.cli import load_or_create_secret_key
 
 
@@ -689,6 +691,41 @@ def test_the_dashboard_runs_the_engine_as_a_child_of_itself(monkeypatch) -> None
     assert opened == []
 
 
+@pytest.mark.anyio
+async def test_the_dashboard_is_three_tabs_and_lets_go_of_its_threads(
+    tmp_path, monkeypatch
+) -> None:
+    """The tabs, and what `q` has to do before the screen goes.
+
+    Its two readers are worker threads on the event loop's own executor, which
+    is joined while the loop closes — so a reader still blocked on the log file
+    or the websocket holds the terminal after the dashboard has gone, which is
+    what `q` used to do.
+    """
+    from textual.widgets import TabbedContent
+
+    from fluksio import cli
+    from fluksio.tui.app import ServeApp
+
+    # Nothing is started under this screen: the engine has its own tests.
+    monkeypatch.setattr(ServeApp, "start_engine", lambda self: None)
+    args = cli._parser().parse_args(["serve", "--data-dir", str(tmp_path / ".fluksio")])
+    app = ServeApp(args)
+    async with app.run_test() as pilot:
+        tabs = app.query_one(TabbedContent)
+        assert tabs.active == "overview-tab"
+        await pilot.press("2")
+        assert tabs.active == "runs-tab"
+        # The run keys live on the table, so they are in the footer only here.
+        assert "app.pick" in {binding[1] for binding in app.query_one("#runs").BINDINGS}
+        await pilot.press("3")
+        assert tabs.active == "logs-tab"
+        await pilot.press("q")
+
+    assert app.stop_log.is_set()
+    assert app.stop_stream.is_set()
+
+
 def test_a_serving_engine_records_itself_until_it_stops(tmp_path) -> None:
     """A pid nobody is running is the same as no pidfile at all."""
     import os
@@ -741,6 +778,41 @@ def test_who_holds_the_port_is_told_apart_by_the_token() -> None:
     assert probe_engine("http://127.0.0.1:1", "t") == "other"
 
 
+def test_a_second_engine_for_one_directory_is_refused(tmp_path, monkeypatch, capsys):
+    """One SQLite file, one engine — whatever port the second was asked for.
+
+    The refusal is worth more than the duplicate it prevents: the second
+    engine signs in on the way up, so `client.json` would point at a port
+    that dies with it and every later command would reach nothing.
+    """
+    import json
+    import os
+
+    from fluksio import cli
+
+    data_dir = tmp_path / ".fluksio"
+    data_dir.mkdir()
+    (data_dir / "client.json").write_text(
+        json.dumps({"url": "http://127.0.0.1:8000", "token": "the-first-one"})
+    )
+    cli.write_pidfile(data_dir, 8000)
+    monkeypatch.setattr(cli, "probe_engine", lambda *a, **k: "ours")
+
+    args = cli._parser().parse_args(
+        ["serve", "--plain", "--port", "8123", "--data-dir", str(data_dir)]
+    )
+    assert cli.cmd_serve(args) == 0
+    said = capsys.readouterr().out
+    assert "already serving" in said and str(os.getpid()) in said
+    # The credential still names the engine that is actually up.
+    stored = json.loads((data_dir / "client.json").read_text())
+    assert stored["url"] == "http://127.0.0.1:8000"
+
+    # A pid that is alive but is not an engine of ours is not a refusal.
+    monkeypatch.setattr(cli, "probe_engine", lambda *a, **k: "foreign")
+    assert cli.already_serving(data_dir, "127.0.0.1") == ""
+
+
 def test_serve_moves_off_a_port_that_is_taken() -> None:
     """A first start should not die on somebody else's dev server."""
     import socket

backend/tests/test_cloud.py

@@ -25,7 +25,8 @@ from fluksio.cloud import config as cloud_config
 from fluksio.core.config import settings
 from fluksio.flow.panels import PanelDef, PanelsConfig, write_config
 from fluksio.models import User
-from tests.utils.portal import ISSUER, portal_token
+from tests.utils.portal import ISSUER, jwks, portal_token
+from tests.utils.user import create_random_user
 
 
 def test_portal_token_is_refused_when_not_enrolled(
@@ -101,6 +102,85 @@ def test_status_reports_not_enrolled(
     assert response.json()["enrolled"] is False
 
 
+def test_enrolling_again_replaces_the_connection(
+    client: TestClient,
+    enrolled: User,
+    portal_key: rsa.RSAPrivateKey,
+    superuser_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    """A new claim code re-pairs rather than being refused.
+
+    Deleting cloud.json by hand used to be the only way through, and it takes
+    every remote connection with it. The mappings of the portal being left go
+    too: nothing in the row says which portal issued the subject, so one kept
+    from the old one would resolve a stranger onto a local account.
+    """
+    stale = create_random_user(db)
+    stale.portal_sub = "portal-user-7"
+    db.add(stale)
+    db.commit()
+
+    elsewhere = "https://other.example.test"
+    reply = Mock(
+        status_code=200,
+        json=Mock(
+            return_value={
+                "ws_url": f"{elsewhere}/api/v1/tunnel/attach",
+                "instance_id": "11111111-2222-3333-4444-555555555555",
+                "instance_token": "the-new-token",
+                "issuer": elsewhere,
+                "jwks": jwks(portal_key),
+                "owner_id": "portal-user-9",
+            }
+        ),
+    )
+    with patch("fluksio.cloud.enroll.httpx.post", return_value=reply):
+        again = client.post(
+            f"{settings.API_V1_STR}/cloud/enroll",
+            headers=superuser_token_headers,
+            json={"portal_url": elsewhere, "claim_code": "ABCD-EFGH"},
+        )
+    assert again.status_code == 200, again.text
+
+    config = cloud_config.load()
+    assert config is not None
+    assert config.portal_url == elsewhere
+    assert config.token == "the-new-token"
+
+    db.expire_all()
+    assert db.get(User, enrolled.id).portal_sub == "portal-user-9"
+    # The other portal's mapping is not carried over to this one.
+    assert db.get(User, stale.id).portal_sub is None
+
+
+def test_a_claim_the_portal_refuses_keeps_the_connection(
+    client: TestClient,
+    enrolled: User,
+    superuser_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    """Nothing is written until the portal has accepted the code.
+
+    That ordering is what makes replacing safe: a mistyped code leaves the
+    instance connected to the portal it was connected to.
+    """
+    before = cloud_config.load()
+    with patch("fluksio.cloud.enroll.httpx.post", return_value=Mock(status_code=404)):
+        refused = client.post(
+            f"{settings.API_V1_STR}/cloud/enroll",
+            headers=superuser_token_headers,
+            json={"portal_url": ISSUER, "claim_code": "NOPE-NOPE"},
+        )
+    assert refused.status_code == 400, refused.text
+
+    after = cloud_config.load()
+    assert before is not None and after is not None
+    assert (before.instance_id, before.token) == (after.instance_id, after.token)
+    db.expire_all()
+    assert db.get(User, enrolled.id).portal_sub == "portal-user-1"
+
+
 def test_enrolling_needs_a_superuser(
     client: TestClient, normal_user_token_headers: dict[str, str]
 ) -> None:
@@ -388,6 +468,39 @@ async def test_a_config_that_appears_while_running_is_dialled(monkeypatch) -> No
     assert started
 
 
+@pytest.mark.anyio
+async def test_an_enrolment_replaced_while_running_is_redialled(monkeypatch) -> None:
+    """`fluksio enroll` is its own process and cannot cancel the live link.
+
+    Without this the tunnel would stay up on the credential that was replaced,
+    since the watcher only ever started a link where there was none.
+    """
+    from types import SimpleNamespace
+
+    from fluksio.cloud import connector
+
+    live = asyncio.create_task(asyncio.sleep(30))
+    app = Mock()
+    app.state = Mock(
+        cloud_task=live, cloud_connector=Mock(), cloud_identity=("old", "old-token")
+    )
+    started: list[object] = []
+    monkeypatch.setattr(connector, "ENROL_POLL_S", 0.01)
+    monkeypatch.setattr(connector, "start", lambda one: started.append(one))
+    monkeypatch.setattr(
+        cloud_config,
+        "load",
+        lambda: SimpleNamespace(instance_id="new", token="new-token"),
+    )
+
+    watcher = asyncio.create_task(connector.watch_enrolment(app))
+    await asyncio.sleep(0.05)
+    watcher.cancel()
+
+    assert live.cancelled() or live.done()
+    assert started
+
+
 @pytest.mark.anyio
 async def test_a_config_that_cannot_be_read_is_not_dialled(monkeypatch) -> None:
     """Otherwise the connector gives up at once and this restarts it forever."""