Find a study wherever it is, and start on a port that is free

Three things the one-folder-per-study layout ran into.

**Discovery walks down.** A plain directory is now walked all the way, so
`fluksio sync dev` finds `dev/s1_baseline/study.py` and naming each study is
no longer the price of the layout. Hidden directories, `__pycache__`,
`node_modules` and virtualenvs are left alone, and a package is taken whole.

Two files that would import under one module name are refused, naming both:
Python keeps one module per name, so the second would silently *be* the first
— and a node's generated body imports by that name, so a worker would run the
wrong study's code. The message says the fix, which is an `__init__.py` per
study directory. A module that raises while importing is now a sentence
naming the file rather than an importlib traceback.

**`run` and `sweep` sync downwards too**, so the flow is found from the
repository root without the sync-then-`--no-sync` two-step. A study that will
not import is a warning rather than a stopped run, since a walk meets every
study and a half-finished one two directories away is not this run's problem.
The upload was already a no-op for a flow nothing changed in, so what the walk
costs is import time — `--sync PATH` narrows it, and skipping unchanged
subtrees would need a cache keyed on file state that is deliberately not here.

**`serve` moves off a busy default port** — 8001, 8002, up to twenty — says
which it took, and writes that one into `client.json`. A port given with
`--port` still fails when it is taken, because naming one is asking for it.

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

backend/fluksio/cli.py

@@ -19,6 +19,7 @@ import argparse
 import copy
 import os
 import secrets
+import socket
 import sys
 from pathlib import Path
 from typing import Any
@@ -230,6 +231,32 @@ CONCURRENCY_FLAGS = {
 }
 
 
+#: What `serve` listens on when nobody says. Taken often enough — another
+#: engine, another framework's dev server — that dying on it is the first
+#: thing a zero-config start would hit.
+DEFAULT_PORT = 8000
+
+#: How far up from it to look before giving up and letting the bind fail.
+PORT_TRIES = 20
+
+
+def _free_port(host: str, start: int) -> int:
+    """The first port from ``start`` that nothing is listening on.
+
+    Probed with the same address and options uvicorn will bind with, so this
+    answers the question uvicorn is about to ask rather than a similar one.
+    """
+    for port in range(start, start + PORT_TRIES):
+        with socket.socket() as probe:
+            probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            try:
+                probe.bind((host, port))
+            except OSError:
+                continue
+            return port
+    return start
+
+
 def cmd_serve(args: argparse.Namespace) -> int:
     data_dir = _data_dir(args.data_dir, args.shared)
     for flag, name in CONCURRENCY_FLAGS.items():
@@ -267,10 +294,16 @@ def cmd_serve(args: argparse.Namespace) -> int:
     from fluksio.flow import modules
     from fluksio.main import app
 
+    port = args.port
+    if port is None:
+        port = _free_port(args.host, DEFAULT_PORT)
+        if port != DEFAULT_PORT:
+            _say(f"Port {DEFAULT_PORT} is in use; serving on {port} instead.")
+
     # The client talks to this engine, and 0.0.0.0 is not an address to talk
     # to — it is a statement about which interfaces to listen on.
     reachable = "127.0.0.1" if args.host in ("0.0.0.0", "::", "") else args.host
-    url = f"http://{reachable}:{args.port}"
+    url = f"http://{reachable}:{port}"
     token_path = _sign_in(admin_id, url, data_dir)
 
     config = cloud_config.load()
@@ -303,7 +336,7 @@ def cmd_serve(args: argparse.Namespace) -> int:
     uvicorn.run(
         app,
         host=args.host,
-        port=args.port,
+        port=port,
         log_level=args.log_level,
         log_config=_log_config(args.log_level),
     )
@@ -374,7 +407,14 @@ def _parser() -> argparse.ArgumentParser:
     serve = subparsers.add_parser("serve", help="run the engine")
     with_data_dir(serve)
     serve.add_argument("--host", default="127.0.0.1")
-    serve.add_argument("--port", type=int, default=8000)
+    # No default: a port nobody asked for may move when it is taken, and one
+    # that was asked for may not.
+    serve.add_argument(
+        "--port",
+        type=int,
+        default=None,
+        help=f"default {DEFAULT_PORT}, or the next free port when it is in use",
+    )
     serve.add_argument("--log-level", default="info")
     serve.add_argument("--admin-email", default=None)
     serve.add_argument("--admin-password", default=None)

backend/fluksio/sdk/cli.py

@@ -115,19 +115,81 @@ def _module_of(path: Path) -> tuple[str, str]:
     return str(directory), ".".join(reversed(parts))
 
 
-def _import(root: str, dotted: str) -> None:
+#: Directories a walk never goes into: none of them is a study, and some of
+#: them are enormous.
+_SKIP_DIRS = frozenset({"__pycache__", "node_modules", "site-packages"})
+
+
+def _import(root: str, dotted: str, expect: Path | None = None) -> None:
     if root not in sys.path:
         sys.path.insert(0, root)
-    importlib.import_module(dotted)
+    module = importlib.import_module(dotted)
+    if expect is None:
+        return
+    # Python keeps one module per name, so a second file importing under a
+    # name already taken is silently the first one — and the generated body
+    # imports by that name too, so a worker would run the wrong study's code.
+    actual = getattr(module, "__file__", "") or ""
+    if actual and Path(actual).resolve() != expect.resolve():
+        raise SyncError(
+            f"two files would both import as '{dotted}':\n"
+            f"  {actual}\n  {expect}\n"
+            "Python keeps one module per name, and a node's generated body "
+            "imports by that name, so the second would run the first's code. "
+            "Put an `__init__.py` in each directory — they become "
+            f"'<dir>.{dotted}' and stop colliding — or rename one of the files."
+        )
 
 
-def discover(targets: list[str]) -> list[Flow]:
+def _walkable(entry: Path) -> bool:
+    """Whether a walk should look inside this directory at all."""
+    return (
+        entry.is_dir()
+        and not entry.name.startswith(".")
+        and entry.name not in _SKIP_DIRS
+        and not (entry / "pyvenv.cfg").exists()
+    )
+
+
+def _below(root: Path) -> Iterator[Path]:
+    """Every module and package under a plain directory, however deep.
+
+    One directory per study — `dev/s1_baseline/study.py` — is a layout people
+    have, and naming each of them on the command line is bookkeeping the tool
+    can do. A package is yielded whole and not descended into: its own walk
+    imports its submodules under the right names.
+    """
+    for entry in sorted(root.iterdir()):
+        if entry.is_file() and entry.suffix == ".py":
+            yield entry
+        elif _walkable(entry):
+            if (entry / "__init__.py").exists():
+                yield entry
+            else:
+                yield from _below(entry)
+
+
+def _import_package(directory: Path) -> None:
+    """A package and every module in it, by their dotted names."""
+    root, dotted = _package_of(directory)
+    _import(root, dotted, directory / "__init__.py")
+    for info in pkgutil.walk_packages(sys.modules[dotted].__path__, f"{dotted}."):
+        importlib.import_module(info.name)
+
+
+def discover(targets: list[str], keep_going: bool = False) -> list[Flow]:
     """Import what was named and hand back the flows it declared.
 
     Imported by dotted name with its root on the path, never from a file
     location: the generated node bodies import the same way, and a module
     loaded under a different name would generate an import that does not
     resolve.
+
+    ``keep_going`` warns about a module that will not import instead of
+    stopping, which is what a *run* wants: a study half-way through an edit
+    two directories away is not a reason to refuse to run this one. A
+    collision between two module names is never skipped — it would produce a
+    node body that imports the wrong file.
     """
     for target in targets:
         path = Path(target)
@@ -136,30 +198,33 @@ def discover(targets: list[str]) -> list[Flow]:
             continue
         path = path.resolve()
         if path.is_file():
-            _import(*_module_of(path))
+            _import(*_module_of(path), path)
             continue
         if (path / "__init__.py").exists():
-            root, dotted = _package_of(path)
-            _import(root, dotted)
-            package = sys.modules[dotted]
-            for info in pkgutil.walk_packages(package.__path__, f"{dotted}."):
-                importlib.import_module(info.name)
+            _import_package(path)
             continue
-        # A plain directory — a repository root, usually. Its own modules,
-        # and the packages inside it: `myresearch/` beside a `README` is the
-        # ordinary shape, and naming it explicitly should not be the price of
-        # keeping your code in a package.
-        for module in sorted(path.glob("*.py")):
-            _import(*_module_of(module))
-        for child in sorted(path.iterdir()):
-            if child.name.startswith(".") or not (child / "__init__.py").exists():
-                continue
-            root, dotted = _package_of(child)
-            _import(root, dotted)
-            for info in pkgutil.walk_packages(
-                sys.modules[dotted].__path__, f"{dotted}."
-            ):
-                importlib.import_module(info.name)
+        # A plain directory — a repository root, or a directory of studies.
+        # Its own modules, the packages inside it, and the same again all the
+        # way down, so one folder per study needs no naming.
+        for entry in _below(path):
+            try:
+                if entry.is_dir():
+                    _import_package(entry)
+                else:
+                    _import(*_module_of(entry), entry)
+            except SyncError:
+                # A name collision is never somebody else's problem: it would
+                # put the wrong file behind a node.
+                raise
+            except Exception as exc:
+                # Importing a module runs it, so this is whatever the study
+                # does at its top level, and a walk meets every study.
+                if keep_going:
+                    _say(f"warning: skipped {entry} — {type(exc).__name__}: {exc}")
+                    continue
+                raise SyncError(
+                    f"{entry} failed to import — {type(exc).__name__}: {exc}"
+                ) from exc
     return list(FLOWS.values())
 
 
@@ -404,7 +469,7 @@ def _params(definition: dict[str, Any], rest: list[str]) -> dict[str, Any]:
     return params
 
 
-def _sync_first(client: Client) -> None:
+def _sync_first(client: Client, targets: list[str] | None = None) -> None:
     """Upload what the working directory declares, before running it.
 
     The reason a run exists is usually the edit that came before it, and
@@ -412,18 +477,24 @@ def _sync_first(client: Client) -> None:
     done. So `run` syncs by default — including the worker refresh, which is
     what makes an edit to your own package take effect at all.
 
+    The whole directory, downwards: from a repository root the flow is usually
+    in a study folder below, and the alternative is syncing by hand and then
+    running with `--no-sync`. The upload is already a no-op for a flow nothing
+    changed in, so the cost is importing the other studies — which `--sync`
+    narrows when that is not free.
+
     A directory that declares nothing is not an error: a flow drawn on the
     canvas is run the same way, and has nothing to upload.
     """
     try:
-        flows = discover(["."])
+        flows = discover(targets or ["."], keep_going=True)
     except (ImportError, SyncError) as exc:
         # Do not fail a run for a module the run may not even need.
         _say(f"warning: nothing synced — {exc}")
         return
     if not flows:
         return
-    repo = repo_root(".")
+    repo = repo_root((targets or ["."])[0])
     reports = sync(flows, client, origin=origin_of(repo))
     changed = [r for r in reports if not r.unchanged]
     if changed:
@@ -494,7 +565,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int:
     try:
         with _client_for(args) as client:
             if not args.no_sync:
-                _sync_first(client)
+                _sync_first(client, args.sync)
             stored = client.get_flow(args.flow)
             if stored is None:
                 return _fail(f"no flow '{args.flow}' on that engine")
@@ -884,7 +955,7 @@ def cmd_sweep(args: argparse.Namespace) -> int:
     try:
         with _client_for(args) as client:
             if not args.no_sync:
-                _sync_first(client)
+                _sync_first(client, args.sync)
             stored = client.get_flow(args.flow)
             if stored is None:
                 return _fail(f"no flow '{args.flow}' on that engine")
@@ -1157,6 +1228,13 @@ def add_parsers(subparsers: Any) -> None:
         action="store_true",
         help="run what is already on the engine, without uploading first",
     )
+    parser.add_argument(
+        "--sync",
+        action="append",
+        default=[],
+        metavar="PATH",
+        help="what to sync first (default: this directory, downwards)",
+    )
     parser.add_argument(
         "--no-cache",
         action="store_true",
@@ -1225,6 +1303,13 @@ def add_parsers(subparsers: Any) -> None:
         action="store_true",
         help="run what is already on the engine, without uploading first",
     )
+    parser.add_argument(
+        "--sync",
+        action="append",
+        default=[],
+        metavar="PATH",
+        help="what to sync first (default: this directory, downwards)",
+    )
     parser.add_argument(
         "--no-cache",
         action="store_true",