Style the engine's own logs, notice enrolment while serving, say more in status

Four things from a testing pass.

`fluksio serve` printed its own lines through the root logger, which has no
handler and falls back to `INFO:fluksio.cloud.connector:...` — beside uvicorn's
aligned output it reads like something went wrong. The engine's loggers and
alembic's now use uvicorn's own handler. Named rather than configuring the
root: httpx logs every portal call at INFO and none of that is printed today.

`fluksio enroll` writes its config from another process, so an engine already
serving never learned it had been paired. It now looks for one every few
seconds and dials when it appears. `load()` rather than `exists()`, or a file
that does not parse would be restarted forever.

`fluksio status` says where the installation stands with its portal — never
paired, linked, or paired and unreachable, which is the one worth acting on.

`--seed` and `--timeout` had no help text at all. Both say what they are for
now, and the docs say what a seed is actually for: recorded on the run, part of
its input digest, and passed to an input named `seed` when the flow declares
one, so the number a run is labelled with is the one the code drew from.

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

backend/fluksio/api/routes/cloud.py

@@ -12,7 +12,6 @@ chain stops with whoever a superuser here typed a code for.
 
 from __future__ import annotations
 
-import asyncio
 import logging
 import secrets
 from typing import Any
@@ -224,13 +223,6 @@ def disconnect(request: Request) -> Message:
 
 
 def _start_connector(app: Any) -> None:
-    from fluksio.cloud.connector import CloudConnector
+    from fluksio.cloud.connector import start
 
-    existing = getattr(app.state, "cloud_task", None)
-    if existing is not None:
-        existing.cancel()
-    connector = CloudConnector(app)
-    app.state.cloud_connector = connector
-    app.state.cloud_task = asyncio.create_task(
-        connector.serve_forever(), name="cloud-connector"
-    )
+    start(app)

backend/fluksio/cli.py

@@ -16,6 +16,7 @@ before any of that happens, which is what `_configure_environment` is for.
 from __future__ import annotations
 
 import argparse
+import copy
 import os
 import secrets
 import sys
@@ -285,16 +286,48 @@ def cmd_serve(args: argparse.Namespace) -> int:
 
     # One process: it holds the flow engine, and a second worker would be a
     # second engine — duplicated subscriptions, cron ticks and webhooks.
-    uvicorn.run(app, host=args.host, port=args.port, log_level=args.log_level)
+    uvicorn.run(
+        app,
+        host=args.host,
+        port=args.port,
+        log_level=args.log_level,
+        log_config=_log_config(args.log_level),
+    )
     return 0
 
 
+def _log_config(level: str) -> dict[str, Any]:
+    """Uvicorn's logging, with the engine's own loggers drawn the same way.
+
+    Without this the engine's lines go to the root logger, which has no handler
+    configured and falls back to `INFO:fluksio.cloud.connector:...` — beside
+    uvicorn's own aligned, coloured output it reads like something went wrong.
+    """
+    from uvicorn.config import LOGGING_CONFIG
+
+    config = copy.deepcopy(LOGGING_CONFIG)
+    # Named rather than configuring the root: everything else that logs — httpx
+    # on every portal call, for one — is at INFO too, and today none of it is
+    # printed at all. Styling the root would turn all of it on.
+    for name in ("fluksio", "alembic"):
+        config["loggers"][name] = {
+            "handlers": ["default"],
+            "level": level.upper(),
+            # It has a handler of its own now; propagating would print each
+            # line twice the moment anything configures the root.
+            "propagate": False,
+        }
+    return config
+
+
 def cmd_enroll(args: argparse.Namespace) -> int:
     data_dir = _data_dir(args.data_dir, args.shared)
     _prepare(data_dir)
     result = _enroll(args.portal, args.code, args.as_email)
     if result == 0:
-        _say("Start it with `fluksio serve`; it dials the portal as it comes up.")
+        # True either way round: one already serving notices within seconds,
+        # and one not yet started dials as it comes up.
+        _say("An engine already running picks this up; otherwise `fluksio serve`.")
     return result
 
 

backend/fluksio/cloud/connector.py

@@ -50,6 +50,49 @@ MAX_WS_MESSAGE = 1024 * 1024
 #: Only the versioned API is served over the tunnel. The MCP mount and the
 #: OAuth endpoints live outside it and stay local-only.
 ALLOWED_PREFIX = "/api/v1/"
+#: How often to look for a config that appeared while the engine was up.
+#: Enrolment is something a person does and then waits on, so this is the
+#: delay they sit through — short enough not to be worth a restart.
+ENROL_POLL_S = 3.0
+
+
+def start(app: FastAPI) -> None:
+    """Dial the portal, replacing any link already up."""
+    existing = getattr(app.state, "cloud_task", None)
+    if existing is not None:
+        existing.cancel()
+    connector = CloudConnector(app)
+    app.state.cloud_connector = connector
+    app.state.cloud_task = asyncio.create_task(
+        connector.serve_forever(), name="cloud-connector"
+    )
+
+
+async def watch_enrolment(app: FastAPI) -> None:
+    """Notice an enrolment that happened outside this process.
+
+    ``fluksio enroll`` writes the config against the database directly, with no
+    idea whether an engine is running — so without this, pairing an installation
+    that is already serving would take a restart to come into effect. Enrolling
+    through the API starts the link itself and this sees a task already there.
+    """
+    while True:
+        await asyncio.sleep(ENROL_POLL_S)
+        task = getattr(app.state, "cloud_task", None)
+        if task is not None and task.done():
+            # It returns of its own accord when the config goes away, which is
+            # what `fluksio disconnect` and the portal's own Disconnect do.
+            app.state.cloud_task = None
+            app.state.cloud_connector = None
+            task = None
+        # `load()`, not `exists()`: a file that cannot be read is not an
+        # enrolment, and the connector would give up on it the instant it
+        # started — which, started from here, is a restart every few seconds.
+        # A config that is fine but unreachable keeps its task, and the
+        # retrying belongs to the connector rather than to this.
+        if task is None and cloud_config.load() is not None:
+            logger.info("Enrolled while running; dialling the portal")
+            start(app)
 
 
 class CloudConnector:

backend/fluksio/main.py

@@ -174,15 +174,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     cloud_task: asyncio.Task[None] | None = None
     app.state.cloud_connector = None
     app.state.cloud_task = None
-    if cloud_config.exists():
-        from fluksio.cloud.connector import CloudConnector
+    from fluksio.cloud import connector as cloud_connector
 
-        connector = CloudConnector(app)
-        app.state.cloud_connector = connector
-        cloud_task = asyncio.create_task(
-            connector.serve_forever(), name="cloud-connector"
-        )
-        app.state.cloud_task = cloud_task
+    if cloud_config.exists():
+        cloud_connector.start(app)
+        cloud_task = app.state.cloud_task
+    # Watched whether or not one exists now: enrolling from the CLI writes the
+    # config from another process entirely, and an engine already serving
+    # should pick it up rather than need restarting.
+    enrol_task = asyncio.create_task(
+        cloud_connector.watch_enrolment(app), name="cloud-enrolment-watch"
+    )
     try:
         # A mounted sub-app gets no lifespan of its own, so the MCP session
         # manager is entered here; without it every /mcp request fails.
@@ -192,6 +194,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         watchdog_task.cancel()
         alerts_task.cancel()
         metrics_task.cancel()
+        enrol_task.cancel()
         # Re-read from app.state: enrolling at runtime replaces this.
         running_cloud = getattr(app.state, "cloud_task", None) or cloud_task
         if running_cloud is not None:

backend/fluksio/sdk/cli.py

@@ -505,6 +505,22 @@ def _flow_state(flow: dict[str, Any]) -> str:
     return "running"
 
 
+def _portal_phrase(portal: dict[str, Any]) -> tuple[str, str]:
+    """Where this installation stands with its portal, and how to colour it.
+
+    Three states worth telling apart: never paired, paired and linked, and
+    paired but not reaching it — the last being the one somebody needs to know
+    about, since the dashboard is served from the other end.
+    """
+    if not portal.get("enrolled"):
+        return "no portal", "dim"
+    if portal.get("connected"):
+        host = str(portal.get("portal_url") or "").split("//")[-1].rstrip("/")
+        return f"portal {host}", "green"
+    trouble = str(portal.get("last_error") or "").strip()
+    return "portal unreachable" + (f" ({trouble[:60]})" if trouble else ""), "red"
+
+
 def _status_screen(client: Client) -> Any:
     """One frame: health, the flows, and the failures under them."""
     from rich.console import Group
@@ -517,6 +533,12 @@ def _status_screen(client: Client) -> Any:
     # fails as an engine event, while a batch run fails on its own row.
     failures = client.events(kind="failure", limit=5)
     runs = client.runs(limit=5)
+    try:
+        portal = client.cloud_status()
+    except (SyncError, ApiError):
+        # Never enrolled, or an engine too old to answer. Neither is worth
+        # failing a status screen over.
+        portal = {}
 
     healthy = summary.get("status") == "ok"
     head = Text()
@@ -527,6 +549,8 @@ def _status_screen(client: Client) -> Any:
     problems = summary.get("problems") or []
     if problems:
         head.append("  " + " · ".join(str(p) for p in problems), style="yellow")
+    phrase, style = _portal_phrase(portal)
+    head.append("   " + phrase, style=style)
 
     counts = summary.get("flows") or {}
     nodes = summary.get("nodes") or {}
@@ -758,14 +782,32 @@ def add_parsers(subparsers: Any) -> None:
         "run", help="sync this directory, then start a run of one of its flows"
     )
     parser.add_argument("flow")
-    parser.add_argument("--seed", type=int, default=None)
+    parser.add_argument(
+        "--seed",
+        type=int,
+        default=None,
+        help=(
+            "the run's seed: recorded on it, part of what tells two runs of one "
+            "configuration apart, and passed to an input named 'seed' when the "
+            "flow declares one"
+        ),
+    )
     parser.add_argument("--wait", action="store_true", help="block until it finishes")
     parser.add_argument(
         "--follow",
         action="store_true",
         help="wait, printing the numbers it reports as they arrive",
     )
-    parser.add_argument("--timeout", type=float, default=0.0)
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=0.0,
+        metavar="SECONDS",
+        help=(
+            "give up waiting after this long and leave the run going; 0, the "
+            "default, waits as long as it takes. Not the node timeout"
+        ),
+    )
     parser.add_argument(
         "--no-sync",
         action="store_true",
@@ -814,10 +856,29 @@ def add_parsers(subparsers: Any) -> None:
         metavar="NAME=V1,V2",
         help="an input and the values to try; repeat for a grid",
     )
-    parser.add_argument("--seed", type=int, default=None)
+    parser.add_argument(
+        "--seed",
+        type=int,
+        default=None,
+        help="the seed every run in the sweep gets; vary it with --param seed=1,2",
+    )
     parser.add_argument("--wait", action="store_true", help="block until all finish")
-    parser.add_argument("--timeout", type=float, default=0.0)
-    parser.add_argument("--no-sync", action="store_true")
-    parser.add_argument("--no-cache", action="store_true")
+    parser.add_argument(
+        "--timeout",
+        type=float,
+        default=0.0,
+        metavar="SECONDS",
+        help="give up waiting after this long; 0, the default, waits them out",
+    )
+    parser.add_argument(
+        "--no-sync",
+        action="store_true",
+        help="run what is already on the engine, without uploading first",
+    )
+    parser.add_argument(
+        "--no-cache",
+        action="store_true",
+        help="execute every node, even one an earlier run already answered",
+    )
     with_engine(parser, local=True)
     parser.set_defaults(func=cmd_sweep)

backend/fluksio/sdk/client.py

@@ -206,6 +206,11 @@ class Client:
         result: dict[str, Any] = self._call("GET", "/observability/summary")
         return result
 
+    def cloud_status(self) -> dict[str, Any]:
+        """Whether this installation is enrolled with a portal, and linked."""
+        result: dict[str, Any] = self._call("GET", "/cloud/status")
+        return result
+
     def events(self, kind: str = "failure", limit: int = 10) -> list[dict[str, Any]]:
         """What went wrong, or who changed what. Newest first."""
         result = self._call(