SDK: sync and run import the declaring module in a child process

The command line no longer imports a research module, so import-time GPU work
ends with a short-lived child instead of holding a CUDA context for the length
of a followed run. fluksio.sdk.declare starts the child, which writes each
flow's document and shims as JSON; Declared stands in for Flow in sync().
Errors keep their type and message, a child's traceback reaches the user.

Claude-Session: https://claude.ai/code/session_01SSi5AvZ7pFmySK8zy3NPMh

backend/fluksio/sdk/__init__.py

@@ -14,6 +14,7 @@ have to resolve inside a worker, where ``fluksio`` is the reporter module.
 from __future__ import annotations
 
 import ast
+import copy
 import inspect
 import json
 import re
@@ -27,6 +28,7 @@ from fluksio.sdk.imports import digest_of, reached
 
 __all__ = [
     "FLOWS",
+    "Declared",
     "Flow",
     "Port",
     "SyncError",
@@ -750,18 +752,27 @@ class Flow:
             ],
             "outputs": self.outputs,
         }
-        if origin is not None:
-            declared_in = _relative(self.declared_in, str(origin.get("repo") or ""))
-            doc["origin"] = origin | {
-                "declared_in": declared_in,
-                "module": self.module if declared_in else "",
-            }
-        return doc
+        return _stamped(doc, origin, self.declared_in, self.module)
 
     def shims(self) -> dict[str, str]:
         """One generated node body per node, keyed by node id."""
         return {spec.id: _shim(spec) for spec in self.nodes}
 
+    def declaration(self) -> dict[str, Any]:
+        """Everything `sync` takes of this flow, as JSON a process can hand on.
+
+        The document is worked out here, where the code was imported, and the
+        origin is stamped onto it later by :class:`Declared`: only the process
+        that runs `sync` knows which repository it is stamping with.
+        """
+        return {
+            "name": self.name,
+            "document": self.document(),
+            "shims": self.shims(),
+            "declared_in": self.declared_in,
+            "module": self.module,
+        }
+
     def submit(
         self, *, seed: int | None = None, client: Any = None, **params: Any
     ) -> Any:
@@ -783,6 +794,47 @@ class Flow:
         return f"Flow({self.name!r}, nodes={[n.id for n in self.nodes]})"
 
 
+def _stamped(
+    doc: dict[str, Any], origin: dict[str, Any] | None, declared_in: str, module: str
+) -> dict[str, Any]:
+    """Where the flow came from, added to its document."""
+    if origin is not None:
+        inside = _relative(declared_in, str(origin.get("repo") or ""))
+        doc["origin"] = origin | {
+            "declared_in": inside,
+            "module": module if inside else "",
+        }
+    return doc
+
+
+class Declared:
+    """A flow another process imported: what `sync` needs, without the code.
+
+    Stands where a :class:`Flow` does for :func:`fluksio.sdk.client.sync`, so
+    `fluksio sync` can read declarations in a child process and never import
+    the module itself. Built from :meth:`Flow.declaration`.
+    """
+
+    def __init__(self, data: dict[str, Any]) -> None:
+        self.name: str = data["name"]
+        self._data = data
+
+    def document(self, origin: dict[str, Any] | None = None) -> dict[str, Any]:
+        return _stamped(
+            copy.deepcopy(self._data["document"]),
+            origin,
+            self._data["declared_in"],
+            self._data["module"],
+        )
+
+    def shims(self) -> dict[str, str]:
+        shims: dict[str, str] = dict(self._data["shims"])
+        return shims
+
+    def __repr__(self) -> str:
+        return f"Declared({self.name!r})"
+
+
 def build(target: Flow, origin: dict[str, Any] | None = None) -> dict[str, Any]:
     """The flow document for a flow — checked, without a version."""
     return target.document(origin)

backend/fluksio/sdk/cli.py

@@ -42,6 +42,7 @@ from fluksio.sdk.client import (
     repo_root,
     sync,
 )
+from fluksio.sdk.declare import ImportFailed, declare, declare_module
 
 __all__ = ["add_parsers", "discover"]
 
@@ -199,6 +200,10 @@ def discover(targets: list[str], keep_going: bool = False) -> list[Flow]:
     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.
+
+    Imports in *this* process. The commands go through
+    :func:`fluksio.sdk.declare.declare`, which runs this in a child so the
+    command line never holds what the modules open.
     """
     for target in targets:
         path = Path(target)
@@ -336,9 +341,14 @@ def cmd_login(args: argparse.Namespace) -> int:
 def cmd_sync(args: argparse.Namespace) -> int:
     targets = args.targets or ["."]
     try:
-        flows = discover(targets)
+        flows = declare(targets)
     except (ImportError, SyncError) as exc:
         return _fail(str(exc))
+    except ImportFailed as exc:
+        # What an uncaught exception from the module printed, back when the
+        # import happened in this process.
+        print(exc.trace, file=sys.stderr, end="")
+        return _fail(str(exc))
     if not flows:
         return _fail(
             f"no flows declared in {', '.join(targets)} — a flow is a `Flow(...)` "
@@ -553,7 +563,8 @@ def _sync_first(
     A flow with no such record — never synced, or an engine older than the
     field — falls back to the whole directory, downwards, skipping what the
     repository ignores; a study there that will not import is a warning.
-    `--sync` names the targets outright.
+    `--sync` names the targets outright. Either way the import happens in a
+    child process (see :mod:`fluksio.sdk.declare`), not in this one.
 
     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.
@@ -561,13 +572,12 @@ def _sync_first(
     declared = None if targets else _declaration(client, flow)
     try:
         if declared is None:
-            flows = discover(targets or ["."], keep_going=True)
+            flows = declare(targets or ["."], keep_going=True)
         else:
             # Under the name it was synced with, so the node bodies it
             # generates are the ones already stored.
-            _import(*declared)
-            flows = list(FLOWS.values())
-    except SyncError as exc:
+            flows = declare_module(*declared)
+    except (SyncError, ImportFailed) as exc:
         # Do not fail a run for a module the run may not even need.
         _say(f"warning: nothing synced — {exc}")
         return

backend/fluksio/sdk/client.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from typing import Any
 from urllib.parse import quote
 
-from fluksio.sdk import MARKER, Flow, SyncError
+from fluksio.sdk import MARKER, Declared, Flow, SyncError
 
 __all__ = [
     "Client",
@@ -784,7 +784,7 @@ def ignored(directory: Path) -> frozenset[Path]:
 
 
 def sync(
-    flows: Iterable[Flow],
+    flows: Iterable[Flow | Declared],
     client: Client,
     *,
     repo: str = "",
@@ -813,7 +813,7 @@ def sync(
 
 
 def _sync_one(
-    target: Flow,
+    target: Flow | Declared,
     client: Client,
     origin: dict[str, Any],
     *,
@@ -908,7 +908,7 @@ def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list
 
 
 def _refuse_on_drift(
-    client: Client, target: Flow, definition: dict[str, Any]
+    client: Client, target: Flow | Declared, definition: dict[str, Any]
 ) -> dict[str, str]:
     """Stop before overwriting work that was done somewhere else.
 

backend/fluksio/sdk/declare.py

@@ -0,0 +1,117 @@
+"""Read the flows a module declares, in a process of its own.
+
+Importing a module runs it. A research module that sets up a GPU framework at
+its top then opens a CUDA context in whichever process imported it — and for
+`fluksio run --follow` that process is the command line, which stays alive for
+the length of the run and holds the card without the `resources.env` the node
+was declared with. So the command line never imports the module: it starts a
+short-lived child, the child imports and declares, writes down what `sync`
+needs of each flow, and exits. Whatever the import opened goes with it.
+
+The request and the answer are files in a directory of their own, not a pipe
+or the command line. The child's output stays the user's:
+a `print` at the top of a study, a warning, a skipped module in a walk, all
+appear on the terminal as they did when the import happened here.
+
+``python -m fluksio.sdk.declare`` is the child. It runs under the same
+interpreter and with the same ``sys.path`` as the command, so the module
+resolves its imports exactly as it did in-process.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import tempfile
+import traceback
+from pathlib import Path
+from typing import Any
+
+from fluksio.sdk import FLOWS, Declared, SyncError
+
+__all__ = ["ImportFailed", "declare", "declare_module"]
+
+
+class ImportFailed(Exception):
+    """A module raised something other than a refusal while it was imported.
+
+    ``str()`` is what `fluksio` printed for it when the import happened in
+    process, ``Type: message``; the child's traceback is kept for the command
+    that wants to show it.
+    """
+
+    def __init__(self, kind: str, message: str, trace: str = "") -> None:
+        super().__init__(f"{kind}: {message}")
+        self.trace = trace
+
+
+#: What the child raised, by name, for the ones a command already handles by
+#: type. Anything else is an :class:`ImportFailed`.
+_RAISED: dict[str, type[Exception]] = {
+    "SyncError": SyncError,
+    "ImportError": ImportError,
+    "ModuleNotFoundError": ModuleNotFoundError,
+}
+
+
+def declare(targets: list[str], keep_going: bool = False) -> list[Declared]:
+    """What importing these targets declares — see ``cli.discover`` for the rules."""
+    return _ask({"targets": targets, "keep_going": keep_going}, ", ".join(targets))
+
+
+def declare_module(root: str, dotted: str, expect: Path) -> list[Declared]:
+    """What importing one module, by name and from this root, declares."""
+    return _ask({"module": [root, dotted, str(expect)]}, dotted)
+
+
+def _ask(request: dict[str, Any], what: str) -> list[Declared]:
+    request["path"] = sys.path
+    with tempfile.TemporaryDirectory() as scratch:
+        (Path(scratch) / "request.json").write_text(json.dumps(request))
+        done = subprocess.run([sys.executable, "-m", __name__, scratch])
+        try:
+            reply = json.loads((Path(scratch) / "answer.json").read_text())
+        except (OSError, ValueError):
+            raise SyncError(
+                f"importing {what} ended without an answer "
+                f"(exit status {done.returncode}); its output is above"
+            ) from None
+    error = reply.get("error")
+    if error:
+        raised = _RAISED.get(error["type"])
+        if raised is not None:
+            raise raised(error["message"])
+        raise ImportFailed(error["type"], error["message"], error["traceback"])
+    return [Declared(entry) for entry in reply["flows"]]
+
+
+def main(scratch: Path) -> None:
+    """The child: read the request, import, declare, write the answer down."""
+    request = json.loads((scratch / "request.json").read_text())
+    # Imported here, not above: `cli` is the one place the walk lives, and it
+    # imports this module for the other half.
+    from fluksio.sdk.cli import _import, discover
+
+    sys.path[:] = request["path"]
+    try:
+        if "module" in request:
+            root, dotted, expect = request["module"]
+            _import(root, dotted, Path(expect))
+            flows = list(FLOWS.values())
+        else:
+            flows = discover(request["targets"], request["keep_going"])
+        reply: dict[str, Any] = {"flows": [flow.declaration() for flow in flows]}
+    except Exception as exc:
+        reply = {
+            "error": {
+                "type": type(exc).__name__,
+                "message": str(exc),
+                "traceback": traceback.format_exc(),
+            }
+        }
+    (scratch / "answer.json").write_text(json.dumps(reply))
+
+
+if __name__ == "__main__":
+    main(Path(sys.argv[1]))

backend/tests/api/routes/test_sync.py

@@ -8,7 +8,7 @@ import pytest
 from fastapi.testclient import TestClient
 
 from fluksio.core.config import settings
-from fluksio.sdk import FLOWS, MARKER, SyncError
+from fluksio.sdk import FLOWS, MARKER, Declared, SyncError
 from fluksio.sdk.client import ApiError, Client, sync
 
 EXAMPLES = Path(__file__).parents[4] / "examples"
@@ -74,6 +74,18 @@ def test_sync_stores_a_runnable_flow_with_its_origin(api, flows):
     assert "from myresearch.train import fit" in source
 
 
+def test_a_flow_read_by_another_process_syncs_as_the_flow_does(api, flows):
+    """`fluksio sync` imports in a child and uploads what it handed back."""
+    carried = Declared(flows["train"].declaration())
+
+    sync([carried], api, origin=ORIGIN)
+    stored = api.get_flow("train")["definition"]
+    assert stored["origin"]["declared_in"] == "myresearch/pipeline.py"
+
+    # Nothing differs from what the flow itself would have stored.
+    assert sync([flows["train"]], api, origin=ORIGIN)[0].unchanged
+
+
 def test_a_node_carries_the_modules_its_code_reaches(api, flows):
     """The shim names one function; the cache has to key on what it calls."""
     sync([flows["train"]], api, origin=ORIGIN)

backend/tests/sdk/test_declare.py

@@ -0,0 +1,141 @@
+"""Reading a module's flows in a child, so the command line never imports it."""
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+from fluksio.sdk import FLOWS, SyncError, cli
+from fluksio.sdk.declare import ImportFailed, declare, declare_module
+
+STUDY = """\
+import os
+
+from fluksio import Flow, Port, node
+
+with open({marker!r}, "a") as marker:
+    marker.write(str(os.getpid()) + "\\n")
+
+
+@node(provides=Port("out", "int"))
+def work():
+    return 1
+
+
+repro = Flow("repro", nodes=[work], outputs=["out"])
+"""
+
+
+def _study(tmp_path: Path, body: str | None = None) -> Path:
+    path = tmp_path / "study.py"
+    path.write_text(body or STUDY.format(marker=str(tmp_path / "marker")))
+    return path
+
+
+def test_the_module_is_imported_by_a_child_and_not_here(tmp_path):
+    """What opens a CUDA context at import must not do it in the command line."""
+    path = _study(tmp_path)
+
+    found = declare_module(str(tmp_path), "study", path)
+
+    assert [flow.name for flow in found] == ["repro"]
+    pids = (tmp_path / "marker").read_text().split()
+    assert pids and str(os.getpid()) not in pids
+    assert "study" not in sys.modules and "repro" not in FLOWS
+    assert str(tmp_path) not in sys.path
+
+
+def test_a_declared_flow_is_what_the_flow_itself_would_sync(tmp_path):
+    """Same document, same shims, same origin: sync cannot tell them apart."""
+    path = _study(tmp_path)
+    origin = {"kind": "python", "repo": str(tmp_path), "commit": "", "dirty": False}
+
+    (declared,) = declare([str(path.parent)])
+    try:
+        (local,) = cli.discover([str(path)])
+        assert declared.document(origin) == local.document(origin)
+        assert declared.shims() == local.shims()
+        assert declared.document(origin)["origin"]["declared_in"] == "study.py"
+    finally:
+        FLOWS.pop("repro", None)
+        sys.modules.pop("study", None)
+        sys.path[:] = [entry for entry in sys.path if entry != str(tmp_path)]
+
+
+def test_what_a_module_raises_reaches_the_user_as_it_did_in_process(tmp_path):
+    path = _study(tmp_path, "raise ValueError('boom')\n")
+    with pytest.raises(ImportFailed, match="^ValueError: boom$") as raised:
+        declare_module(str(tmp_path), "study", path)
+    assert "study.py" in raised.value.trace
+
+    path.write_text("import not_a_module_anywhere\n")
+    with pytest.raises(ModuleNotFoundError, match="not_a_module_anywhere"):
+        declare_module(str(tmp_path), "study", path)
+
+    # A refusal from the declaration itself keeps its own type, and message.
+    path.write_text(
+        "from fluksio import Flow\nFlow('Not A Name', nodes=[])\n",
+    )
+    with pytest.raises(SyncError, match="flow name"):
+        declare_module(str(tmp_path), "study", path)
+
+
+def test_a_child_that_ends_without_an_answer_is_named(tmp_path):
+    path = _study(tmp_path, "import os\nos._exit(3)\n")
+    with pytest.raises(SyncError, match="exit status 3"):
+        declare_module(str(tmp_path), "study", path)
+
+
+def test_a_walk_keeps_going_past_a_study_that_will_not_import(tmp_path, capfd):
+    (tmp_path / "bad.py").write_text("raise RuntimeError('half an edit')\n")
+    _study(tmp_path)
+
+    found = declare([str(tmp_path)], keep_going=True)
+
+    assert [flow.name for flow in found] == ["repro"]
+    assert "warning: skipped" in capfd.readouterr().out
+    with pytest.raises(SyncError, match="failed to import"):
+        declare([str(tmp_path)])
+
+
+def test_sync_prints_what_the_child_declared(tmp_path, capsys):
+    from fluksio.cli import _parser
+
+    path = _study(tmp_path)
+    args = _parser().parse_args(["sync", "--dry-run", str(path.parent)])
+
+    assert cli.cmd_sync(args) == 0
+    out = capsys.readouterr().out
+    assert "=== flow repro" in out
+    assert '"declared_in": "study.py"' in out
+    assert "=== repro.work" in out
+
+
+def test_a_failing_import_is_reported_by_sync_and_warned_about_by_run(
+    tmp_path, monkeypatch, capsys
+):
+    from fluksio.cli import _parser
+
+    path = _study(tmp_path, "raise ValueError('boom')\n")
+
+    # Found by a walk, it is the walk's sentence.
+    assert cli.cmd_sync(_parser().parse_args(["sync", str(tmp_path)])) == 1
+    assert "study.py failed to import — ValueError: boom" in capsys.readouterr().err
+
+    # Named outright, it is what the module raised, with the child's traceback
+    # where an uncaught exception used to print its own.
+    assert cli.cmd_sync(_parser().parse_args(["sync", str(path)])) == 1
+    seen = capsys.readouterr()
+    assert "fluksio: ValueError: boom" in seen.err
+    assert "Traceback" in seen.err and "study.py" in seen.err
+
+    class Engine:
+        def get_flow(self, name):
+            return {
+                "definition": {"origin": {"declared_in": "study.py", "module": "study"}}
+            }
+
+    monkeypatch.chdir(tmp_path)
+    cli._sync_first(Engine(), [], "repro")
+    assert "warning: nothing synced — ValueError: boom" in capsys.readouterr().out

backend/tests/test_cli.py

@@ -687,10 +687,12 @@ def test_run_imports_the_module_that_declared_the_flow(tmp_path, monkeypatch):
 
     imported: list[tuple] = []
     walked: list[list[str]] = []
-    monkeypatch.setattr(sdk_cli, "_import", lambda *args: imported.append(args))
+    monkeypatch.setattr(
+        sdk_cli, "declare_module", lambda *args: imported.append(args) or []
+    )
     monkeypatch.setattr(
         sdk_cli,
-        "discover",
+        "declare",
         lambda targets, keep_going=False: walked.append(targets) or [],
     )