Header of process and node ports follow each other both ways: a panel edit rewrites only the def line of edited code, a code save reads the header back into input ports and settings (draft and response); Space types in the node editor again

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/fluksio/api/routes/flows.py

@@ -34,7 +34,7 @@ from fluksio.api.deps import (
 from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
 from fluksio.core.db import engine
 from fluksio.flow.artifacts import is_reference
-from fluksio.flow.controller import NODE_TYPES, FlowController
+from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.events import event_bus
 from fluksio.flow.messages import flow_of, qualify
@@ -54,6 +54,7 @@ from fluksio.flow.schemas import (
     MessageValue,
     NodeDef,
     NodeSource,
+    NodeSourceSaved,
     NodeStatusPublic,
     NodeTypeInfo,
 )
@@ -672,9 +673,41 @@ def read_node_source(
     )
 
 
+def _follow_header(
+    controller: FlowController, name: str, node_id: str, code: str
+) -> FlowDef | None:
+    """Bring a function node's ports and settings in line with its header.
+
+    The draft is what the editor is on, so that is what moves, and only for a
+    node with its own code: a shared one serves other flows with other ports,
+    and a synced flow's ports are declared in its repository. Returns the
+    draft when it changed.
+    """
+    definition = controller.store.read_flow(name, draft=True)
+    node = next((n for n in definition.nodes if n.id == node_id), None)
+    spec = NODE_TYPES.get(node.type) if node is not None else None
+    if (
+        node is None
+        or spec is None
+        or not spec.free_params
+        or node.source_ref
+        or definition.origin
+    ):
+        return None
+    followed = follow_header(node, code)
+    if followed == node:
+        return None
+    definition.nodes = [followed if n.id == node_id else n for n in definition.nodes]
+    try:
+        return controller.store.write_draft(definition, definition.version)
+    except StaleVersion:
+        # Someone saved in between; the next save of this code tries again.
+        return None
+
+
 @router.put(
     "/{name}/nodes/{node_id}/source",
-    response_model=NodeStatusPublic,
+    response_model=NodeSourceSaved,
     dependencies=[Depends(get_code_author)],
 )
 async def save_node_source(
@@ -694,6 +727,10 @@ async def save_node_source(
     it — and that one is live immediately rather than waiting for a publish,
     because the copy is not any single flow's to hold back. Being live, it is
     audited like a publish.
+
+    A node with its own code takes its input ports and settings from the
+    header of ``process`` (see `follow_header`), and the answer carries the
+    node when that changed it.
     """
     definition = _read_flow(controller, name)
     ref = _source_ref(definition, node_id)
@@ -708,6 +745,9 @@ async def save_node_source(
         await run_in_threadpool(
             controller.store.write_node_source, name, node_id, source.code, True
         )
+    followed = await run_in_threadpool(
+        _follow_header, controller, name, node_id, source.code
+    )
     node_def = next((n for n in definition.nodes if n.id == node_id), None)
     device = (
         node_def.device
@@ -717,10 +757,14 @@ async def save_node_source(
     error = await run_in_threadpool(
         controller.compile_check, name, node_id, source.code, device
     )
-    return NodeStatusPublic(
+    return NodeSourceSaved(
         id=f"{name}.{node_id}",
         status="error" if error else "active",
         error=error,
+        node=next((n for n in followed.nodes if n.id == node_id), None)
+        if followed
+        else None,
+        version=followed.version if followed else None,
     )
 
 

backend/fluksio/flow/controller.py

@@ -1837,6 +1837,93 @@ def _signature_gap(
     )
 
 
+def follow_header(node: NodeDef, code: str) -> NodeDef:
+    """The node with its input ports and settings read off ``def process(...)``.
+
+    The inverse of the canvas's scaffold, which writes the input ports first,
+    in their order, and the settings after them. A parameter already declared
+    keeps its kind. A new one is a setting when it has a default, which becomes
+    its value, or when it follows a setting, and an input port otherwise. One
+    name swapped for one other is a rename, so the port keeps its message and
+    the setting its value. A declared name the header no longer has is dropped.
+
+    A header this cannot read for certain leaves the node as it is: a syntax
+    error, no top-level ``def process``, a decorator, ``*args``, ``**kwargs``,
+    a bare ``*`` or ``/``.
+    """
+    try:
+        tree = ast.parse(code)
+    except SyntaxError:
+        return node
+    defs = [
+        entry
+        for entry in tree.body
+        if isinstance(entry, ast.FunctionDef) and entry.name == "process"
+    ]
+    if not defs:
+        return node
+    args = defs[-1].args
+    if (
+        defs[-1].decorator_list
+        or args.vararg
+        or args.kwarg
+        or args.posonlyargs
+        or args.kwonlyargs
+    ):
+        return node
+    # Defaults belong to the last parameters, so they line up from the end.
+    defaults: list[ast.expr | None] = [None] * (
+        len(args.args) - len(args.defaults)
+    ) + list(args.defaults)
+    header = [
+        (arg.arg, default)
+        for arg, default in zip(args.args, defaults, strict=True)
+        if arg.arg not in RESERVED_SETTINGS
+    ]
+
+    def written(name: str) -> bool:
+        return name.isascii() and name.isidentifier()
+
+    ports = {spec.port: spec for spec in node.requires if written(spec.port)}
+    settings = {
+        key: value
+        for key, value in node.params.items()
+        if key not in RESERVED_SETTINGS and written(key)
+    }
+    names = [name for name, _ in header]
+    gone = [name for name in [*ports, *settings] if name not in names]
+    new = [name for name in names if name not in ports and name not in settings]
+    renamed = {new[0]: gone[0]} if len(gone) == len(new) == 1 else {}
+
+    requires: list[MessageSpec] = []
+    params = {key: value for key, value in node.params.items() if key not in settings}
+    after_setting = False
+    for name, default in header:
+        was = renamed.get(name, name)
+        if was in ports:
+            requires.append(ports[was].model_copy(update={"port": name}))
+        elif was in settings or default is not None or after_setting:
+            params[name] = settings[was] if was in settings else _literal(default)
+            after_setting = True
+        else:
+            requires.append(MessageSpec(port=name))
+    # Ports the header cannot name, such as a row not named yet, stay.
+    requires += [spec for spec in node.requires if not written(spec.port)]
+    return node.model_copy(update={"requires": requires, "params": params})
+
+
+def _literal(default: ast.expr | None) -> Any:
+    """A parameter's default as a setting's value, or empty when it is not JSON."""
+    if default is None:
+        return ""
+    try:
+        value = ast.literal_eval(default)
+        json.dumps(value)
+    except (ValueError, TypeError, SyntaxError, RecursionError):
+        return ""
+    return value
+
+
 def with_settings(
     function: Callable[..., Any], params: dict[str, Any]
 ) -> Callable[..., Any]:

backend/fluksio/flow/schemas.py

@@ -343,6 +343,17 @@ class NodeStatusPublic(BaseModel):
     last_error_ts: float | None = None
 
 
+class NodeSourceSaved(NodeStatusPublic):
+    """Whether saved code loads, and what its header changed.
+
+    `node` is set when the parameters of ``process`` added, dropped or renamed
+    an input port or setting. The draft moved on to `version` with it.
+    """
+
+    node: NodeDef | None = None
+    version: int | None = None
+
+
 class MessageValue(BaseModel):
     """The last payload seen on a message."""
 

backend/fluksio/mcp/server.py

@@ -208,7 +208,9 @@ async def save_node_source(name: str, node_id: str, code: str) -> Any:
     """Save a node's Python source and report whether it compiles.
 
     A node defines ``process(...)``, taking one argument per input port and
-    one per setting, and returns a dict keyed by output port.
+    one per setting, and returns a dict keyed by output port. A parameter added
+    or dropped there adds or drops the input port or setting in the draft,
+    which moves its version on: read the flow again before saving it.
     """
     return await _call(
         "PUT", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/source", json={"code": code}

backend/tests/api/routes/test_flows.py

@@ -557,6 +557,33 @@ async def test_the_live_socket_ends_when_its_client_goes(
     assert parked == []
 
 
+def test_saving_a_header_moves_the_node_ports_in_the_draft(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    saved = client.put(
+        f"{PREFIX}/header", headers=superuser_token_headers, json=a_flow("header")
+    ).json()
+
+    response = client.put(
+        f"{PREFIX}/header/nodes/logger/source",
+        headers=superuser_token_headers,
+        json={"code": "def process(reading, level, gain=2):\n    return {}\n"},
+    ).json()
+
+    assert [spec["port"] for spec in response["node"]["requires"]] == [
+        "reading",
+        "level",
+    ]
+    assert response["node"]["params"] == {"gain": 2}
+    draft = client.get(f"{PREFIX}/header", headers=superuser_token_headers).json()
+    assert draft["definition"]["version"] == response["version"]
+    assert response["version"] == saved["definition"]["version"] + 1
+    logger = next(n for n in draft["definition"]["nodes"] if n["id"] == "logger")
+    assert logger["params"] == {"gain": 2}
+
+    client.delete(f"{PREFIX}/header", headers=superuser_token_headers)
+
+
 def test_a_shared_node_in_use_is_retired_only_with_detach(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None:

backend/tests/flow/test_node_settings.py

@@ -9,7 +9,7 @@ from pathlib import Path
 
 import pytest
 
-from fluksio.flow.controller import FlowController
+from fluksio.flow.controller import FlowController, follow_header
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.pipeline import Pipeline
 from fluksio.flow.schemas import FlowDef, NodeDef
@@ -148,3 +148,66 @@ def test_a_shared_body_is_checked_against_each_user(store: FlowStore):
 
     assert [n.status for n in nodes] == ["error"]
     assert "'level' is an input of 'scale'" in (nodes[0].error or "")
+
+
+def _scaffold_header(node: NodeDef) -> str:
+    """The header the canvas writes: input ports in order, then settings."""
+    names = [spec.port for spec in node.requires] + [
+        key for key in node.params if key != "synchronous"
+    ]
+    return f"def process({', '.join(names)}):\n    return {{}}\n"
+
+
+def test_a_scaffold_header_reads_back_as_the_node_it_came_from():
+    node = _flow(
+        [*READING, MessageSpec(name="garden.level", port="depth")],
+        factor=3,
+        synchronous=True,
+    ).nodes[0]
+
+    assert follow_header(node, _scaffold_header(node)) == node
+
+
+def test_a_header_declares_ports_and_settings_that_read_back_the_same():
+    header = "def process(reading, level, gain=1.5, mode='eco'):\n    return {}\n"
+
+    node = follow_header(_flow([]).nodes[0], header)
+
+    assert [spec.port for spec in node.requires] == ["reading", "level"]
+    assert node.params == {"gain": 1.5, "mode": "eco"}
+    assert _scaffold_header(node).split(":")[0] == (
+        "def process(reading, level, gain, mode)"
+    )
+
+
+def test_a_renamed_parameter_keeps_the_message_and_a_dropped_one_goes():
+    node = _flow(READING, factor=3, offset=1).nodes[0]
+
+    renamed = follow_header(node, "def process(value, factor, offset): pass\n")
+    dropped = follow_header(node, "def process(reading, factor): pass\n")
+
+    assert renamed.requires == [READING[0].model_copy(update={"port": "value"})]
+    assert renamed.params == {"factor": 3, "offset": 1}
+    assert dropped.requires == READING
+    assert dropped.params == {"factor": 3}
+
+
+def test_a_new_parameter_after_a_setting_is_a_setting():
+    node = _flow(READING, factor=3).nodes[0]
+
+    followed = follow_header(node, "def process(reading, factor, offset): pass\n")
+
+    assert followed.requires == READING
+    assert followed.params == {"factor": 3, "offset": ""}
+
+
+def test_a_header_it_cannot_read_for_certain_leaves_the_node():
+    node = _flow(READING, factor=3).nodes[0]
+
+    for code in (
+        "def process(reading, factor",
+        "def process(**kw): pass\n",
+        "@cache\ndef process(value): pass\n",
+        "def run(value): pass\n",
+    ):
+        assert follow_header(node, code) == node