diff --git a/.env.example b/.env.example index 6a18345..0ed4641 100644 --- a/.env.example +++ b/.env.example @@ -5,7 +5,7 @@ # Domain Traefik routes on and acquires TLS certificates for. DOMAIN=localhost -# The address people and devices reach this instance at: panel links, mails +# The address people and devices reach this instance at: device links, mails # and the MCP issuer are built from it. FRONTEND_HOST=http://localhost:5173 diff --git a/backend/README.md b/backend/README.md index e2f0413..001b037 100644 --- a/backend/README.md +++ b/backend/README.md @@ -23,7 +23,7 @@ from fluksio import Port, node @node( requires=["dataset", Port("lr", "float")], provides=[Port("loss", "float", stream=True), Port("weights", "artifact")], - device="gpu", device_policy="prefer", + worker="gpu", worker_policy="prefer", ) def fit(dataset, lr, epochs=25): for epoch in range(epochs): diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 76eca9f..0785f8f 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -745,13 +745,13 @@ async def save_node_source( _follow_header, controller, name, node_id, source.code, drop ) node_def = next((n for n in definition.nodes if n.id == node_id), None) - device = ( - node_def.device - if node_def is not None and node_def.device_policy == "require" + worker = ( + node_def.worker + if node_def is not None and node_def.worker_policy == "require" else None ) error = await run_in_threadpool( - controller.compile_check, name, node_id, source.code, device + controller.compile_check, name, node_id, source.code, worker ) return NodeSourceSaved( id=f"{name}.{node_id}", diff --git a/backend/fluksio/api/routes/workers.py b/backend/fluksio/api/routes/workers.py index 2388ef9..3dca9d0 100644 --- a/backend/fluksio/api/routes/workers.py +++ b/backend/fluksio/api/routes/workers.py @@ -83,7 +83,7 @@ class WorkerInfo(BaseModel): class TokenRequest(BaseModel): name: str = Field(min_length=1, max_length=255) - #: What a node's `device` may match on this worker; it cannot add to them. + #: What a node's `worker` may match on this worker; it cannot add to them. labels: list[str] = Field(default_factory=list) diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 5741e5e..cbe2039 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -175,8 +175,8 @@ class LoadedNode: #: Published, and with no body of its own — so what runs is the new-node #: template, which returns nothing at all. missing_source: bool = False - #: The device this node names although its type never runs on a worker. - ignored_device: str = "" + #: The worker this node names although its type never runs on one. + ignored_worker: str = "" @dataclass @@ -482,7 +482,7 @@ class FlowController: # Without a pool, python nodes are compiled and run in this process — # which is what the tests do, and what a bare `Pipeline` has always done. self.workers = workers - # Workers on other hosts. A node without a device never touches it. + # Workers on other hosts. A node naming no worker never touches it. self.remote = remote # This machine's own books. The pool sizes its fair share off them. self.resources = resources @@ -1005,7 +1005,7 @@ class FlowController: def _placed( self, wanted: Resources, - device: str | None, + label: str | None, policy: str, owner: str, local: str, @@ -1034,7 +1034,7 @@ class FlowController: def call(**kwargs: Any) -> Any: with placer.claim( # type: ignore[union-attr] - wanted, device=device, policy=policy, node=node_id, run=run_id + wanted, label=label, policy=policy, node=node_id, run=run_id ) as (target, allocation): env = derive_env(wanted, allocation, local=target.worker is None) if target.worker is None: @@ -1100,8 +1100,8 @@ class FlowController: node_type = NODE_TYPES.get(node_def.type) if node_type is None: raise ValueError(f"Unknown node type '{node_def.type}'") - if node_def.device and not node_type.has_source: - entry.ignored_device = node_def.device + if node_def.worker and not node_type.has_source: + entry.ignored_worker = node_def.worker params = resolve_params(node_def.params) @@ -1153,7 +1153,7 @@ class FlowController: else: # The code never runs here: it is loaded in a worker, and # the node calls that worker instead of a local function. - # A node bound to a device is loaded on *that* machine — + # A node bound to a worker is loaded on *that* machine — # one importing torch is correct on the GPU box and a # missing module here, so checking it here would fail a # node that is fine. @@ -1170,17 +1170,17 @@ class FlowController: entry.error = str(exc) return entry remote_only = ( - node_def.device - and node_def.device_policy == "require" + node_def.worker + and node_def.worker_policy == "require" and self.remote is not None ) if remote_only and self.remote is not None: problem = self.remote.compile( - node_def.device or "", owner, local, code + node_def.worker or "", owner, local, code ) elif self._runs_elsewhere(wanted): # Asks for more than this machine has, so it will run on - # one that has it. Same reason as a device: checking the + # one that has it. Same reason as a worker label: checking the # import here would fail a node that is fine there. problem = None else: @@ -1199,12 +1199,12 @@ class FlowController: if wanted is not None and self.placer is not None: # Says how much of a machine it takes, so which machine # and how much of it are one decision — including when - # it also names a device, which used to mean the two + # it also names a worker, which used to mean the two # answers disagreed and nothing was accounted at all. function = self._placed( wanted, - node_def.device, - node_def.device_policy, + node_def.worker, + node_def.worker_policy, owner, local, code, @@ -1213,11 +1213,11 @@ class FlowController: run_id=run.run_id if run else "", on_event=emissions.handle, ) - elif node_def.device and self.remote is not None: - # A device and nothing about size: the label alone + elif node_def.worker and self.remote is not None: + # A worker and nothing about size: the label alone # decides, least busy first, as it always has. function = self.remote.proxy( - node_def.device, + node_def.worker, owner, local, code, @@ -1235,7 +1235,7 @@ class FlowController: run_id=run.run_id if run else "", on_event=emissions.handle, ) - if node_def.device_policy == "prefer" + if node_def.worker_policy == "prefer" else None ), ) @@ -1442,11 +1442,11 @@ class FlowController: ) def compile_check( - self, flow: str, node_id: str, code: str, device: str | None = None + self, flow: str, node_id: str, code: str, worker: str | None = None ) -> str | None: """Does this source load? Returns what to show the author, or None. - Loaded on the machine that will run it: a node bound to a device is + Loaded on the machine that will run it: a node bound to a worker is checked against that worker's packages, because a missing import here says nothing about whether it is missing there. @@ -1454,8 +1454,8 @@ class FlowController: the published one — so ``keep=False``: compile it to answer the question, and leave the worker holding what it is serving calls from. """ - if device and self.remote is not None: - return self.remote.compile(device, flow, node_id, code) + if worker and self.remote is not None: + return self.remote.compile(worker, flow, node_id, code) if self.workers is not None: return self.workers.compile(flow, node_id, code, keep=False) try: @@ -2082,21 +2082,21 @@ def _collect_issues( for entry in loaded.values() if entry.missing_source ] - # Only code travels to a worker, so a device on any other node is a field + # Only code travels to a worker, so a worker on any other node is a field # nothing reads. Saying so beats a setting that looks like it took. issues += [ ValidationIssue( - code="ignored_device", + code="ignored_worker", message=( - f"Node '{entry.id.rpartition('.')[2]}' names device " - f"'{entry.ignored_device}', but only a Function node runs on a " + f"Node '{entry.id.rpartition('.')[2]}' names worker " + f"'{entry.ignored_worker}', but only a Function node runs on a " "worker. The setting is ignored." ), flow=entry.flow, node=entry.id, ) for entry in loaded.values() - if entry.ignored_device + if entry.ignored_worker ] return issues diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 2fcfaf9..c0e8504 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -47,7 +47,7 @@ MANUAL_RUN_PREFIX = "manual-" # neither the brain graph nor the health summary treats them as a fault. # Lives here rather than beside the controller so `ValidationIssue` can carry # the distinction itself, and every reader gets it for free. -ADVISORY_ISSUES = frozenset({"unauthenticated_hook", "ignored_device"}) +ADVISORY_ISSUES = frozenset({"unauthenticated_hook", "ignored_worker"}) class ValidationIssue(BaseModel): @@ -62,7 +62,7 @@ class ValidationIssue(BaseModel): "self_loop_needs_initial", "missing_source", "node_unhealthy", - "ignored_device", + "ignored_worker", "unknown_device", "device_port", ] diff --git a/backend/fluksio/flow/placement.py b/backend/fluksio/flow/placement.py index 4cc7330..f35d2fe 100644 --- a/backend/fluksio/flow/placement.py +++ b/backend/fluksio/flow/placement.py @@ -1,7 +1,7 @@ """Which machine a node runs on, and the waiting when none of them has room. The engine used to answer this twice, in two ways that could not see each -other. A node with a ``device`` went to a worker carrying that label, chosen by +other. A node with a ``worker`` went to a worker carrying that label, chosen by which of them had least in flight — a machine's *name*, never its size. A node with ``resources`` was accounted against the engine's own cores, and only the engine's: declaring both meant the second answer silently won and nothing was @@ -63,15 +63,15 @@ class Target: #: None is the engine itself, which runs nodes in its own worker pool. worker: RemoteWorker | None = None - def carries(self, device: str) -> bool: - """Whether a node asking for this device may run here. + def carries(self, label: str) -> bool: + """Whether a node asking for this label may run here. The engine carries no label: a label names a machine somebody attached for the purpose, and the engine is where the node would have run anyway. """ if self.worker is None: return False - return device == self.name or device in self.worker.labels + return label == self.name or label in self.worker.labels @dataclass @@ -138,7 +138,7 @@ class Placer: cpus: int, gpus: int, ram_mb: int, - device: str | None, + label: str | None, policy: str, ) -> list[Target]: """The machines that could run this node, best first.""" @@ -146,8 +146,8 @@ class Placer: remote = [target for target in targets if target.worker is not None] engine = [target for target in targets if target.worker is None] - if device: - labelled = [target for target in remote if target.carries(device)] + if label: + labelled = [target for target in remote if target.carries(label)] fitting = self._fitting(labelled, cpus, gpus, ram_mb) if fitting or policy != "prefer": # `require` waits for the labelled machine even when the engine @@ -174,7 +174,7 @@ class Placer: # -- what can actually be granted ------------------------------------------ - def _shapes(self, device: str | None, policy: str) -> list[tuple[int, int, int]]: + def _shapes(self, label: str | None, policy: str) -> list[tuple[int, int, int]]: """Every machine's size, as something to measure a request against. A provisioner's job shapes count: a machine it can start on demand is @@ -182,18 +182,18 @@ class Placer: """ shapes = [] for target in self.targets(): - if device and not target.carries(device): + if label and not target.carries(label): # A `prefer` node may end up here, so the engine still counts. if policy == "require" or target.worker is not None: continue books = target.accountant shapes.append((books.cpus, books.gpus, books.ram_mb or 0)) for provisioner in self.provisioners: - shapes.extend(provisioner.shapes(device)) + shapes.extend(provisioner.shapes(label)) return shapes or [(self.local.cpus, self.local.gpus, self.local.ram_mb or 0)] def _clamp( - self, wanted: Resources, node: str, device: str | None, policy: str + self, wanted: Resources, node: str, label: str | None, policy: str ) -> tuple[int, int, int]: """What can be granted somewhere, which may be less than was asked. @@ -206,7 +206,7 @@ class Placer: measured only against machines that carry that many. """ ram = wanted.ram or 0 - shapes = self._shapes(device, policy) + shapes = self._shapes(label, policy) gpus = min(wanted.gpus, max(shape[1] for shape in shapes)) capable = [shape for shape in shapes if shape[1] >= gpus] cpus = min(wanted.cpus, max(shape[0] for shape in capable)) @@ -246,7 +246,7 @@ class Placer: self, wanted: Resources, *, - device: str | None = None, + label: str | None = None, policy: str = "require", node: str = "", run: str = "", @@ -259,8 +259,8 @@ class Placer: with self._condition: while chosen is None: - cpus, gpus, ram = self._clamp(wanted, node, device, policy) - for target in self._candidates(cpus, gpus, ram, device, policy): + cpus, gpus, ram = self._clamp(wanted, node, label, policy) + for target in self._candidates(cpus, gpus, ram, label, policy): got = target.accountant.try_take(cpus, gpus, ram) if got is not None: chosen, allocation = target, got @@ -269,9 +269,9 @@ class Placer: break if not ticket: ticket = self._announce( - node, run, waited_from, self._reason(cpus, gpus, device) + node, run, waited_from, self._reason(cpus, gpus, label) ) - self._provision(cpus, gpus, ram, device) + self._provision(cpus, gpus, ram, label) self._condition.wait(timeout=WAKE_S) if ticket: self._waiting.pop(ticket, None) @@ -291,21 +291,21 @@ class Placer: # timeout, a cancellation, or the worker going away mid-call. chosen.accountant.give_back(allocation) - def _reason(self, cpus: int, gpus: int, device: str | None) -> str: + def _reason(self, cpus: int, gpus: int, label: str | None) -> str: parts = [] if gpus: parts.append(f"{gpus} gpu(s)") parts.append(f"{cpus} cpu(s)") wants = " and ".join(parts) - if device: - return f"waiting for a worker labelled '{device}' with {wants} free" + if label: + return f"waiting for a worker labelled '{label}' with {wants} free" return f"waiting for {wants}" - def _provision(self, cpus: int, gpus: int, ram_mb: int, device: str | None) -> None: + def _provision(self, cpus: int, gpus: int, ram_mb: int, label: str | None) -> None: """Ask for a machine, if something here can start one.""" for provisioner in self.provisioners: - if provisioner.covers(cpus, gpus, ram_mb, device): - provisioner.provision(cpus, gpus, ram_mb, device) + if provisioner.covers(cpus, gpus, ram_mb, label): + provisioner.provision(cpus, gpus, ram_mb, label) return def provision_for(self, needs: dict[str, Any] | None) -> None: @@ -321,7 +321,7 @@ class Placer: int(needs.get("cpus") or 1), int(needs.get("gpus") or 0), int(needs.get("ram_mb") or 0), - needs.get("device") or None, + needs.get("worker") or None, ) def _announce(self, node: str, run: str, since: float, reason: str) -> int: @@ -362,15 +362,15 @@ class Placer: cpus = int(needs.get("cpus") or 1) gpus = int(needs.get("gpus") or 0) ram = int(needs.get("ram_mb") or 0) - device = needs.get("device") or None + label = needs.get("worker") or None for target in self.targets(): - if device and not target.carries(device): + if label and not target.carries(label): continue if target.accountant.fits(cpus, gpus, ram): # Something here could take it. Busy is queueing, not waiting. return None if not any( - provisioner.covers(cpus, gpus, ram, device) + provisioner.covers(cpus, gpus, ram, label) for provisioner in self.provisioners ): # Nothing attached fits and nothing can be started, so the node diff --git a/backend/fluksio/flow/provision.py b/backend/fluksio/flow/provision.py index ea4520b..b031cb3 100644 --- a/backend/fluksio/flow/provision.py +++ b/backend/fluksio/flow/provision.py @@ -91,7 +91,7 @@ class SlurmProfile: cpus: int = 1 gpus: int = 0 ram_mb: int = 0 - #: What a worker started this way advertises, so a node bound to a device + #: What a worker started this way advertises, so a node naming one of them #: can be the reason one is started. labels: list[str] = field(default_factory=list) #: ``#SBATCH`` lines, verbatim: partition, gres, walltime, account. diff --git a/backend/fluksio/flow/remote.py b/backend/fluksio/flow/remote.py index fcf638d..8b31f3b 100644 --- a/backend/fluksio/flow/remote.py +++ b/backend/fluksio/flow/remote.py @@ -125,7 +125,7 @@ class RemoteWorker: self._lock = threading.Lock() self._gone = False # Source this worker has already loaded, by digest. A rebuild asks - # every device-bound node whether it compiles, and without this that + # every worker-bound node whether it compiles, and without this that # is a network round trip each time — which is how attaching a GPU box # makes rebuilding the flows slow enough to fail a health check. A # worker that reconnects is a new object, so this empties with it. @@ -462,7 +462,7 @@ class RemoteWorkerHub: fallback: Callable[..., Any] | None = None, env: dict[str, str] | None = None, ) -> Callable[..., Any]: - """What a node with a device runs instead of its own function. + """What a node bound to a worker runs instead of its own function. ``fallback`` is the local pool's proxy, used when the node only prefers the label rather than requiring it. diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 4f5b3de..5145ff8 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -181,7 +181,7 @@ def _runs_on_a_worker(node: NodeDef) -> bool: """Whether this node's body could go to a worker at all. Only a node whose source travels can: a connector is an entry point loaded - in this process, so a device on one is a field nothing reads. Holding a run + in this process, so a worker on one is a field nothing reads. Holding a run for a worker that could never take that node is a run that never starts. """ node_type = NODE_TYPES.get(node.type) @@ -191,17 +191,17 @@ def _runs_on_a_worker(node: NodeDef) -> bool: def required_labels(flow: FlowDef) -> list[str]: """Worker labels this flow cannot run without. - Only the nodes that *require* their device. One that merely prefers it runs + Only the nodes that *require* their worker. One that merely prefers it runs on the engine when no such worker is attached, so holding the whole run back to wait for one would be waiting for something it does not need. """ return sorted( { - node.device + node.worker for node in flow.nodes - if node.device - and node.device.strip() - and node.device_policy == "require" + if node.worker + and node.worker.strip() + and node.worker_policy == "require" and _runs_on_a_worker(node) } ) @@ -212,11 +212,11 @@ def required_resources(flow: FlowDef) -> dict[str, Any] | None: Dimension by dimension rather than per node, which is deliberately blunt: it answers "could this run start at all", not "in what order". A node that - only *prefers* its device is left out — it runs here when nothing carries + only *prefers* its worker is left out — it runs here when nothing carries the label, so it is never what a run is waiting for. """ cpus, gpus, ram = 0, 0, 0 - device = "" + worker = "" for node in flow.nodes: wanted = node.resources if wanted is None or not _runs_on_a_worker(node): @@ -224,11 +224,11 @@ def required_resources(flow: FlowDef) -> dict[str, Any] | None: cpus = max(cpus, wanted.cpus) gpus = max(gpus, wanted.gpus) ram = max(ram, wanted.ram or 0) - if node.device and node.device_policy == "require": - device = node.device + if node.worker and node.worker_policy == "require": + worker = node.worker if not (cpus or gpus or ram): return None - return {"cpus": cpus, "gpus": gpus, "ram_mb": ram, "device": device} + return {"cpus": cpus, "gpus": gpus, "ram_mb": ram, "worker": worker} #: What a run's output is called from outside it: ``@run:.``. @@ -1310,6 +1310,9 @@ class RunService: run = session.get(Run, run_id) needed = list(run.labels) if run else [] needs = dict(run.needs) if run and run.needs else None + if needs and "device" in needs: + # Queued before the label was called `worker`. + needs["worker"] = needs.pop("device") hub = self.controller.remote available = hub.labels() | {w.name for w in hub.workers()} if hub else set() # A node that only prefers its label runs locally instead, so it is not diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index afe4397..2962f12 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -177,7 +177,7 @@ class NodeDef(BaseModel): "inherits the engine default." ), ) - device: str | None = Field( + worker: str | None = Field( default=None, description=( "Label of the worker this node's code must run on, such as 'gpu'. " @@ -185,10 +185,10 @@ class NodeDef(BaseModel): "attached worker carries waits rather than failing." ), ) - device_policy: Literal["require", "prefer"] = Field( + worker_policy: Literal["require", "prefer"] = Field( default="require", description=( - "What to do when no worker carries `device`: wait for one, or run " + "What to do when no worker carries `worker`: wait for one, or run " "locally anyway." ), ) diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index c239422..f2ea6b2 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -172,8 +172,10 @@ class FlowStore: """Rewrite node types and settings stored under a name they no longer have. On 2026-10-15 the node type ``trigger`` became ``debounce`` and the mqtt - setting ``publish_queue_size`` became ``send_queue_size``, with no alias - left for the old names. Each published and draft document is read as + setting ``publish_queue_size`` became ``send_queue_size``. On 2026-10-16 + a node's ``device`` and ``device_policy`` became ``worker`` and + ``worker_policy``, so that a device is only ever a paired one. No alias + is left for the old names. Each published and draft document is read as plain JSON and only the matching node changes, so a field this version does not model survives, and the old document stays in git. The version goes up by one, so an editor still holding the old document gets a @@ -192,25 +194,36 @@ class FlowStore: if n.get("type") == "mqtt" and "publish_queue_size" in (n.get("params") or {}) ] + placed = [ + n for n in data["nodes"] if "device" in n or "device_policy" in n + ] except Exception: # Unreadable to the engine too, which reports it when it loads. continue - if not nodes and not queues: + if not nodes and not queues and not placed: continue for node in nodes: node["type"] = "debounce" for params in queues: params["send_queue_size"] = params.pop("publish_queue_size") + for node in placed: + for old in ("device", "device_policy"): + if old in node: + node[old.replace("device", "worker")] = node.pop(old) data["version"] = data.get("version", 1) + 1 path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") changed.append(f"{path.parent.name}/{path.name}") if changed: logger.warning( - "Renamed node type 'trigger' to 'debounce' and mqtt " - "'publish_queue_size' to 'send_queue_size' in %s", + "Renamed node type 'trigger' to 'debounce', mqtt " + "'publish_queue_size' to 'send_queue_size' and a node's " + "'device' to 'worker' in %s", ", ".join(changed), ) - self._commit("Rename 'trigger' nodes and mqtt 'publish_queue_size'") + self._commit( + "Rename 'trigger' nodes, mqtt 'publish_queue_size' and a " + "node's 'device'" + ) # ------------------------------------------------------------------------- # git diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index ff6902e..f40b9cd 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -398,7 +398,7 @@ class Run(SQLModel, table=True): status_reason: str = Field(default="", max_length=1024) #: Worker labels its nodes need, so a run with nowhere to go can say so. labels: list[str] = Field(sa_column=Column(JSON), default_factory=list) - #: The largest machine its nodes ask for, as cpus, gpus, ram_mb and device. + #: The largest machine its nodes ask for, as cpus, gpus, ram_mb and worker. #: Read while the run is queued, to tell waiting for a machine from having #: nowhere to run at all. needs: dict[str, Any] | None = Field(sa_column=Column(JSON), default=None) diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index 888edf8..2a64d95 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -193,8 +193,8 @@ class NodeSpec: settings: dict[str, Any], title: str, timeout: float | None, - device: str | None, - device_policy: str, + worker: str | None, + worker_policy: str, cache: bool = True, resources: dict[str, Any] | None = None, ) -> None: @@ -206,8 +206,8 @@ class NodeSpec: self.settings = settings self.title = title self.timeout = timeout - self.device = device - self.device_policy = device_policy + self.worker = worker + self.worker_policy = worker_policy self.cache = cache self.resources = resources @@ -238,8 +238,8 @@ class NodeSpec: settings={**self.settings, **settings}, title=self.title, timeout=self.timeout, - device=self.device, - device_policy=self.device_policy, + worker=self.worker, + worker_policy=self.worker_policy, cache=self.cache, resources=self.resources, ) @@ -270,8 +270,8 @@ def node( settings: dict[str, Any] | None = None, title: str = "", timeout: float | None = None, - device: str | None = None, - device_policy: str = "require", + worker: str | None = None, + worker_policy: str = "require", cache: bool = True, resources: dict[str, Any] | str | None = None, ) -> Callable[[F], F]: @@ -282,8 +282,8 @@ def node( shorthand for a ``float`` port of that name. Parameters with a default that are not ports become the node's settings, so - the canvas can tune them without editing code. ``device`` picks the worker - the code runs on — ``"gpu"``, say — and ``device_policy="prefer"`` runs it + the canvas can tune them without editing code. ``worker`` picks the worker + the code runs on — ``"gpu"``, say — and ``worker_policy="prefer"`` runs it locally when no such worker is attached rather than waiting for one. A batch run skips this node when an earlier one already ran the same source @@ -312,8 +312,8 @@ def node( thing as ``{"flavor": "gpu-small"}`` and stands in for cpus, gpus and ram. Sizes accept ``"2G"``; ``duration_s`` accepts ``"30m"``. """ - if device_policy not in ("require", "prefer"): - raise SyncError("device_policy is 'require' or 'prefer'") + if worker_policy not in ("require", "prefer"): + raise SyncError("worker_policy is 'require' or 'prefer'") if isinstance(resources, str): resources = {"flavor": resources} if resources is not None: @@ -341,8 +341,8 @@ def node( settings=dict(settings or {}), title=title, timeout=timeout, - device=device, - device_policy=device_policy, + worker=worker, + worker_policy=worker_policy, cache=cache, resources=dict(resources) if resources is not None else None, ) @@ -547,8 +547,8 @@ def _check(spec: NodeSpec, mode: str) -> dict[str, Any]: "requires": [port.spec() for port in spec.requires], "provides": [port.spec() for port in spec.provides], "timeout": spec.timeout, - "device": spec.device, - "device_policy": spec.device_policy, + "worker": spec.worker, + "worker_policy": spec.worker_policy, "cache": spec.cache, "resources": spec.resources, } diff --git a/backend/tests/flow/test_drafts.py b/backend/tests/flow/test_drafts.py index 03d7d58..950ad05 100644 --- a/backend/tests/flow/test_drafts.py +++ b/backend/tests/flow/test_drafts.py @@ -125,8 +125,8 @@ def test_a_published_node_with_no_body_is_reported(store: FlowStore): assert [i for i in controller.issues if i.code == "missing_source"] == [] -def test_a_device_on_a_node_without_code_is_reported(store: FlowStore): - """Only code travels to a worker, so the device would be ignored silently.""" +def test_a_worker_on_a_node_without_code_is_reported(store: FlowStore): + """Only code travels to a worker, so the worker would be ignored silently.""" store.write_flow( FlowDef( name="heating", @@ -134,7 +134,7 @@ def test_a_device_on_a_node_without_code_is_reported(store: FlowStore): NodeDef( id="relay", type="change", - device="gpu", + worker="gpu", provides=[MessageSpec(name="temp")], ) ], @@ -143,7 +143,7 @@ def test_a_device_on_a_node_without_code_is_reported(store: FlowStore): controller = FlowController(store) asyncio.run(controller.reload()) - (issue,) = [i for i in controller.issues if i.code == "ignored_device"] + (issue,) = [i for i in controller.issues if i.code == "ignored_worker"] assert issue.node == "heating.relay" assert issue.advisory diff --git a/backend/tests/flow/test_placement.py b/backend/tests/flow/test_placement.py index 27705ed..fe2b61e 100644 --- a/backend/tests/flow/test_placement.py +++ b/backend/tests/flow/test_placement.py @@ -108,7 +108,7 @@ def test_a_required_label_waits_for_its_machine_rather_than_running_here(loop): placer = placer_over(hub) worker, _ = attach(hub, loop, labels=("gpu",)) - with placer.claim(Resources(cpus=2), device="gpu", policy="require") as ( + with placer.claim(Resources(cpus=2), label="gpu", policy="require") as ( target, _, ): @@ -120,7 +120,7 @@ def test_a_preferred_label_falls_back_here_and_is_still_accounted(loop): hub = RemoteWorkerHub() placer = placer_over(hub) - with placer.claim(Resources(cpus=3), device="gpu", policy="prefer") as ( + with placer.claim(Resources(cpus=3), label="gpu", policy="prefer") as ( target, allocation, ): diff --git a/backend/tests/flow/test_provision.py b/backend/tests/flow/test_provision.py index 966d54f..33d0405 100644 --- a/backend/tests/flow/test_provision.py +++ b/backend/tests/flow/test_provision.py @@ -231,7 +231,7 @@ def test_a_run_held_before_it_starts_still_gets_a_machine_asked_for(monkeypatch) hpc = cluster() placer = Placer(local=ResourceAccountant(cpus=4, gpus=0)) placer.provisioners = [hpc] - needs = {"cpus": 2, "gpus": 1, "ram_mb": 0, "device": ""} + needs = {"cpus": 2, "gpus": 1, "ram_mb": 0, "worker": ""} assert placer.satisfiable(needs) == "1 gpu(s) and 2 cpu(s)" placer.provision_for(needs) diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index b36a9d7..f1abf65 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -378,23 +378,23 @@ def test_the_digest_identifies_the_inputs_not_their_order(): assert digest_of({"a": 1}, 3) != digest_of({"a": 1}, 4) -def test_labels_come_from_the_nodes_that_ask_for_a_device(): +def test_labels_come_from_the_nodes_that_ask_for_a_worker(): flow = double_flow() assert required_labels(flow) == [] - flow.nodes[0].device = "gpu" + flow.nodes[0].worker = "gpu" assert required_labels(flow) == ["gpu"] -def test_a_preferred_device_does_not_hold_a_run_back(): +def test_a_preferred_worker_does_not_hold_a_run_back(): flow = double_flow() - flow.nodes[0].device = "gpu" - flow.nodes[0].device_policy = "prefer" + flow.nodes[0].worker = "gpu" + flow.nodes[0].worker_policy = "prefer" # It runs on the engine when no such worker is attached, so waiting for # one would be waiting for something the run does not need. assert required_labels(flow) == [] -def test_a_device_on_a_node_no_worker_could_run_is_not_waited_for(): +def test_a_worker_on_a_node_no_worker_could_run_is_not_waited_for(): """A connector is an entry point loaded here, so a worker cannot take it. Holding the run for one is a run that never starts — and it looked exactly @@ -402,7 +402,7 @@ def test_a_device_on_a_node_no_worker_could_run_is_not_waited_for(): """ flow = double_flow() flow.nodes[0].type = "mqtt" - flow.nodes[0].device = "gpu" + flow.nodes[0].worker = "gpu" assert required_labels(flow) == [] @@ -416,7 +416,7 @@ def test_what_a_run_needs_is_the_largest_single_machine_it_asks_for(): "cpus": 8, "gpus": 1, "ram_mb": 0, - "device": "", + "worker": "", } diff --git a/backend/tests/flow/test_store.py b/backend/tests/flow/test_store.py index a5e37d3..6fee24d 100644 --- a/backend/tests/flow/test_store.py +++ b/backend/tests/flow/test_store.py @@ -1,3 +1,4 @@ +import json import os import subprocess from datetime import UTC, datetime @@ -173,3 +174,22 @@ def test_opening_the_store_renames_the_mqtt_publish_queue_size(store: FlowStore) assert published.nodes[0].params == {"send_queue_size": 8} assert published.version == 2 assert commit_count(store) == commits + 1 + + +def test_opening_the_store_renames_a_nodes_device_to_worker(store: FlowStore): + """`device` names a paired device now, so a node's worker label moves over.""" + store.write_flow(FlowDef(name="lab", nodes=[NodeDef(id="fit")])) + path = store.root / "lab" / "flow.json" + data = json.loads(path.read_text()) + node = data["nodes"][0] + del node["worker"], node["worker_policy"] + node.update(device="gpu", device_policy="prefer") + path.write_text(json.dumps(data)) + + published = FlowStore(store.root).read_flow("lab") + + assert (published.nodes[0].worker, published.nodes[0].worker_policy) == ( + "gpu", + "prefer", + ) + assert published.version == 2 diff --git a/backend/tests/sdk/test_build.py b/backend/tests/sdk/test_build.py index 672bc5b..65e1178 100644 --- a/backend/tests/sdk/test_build.py +++ b/backend/tests/sdk/test_build.py @@ -17,8 +17,8 @@ def prepare(limit=8): @node( requires=["dataset", Port("lr", "float")], provides=[Port("loss", "float", stream=True), Port("weights", "artifact")], - device="gpu", - device_policy="prefer", + worker="gpu", + worker_policy="prefer", ) def fit(dataset, lr, epochs=3): for step in range(epochs): @@ -68,12 +68,12 @@ def test_defaults_become_settings_and_ports_do_not(): assert {p.port for p in nodes["fit"].requires} == {"dataset", "lr"} -def test_device_travels_to_the_node(): +def test_worker_travels_to_the_node(): definition = FlowDef.model_validate(a_flow().document()) node_def = next(n for n in definition.nodes if n.id == "fit") - assert node_def.device == "gpu" - assert node_def.device_policy == "prefer" + assert node_def.worker == "gpu" + assert node_def.worker_policy == "prefer" def test_a_required_port_takes_the_type_it_is_provided_as(): diff --git a/compose.yml b/compose.yml index 8681ccf..3b1b490 100644 --- a/compose.yml +++ b/compose.yml @@ -23,7 +23,7 @@ services: volumes: - data:/data environment: - # The address people and devices reach this instance at. Panel links and + # The address people and devices reach this instance at. Device links and # mails are built from it. FRONTEND_HOST: http://localhost:8080 # Schedules run in this timezone, such as Europe/Berlin. diff --git a/development.md b/development.md index 64fb790..d8b2f71 100644 --- a/development.md +++ b/development.md @@ -126,7 +126,7 @@ HTML report lands in `backend/htmlcov`. Nothing has to be running — mail is pa `tests/`, and the flow engine falls back to `MemoryState` while `REDIS_HOST` is empty. `tests/__init__.py` pins the suite's own environment before anything imports the settings: a SQLite file in a temp directory (deleted per session, so a run never touches development -data), its own superuser, and `ENVIRONMENT=local` with `DOMAIN=localhost`, so a checkout +data), its own superuser, and `ENVIRONMENT=local`, so a checkout configured for a deployment cannot drag that deployment's configuration into the run. **Frontend (e2e).** `make test-frontend` runs the suite against a stack of its own, because @@ -155,9 +155,9 @@ in once into `playwright/.auth/user.json`, which the browser projects reuse; `mo `mobile.spec.ts` only, on a Pixel 5 at 393px. The origin comes from `PLAYWRIGHT_BASE_URL`, with the API on it unless `PLAYWRIGHT_API_URL` -names another, deliberately never from `VITE_API_URL` — that one belongs to the app build and in a deployment checkout it -names the deployment, which would mean driving a browser at the local stack while sending -teardown `DELETE`s to the live instance. `tests/guard.ts` runs first and refuses the whole +names another, deliberately never from `.env`. In a deployment checkout that file names the +deployment, which would mean driving a browser at the local stack while sending teardown +`DELETE`s to the live instance. `tests/guard.ts` runs first and refuses the whole run when either origin resolves outside loopback or the private ranges, because the suite creates and deletes flows, dashboards and users. `PLAYWRIGHT_ALLOW_PUBLIC=1` overrides it. diff --git a/docker/compose.lan.yml b/docker/compose.lan.yml index a652b29..3846cc6 100644 --- a/docker/compose.lan.yml +++ b/docker/compose.lan.yml @@ -10,7 +10,7 @@ services: ports: !override - "${APP_PORT:-8080}:8000" environment: - # The address the instance gives out: the link the Panels dialog shows a + # The address the instance gives out: the link a device's page shows a # screen and the one its emails carry. app.${DOMAIN} resolves on this host # only, so `make dev-lan` passes the published port on the LAN address. - FRONTEND_HOST=${LAN_URL:-http://app.${DOMAIN:-localhost}} diff --git a/docs/code/agents.md b/docs/code/agents.md index 5876a73..9686821 100644 --- a/docs/code/agents.md +++ b/docs/code/agents.md @@ -18,8 +18,8 @@ issuer of the client's tokens. The client registers, opens browser authorization, and receives an OAuth token after you approve access. The issuer and browser redirects use -`DOMAIN`, `ENVIRONMENT`, and `FRONTEND_HOST`, so configure those for the -addresses your client and browser can reach. +`FRONTEND_HOST` and `ENVIRONMENT`, so configure those for the addresses your +client and browser can reach. Manage registered clients under **Admin → Agents**. Revoking a client invalidates its access. MCP uses a separate signing key at diff --git a/docs/code/api.md b/docs/code/api.md index 1a27267..0486087 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -195,6 +195,7 @@ on its own, so this is for reclaiming space sooner. | `POST` | `/devices/pair` | start device pairing, optionally with `{name, board, firmware}`; unauthenticated | | `GET` | `/devices/pair/{code}` | poll pairing with its `secret`; answers `device` (and `panel` for older firmware); unauthenticated | | `GET` | `/devices/pending` | devices waiting for a code, as they describe themselves | +| `GET` | `/devices/pair/{code}/device` | what is waiting on a code, before it is approved | | `POST` | `/devices/{id}/pair` | adopt by `{code}`, or by `{request}` from the pending list for a request from your network | | `POST` | `/devices/{id}/unpair` | revoke the device credential | diff --git a/docs/code/cli.md b/docs/code/cli.md index 42454b3..8812c73 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -340,7 +340,8 @@ The main files and directories are: |---|---| | `client.json` | Local API URL and token | | `fluksio.db` | SQLite database: accounts, runs, metrics and operations history | -| `flows/` | Git repository of flows, node sources, dashboards and module requirements | +| `flows/` | Git repository of flows, node sources, dashboards, devices (`_devices/`) and module requirements | +| `devices.json` | Device credential nonces, outside the flow repository | | `artifacts/` | Files stored by content digest | | `user-venv/` | Managed node environment, when used | | `secrets.enc` | Encrypted credentials, outside the flow repository | diff --git a/docs/code/nodes.md b/docs/code/nodes.md index 5597234..1332173 100644 --- a/docs/code/nodes.md +++ b/docs/code/nodes.md @@ -324,10 +324,10 @@ waiting for that deadline. ## Running a node somewhere else -Set a device label to select an attached worker: +Set a worker label to select an attached worker: ```json -{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200} +{"id": "train", "worker": "gpu", "worker_policy": "require", "timeout": 7200} ``` `require` is the default and waits for a worker with that label. `prefer` uses diff --git a/docs/code/workers.md b/docs/code/workers.md index 0c456bc..ad2a643 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -76,16 +76,16 @@ Set these fields on a node through `PUT /api/v1/flows/{name}`: ```json { "id": "train", - "device": "gpu", - "device_policy": "require", + "worker": "gpu", + "worker_policy": "require", "timeout": 7200 } ``` This is a node fragment; keep the node's other fields and the rest of the flow -when updating it. `device` and `device_policy` are not in the node panel yet. +when updating it. `worker` and `worker_policy` are not in the node panel yet. -| `device_policy` | If no matching worker is attached | +| `worker_policy` | If no matching worker is attached | |---|---| | `require` (default) | queues until a matching worker is available | | `prefer` | falls back to the engine | diff --git a/docs/concepts/flows.md b/docs/concepts/flows.md index e0e43c5..8305248 100644 --- a/docs/concepts/flows.md +++ b/docs/concepts/flows.md @@ -137,7 +137,7 @@ The canvas reports issues on the affected nodes: | `missing_source` | a Function node has no saved code | | `node_error` | code or configuration could not load | | `unauthenticated_hook` | a webhook or served websocket has no shared secret; advisory | -| `ignored_device` | a node other than a Function node names a device, which nothing reads; advisory | +| `ignored_worker` | a node other than a Function node names a worker, which nothing reads; advisory | | `unknown_device` | a Device node names no device, or one that does not exist | | `device_port` | a device port is wired in two places, or against the device's report the wrong way or type; a port the report does not list is advisory | | `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection, or a device being offline | diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index ecee531..6580940 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -157,7 +157,7 @@ artifact and rebuild the object in the next node. ## Running a node somewhere else Attach a [remote worker](../code/workers.md) to the engine and declare the -worker's label as the node's `device`. `device_policy="require"` waits for +worker's label as the node's `worker`. `worker_policy="require"` waits for that label; `"prefer"` uses the local engine when none is attached. The selected machine needs your imported packages. Code-defined flows also diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 6ba487a..fe3cac3 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -322,10 +322,10 @@ face, and the editor marks it **Poor fit for round**. | Gesture | Effect | |---|---| | Swipe left / right | next / previous face of this dashboard; stops at either end | -| Swipe up / down | next / previous dashboard of the panel (`/panel/{id}` only); stops at either end | +| Swipe up / down | next / previous dashboard of the device (`/panel/{id}` only); stops at either end | | Page dot | that face | | PageDown / PageUp | next / previous face | -| Shift+PageDown / Shift+PageUp | next / previous dashboard of the panel | +| Shift+PageDown / Shift+PageUp | next / previous dashboard of the device | | Arrow keys | turn the ring, not the page | A swipe carries the page with the finger ([Swiping](#swiping)); a drag that @@ -423,7 +423,7 @@ angle, which is what the browser's reveal amounts to. ### Swiping -Faces are swiped sideways and a panel's dashboards up and down, with the page +Faces are swiped sideways and a device's dashboards up and down, with the page under the finger, as a tile view moves. | Part | Value | diff --git a/docs/interface/flow-editor.md b/docs/interface/flow-editor.md index 5c86556..64975b2 100644 --- a/docs/interface/flow-editor.md +++ b/docs/interface/flow-editor.md @@ -68,7 +68,7 @@ yielding regularly can run longer than its timeout. Empty or `0` means no limit, the default. **Resources** selects a shared worker pool, a named flavor or custom CPU, -memory and GPU requirements. Placement by `device` and `device_policy` is +memory and GPU requirements. Placement by `worker` and `worker_policy` is configured through the API or flow JSON; see [Remote workers](../code/workers.md). ### Code and reuse diff --git a/docs/interface/portal.md b/docs/interface/portal.md index e09364c..1e9afcc 100644 --- a/docs/interface/portal.md +++ b/docs/interface/portal.md @@ -92,7 +92,7 @@ proxy on that route must support WebSockets. the portal to be online. It also deletes the users the portal let in; the account that enrolled and every superuser stay. - To replace the pairing, use **Connect with a new code**. Remote users and - panels may need to pair again. + devices paired through the portal may need to pair again. If the instance is removed at the portal, or its code is renewed there, the status reads **Removed at the portal** and the instance stops dialling. Connect diff --git a/docs/llms.txt b/docs/llms.txt index 5e3b941..6ac2a53 100644 --- a/docs/llms.txt +++ b/docs/llms.txt @@ -31,7 +31,7 @@ - [The fluksio command](/code/cli/): serve, enroll, worker, and the data directory - [Writing node code](/code/nodes/): process(), generators, fluksio.emit, artifacts - [The HTTP API](/code/api/): auth, flows, messages, runs, observability, websocket -- [Remote workers](/code/workers/): fluksio-worker, device labels, artifacts across hosts +- [Remote workers](/code/workers/): fluksio-worker, worker labels, artifacts across hosts - [Agents over MCP](/code/agents/): the tool list and the security model - [Writing a connector](/code/connectors/): packaging a device integration as a node type diff --git a/examples/myresearch/train.py b/examples/myresearch/train.py index 2761713..3de9c3b 100644 --- a/examples/myresearch/train.py +++ b/examples/myresearch/train.py @@ -24,8 +24,8 @@ def train_curve(lr: float, epochs: int) -> list[float]: Port("weights", "artifact"), Port("final_loss", "float"), ], - device="gpu", - device_policy="prefer", + worker="gpu", + worker_policy="prefer", timeout=600, resources={"cpus": 2}, ) @@ -34,8 +34,8 @@ def fit(dataset, lr, epochs=25): Yielding is the reporting: each one publishes on the `loss` port the instant it happens, and the run keeps every value as a series — which is - why there is no `log_metric()` to call. `device="gpu"` with - `device_policy="prefer"` sends this to a worker carrying that label when + why there is no `log_metric()` to call. `worker="gpu"` with + `worker_policy="prefer"` sends this to a worker carrying that label when one is attached, and runs it here when none is. `resources` is what a real training node needs and the rest of a flow does diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 69e6d65..f1767a3 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -2606,7 +2606,7 @@ export const NodeDef_InputSchema = { title: 'Timeout', description: "Seconds this node's code may be silent before it is stopped. A yield or an emit resets the clock, and the first call's imports are not charged to it. 0 disables the limit: the node runs until it finishes, and only a dead worker fails the call. Empty inherits the engine default." }, - device: { + worker: { anyOf: [ { type: 'string' @@ -2615,14 +2615,14 @@ export const NodeDef_InputSchema = { type: 'null' } ], - title: 'Device', + title: 'Worker', description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing." }, - device_policy: { + worker_policy: { type: 'string', enum: ['require', 'prefer'], - title: 'Device Policy', - description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', + title: 'Worker Policy', + description: 'What to do when no worker carries `worker`: wait for one, or run locally anyway.', default: 'require' }, cache: { @@ -2726,7 +2726,7 @@ export const NodeDef_OutputSchema = { title: 'Timeout', description: "Seconds this node's code may be silent before it is stopped. A yield or an emit resets the clock, and the first call's imports are not charged to it. 0 disables the limit: the node runs until it finishes, and only a dead worker fails the call. Empty inherits the engine default." }, - device: { + worker: { anyOf: [ { type: 'string' @@ -2735,14 +2735,14 @@ export const NodeDef_OutputSchema = { type: 'null' } ], - title: 'Device', + title: 'Worker', description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing." }, - device_policy: { + worker_policy: { type: 'string', enum: ['require', 'prefer'], - title: 'Device Policy', - description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', + title: 'Worker Policy', + description: 'What to do when no worker carries `worker`: wait for one, or run locally anyway.', default: 'require' }, cache: { @@ -4727,7 +4727,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_device', 'unknown_device', 'device_port'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source', 'node_unhealthy', 'ignored_worker', 'unknown_device', 'device_port'], title: 'Code' }, message: { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 22273f2..526e95c 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -887,11 +887,11 @@ export type NodeDef_Input = { /** * Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing. */ - device?: (string | null); + worker?: (string | null); /** - * What to do when no worker carries `device`: wait for one, or run locally anyway. + * What to do when no worker carries `worker`: wait for one, or run locally anyway. */ - device_policy?: 'require' | 'prefer'; + worker_policy?: 'require' | 'prefer'; /** * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. */ @@ -913,9 +913,9 @@ export type NodeDef_Input = { }; /** - * What to do when no worker carries `device`: wait for one, or run locally anyway. + * What to do when no worker carries `worker`: wait for one, or run locally anyway. */ -export type device_policy = 'require' | 'prefer'; +export type worker_policy = 'require' | 'prefer'; /** * A node as stored: identity, configuration and ports. @@ -941,11 +941,11 @@ export type NodeDef_Output = { /** * Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing. */ - device?: (string | null); + worker?: (string | null); /** - * What to do when no worker carries `device`: wait for one, or run locally anyway. + * What to do when no worker carries `worker`: wait for one, or run locally anyway. */ - device_policy?: 'require' | 'prefer'; + worker_policy?: 'require' | 'prefer'; /** * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. */ @@ -1514,7 +1514,7 @@ export type ValidationError = { * Something wrong with a flow — a fault, or merely advisory. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_worker' | 'unknown_device' | 'device_port'; message: string; flow?: string; nodes?: Array<(string)>; @@ -1524,7 +1524,7 @@ export type ValidationIssue = { advisory?: boolean; }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_device' | 'unknown_device' | 'device_port'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source' | 'node_unhealthy' | 'ignored_worker' | 'unknown_device' | 'device_port'; export type ValidationResult = { issues?: Array; diff --git a/frontend/src/components/Dashboard/settings.tsx b/frontend/src/components/Dashboard/settings.tsx index 0abd77f..c0f5ae1 100644 --- a/frontend/src/components/Dashboard/settings.tsx +++ b/frontend/src/components/Dashboard/settings.tsx @@ -23,7 +23,7 @@ import { isLight, parsePalette, type Roles, rolesOf } from "./ui/core/theme" * * A bound setting is read the way a widget reads a value — the same live store * and the same socket — so a panel's own settings messages are part of what - * its credential is entitled to (`flow/panels.py`, `messages_for`). + * its credential is entitled to (`flow/devices.py`, `messages_for`). */ /** diff --git a/frontend/tests/config.ts b/frontend/tests/config.ts index 028edf0..2ec50e8 100644 --- a/frontend/tests/config.ts +++ b/frontend/tests/config.ts @@ -22,10 +22,9 @@ export const firstSuperuserPassword = getEnvVar("FIRST_SUPERUSER_PASSWORD") * The two origins a run talks to: the app in the browser, and the API for the * setup and teardown around a spec. * - * Deliberately *not* taken from `VITE_API_URL`, even though `../../.env` above - * has one. That variable belongs to the app build, and in a checkout configured - * for a deployment it names the deployment — so a suite reading it would drive - * a browser at the local stack while sending its `DELETE`s to the live + * Deliberately *not* taken from `../../.env` above, which in a checkout + * configured for a deployment names the deployment. A suite reading it would + * drive a browser at the local stack while sending its `DELETE`s to the live * instance. Both origins come from the same place instead: whatever the browser * is pointed at is what teardown may write to. */ diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py index 3a428f2..2f79fc8 100644 --- a/worker/fluksio_worker/agent.py +++ b/worker/fluksio_worker/agent.py @@ -463,7 +463,7 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument( "--labels", default="", - help="comma-separated, e.g. gpu,cuda12 — what a node's device matches", + help="comma-separated, e.g. gpu,cuda12 — what a node's worker matches", ) parser.add_argument( "--python",