diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 37eb6eb..af7fda6 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/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, ) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 8c8bf1c..ba631d1 100644 --- a/backend/fluksio/flow/controller.py +++ b/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]: diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 00d5f1d..afe4397 100644 --- a/backend/fluksio/flow/schemas.py +++ b/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.""" diff --git a/backend/fluksio/mcp/server.py b/backend/fluksio/mcp/server.py index 5c7b5a8..4cffc76 100644 --- a/backend/fluksio/mcp/server.py +++ b/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} diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 692a3a2..a83a281 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/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: diff --git a/backend/tests/flow/test_node_settings.py b/backend/tests/flow/test_node_settings.py index 6fabb30..96a2650 100644 --- a/backend/tests/flow/test_node_settings.py +++ b/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 diff --git a/docs/code/nodes.md b/docs/code/nodes.md index cee3609..4b6dec3 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -29,6 +29,24 @@ Inside a node, `import fluksio` exposes reporting and artifact helpers. It does not expose the engine's internal modules. Other imports come from the node's Python environment; see [Packages](#packages). +## The header follows the ports + +On the canvas, the parameters of `process` and the node's input ports and +settings are kept in step, in both directions. + +- Changing ports or settings in the node panel rewrites the + `def process(...)` line and leaves the rest of the code as it is. Input + ports come first, in their order, then the settings. A parameter that stays + keeps its annotation and default. +- Saving the code reads that line back. A name the node already has keeps + its kind. A new parameter becomes a setting when it has a default, which + becomes its value, or when it comes after a setting. Any other new + parameter becomes an input port. Removing a parameter removes its port or + setting, and renaming a single one renames the port (keeping its message) + or the setting (keeping its value). +- Shared code, code from `fluksio sync`, and a header with a decorator, + `*args`, `**kwargs`, a bare `*` or `/` are left alone in both directions. + ## Producing values over time Yield dictionaries to publish outputs before the node finishes: diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 4a9df04..222746c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -2543,6 +2543,92 @@ export const NodeSourceSchema = { description: 'The Python source of a node.' } as const; +export const NodeSourceSavedSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + status: { + type: 'string', + title: 'Status', + default: 'active' + }, + error: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Error' + }, + health: { + type: 'string', + enum: ['ok', 'degraded', 'down'], + title: 'Health', + default: 'ok' + }, + health_detail: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Health Detail' + }, + last_error: { + type: 'string', + title: 'Last Error', + default: '' + }, + last_error_ts: { + anyOf: [ + { + type: 'number' + }, + { + type: 'null' + } + ], + title: 'Last Error Ts' + }, + node: { + anyOf: [ + { + '$ref': '#/components/schemas/NodeDef-Output' + }, + { + type: 'null' + } + ] + }, + version: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Version' + } + }, + type: 'object', + required: ['id'], + title: 'NodeSourceSaved', + description: `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.` +} as const; + export const NodeStatusPublicSchema = { properties: { id: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 8e727a1..221429d 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1116,11 +1116,15 @@ export class FlowsService { * 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. * @param data The data for the request. * @param data.name * @param data.nodeId * @param data.requestBody - * @returns NodeStatusPublic Successful Response + * @returns NodeSourceSaved Successful Response * @throws ApiError */ public static saveNodeSource(data: FlowsSaveNodeSourceData): CancelablePromise { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index a599c09..f825088 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -888,6 +888,26 @@ export type NodeSource = { missing?: boolean; }; +/** + * 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. + */ +export type NodeSourceSaved = { + id: string; + status?: string; + error?: (string | null); + health?: 'ok' | 'degraded' | 'down'; + health_detail?: (string | null); + last_error?: string; + last_error_ts?: (number | null); + node?: (NodeDef_Output | null); + version?: (number | null); +}; + +export type health = 'ok' | 'degraded' | 'down'; + /** * Whether a node loaded, and how its connection is doing. */ @@ -901,8 +921,6 @@ export type NodeStatusPublic = { last_error_ts?: (number | null); }; -export type health = 'ok' | 'degraded' | 'down'; - /** * A node type the editor can offer, with its parameter schema. */ @@ -1836,7 +1854,7 @@ export type FlowsSaveNodeSourceData = { requestBody: NodeSource; }; -export type FlowsSaveNodeSourceResponse = (NodeStatusPublic); +export type FlowsSaveNodeSourceResponse = (NodeSourceSaved); export type FlowsShareNodeData = { name: string; diff --git a/frontend/src/components/Flow/FlowEditor.tsx b/frontend/src/components/Flow/FlowEditor.tsx index 239f751..34c286f 100644 --- a/frontend/src/components/Flow/FlowEditor.tsx +++ b/frontend/src/components/Flow/FlowEditor.tsx @@ -273,6 +273,7 @@ function FlowEditorInner({ flush, conflict, resolveConflict, + adopt, mutation: saving, } = useAutosave(flowName) const publish = usePublish(flowName) @@ -855,13 +856,34 @@ function FlowEditorInner({ }) const sourceMutation = useMutation({ - mutationFn: ({ nodeId, code }: { nodeId: string; code: string }) => - FlowsService.saveNodeSource({ + mutationFn: async ({ nodeId, code }: { nodeId: string; code: string }) => { + // The server reads the ports off the header and compares them with its + // draft, so the ports on the canvas have to be there first. + await flush() + return FlowsService.saveNodeSource({ name: flowName, nodeId, requestBody: { code }, - }), - onSuccess: () => { + }) + }, + onSuccess: (saved) => { + const followed = saved.node + if (followed && saved.version) { + // The header changed the ports in the draft: take them, and the + // version they moved it to, so the next save is not a conflict. + adopt(saved.version) + commit( + (latest.current.nodes ?? []).map((node) => + node.id === followed.id + ? { + ...node, + requires: followed.requires, + params: followed.params, + } + : node, + ), + ) + } queryClient.invalidateQueries({ queryKey: ["flows", flowName] }) }, }) diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index b5cdffc..f09bfe4 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -49,6 +49,13 @@ import useCustomToast from "@/hooks/useCustomToast" import { inCodeEditor, useShortcuts } from "@/lib/shortcuts" import { cn } from "@/lib/utils" import { DtypeValue } from "./FlowBoundary" +import { + IDENTIFIER, + portName, + processArgs, + RESERVED_PARAMS, + withHeader, +} from "./header" import { JsonField } from "./JsonField" import { MessageSparkline } from "./MessageSparkline" import { @@ -92,12 +99,6 @@ const NO_SECRET = "__none__" const SECTION = PANEL_SECTION -/** Settings the engine reads itself, so they are not the author's to name. */ -const RESERVED_PARAMS = new Set(["synchronous"]) - -/** A setting reaches the function by name, so the name has to be one. */ -const IDENTIFIER = /^[A-Za-z_]\w*$/ - /** Sentinels for the resources dropdown, which has two options that are not sizes. */ const NO_RESOURCES = "__shared__" const CUSTOM_RESOURCES = "__custom__" @@ -582,6 +583,8 @@ function FreeParamsForm({ // remount its row and take the half-typed value with it.
- spec.port || (spec.name ?? "").split(".").pop() || "" - /** * Whether a port was named by hand rather than after its message. Nothing * stores that: a port whose name is not the one its message gives it is one. @@ -1376,25 +1375,11 @@ const ownsName = (spec: MessageSpec) => Boolean(spec.port) && spec.port !== portName({ name: spec.name }) /** - * A `process` that takes this node's inputs and settings, and returns its - * outputs. - * - * Both arrive by name, so both are arguments — the ports first, in the order - * they are declared, then the settings. + * A `process` that takes this node's inputs and settings (`processArgs`), and + * returns its outputs. */ function scaffoldFor(node: NodeDef_Input): string { - const args = [ - ...new Set( - [ - ...(node.requires ?? []).map(portName), - ...Object.keys(node.params ?? {}).filter( - (name) => !RESERVED_PARAMS.has(name), - ), - ] - // Anything else cannot be a keyword argument, so it cannot be one. - .filter((name) => IDENTIFIER.test(name)), - ), - ] + const args = processArgs(node) const returns = (node.provides ?? []) .filter((spec) => portName(spec)) .map((spec) => `"${portName(spec)}": ${PLACEHOLDER[spec.dtype ?? "float"]}`) @@ -1470,11 +1455,12 @@ function PanelBody({ } /** - * Change the node, and keep an untouched scaffold in step with its ports. + * Change the node, and keep its code in step with its ports and settings. * - * Only code that is still exactly what this panel generates is rewritten — - * one edit of your own and it is yours, ports or no ports. Shared code is - * never rewritten either: the other flows using it have their own ports. + * Code that is still exactly what this panel generates is generated again, + * outputs included. Code of your own only has the header of `process` + * rewritten, and the body is left as it is. Shared code is never rewritten: + * the other flows using it have their own ports. */ const editNode = (next: NodeDef_Input) => { onChange(next) @@ -1487,8 +1473,10 @@ function PanelBody({ current === undefined ) return - const wanted = scaffoldFor(next) - if (current !== wanted && SCAFFOLD_SHAPE.test(current)) editCode(wanted) + const wanted = SCAFFOLD_SHAPE.test(current) + ? scaffoldFor(next) + : withHeader(current, processArgs(next)) + if (current !== wanted) editCode(wanted) } /** @@ -1705,7 +1693,9 @@ function PanelBody({ {CODE_LOCKED}

)} -
+ {/* `nokey`: Monaco types through a plain div (EditContext), which + the canvas takes for itself, so Space would pan instead. */} +
diff --git a/frontend/src/components/Flow/header.check.ts b/frontend/src/components/Flow/header.check.ts new file mode 100644 index 0000000..bebc171 --- /dev/null +++ b/frontend/src/components/Flow/header.check.ts @@ -0,0 +1,79 @@ +/** + * The header of `process` kept in step with a node's ports and settings, + * checked. The server's half, reading them back, is `follow_header` in + * `backend/tests/flow/test_node_settings.py`. + * + * cd frontend && bun run src/components/Flow/header.check.ts + */ + +import assert from "node:assert/strict" + +import { processArgs, readHeader, withHeader } from "./header" + +const BODY = ` # the body stays exactly as written + return {"out": reading * gain} +` +const code = (params: string) => + `import math\n\n\ndef process(${params}) -> dict:\n${BODY}` + +// Inputs in order, then settings; the engine's own setting is not one. +assert.deepEqual( + processArgs({ + id: "n", + requires: [{ name: "house.reading" }, { name: "x", port: "level" }], + params: { gain: 2, synchronous: true }, + }), + ["reading", "level", "gain"], +) + +// Config to header and back: the names written are the names read. +const written = withHeader(code("reading, gain=2.0"), [ + "reading", + "level", + "gain", +]) +assert.equal(written, code("reading, level, gain=2.0")) +assert.deepEqual( + readHeader(written)?.params.map((param) => param.name), + ["reading", "level", "gain"], +) + +// A rename keeps the annotation; a dropped name goes; the body never moves. +assert.equal( + withHeader(code("reading: float, gain=2.0"), ["value", "gain"]), + code("value: float, gain=2.0"), +) +assert.equal(withHeader(code("reading, gain"), ["reading"]), code("reading")) + +// A new setting without a default cannot follow one with a default. +assert.equal( + withHeader(code("reading, gain=2.0"), ["reading", "gain", "offset"]), + code("reading, offset, gain=2.0"), +) + +// Nothing to change is no rewrite, not even of the spacing. +assert.equal( + withHeader(code("reading ,gain"), ["reading", "gain"]), + code("reading ,gain"), +) + +// Commas inside a default or a string are not parameter boundaries. +assert.deepEqual( + readHeader(code('reading, sep=", ", sizes=(1, 2)'))?.params.map( + (param) => param.name, + ), + ["reading", "sep", "sizes"], +) + +// What the server will not read is not rewritten either. +for (const params of [ + "*values", + "reading, **settings", + "reading, /", + "a, # why\n b", +]) + assert.equal(withHeader(code(params), ["other"]), code(params)) +const decorated = `@cache\n${code("reading").slice("import math\n\n\n".length)}` +assert.equal(withHeader(decorated, ["other"]), decorated) + +console.log("header: ok") diff --git a/frontend/src/components/Flow/header.ts b/frontend/src/components/Flow/header.ts new file mode 100644 index 0000000..30b459d --- /dev/null +++ b/frontend/src/components/Flow/header.ts @@ -0,0 +1,140 @@ +/** + * The header of a function node's `process`, and keeping it in step with the + * node's ports and settings. + * + * The other direction, reading ports and settings back off a header, is the + * server's: `follow_header` runs on every save of the code. + */ + +import type { MessageSpec, NodeDef_Input } from "@/client" + +/** Settings the engine reads itself, so they are not the author's to name. */ +export const RESERVED_PARAMS = new Set(["synchronous"]) + +/** A setting reaches the function by name, so the name has to be one. */ +export const IDENTIFIER = /^[A-Za-z_]\w*$/ + +/** The name the node function sees; the engine derives it the same way. */ +export const portName = (spec: MessageSpec) => + spec.port || (spec.name ?? "").split(".").pop() || "" + +/** + * The parameters `process` takes for this node. + * + * Inputs and settings both arrive by name, so both are parameters, the inputs + * first in the order they are declared, then the settings. + */ +export function processArgs(node: NodeDef_Input): string[] { + return [ + ...new Set( + [ + ...(node.requires ?? []).map(portName), + ...Object.keys(node.params ?? {}).filter( + (name) => !RESERVED_PARAMS.has(name), + ), + ] + // Anything else cannot be a keyword argument, so it cannot be one. + .filter((name) => IDENTIFIER.test(name)), + ), + ] +} + +type Param = { name: string; text: string; hasDefault: boolean } + +/** Where the parameters of the last top-level `def process(` sit. */ +const PROCESS_DEF = /^def process\(/gm + +/** + * The parameters of `process` as written, and the span they take up. + * + * Null for a header this cannot read for certain, the same ones the server + * leaves alone: no top-level `def process`, a decorator, `*args`, `**kwargs`, + * a bare `*` or `/`, or a comment inside the parentheses. + */ +export function readHeader( + code: string, +): { from: number; to: number; params: Param[] } | null { + // Python keeps the last definition, so that is the one that runs. + const match = [...code.matchAll(PROCESS_DEF)].pop() + if (match?.index === undefined) return null + const above = code.slice(0, match.index).trimEnd() + if (above.slice(above.lastIndexOf("\n") + 1).startsWith("@")) return null + + const from = match.index + match[0].length + const params: Param[] = [] + let start = from + let depth = 0 + let hasDefault = false + const take = (end: number) => { + const text = code.slice(start, end).trim() + const name = /^[A-Za-z_]\w*/.exec(text)?.[0] + params.push({ name: name ?? "", text, hasDefault }) + start = end + 1 + hasDefault = false + return Boolean(name) + } + for (let i = from; i < code.length; i++) { + const char = code[i] + if (char === '"' || char === "'") { + const quote = code.startsWith(char.repeat(3), i) ? char.repeat(3) : char + let j = i + quote.length + while (j < code.length && !code.startsWith(quote, j)) + j += code[j] === "\\" ? 2 : 1 + if (j >= code.length) return null + i = j + quote.length - 1 + } else if (char === "#") { + return null + } else if ("([{".includes(char)) { + depth++ + } else if (")]}".includes(char) && depth > 0) { + depth-- + } else if (char === ")") { + // A trailing comma leaves nothing after it, which is not a parameter. + if (code.slice(start, i).trim() && !take(i)) return null + return { from, to: i, params } + } else if (depth === 0 && char === ",") { + if (!take(i)) return null + } else if (depth === 0 && char === "=") { + hasDefault = true + } + } + return null +} + +/** + * `code` with `process` taking `args`, and every other byte as it was. + * + * A parameter that stays keeps its annotation and default, and so does one + * renamed (one name gone, one new). Python wants the parameters without a + * default first, so a new one goes ahead of any that have one. A header this + * cannot read is left alone. + */ +export function withHeader(code: string, args: string[]): string { + const header = readHeader(code) + if (!header) return code + if (header.params.map((param) => param.name).join() === args.join()) + return code + const written = new Map(header.params.map((param) => [param.name, param])) + const gone = header.params.filter((param) => !args.includes(param.name)) + const fresh = args.filter((name) => !written.has(name)) + if (gone.length === 1 && fresh.length === 1) { + const [was] = gone + written.set(fresh[0], { + ...was, + name: fresh[0], + text: fresh[0] + was.text.slice(was.name.length), + }) + } + const params = args.map( + (name) => written.get(name) ?? { name, text: name, hasDefault: false }, + ) + const ordered = [ + ...params.filter((param) => !param.hasDefault), + ...params.filter((param) => param.hasDefault), + ] + return ( + code.slice(0, header.from) + + ordered.map((param) => param.text).join(", ") + + code.slice(header.to) + ) +} diff --git a/frontend/src/components/Flow/queries.ts b/frontend/src/components/Flow/queries.ts index 66a748d..8c82c78 100644 --- a/frontend/src/components/Flow/queries.ts +++ b/frontend/src/components/Flow/queries.ts @@ -172,6 +172,8 @@ export function useAutosave(name: string): { flush: () => Promise conflict: boolean resolveConflict: (mode: "theirs" | "mine") => Promise + /** Take a version the server moved the draft to by itself. */ + adopt: (next: number) => void mutation: UseMutationResult } { const queryClient = useQueryClient() @@ -283,5 +285,9 @@ export function useAutosave(name: string): { } }, [flush]) - return { save, flush, conflict, resolveConflict, mutation } + const adopt = useCallback((next: number) => { + version.current = next + }, []) + + return { save, flush, conflict, resolveConflict, adopt, mutation } }