From fa2a213631a91af29b188aa064e43692b4930d5f Mon Sep 17 00:00:00 2001 From: stroblme Date: Mon, 2 Nov 2026 21:06:39 +0100 Subject: [PATCH] Rename planner and document rewrite: rewrite_documents moves qualified names of the renamed flow in every flow (ports, inputs, outputs) and dashboard binding (one key table, WIDGET_BINDINGS, shared with the WidgetDef accessors, legacy pages included), published and draft, as plain JSON with both variants above either version; plan_rename reports documents, counts, fingerprint, warnings and refusals; superuser plan routes for flows and dashboards; key residue classifies state as the mapper does; duplicate retargets inputs and outputs too; queue reports its Redis version and other consumers --- backend/fluksio/api/routes/dashboards.py | 34 ++ backend/fluksio/api/routes/flows.py | 36 ++ backend/fluksio/flow/controller.py | 26 +- backend/fluksio/flow/dashboards.py | 30 + backend/fluksio/flow/namespace.py | 68 ++- backend/fluksio/flow/queue.py | 32 + backend/fluksio/flow/rename.py | 579 +++++++++++++++++- backend/fluksio/flow/store.py | 20 +- backend/tests/flow/test_drafts.py | 12 +- backend/tests/flow/test_rename_runtime.py | 17 + backend/tests/test_rename_plan.py | 687 ++++++++++++++++++++++ frontend/src/client/schemas.gen.ts | 271 +++++++++ frontend/src/client/sdk.gen.ts | 57 +- frontend/src/client/types.gen.ts | 94 +++ 14 files changed, 1922 insertions(+), 41 deletions(-) create mode 100644 backend/tests/test_rename_plan.py diff --git a/backend/fluksio/api/routes/dashboards.py b/backend/fluksio/api/routes/dashboards.py index 2c1a184..05e8bea 100644 --- a/backend/fluksio/api/routes/dashboards.py +++ b/backend/fluksio/api/routes/dashboards.py @@ -16,8 +16,10 @@ from fluksio.api.deps import ( DashboardStoreDep, FlowControllerDep, check_pin, + get_current_active_superuser, get_current_user, ) +from fluksio.api.routes.alerts import read_config as read_alerts_config from fluksio.flow.dashboards import ( DashboardCreate, DashboardDef, @@ -28,6 +30,7 @@ from fluksio.flow.dashboards import ( results_dashboard, ) from fluksio.flow.events import event_bus +from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename from fluksio.flow.store import FlowNotFound, IdTaken, StaleVersion from fluksio.models import Message @@ -245,3 +248,34 @@ async def duplicate_dashboard(dashboard_id: str, store: DashboardStoreDep) -> An return await run_in_threadpool(store.duplicate, dashboard_id) except DashboardNotFound: raise _missing(dashboard_id) + + +@router.post( + "/{dashboard_id}/rename/plan", + response_model=RenamePlan, + dependencies=[Depends(get_current_active_superuser)], +) +async def plan_dashboard_rename( + dashboard_id: str, + body: RenamePlanRequest, + store: DashboardStoreDep, + controller: FlowControllerDep, +) -> Any: + """What renaming this dashboard's key would change, and what refuses it. + + Changes nothing. Nothing binds to a dashboard's key, so the plan is the + dashboard alone; a superuser's all the same, as a flow's is. + """ + try: + return await run_in_threadpool( + lambda: plan_rename( + controller, + store, + "dashboard", + dashboard_id, + body.new_name, + read_alerts_config(), + ) + ) + except DashboardNotFound: + raise _missing(dashboard_id) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 93a3be6..cf20e7d 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -32,6 +32,7 @@ from fluksio.api.deps import ( CAMERA_IS_SUPERUSERS, CODE_IS_SUPERUSERS, CurrentUser, + DashboardStoreDep, FlowControllerDep, HiddenDep, SessionDep, @@ -39,10 +40,12 @@ from fluksio.api.deps import ( check_pin, decode_token, get_code_author, + get_current_active_superuser, get_current_user, hidden_from, user_from_token, ) +from fluksio.api.routes.alerts import read_config as read_alerts_config from fluksio.api.routes.messages import port_source from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token from fluksio.core.config import settings @@ -66,6 +69,7 @@ from fluksio.flow.events import event_bus from fluksio.flow.messages import MEDIA_FAMILIES, DType, flow_of, qualify from fluksio.flow.nodes.wren import TRAIN_FLOW, WrenNode, pack_meta from fluksio.flow.pipeline import EngineHeld, ValidationIssue +from fluksio.flow.rename import RenamePlan, RenamePlanRequest, plan_rename from fluksio.flow.runs import RunRejected from fluksio.flow.schemas import ( NAME_PATTERN, @@ -740,6 +744,38 @@ async def duplicate_flow( return await run_in_threadpool(_detail, controller, copy) +@router.post( + "/{flow_id}/rename/plan", + response_model=RenamePlan, + dependencies=[Depends(get_current_active_superuser)], +) +async def plan_flow_rename( + flow_id: str, + body: RenamePlanRequest, + controller: FlowControllerDep, + dashboards: DashboardStoreDep, +) -> Any: + """What renaming this flow's key would change, and what refuses it. + + Changes nothing. A superuser's, as the rename is: the key names the + flow's messages, and renaming it rewrites every flow and dashboard that + names one, its history and its live state. + """ + try: + return await run_in_threadpool( + lambda: plan_rename( + controller, + dashboards, + "flow", + flow_id, + body.new_name, + read_alerts_config(), + ) + ) + except FlowNotFound: + raise _missing(flow_id) + + # ----------------------------------------------------------------------------- # Node source # ----------------------------------------------------------------------------- diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 963141a..981f9c5 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -39,6 +39,7 @@ from fluksio.flow.demand import demand from fluksio.flow.events import EventBus from fluksio.flow.executor import ExecutionService from fluksio.flow.messages import MessageSpec, flow_of, qualify +from fluksio.flow.namespace import touches from fluksio.flow.nodes import ( RESERVED_SETTINGS, AssistantNode, @@ -1894,20 +1895,21 @@ class FlowController: if self.execution is not None: self.execution.queue.clear_flow(flow) with self.state.lock(): - for key in [k for k in self.state.keys() if _in_namespace(k, flow)]: + for key in [k for k in self.state.keys() if touches(flow, k)]: self.state.delete(key) def key_residue(self, flow: str) -> list[str]: """What still lives under the flow key *flow*. Blocking. - State and live queue work in its namespace: left by a flow that held - the key before, or put there by anything still writing to it. A new - flow taking the key would inherit it, so creating one is refused while - this is not empty, and so is renaming a flow onto it. Work that was - acknowledged, dead-lettered or marked done is history and not counted. - Names what it found, state keys first. + State and live queue work in its namespace, judged as a rename maps + it (`namespace.touches`): left by a flow that held the key before, or + put there by anything still writing to it. A new flow taking the key + would inherit it, so creating one is refused while this is not empty, + and so is renaming a flow onto it. Work that was acknowledged, + dead-lettered or marked done is history and not counted. Names what it + found, state keys first. """ - found = [key for key in self.state.keys() if _in_namespace(key, flow)] + found = [key for key in self.state.keys() if touches(flow, key)] if self.execution is not None: found += [ f"queued work of '{item.node}'" @@ -2010,14 +2012,6 @@ class FlowController: self.events.publish(event) -def _in_namespace(state_key: str, flow: str) -> bool: - """Whether a state key is about the flow key *flow*: one of its messages, - or the engine's own bookkeeping on one of its messages or nodes, which is - keyed by those names too.""" - prefix = f"{flow}." - return state_key.startswith(prefix) or f":{prefix}" in state_key - - def _touches(item: WorkItem, flow: str) -> bool: """Whether a queued item names anything of the flow key *flow*: as its flow, its node or one of the messages it carries.""" diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 8a9b252..26c7efb 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/backend/fluksio/flow/dashboards.py @@ -219,6 +219,36 @@ class WidgetFootprint(Placement): type: WidgetType +#: Where a widget's config names a message, as paths into it +#: (`namespace.rebind`), each with the accessor of `WidgetDef` that reads it. A +#: rename rewrites exactly these, and leaves titles, labels and every other +#: config string alone, so a binding the accessors read and this does not list +#: would keep its old name. A registry test holds the two to each other. +WIDGET_BINDINGS: tuple[tuple[str, ...], ...] = ( + ("message",), # `messages`, and a bar's first row (`bar_rows`) + ("target",), # `target` + ("request",), # `target`, of a querying chart + ("series", "*", "message"), # `messages`, of a chart + ("rows", "*", "message"), # `bar_rows` + ("inner",), # `inner_bindings`, as one name + ("inner", "*", "message"), # `inner_bindings`, as a list + ("reading", "message"), # `reading` + ("runs", "metric"), # a chart of runs: a recorded series, named as a message +) + +#: The same for a whole stored document: the bound settings +#: (`DashboardDef.setting_messages`), the widgets, and the widgets of a +#: document still written as pages and sections (`_flatten_pages`). +DASHBOARD_BINDINGS: tuple[tuple[str, ...], ...] = ( + ("settings", "*", "message"), + *(("widgets", "*", "config", *path) for path in WIDGET_BINDINGS), + *( + ("pages", "*", "sections", "*", "widgets", "*", "config", *path) + for path in WIDGET_BINDINGS + ), +) + + class WidgetDef(BaseModel): """One tile: what it shows or does, and where it sits. diff --git a/backend/fluksio/flow/namespace.py b/backend/fluksio/flow/namespace.py index e740232..3110ce7 100644 --- a/backend/fluksio/flow/namespace.py +++ b/backend/fluksio/flow/namespace.py @@ -9,12 +9,17 @@ the new name, so every family is registered here, beside the builders that take their prefix from it, and a key of a family that is not is refused rather than left behind. +Documents name messages too: a flow its ports, inputs and outputs, a dashboard +its bindings. Where they do is registered beside the paths into them +(`FLOW_FIELDS`, `dashboards.DASHBOARD_BINDINGS`), and `rebind` rewrites them. + Nothing here does I/O. It maps names; moving the keys and items is the state -and queue backends' work. +and queue backends' work, and writing documents the rename's. """ from __future__ import annotations +from collections.abc import Callable, Iterable from dataclasses import dataclass, replace from typing import TYPE_CHECKING, Any @@ -87,6 +92,62 @@ ITEM_FIELDS: dict[str, str | None] = { "parent": None, } +#: Where a flow document names messages, as paths into its JSON (`rebind`). +#: A bare name is the flow's own and follows its key by itself, so a rename +#: moves only the qualified ones: those the flow names in another's +#: namespace, or explicitly in its own. +FLOW_FIELDS: tuple[tuple[str, ...], ...] = ( + ("nodes", "*", "requires", "*", "name"), + ("nodes", "*", "provides", "*", "name"), + ("inputs", "*", "spec", "name"), + ("outputs", "*"), +) + + +def rebind( + document: Any, paths: Iterable[tuple[str, ...]], rename: Callable[[str], str] +) -> int: + """Pass every string *paths* lead to in *document* through *rename*, in + place, and count those that changed. + + A path is a key per level, or ``*`` for every item of a list or value of + an object. One that leads nowhere, or to anything but a string, is passed + over, so a document missing a field or holding an older shape of it is + read as far as it goes. + """ + return sum(_rebind(document, path, rename) for path in paths) + + +def names(document: Any, paths: Iterable[tuple[str, ...]]) -> list[str]: + """Every string *paths* lead to in *document* (`rebind`).""" + found: list[str] = [] + + def keep(name: str) -> str: + found.append(name) + return name + + rebind(document, paths, keep) + return found + + +def _rebind(node: Any, path: tuple[str, ...], rename: Callable[[str], str]) -> int: + head, rest = path[0], path[1:] + keys: Iterable[Any] = () + if head == "*" and isinstance(node, list): + keys = range(len(node)) + elif head == "*" and isinstance(node, dict): + keys = list(node) + elif isinstance(node, dict) and head in node: + keys = (head,) + changed = 0 + for key in keys: + if rest: + changed += _rebind(node[key], rest, rename) + elif isinstance(node[key], str) and (moved := rename(node[key])) != node[key]: + node[key] = moved + changed += 1 + return changed + def _parts(key: str) -> tuple[str, list[tuple[str, str]]]: """A state key's family prefix and its parts, each with what it holds. @@ -129,6 +190,11 @@ class Rename: def flow(self, key: str) -> str: return self.new if key == self.old else key + def binding(self, name: str) -> str: + """A message name as a document holds it. A bare one is local to the + flow the document is, so only a qualified one moves.""" + return self.name(name) if "." in name else name + def state_key(self, key: str) -> str | None: """Where the state key *key* moves to, or None if it stays. diff --git a/backend/fluksio/flow/queue.py b/backend/fluksio/flow/queue.py index 6ad519c..2168f08 100644 --- a/backend/fluksio/flow/queue.py +++ b/backend/fluksio/flow/queue.py @@ -457,6 +457,17 @@ class WorkQueue(ABC): """Release a held list left by a process that stopped mid-rename. For startup, before anything claims. Returns how many items moved.""" + def server_version(self) -> tuple[int, int] | None: + """The major and minor version of the server the queue lives on, + None for one in this process. A rename rebuilds streams with commands + Redis has had since 7.0.""" + return None + + def other_consumers(self, within_ms: int) -> list[str]: + """Consumers but this one that read the queue in the last *within_ms*: + another engine on the same server, which a rename cannot hold.""" + return [] + @abstractmethod def rename( self, rename: Rename, progress: Callable[[str], None] | None = None @@ -1132,6 +1143,27 @@ class RedisWorkQueue(WorkQueue): "durable": True, } + def server_version(self) -> tuple[int, int] | None: + info = cast(dict[str, Any], self._redis.info("server")) + major, minor, *_ = str(info["redis_version"]).split(".") + return int(major), int(minor) + + def other_consumers(self, within_ms: int) -> list[str]: + found: set[str] = set() + for stream in (self._stream, self._due_stream): + try: + rows = cast( + list[dict[str, Any]], self._redis.xinfo_consumers(stream, GROUP) + ) + except redis.ResponseError: + continue + found |= { + str(row["name"]) + for row in rows + if row["name"] != self._consumer and int(row["idle"]) < within_ms + } + return sorted(found) + def dead_letters(self, count: int = 50) -> list[dict[str, Any]]: """What was set aside, newest first. The entry id carries the time.""" entries = cast( diff --git a/backend/fluksio/flow/rename.py b/backend/fluksio/flow/rename.py index 536f55d..9b4a4bf 100644 --- a/backend/fluksio/flow/rename.py +++ b/backend/fluksio/flow/rename.py @@ -1,23 +1,50 @@ -"""Renaming a flow's key: the door that keeps the API out while it runs. +"""Renaming a flow's or a dashboard's key. A rename moves everything named under a flow's key, in documents, history, live state and queued work, so nothing may write through the API meanwhile. -`RenameGate` is that door. ``docs/private/id_refac_rename.md`` has the whole -protocol. +`RenameGate` is the door that keeps it out. `plan_rename` says beforehand what +a rename would change and what refuses it, and `rewrite_documents` is its +document part. ``docs/private/id_refac_rename.md`` has the whole protocol. """ from __future__ import annotations +import hashlib +import json import re import threading from collections import Counter -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Literal, cast +import orjson +from pydantic import BaseModel +from sqlalchemy import String +from sqlmodel import Session, col, func, select from starlette.types import Scope from fluksio.core.config import settings -from fluksio.flow.controller import HOOK_PREFIX +from fluksio.core.db import engine as db_engine +from fluksio.flow.alerts import AlertsConfig +from fluksio.flow.controller import HOOK_PREFIX, FlowController +from fluksio.flow.dashboards import ( + DASHBOARD_BINDINGS, + DASHBOARD_DIR, + DashboardDef, + DashboardStore, + results_name, +) from fluksio.flow.executor import wait_out +from fluksio.flow.messages import flow_of +from fluksio.flow.namespace import FLOW_FIELDS, NAME, OUTPUTS, Rename, names, rebind +from fluksio.flow.schemas import FlowDef, _validate_name +from fluksio.flow.state import MemoryState +from fluksio.flow.store import ISSUED_ID, FlowStore, InvalidName, ReservedKey +from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode + +Kind = Literal["flow", "dashboard"] #: What reading looks like. Everything else under the API changes something. _READS = frozenset({"GET", "HEAD", "OPTIONS"}) @@ -120,3 +147,545 @@ class RenameGate: self._closed = False self._affected = frozenset() self.op = "" + + +# ----------------------------------------------------------------------------- +# Documents +# ----------------------------------------------------------------------------- + + +@dataclass +class Rewrite: + """A stored document a rename changes, as texts before and after.""" + + kind: Kind + id: str + name: str + published_version: int | None + draft_version: int | None + #: Bindings moved, in whichever variant moves more of them. + fields: int + #: Bindings moved in the draft, which stays unpublished. + draft_fields: int + #: Declared in code (`FlowDef.origin`), so a sync puts the old names back. + code_defined: bool + #: ``(path, before, after)`` per variant, the path relative to the store. + files: list[tuple[str, str, str]] + + +_VARIANTS = (("published", ".json"), ("draft", ".draft.json")) + + +def _documents( + store: FlowStore, dashboards: DashboardStore +) -> Iterator[tuple[Kind, str, str, tuple[tuple[str, ...], ...]]]: + """Every stored flow and dashboard: kind, id, path without its variant's + suffix, and where it names messages.""" + for flow_id in store.list_flows(): + yield "flow", flow_id, f"{flow_id}/flow", FLOW_FIELDS + found = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")} + for dash_id in sorted(found): + yield ( + "dashboard", + dash_id, + f"{DASHBOARD_DIR}/{dash_id}/dashboard", + DASHBOARD_BINDINGS, + ) + + +def _variants(root: Path, stem: str) -> dict[str, tuple[str, dict[str, Any]]]: + """A document's published and draft variants, as text and as JSON. + + Plain JSON rather than the model, so a field this version does not model, + or a dashboard still written as pages, comes through as it was. Empty for + one that does not parse, which the engine cannot read either. + """ + found: dict[str, tuple[str, dict[str, Any]]] = {} + for variant, suffix in _VARIANTS: + path = root / f"{stem}{suffix}" + if not path.exists(): + continue + text = path.read_text() + try: + data = orjson.loads(text) + except orjson.JSONDecodeError: + return {} + if not isinstance(data, dict): + return {} + found[variant] = (text, data) + return found + + +def rewrite_documents( + store: FlowStore, + dashboards: DashboardStore, + kind: Kind, + obj_id: str, + rename: Rename, +) -> list[Rewrite]: + """Every document the rename of *obj_id* changes, published and draft. + Reads only; writing the texts is the rename's. + + The object itself takes the new key. Renaming a flow also moves every + qualified name of its namespace that a flow names (`FLOW_FIELDS`, its own + included) or a dashboard binds (`DASHBOARD_BINDINGS`). Bare names are a + flow's own and follow its key by themselves, and no other field changes. + Both variants of a document that changes go to one version above either, + so an editor holding the old text is refused rather than saving it back, + and a draft discarded later cannot bring an older version back. + """ + rewrites = [] + for doc_kind, doc_id, stem, paths in _documents(store, dashboards): + own = doc_kind == kind and doc_id == obj_id + if kind == "dashboard" and not own: + continue + variants = _variants(store.root, stem) + if not variants: + continue + moved = { + variant: rebind(data, paths, rename.binding) if kind == "flow" else 0 + for variant, (_, data) in variants.items() + } + if not own and not any(moved.values()): + continue + versions = { + variant: int(data.get("version", 1)) + for variant, (_, data) in variants.items() + } + working = variants.get("draft") or variants["published"] + rewrite = Rewrite( + kind=doc_kind, + id=doc_id, + name=str(working[1].get("name", doc_id)), + published_version=versions.get("published"), + draft_version=versions.get("draft"), + fields=max(moved.values()), + draft_fields=moved.get("draft", 0), + code_defined=any(data.get("origin") for _, data in variants.values()), + files=[], + ) + top = max(versions.values()) + 1 + for variant, suffix in _VARIANTS: + if variant not in variants: + continue + text, data = variants[variant] + if own: + data["name"] = rename.new + data["version"] = top + # The stores write what pydantic writes, which this matches. + after = orjson.dumps(data, option=orjson.OPT_INDENT_2).decode() + if text.endswith("\n"): + after += "\n" + rewrite.files.append((f"{stem}{suffix}", text, after)) + rewrites.append(rewrite) + return rewrites + + +def references(store: FlowStore, dashboards: DashboardStore, key: str) -> list[str]: + """The flows and dashboards that name a message of the flow key *key*, + published or draft, by id. + + A rename onto *key* would hand those bindings to the renamed flow, and a + rename back would take them along. Creating a flow under *key* is another + matter: that is how bindings declared ahead of their flow are met. + """ + found = [] + for _, doc_id, stem, paths in _documents(store, dashboards): + bound = ( + name + for _, data in _variants(store.root, stem).values() + for name in names(data, paths) + ) + if any("." in name and flow_of(name) == key for name in bound): + found.append(doc_id) + return found + + +# ----------------------------------------------------------------------------- +# Plan +# ----------------------------------------------------------------------------- + +#: An alert's dedup key (`alerts.dedup_key`), whose tail is a node or a flow. +DEDUP = "dedup" + +#: Where the history names messages, by table and column: a qualified name +#: (`namespace.NAME`), a JSON object keyed by them (`namespace.OUTPUTS`) or an +#: alert's dedup key. A rename rewrites exactly these. Every other column +#: holds ids, which never change, or payload and free text, which are history, +#: and a registry test fails when a column is added without saying which. +SQL_NAMES: dict[tuple[str, str], str] = { + ("run_metric", "name"): NAME, + ("run_artifact", "name"): NAME, + ("run_node", "outputs"): OUTPUTS, + ("alert_sent", "key"): DEDUP, +} + +#: How recently another consumer must have read the queue to count as an +#: engine still running: it blocks a second at a time. +OTHER_ENGINE_MS = 10_000 + +#: Run states a rename refuses: the run would read names mid-move. +_ACTIVE_RUNS = ("queued", "running") + + +class RenamePlanRequest(BaseModel): + """The key a flow or dashboard would be renamed to.""" + + new_name: str + + +class RenameRefusal(BaseModel): + """Why a rename cannot go ahead, with the ids of what is in the way.""" + + code: str + detail: str + ids: list[str] = [] + + +class RenameWarning(BaseModel): + """Something a rename changes, or leaves, that its caller should know.""" + + code: str + detail: str + + +class PlannedDocument(BaseModel): + """A document a rename rewrites, and the versions it was planned against.""" + + kind: Kind + id: str + name: str + published_version: int | None + draft_version: int | None + #: Bindings moved, in whichever variant moves more of them. + fields: int + + +class AlertCounts(BaseModel): + channels: int + ledger_rows: int + + +class HistoryCounts(BaseModel): + runs: int + metric_rows: int + artifact_rows: int + output_rows: int + + +class RuntimeCounts(BaseModel): + #: False with memory backends, whose state and queue a restart loses anyway. + durable: bool + state_keys: int + queue_items: int + delayed: int + parked: int + done_markers: int + + +class RenamePlan(BaseModel): + """What renaming a flow's or dashboard's key would change, and whether it + may. Read-only; the rename request repeats ``version`` and + ``fingerprint``, and is refused when either moved since.""" + + kind: Kind + id: str + name: str + new_name: str + #: The working copy's version, the draft's when there is one. + version: int + #: Over the kind, id and versions of every listed document and both + #: keys, so a document changing between plan and rename is noticed. + fingerprint: str = "" + allowed: bool = False + refusals: list[RenameRefusal] = [] + documents: list[PlannedDocument] = [] + #: Dry-run counts, a flow's only, which inform and are not checked again. + alerts: AlertCounts | None = None + history: HistoryCounts | None = None + runtime: RuntimeCounts | None = None + warnings: list[RenameWarning] = [] + + +def plan_rename( + controller: FlowController, + dashboards: DashboardStore, + kind: Kind, + obj_id: str, + new_name: str, + alerts: AlertsConfig, +) -> RenamePlan: + """What renaming the flow or dashboard *obj_id* to *new_name* would change, + and what refuses it. Blocking, and changes nothing. + + :raises FlowNotFound: For a flow that does not exist. + :raises DashboardNotFound: For a dashboard that does not exist. + """ + store = controller.store + current: FlowDef | DashboardDef + if kind == "flow": + current = store.read_flow(obj_id, draft=True) + keys, ids = store.key_index(), set(store.list_flows()) + else: + current = dashboards.read(obj_id, draft=True) + keys = dashboards.key_index() + ids = {path.parent.name for path in dashboards.root.glob("*/dashboard*.json")} + old = current.name + plan = RenamePlan( + kind=kind, id=obj_id, name=old, new_name=new_name, version=current.version + ) + + def refuse(code: str, detail: str, refused: Iterable[str] = ()) -> None: + plan.refusals.append(RenameRefusal(code=code, detail=detail, ids=[*refused])) + + def warn(code: str, detail: str) -> None: + plan.warnings.append(RenameWarning(code=code, detail=detail)) + + try: + _validate_name(new_name) + except ValueError: + refuse("invalid_name", str(InvalidName(new_name))) + return plan + if ISSUED_ID.fullmatch(new_name): + refuse("reserved_shape", str(ReservedKey(new_name))) + return plan + if new_name == old: + refuse("same_name", f"The {kind} is already called '{old}'") + return plan + + rename = Rename(old, new_name) + rewrites = rewrite_documents(store, dashboards, kind, obj_id, rename) + plan.documents = [ + PlannedDocument( + kind=r.kind, + id=r.id, + name=r.name, + published_version=r.published_version, + draft_version=r.draft_version, + fields=r.fields, + ) + for r in rewrites + ] + plan.fingerprint = _fingerprint(plan) + if new_name in keys: + refuse("name_taken", f"Another {kind} has the key '{new_name}'") + if new_name in ids - {obj_id}: + warn( + "new_name_is_id", + f"'{new_name}' is also the id of another {kind}, so a client naming " + "it by either is asked which one it means", + ) + if kind == "flow": + _plan_flow(plan, controller, dashboards, rename, rewrites, alerts, refuse, warn) + plan.allowed = not plan.refusals + return plan + + +def _plan_flow( + plan: RenamePlan, + controller: FlowController, + dashboards: DashboardStore, + rename: Rename, + rewrites: list[Rewrite], + alerts: AlertsConfig, + refuse: Callable[..., None], + warn: Callable[[str, str], None], +) -> None: + """The part of a plan only a flow has: what holds its namespace, and + what moves outside the documents.""" + old, new = rename.old, rename.new + own = next(r for r in rewrites if r.kind == "flow" and r.id == plan.id) + if own.code_defined: + refuse( + "code_defined", + f"'{old}' is declared in code: rename it there and sync it again", + ) + coded = [r.id for r in rewrites if r is not own and r.code_defined] + if coded: + refuse( + "code_defined_dependent", + f"{len(coded)} flows declared in code name messages of '{old}', and " + "a sync would put the old names back: rename them in code first", + coded, + ) + flow_ids = [r.id for r in rewrites if r.kind == "flow"] + with Session(db_engine) as session: + active = session.exec( + select(Run.id, Run.flow_id, Run.for_node).where( + col(Run.status).in_(_ACTIVE_RUNS) + ) + ).all() + running = [ + run_id + for run_id, flow_id, for_node in active + if flow_id in flow_ids or flow_of(for_node or "") in flow_ids + ] + if running: + refuse( + "run_active", + f"{len(running)} runs of affected flows are queued or running", + running, + ) + history, collisions, too_long = _history(session, rename, flow_ids) + ledger = [ + rename.dedup_key(key) + for key in session.exec(select(AlertSent.key)).all() + if rename.dedup_key(key) != key + ] + if collisions: + refuse( + "history_collision", + f"{collisions} history rows of the affected flows are already " + f"named '{new}.*'", + ) + too_long += [key for key in ledger if len(key) > _length(col(AlertSent.key))] + if too_long: + refuse( + "name_too_long", + f"'{too_long[0]}' would be longer than its column holds", + ) + + residue = controller.key_residue(new) + referencing = references(controller.store, dashboards, new) + if residue or referencing: + named = [*residue, *(f"bindings in '{one}'" for one in referencing)] + refuse( + "target_in_use", + f"'{new}' still has {', '.join(named[:3])} behind it", + referencing, + ) + + state = controller.state + queue = controller.execution.queue if controller.execution else None + counts = queue.count(rename) if queue else {} + plan.runtime = RuntimeCounts( + durable=not isinstance(state, MemoryState), + state_keys=state.count(rename), + queue_items=counts.get("queue_items", 0), + delayed=counts.get("delayed", 0), + parked=counts.get("parked", 0), + done_markers=counts.get("done_markers", 0), + ) + if queue is not None: + others = queue.other_consumers(OTHER_ENGINE_MS) + if others: + refuse( + "other_engine", + f"Another engine reads the queue ({', '.join(others)}), which a " + "rename cannot hold", + ) + version = queue.server_version() + if version is not None and version < (7, 0): + refuse( + "redis_too_old", + f"Redis {version[0]}.{version[1]} is older than 7.0, which " + "a rename needs", + ) + + plan.history = history + plan.alerts = AlertCounts( + channels=sum( + channel.kind == "dashboard" + and rename.binding(str(channel.config.get("message") or "")) + != channel.config.get("message") + for channel in alerts.channels + ), + ledger_rows=len(ledger), + ) + + warn( + "external_references", + f"REST /messages/{old}.*, scripts, CLI, SDK and MCP key arguments and " + "hand-written bindings keep the old name", + ) + if results_name(old) in dashboards.key_index(): + warn( + "results_dashboard", + f"The dashboard '{results_name(old)}' keeps its name", + ) + drafts = [r.id for r in rewrites if r.draft_fields] + if drafts: + warn( + "drafts_rewritten", + f"{len(drafts)} unpublished drafts change and stay unpublished", + ) + pipeline = controller.pipeline + if (pipeline is not None and pipeline.is_paused(old)) or plan.runtime.parked: + warn("paused", f"'{old}' is paused, and its parked work moves with it") + if not plan.runtime.durable: + warn( + "not_durable", + "State and queued work are kept in memory, so a restart during the " + "rename loses them, as any restart does", + ) + rows = history.metric_rows + history.artifact_rows + history.output_rows + warn( + "engine_pause", + f"Automations wait while {rows} history rows and " + f"{plan.runtime.state_keys} state keys move", + ) + + +def _history( + session: Session, rename: Rename, flow_ids: list[str] +) -> tuple[HistoryCounts, int, list[str]]: + """What moves in the history of the flows *flow_ids*, how many of its + names are in the new namespace already, and the names that would no + longer fit their column.""" + old, new = f"{rename.old}.", f"{rename.new}." + grow = len(rename.new) - len(rename.old) + runs = select(Run.id).where(col(Run.flow_id).in_(flow_ids)) + counts = {} + collisions = 0 + too_long = [] + for model in (RunMetric, RunArtifact): + name = col(model.name) + scope = col(model.run_id).in_(runs) + moving = func.substr(name, 1, len(old)) == old + counts[model] = session.exec( + select(func.count()).select_from(model).where(scope, moving) + ).one() + collisions += session.exec( + select(func.count()) + .select_from(model) + .where(scope, func.substr(name, 1, len(new)) == new) + ).one() + longest = session.exec( + select(name).where(scope, moving).order_by(func.length(name).desc()) + ).first() + if longest is not None and len(longest) + grow > _length(name): + too_long.append(rename.name(longest)) + outputs = session.exec( + select(RunNode.outputs).where( + col(RunNode.run_id).in_(runs), + col(RunNode.outputs).contains(f'"{old}', autoescape=True), + ) + ).all() + moved_outputs = sum( + any(rename.binding(key) != key for key in json.loads(text or "null") or {}) + for text in outputs + ) + total = session.exec( + select(func.count()).select_from(Run).where(col(Run.flow_id).in_(flow_ids)) + ).one() + history = HistoryCounts( + runs=total, + metric_rows=counts[RunMetric], + artifact_rows=counts[RunArtifact], + output_rows=moved_outputs, + ) + return history, collisions, too_long + + +def _length(column: Any) -> int: + """How many characters a text column holds.""" + return cast(int, cast(String, column.type).length) + + +def _fingerprint(plan: RenamePlan) -> str: + seen = sorted( + (d.kind, d.id, d.published_version or 0, d.draft_version or 0) + for d in plan.documents + ) + body = json.dumps([seen, plan.name, plan.new_name]).encode() + return f"sha256:{hashlib.sha256(body).hexdigest()}" diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 3374ec9..0cc3fb6 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -34,6 +34,7 @@ from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Literal +from fluksio.flow.namespace import FLOW_FIELDS, Rename, rebind from fluksio.flow.schemas import FlowDef, NodeDef, _validate_name if TYPE_CHECKING: @@ -971,7 +972,9 @@ class FlowStore: ) copy = self._claim(copy, create=True) # Its own messages named with its own key are the copy's now. - self._retarget(copy, f"{source.name}.", f"{new_name}.") + document = copy.model_dump(mode="json") + rebind(document, FLOW_FIELDS, Rename(source.name, new_name).binding) + copy = FlowDef.model_validate(document) path = self._draft_file(copy.id) path.parent.mkdir(parents=True) path.write_text(copy.model_dump_json(indent=2) + "\n") @@ -993,21 +996,6 @@ class FlowStore: ) return copy - @staticmethod - def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool: - """Point this flow's cross-flow message names at a renamed flow.""" - changed = False - for node in flow.nodes: - for specs in (node.requires, node.provides): - for position, spec in enumerate(specs): - if spec.name.startswith(old_prefix): - tail = spec.name[len(old_prefix) :] - specs[position] = spec.model_copy( - update={"name": new_prefix + tail} - ) - changed = True - return changed - # ------------------------------------------------------------------------- # Node source # ------------------------------------------------------------------------- diff --git a/backend/tests/flow/test_drafts.py b/backend/tests/flow/test_drafts.py index 1c163ba..a9ef65d 100644 --- a/backend/tests/flow/test_drafts.py +++ b/backend/tests/flow/test_drafts.py @@ -7,7 +7,7 @@ import pytest from fluksio.flow.controller import FlowController from fluksio.flow.messages import MessageSpec -from fluksio.flow.schemas import FlowDef, FlowOrigin, NodeDef +from fluksio.flow.schemas import FlowDef, FlowInput, FlowOrigin, NodeDef from fluksio.flow.store import ISSUED_ID, FlowStore, StaleVersion from tests.utils.stores import created @@ -166,7 +166,13 @@ def test_resaving_the_published_source_creates_no_draft(store: FlowStore): def test_a_duplicate_is_a_local_draft_of_what_the_editor_shows(store: FlowStore): - flow = a_flow(title="Warm").model_copy(update={"origin": FlowOrigin()}) + flow = a_flow(title="Warm").model_copy( + update={ + "origin": FlowOrigin(), + "inputs": [FlowInput(spec=MessageSpec(name="heating.target"))], + "outputs": ["heating.temp", "temp"], + } + ) flow.nodes.append(NodeDef(id="reader", requires=[MessageSpec(name="heating.temp")])) store.write_flow(flow) store.write_node_source("heating", "sensor", SOURCE) @@ -180,6 +186,8 @@ def test_a_duplicate_is_a_local_draft_of_what_the_editor_shows(store: FlowStore) assert ISSUED_ID.fullmatch(copy.id) assert copy.origin is None assert copy.nodes[1].requires[0].name == "heating_copy.temp" + assert copy.inputs[0].spec.name == "heating_copy.target" + assert copy.outputs == ["heating_copy.temp", "temp"] assert store.has_draft(copy.id) and not store.is_published(copy.id) assert store.read_node_source(copy.id, "sensor", draft=True) == EDITED assert not (store.root / copy.id / "runtime.json").exists() diff --git a/backend/tests/flow/test_rename_runtime.py b/backend/tests/flow/test_rename_runtime.py index aa5e281..0356acd 100644 --- a/backend/tests/flow/test_rename_runtime.py +++ b/backend/tests/flow/test_rename_runtime.py @@ -4,6 +4,7 @@ FLUKSIO_TEST_REDIS_HOST (and optional PORT).""" import os import threading +import time import uuid import pytest @@ -487,3 +488,19 @@ def test_state_is_never_moved_onto_a_key_that_exists(state): with pytest.raises(RuntimeError, match="exists"): state.rename_keys(HEAT) assert state.get("warm.temp") == 2 + + +def test_another_engine_reading_the_queue_is_seen(redis_queue): + assert redis_queue.server_version() >= (7, 0) + redis_queue.claim(1, 10) + assert redis_queue.other_consumers(10_000) == [] + + other = _redis_queue(redis_queue._ns, "engine-b") + try: + other.claim(1, 10) + assert redis_queue.other_consumers(10_000) == ["engine-b"] + # Gone quiet for longer than asked about: a stopped engine. + time.sleep(0.05) + assert redis_queue.other_consumers(10) == [] + finally: + other.close() diff --git a/backend/tests/test_rename_plan.py b/backend/tests/test_rename_plan.py new file mode 100644 index 0000000..6f5ee2c --- /dev/null +++ b/backend/tests/test_rename_plan.py @@ -0,0 +1,687 @@ +"""Planning a rename and its document part (sections 1.5 and 5.1 of +docs/private/id_refac_rename.md, "The key rename operation" in id_refac.md).""" + +import ast +import inspect +import json +import time +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import JSON, String +from sqlmodel import Session, SQLModel, delete +from sqlmodel.sql.sqltypes import AutoString + +from fluksio.core.config import settings +from fluksio.core.db import engine +from fluksio.flow.alerts import AlertsConfig, Channel +from fluksio.flow.controller import FlowController +from fluksio.flow.dashboards import ( + DASHBOARD_BINDINGS, + WIDGET_BINDINGS, + DashboardDef, + DashboardStore, + SettingDef, + WidgetDef, +) +from fluksio.flow.executor import ExecutionService +from fluksio.flow.messages import MessageSpec +from fluksio.flow.namespace import FLOW_FIELDS, Rename, names +from fluksio.flow.queue import MemoryWorkQueue, WorkItem +from fluksio.flow.rename import SQL_NAMES, RenamePlan, plan_rename, rewrite_documents +from fluksio.flow.schemas import FlowDef, FlowInput, FlowOrigin, NodeDef +from fluksio.flow.state import MemoryState +from fluksio.flow.store import FlowStore +from fluksio.models import AlertSent, Run, RunArtifact, RunMetric, RunNode +from tests.utils.stores import created + +NOW = datetime.now(UTC) + + +def _flow(name: str, *nodes: NodeDef, **fields) -> FlowDef: + return FlowDef(name=name, nodes=list(nodes), **fields) + + +def _reads(node_id: str, *messages: str) -> NodeDef: + return NodeDef(id=node_id, requires=[MessageSpec(name=m) for m in messages]) + + +def _writes(node_id: str, *messages: str) -> NodeDef: + return NodeDef(id=node_id, provides=[MessageSpec(name=m) for m in messages]) + + +@pytest.fixture +def world(tmp_path: Path) -> Iterator[SimpleNamespace]: + """`heating`, published with a draft over it; `lights` reading it, published; + `notes` reading it, only a draft; `heating_extra`, which only looks alike; + and `home`, a dashboard bound to it, published with a draft over it.""" + store = FlowStore(tmp_path / "flows") + dashboards = DashboardStore(store) + queue = MemoryWorkQueue() + controller = FlowController( + store, state=MemoryState(), execution=ExecutionService(queue) + ) + + heating = _flow( + "heating", + _writes("sensor", "temp", "heating.setpoint"), + inputs=[FlowInput(spec=MessageSpec(name="heating.target"))], + outputs=["temp", "heating.setpoint"], + ) + store.write_flow(heating) + draft = store.read_flow("heating", draft=True) + draft.nodes.append(_reads("valve", "heating.temp")) + store.write_draft(draft, draft.version) + store.write_flow(_flow("lights", _reads("lamp", "heating.temp", "lights.level"))) + created(store, _flow("notes", _reads("note", "heating.mode"))) + store.write_flow(_flow("heating_extra", _writes("x", "heating_extra.x"))) + + home = DashboardDef( + name="home", + settings={"theme": SettingDef(message="heating.mode", dtype="str")}, + widgets=[ + WidgetDef( + id="t", + type="stat", + title="heating.temp", + config={"message": "heating.temp"}, + ), + WidgetDef( + id="s", + type="slider", + config={ + "target": "heating.setpoint", + "reading": {"message": "heating.temp", "label": "heating.temp"}, + }, + ), + WidgetDef(id="n", type="markdown", config={"text": "heating.temp"}), + ], + ) + dashboards.write(home) + current = dashboards.read("home", draft=True) + dashboards.write_draft( + current.model_copy(update={"title": "Home"}), current.version + ) + dashboards.write(DashboardDef(name="away")) + + yield SimpleNamespace( + store=store, dashboards=dashboards, controller=controller, queue=queue + ) + + +def _plan( + world, new: str = "warmth", kind: str = "flow", obj: str = "heating", alerts=None +) -> RenamePlan: + return plan_rename( + world.controller, world.dashboards, kind, obj, new, alerts or AlertsConfig() + ) + + +def _codes(found) -> set[str]: + return {one.code for one in found} + + +def _files(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() and ".git" not in path.parts + } + + +def _apply(world, rewrites) -> None: + for rewrite in rewrites: + for path, _, after in rewrite.files: + (world.store.root / path).write_text(after) + + +def test_a_plan_lists_the_flow_what_names_it_and_what_moves(world): + world.controller.state.set("heating.temp", 20.5) + world.controller.state.set("__version__:heating.temp", 3) + world.queue.add( + WorkItem( + kind="cascade", + node="lights.lamp", + flow="lights", + outputs={"heating.temp": 1}, + ) + ) + alerts = AlertsConfig( + channels=[ + Channel( + name="board", kind="dashboard", config={"message": "heating.alerts"} + ), + Channel(name="mail", kind="smtp", config={"to": "heating.x@example.com"}), + ] + ) + with Session(engine) as session: + session.add(Run(id="r-plan", created_at=NOW, flow_id="heating", status="ok")) + session.add(RunMetric(run_id="r-plan", name="heating.loss", step=0)) + session.add( + RunArtifact( + run_id="r-plan", + name="heating.plot", + digest="d", + size=1, + media_type="image/png", + ) + ) + session.add( + RunNode( + run_id="r-plan", node="heating.sensor", outputs='{"heating.temp": 1}' + ) + ) + session.add(AlertSent(key="node_failed:heating.sensor", ts=time.time())) + session.add(AlertSent(key="node_failed:heating_extra.x", ts=time.time())) + session.commit() + try: + plan = _plan(world, alerts=alerts) + finally: + with Session(engine) as session: + for model in (RunMetric, RunArtifact, RunNode, Run, AlertSent): + session.execute(delete(model)) + session.commit() + + assert plan.allowed, plan.refusals + assert (plan.name, plan.new_name, plan.version) == ("heating", "warmth", 2) + listed = { + (d.kind, d.id): (d.published_version, d.draft_version, d.fields) + for d in plan.documents + } + assert listed == { + # Its own qualified names: three in the published variant, four in + # the draft. + ("flow", "heating"): (1, 2, 4), + ("flow", "lights"): (1, None, 1), + ("flow", "notes"): (None, 1, 1), + ("dashboard", "home"): (1, 2, 4), + } + assert plan.fingerprint.startswith("sha256:") + assert plan.history is not None and plan.history.model_dump() == { + "runs": 1, + "metric_rows": 1, + "artifact_rows": 1, + "output_rows": 1, + } + assert plan.alerts is not None and plan.alerts.model_dump() == { + "channels": 1, + "ledger_rows": 1, + } + assert plan.runtime is not None + assert ( + plan.runtime.durable, + plan.runtime.state_keys, + plan.runtime.queue_items, + ) == (False, 2, 1) + assert _codes(plan.warnings) == { + "external_references", + "drafts_rewritten", + "not_durable", + "engine_pause", + } + + +def test_a_dry_run_changes_no_bytes(world): + before = _files(world.store.root) + head = world.store._git("rev-parse", "HEAD").stdout + + _plan(world) + _plan(world, "house", "dashboard", "home") + + assert _files(world.store.root) == before + assert world.store._git("rev-parse", "HEAD").stdout == head + + +def test_the_fingerprint_follows_every_listed_documents_version(world): + first = _plan(world) + assert _plan(world).fingerprint == first.fingerprint + + lights = world.store.read_flow("lights") + world.store.write_draft( + lights.model_copy(update={"title": "Lights"}), lights.version + ) + + assert _plan(world).fingerprint != first.fingerprint + + +def test_a_rename_rewrites_bindings_and_nothing_else(world): + legacy = world.dashboards.root / "legacy" / "dashboard.json" + legacy.parent.mkdir() + legacy.write_text( + json.dumps( + { + "name": "legacy", + "version": 4, + "pages": [ + { + "sections": [ + { + "widgets": [ + { + "id": "w", + "type": "stat", + "config": {"message": "heating.temp"}, + } + ] + } + ] + } + ], + }, + indent=2, + ) + ) + device = world.store.root / "_devices" / "panel" / "device.json" + device.parent.mkdir(parents=True) + device.write_text('{"name": "panel", "dashboards": ["home", "legacy"]}\n') + before = _files(world.store.root) + + rewrites = rewrite_documents( + world.store, world.dashboards, "flow", "heating", Rename("heating", "warmth") + ) + _apply(world, rewrites) + + changed = {path for r in rewrites for path, _, _ in r.files} + after = _files(world.store.root) + # The stability set: no file appears or goes, and nothing else changes. + assert after.keys() == before.keys() + assert {path for path in after if after[path] != before[path]} == changed + assert "_devices/panel/device.json" not in changed + for kind, stem, paths in ( + ("flow", "*/flow*.json", FLOW_FIELDS), + ("dashboard", "_dashboards/*/dashboard*.json", DASHBOARD_BINDINGS), + ): + for path in world.store.root.glob(stem): + data = json.loads(path.read_text()) + assert data.get("id", path.parent.name) == path.parent.name + bound = names(data, paths) + assert not [n for n in bound if n.startswith("heating.")], (kind, path) + assert ( + world.store.read_flow("heating_extra").nodes[0].provides[0].name + == "heating_extra.x" + ) + + renamed = world.store.read_flow("heating", draft=True) + assert renamed.name == "warmth" and renamed.version == 3 + assert [s.name for s in renamed.nodes[0].provides] == ["temp", "warmth.setpoint"] + assert renamed.inputs[0].spec.name == "warmth.target" + assert renamed.outputs == ["temp", "warmth.setpoint"] + assert world.store.read_flow("heating").version == 3 + assert ( + world.store.read_flow("notes", draft=True).nodes[0].requires[0].name + == "warmth.mode" + ) + + home = world.dashboards.read("home", draft=True) + assert (home.version, home.title) == (3, "Home") + assert home.settings["theme"].message == "warmth.mode" + assert home.widgets[0].messages == ["warmth.temp"] + # Labels and free text are the person's. + assert home.widgets[0].title == "heating.temp" + assert home.widgets[1].target == "warmth.setpoint" + assert home.widgets[1].reading == { + "message": "warmth.temp", + "label": "heating.temp", + } + assert home.widgets[2].config == {"text": "heating.temp"} + assert world.dashboards.read("home").version == 3 + assert json.loads(legacy.read_text())["pages"][0]["sections"][0]["widgets"][0][ + "config" + ] == {"message": "warmth.temp"} + + +def test_renaming_back_restores_the_documents_but_for_versions(world): + def unversioned() -> dict: + return { + path: {k: v for k, v in json.loads(text).items() if k != "version"} + for path, text in _files(world.store.root).items() + if path.endswith(".json") and not path.endswith("runtime.json") + } + + before = unversioned() + _apply( + world, + rewrite_documents( + world.store, + world.dashboards, + "flow", + "heating", + Rename("heating", "warmth"), + ), + ) + assert unversioned() != before + _apply( + world, + rewrite_documents( + world.store, + world.dashboards, + "flow", + "heating", + Rename("warmth", "heating"), + ), + ) + assert unversioned() == before + + +def test_a_dashboard_rename_changes_its_key_and_versions_only(world): + plan = _plan(world, "house", "dashboard", "home") + assert plan.allowed and plan.version == 2 + assert [ + (d.kind, d.id, d.published_version, d.draft_version, d.fields) + for d in plan.documents + ] == [("dashboard", "home", 1, 2, 0)] + assert (plan.alerts, plan.history, plan.runtime, plan.warnings) == ( + None, + None, + None, + [], + ) + assert _codes(_plan(world, "away", "dashboard", "home").refusals) == {"name_taken"} + + (rewrite,) = rewrite_documents( + world.store, world.dashboards, "dashboard", "home", Rename("home", "house") + ) + _apply(world, [rewrite]) + published, draft = ( + world.dashboards.read("home"), + world.dashboards.read("home", draft=True), + ) + assert (published.name, published.version, draft.name, draft.version) == ( + "house", + 3, + "house", + 3, + ) + assert draft.widgets[0].messages == ["heating.temp"] + + +@pytest.mark.parametrize( + ("new", "code"), + [ + ("Warmth", "invalid_name"), + ("f_abcdefgh", "reserved_shape"), + ("heating", "same_name"), + ("lights", "name_taken"), + ("notes", "name_taken"), + ], +) +def test_a_key_that_cannot_be_taken_is_refused(world, new, code): + plan = _plan(world, new) + assert not plan.allowed + assert code in _codes(plan.refusals) + + +def test_code_defined_flows_are_refused(world): + lights = world.store.read_flow("lights") + world.store.write_flow(lights.model_copy(update={"origin": FlowOrigin()})) + plan = _plan(world) + assert [(r.code, r.ids) for r in plan.refusals] == [ + ("code_defined_dependent", ["lights"]) + ] + + heating = world.store.read_flow("heating") + world.store.write_flow(heating.model_copy(update={"origin": FlowOrigin()})) + assert _codes(_plan(world).refusals) == {"code_defined", "code_defined_dependent"} + + +def test_an_active_run_of_an_affected_flow_is_refused(world): + with Session(engine) as session: + session.add( + Run(id="r-busy", created_at=NOW, flow_id="lights", status="running") + ) + session.add( + Run( + id="r-train", + created_at=NOW, + flow_id="trainer", + for_node="heating.model", + status="queued", + ) + ) + session.add( + Run(id="r-other", created_at=NOW, flow_id="heating_extra", status="running") + ) + session.commit() + try: + plan = _plan(world) + finally: + with Session(engine) as session: + session.execute(delete(Run)) + session.commit() + assert [(r.code, sorted(r.ids)) for r in plan.refusals] == [ + ("run_active", ["r-busy", "r-train"]) + ] + + +def test_what_still_lives_under_the_new_key_is_refused(world): + state = world.controller.state + # A node's own key is its data, even when it reads like a name. + state.set("__node__:lights.lamp:last:warmth.temp", 1) + assert _plan(world).allowed + + state.set("__last_seen__:lights.lamp:warmth.temp", 1) + world.queue.add(WorkItem(kind="cascade", node="warmth.sensor", flow="warmth")) + plan = _plan(world) + assert [r.code for r in plan.refusals] == ["target_in_use"] + assert "__last_seen__:lights.lamp:warmth.temp" in plan.refusals[0].detail + assert world.controller.key_residue("warmth") == [ + "__last_seen__:lights.lamp:warmth.temp", + "queued work of 'warmth.sensor'", + ] + + state.delete("__last_seen__:lights.lamp:warmth.temp") + world.queue.clear_flow("warmth") + away = world.dashboards.read("away") + world.dashboards.write( + away.model_copy( + update={ + "widgets": [ + WidgetDef(id="w", type="stat", config={"message": "warmth.temp"}) + ] + } + ) + ) + plan = _plan(world) + assert [(r.code, r.ids) for r in plan.refusals] == [("target_in_use", ["away"])] + + +def test_history_that_would_collide_or_overflow_is_refused(world): + long = "heating." + "x" * (128 - len("heating.")) + with Session(engine) as session: + session.add(Run(id="r-hist", created_at=NOW, flow_id="heating", status="ok")) + session.add(RunMetric(run_id="r-hist", name="heating_v2.loss", step=0)) + session.add(RunMetric(run_id="r-hist", name=long, step=0)) + session.commit() + try: + assert _codes(_plan(world, "heating_v2").refusals) == { + "history_collision", + "name_too_long", + } + # A key no longer than the old one fits. + assert _codes(_plan(world, "warmth").refusals) == set() + finally: + with Session(engine) as session: + for model in (RunMetric, Run): + session.execute(delete(model)) + session.commit() + + +def test_another_engine_or_an_old_redis_is_refused(world, monkeypatch): + monkeypatch.setattr( + MemoryWorkQueue, "other_consumers", lambda self, ms: ["engine-b"] + ) + monkeypatch.setattr(MemoryWorkQueue, "server_version", lambda self: (6, 2)) + assert _codes(_plan(world).refusals) == {"other_engine", "redis_too_old"} + + +def test_what_a_rename_leaves_behind_is_said(world): + # A flow whose id is the new key, under a key of its own. + world.store.create(_flow("elsewhere"), "warmth") + world.dashboards.write(DashboardDef(name="heating_results")) + world.queue.park( + "heating", WorkItem(kind="cascade", node="heating.sensor", flow="heating") + ) + + plan = _plan(world) + assert plan.allowed + assert {"new_name_is_id", "results_dashboard", "paused"} <= _codes(plan.warnings) + + +def test_every_binding_a_widget_reads_is_in_the_key_table(): + """A new binding in an accessor fails here until the table knows it, so a + rename cannot leave it behind. Keys that hold no message name are listed + as such.""" + plain = { + "source", + "dtype", + "inner_dtype", + "request_dtype", + "history", + "points", + "format", + "counters", + "ids", + "group", + "flow", + "latest", + } + tree = ast.parse(inspect.getsource(WidgetDef)) + read = { + call.args[0].value + for call in ast.walk(tree) + if isinstance(call, ast.Call) + and isinstance(call.func, ast.Attribute) + and call.func.attr == "get" + and call.args + and isinstance(call.args[0], ast.Constant) + } + bound = {part for path in WIDGET_BINDINGS for part in path if part != "*"} + assert read - plain == bound + assert set(SettingDef.model_fields) - {"value", "dtype"} == {"message"} + assert ("settings", "*", "message") in DASHBOARD_BINDINGS + + +def test_every_history_column_is_classified(): + """A column added to the history fails here until it says whether it + names a message, which a rename then rewrites (`SQL_NAMES`), or not.""" + unnamed = { + # Ids, which never change: flows and nodes are filed by id. + "run": { + "id", + "flow", + "commit", + "origin_commit", + "code_digest", + "params", + "passed", + "declared", + "params_digest", + "group_id", + "parent_id", + "idempotency_key", + "external_id", + "cause", + "for_node", + "status", + "status_reason", + "labels", + "needs", + "result", + "engine", + "actor", + }, + "run_node": { + "run_id", + "node", + "status", + "worker", + "error", + "logs", + "cache_key", + "cached_from", + }, + "run_metric": {"run_id", "node"}, + "run_artifact": {"run_id", "digest", "media_type", "filename", "node"}, + "flow_run": {"id", "flow", "source", "parent", "status"}, + "metric_minute": {"flow", "node"}, + "engine_event": {"type", "flow", "node", "detail", "actor", "run"}, + "flow_tombstone": {"id", "name"}, + } + # Tables that hold no flow, node or message at all. + unrelated = { + "user", + "oauth_client", + "oauth_authorization_code", + "oauth_refresh_token", + "flavor", + "worker_token", + "connector_state", + "alembic_version", + } + found: dict[str, set[str]] = {} + for table in SQLModel.metadata.sorted_tables: + if table.name in unrelated: + continue + found[table.name] = { + column.name + for column in table.columns + if isinstance(column.type, String | AutoString | JSON) + } + classified = dict(unnamed) + for table, column in SQL_NAMES: + classified.setdefault(table, set()).add(column) + assert found == classified + + +def test_planning_is_a_superusers( + client: TestClient, superuser_token_headers, normal_user_token_headers +): + flows = f"{settings.API_V1_STR}/flows" + made = client.post( + f"{flows}/", headers=superuser_token_headers, json={"name": "plan_me"} + ) + assert made.status_code == 201, made.text + flow_id = made.json()["definition"]["id"] + try: + body = {"new_name": "plan_me_too"} + assert ( + client.post( + f"{flows}/{flow_id}/rename/plan", + headers=normal_user_token_headers, + json=body, + ).status_code + == 403 + ) + assert ( + client.post( + f"{flows}/f_nothere0/rename/plan", + headers=superuser_token_headers, + json=body, + ).status_code + == 404 + ) + plan = client.post( + f"{flows}/{flow_id}/rename/plan", headers=superuser_token_headers, json=body + ) + assert plan.status_code == 200, plan.text + assert (plan.json()["name"], plan.json()["allowed"]) == ("plan_me", True) + + boards = f"{settings.API_V1_STR}/dashboards" + board = client.post( + f"{boards}/", headers=superuser_token_headers, json={"name": "plan_board"} + ).json() + try: + plan = client.post( + f"{boards}/{board['id']}/rename/plan", + headers=superuser_token_headers, + json=body, + ) + assert plan.status_code == 200 and plan.json()["kind"] == "dashboard" + finally: + client.delete(f"{boards}/{board['id']}", headers=superuser_token_headers) + finally: + client.delete(f"{flows}/{flow_id}", headers=superuser_token_headers) diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index b778fd8..38c0d00 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -123,6 +123,22 @@ export const AIStatusSchema = { title: 'AIStatus' } as const; +export const AlertCountsSchema = { + properties: { + channels: { + type: 'integer', + title: 'Channels' + }, + ledger_rows: { + type: 'integer', + title: 'Ledger Rows' + } + }, + type: 'object', + required: ['channels', 'ledger_rows'], + title: 'AlertCounts' +} as const; + export const AlertsConfigSchema = { properties: { enabled: { @@ -2183,6 +2199,30 @@ export const HealthSummarySchema = { title: 'HealthSummary' } as const; +export const HistoryCountsSchema = { + properties: { + runs: { + type: 'integer', + title: 'Runs' + }, + metric_rows: { + type: 'integer', + title: 'Metric Rows' + }, + artifact_rows: { + type: 'integer', + title: 'Artifact Rows' + }, + output_rows: { + type: 'integer', + title: 'Output Rows' + } + }, + type: 'object', + required: ['runs', 'metric_rows', 'artifact_rows', 'output_rows'], + title: 'HistoryCounts' +} as const; + export const HistoryPointSchema = { properties: { ts: { @@ -3647,6 +3687,54 @@ export const PlacementSchema = { description: "Where a widget sits in its section's grid, in grid units." } as const; +export const PlannedDocumentSchema = { + properties: { + kind: { + type: 'string', + enum: ['flow', 'dashboard'], + title: 'Kind' + }, + id: { + type: 'string', + title: 'Id' + }, + name: { + type: 'string', + title: 'Name' + }, + published_version: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Published Version' + }, + draft_version: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Draft Version' + }, + fields: { + type: 'integer', + title: 'Fields' + } + }, + type: 'object', + required: ['kind', 'id', 'name', 'published_version', 'draft_version', 'fields'], + title: 'PlannedDocument', + description: 'A document a rename rewrites, and the versions it was planned against.' +} as const; + export const PortalMemberSchema = { properties: { user_id: { @@ -3797,6 +3885,157 @@ export const RemoteUserBodySchema = { title: 'RemoteUserBody' } as const; +export const RenamePlanSchema = { + properties: { + kind: { + type: 'string', + enum: ['flow', 'dashboard'], + title: 'Kind' + }, + id: { + type: 'string', + title: 'Id' + }, + name: { + type: 'string', + title: 'Name' + }, + new_name: { + type: 'string', + title: 'New Name' + }, + version: { + type: 'integer', + title: 'Version' + }, + fingerprint: { + type: 'string', + title: 'Fingerprint', + default: '' + }, + allowed: { + type: 'boolean', + title: 'Allowed', + default: false + }, + refusals: { + items: { + '$ref': '#/components/schemas/RenameRefusal' + }, + type: 'array', + title: 'Refusals', + default: [] + }, + documents: { + items: { + '$ref': '#/components/schemas/PlannedDocument' + }, + type: 'array', + title: 'Documents', + default: [] + }, + alerts: { + anyOf: [ + { + '$ref': '#/components/schemas/AlertCounts' + }, + { + type: 'null' + } + ] + }, + history: { + anyOf: [ + { + '$ref': '#/components/schemas/HistoryCounts' + }, + { + type: 'null' + } + ] + }, + runtime: { + anyOf: [ + { + '$ref': '#/components/schemas/RuntimeCounts' + }, + { + type: 'null' + } + ] + }, + warnings: { + items: { + '$ref': '#/components/schemas/RenameWarning' + }, + type: 'array', + title: 'Warnings', + default: [] + } + }, + type: 'object', + required: ['kind', 'id', 'name', 'new_name', 'version'], + title: 'RenamePlan', + description: `What renaming a flow's or dashboard's key would change, and whether it +may. Read-only; the rename request repeats \`\`version\`\` and +\`\`fingerprint\`\`, and is refused when either moved since.` +} as const; + +export const RenamePlanRequestSchema = { + properties: { + new_name: { + type: 'string', + title: 'New Name' + } + }, + type: 'object', + required: ['new_name'], + title: 'RenamePlanRequest', + description: 'The key a flow or dashboard would be renamed to.' +} as const; + +export const RenameRefusalSchema = { + properties: { + code: { + type: 'string', + title: 'Code' + }, + detail: { + type: 'string', + title: 'Detail' + }, + ids: { + items: { + type: 'string' + }, + type: 'array', + title: 'Ids', + default: [] + } + }, + type: 'object', + required: ['code', 'detail'], + title: 'RenameRefusal', + description: 'Why a rename cannot go ahead, with the ids of what is in the way.' +} as const; + +export const RenameWarningSchema = { + properties: { + code: { + type: 'string', + title: 'Code' + }, + detail: { + type: 'string', + title: 'Detail' + } + }, + type: 'object', + required: ['code', 'detail'], + title: 'RenameWarning', + description: 'Something a rename changes, or leaves, that its caller should know.' +} as const; + export const ResourceLevelSchema = { properties: { total: { @@ -4260,6 +4499,38 @@ export const RunRequestSchema = { title: 'RunRequest' } as const; +export const RuntimeCountsSchema = { + properties: { + durable: { + type: 'boolean', + title: 'Durable' + }, + state_keys: { + type: 'integer', + title: 'State Keys' + }, + queue_items: { + type: 'integer', + title: 'Queue Items' + }, + delayed: { + type: 'integer', + title: 'Delayed' + }, + parked: { + type: 'integer', + title: 'Parked' + }, + done_markers: { + type: 'integer', + title: 'Done Markers' + } + }, + type: 'object', + required: ['durable', 'state_keys', 'queue_items', 'delayed', 'parked', 'done_markers'], + title: 'RuntimeCounts' +} as const; + export const SearchEntrySchema = { properties: { category: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 6637662..cb4e815 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsStartTelegramLinkData, AlertsStartTelegramLinkResponse, AlertsReadTelegramLinkData, AlertsReadTelegramLinkResponse, AlertsCancelTelegramLinkData, AlertsCancelTelegramLinkResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsData, DashboardsReadDashboardsResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsData, FlowsReadFlowsResponse, FlowsCreateFlowData, FlowsCreateFlowResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsStartTelegramLinkData, AlertsStartTelegramLinkResponse, AlertsReadTelegramLinkData, AlertsReadTelegramLinkResponse, AlertsCancelTelegramLinkData, AlertsCancelTelegramLinkResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, ArtifactsDeleteArtifactData, ArtifactsDeleteArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudReadMembersResponse, CloudLinkMemberData, CloudLinkMemberResponse, CloudRevokeMemberData, CloudRevokeMemberResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsData, DashboardsReadDashboardsResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsDuplicateDashboardData, DashboardsDuplicateDashboardResponse, DashboardsPlanDashboardRenameData, DashboardsPlanDashboardRenameResponse, DevicesStartPairingData, DevicesStartPairingResponse, DevicesPollPairingData, DevicesPollPairingResponse, DevicesReadDevicesResponse, DevicesReadPendingResponse, DevicesPendingDeviceData, DevicesPendingDeviceResponse, DevicesCreateDeviceData, DevicesCreateDeviceResponse, DevicesReadDeviceData, DevicesReadDeviceResponse, DevicesSaveDeviceData, DevicesSaveDeviceResponse, DevicesDeleteDeviceData, DevicesDeleteDeviceResponse, DevicesReadIceData, DevicesReadIceResponse, DevicesPutReportData, DevicesPutReportResponse, DevicesApprovePairingData, DevicesApprovePairingResponse, DevicesUnpairDeviceData, DevicesUnpairDeviceResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsData, FlowsReadFlowsResponse, FlowsCreateFlowData, FlowsCreateFlowResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsDuplicateFlowData, FlowsDuplicateFlowResponse, FlowsPlanFlowRenameData, FlowsPlanFlowRenameResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsTrainNodeData, FlowsTrainNodeResponse, FlowsReadTrainingData, FlowsReadTrainingResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadNodeTypeTimingsData, ObservabilityReadNodeTypeTimingsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, ObservabilityFlushMetricsResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsReadNodeOutputsData, RunsReadNodeOutputsResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -865,6 +865,33 @@ export class DashboardsService { } }); } + + /** + * Plan Dashboard Rename + * What renaming this dashboard's key would change, and what refuses it. + * + * Changes nothing. Nothing binds to a dashboard's key, so the plan is the + * dashboard alone; a superuser's all the same, as a flow's is. + * @param data The data for the request. + * @param data.dashboardId + * @param data.requestBody + * @returns RenamePlan Successful Response + * @throws ApiError + */ + public static planDashboardRename(data: DashboardsPlanDashboardRenameData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/dashboards/{dashboard_id}/rename/plan', + path: { + dashboard_id: data.dashboardId + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } } export class DevicesService { @@ -1518,6 +1545,34 @@ export class FlowsService { }); } + /** + * Plan Flow Rename + * What renaming this flow's key would change, and what refuses it. + * + * Changes nothing. A superuser's, as the rename is: the key names the + * flow's messages, and renaming it rewrites every flow and dashboard that + * names one, its history and its live state. + * @param data The data for the request. + * @param data.flowId + * @param data.requestBody + * @returns RenamePlan Successful Response + * @throws ApiError + */ + public static planFlowRename(data: FlowsPlanFlowRenameData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{flow_id}/rename/plan', + path: { + flow_id: data.flowId + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Node Source * Read a node's Python source, including unpublished edits. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 6423580..15b4019 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -25,6 +25,11 @@ export type AIStatus = { model: (string | null); }; +export type AlertCounts = { + channels: number; + ledger_rows: number; +}; + /** * The whole alerting setup, as stored and as the API sees it. */ @@ -719,6 +724,13 @@ export type HealthSummary = { version?: string; }; +export type HistoryCounts = { + runs: number; + metric_rows: number; + artifact_rows: number; + output_rows: number; +}; + /** * One numeric value a message carried, and when. */ @@ -1190,6 +1202,20 @@ export type Placement = { h?: number; }; +/** + * A document a rename rewrites, and the versions it was planned against. + */ +export type PlannedDocument = { + kind: 'flow' | 'dashboard'; + id: string; + name: string; + published_version: (number | null); + draft_version: (number | null); + fields: number; +}; + +export type kind2 = 'flow' | 'dashboard'; + export type PortalMember = { user_id: string; email: string; @@ -1232,6 +1258,51 @@ export type RemoteUserBody = { code: string; }; +/** + * What renaming a flow's or dashboard's key would change, and whether it + * may. Read-only; the rename request repeats ``version`` and + * ``fingerprint``, and is refused when either moved since. + */ +export type RenamePlan = { + kind: 'flow' | 'dashboard'; + id: string; + name: string; + new_name: string; + version: number; + fingerprint?: string; + allowed?: boolean; + refusals?: Array; + documents?: Array; + alerts?: (AlertCounts | null); + history?: (HistoryCounts | null); + runtime?: (RuntimeCounts | null); + warnings?: Array; +}; + +/** + * The key a flow or dashboard would be renamed to. + */ +export type RenamePlanRequest = { + new_name: string; +}; + +/** + * Why a rename cannot go ahead, with the ids of what is in the way. + */ +export type RenameRefusal = { + code: string; + detail: string; + ids?: Array<(string)>; +}; + +/** + * Something a rename changes, or leaves, that its caller should know. + */ +export type RenameWarning = { + code: string; + detail: string; +}; + export type ResourceLevel = { total: number; free: number; @@ -1373,6 +1444,15 @@ export type RunRequest = { }; }; +export type RuntimeCounts = { + durable: boolean; + state_keys: number; + queue_items: number; + delayed: number; + parked: number; + done_markers: number; +}; + /** * One thing somebody might be looking for. * @@ -1937,6 +2017,13 @@ export type DashboardsDuplicateDashboardData = { export type DashboardsDuplicateDashboardResponse = (DashboardDef_Output); +export type DashboardsPlanDashboardRenameData = { + dashboardId: string; + requestBody: RenamePlanRequest; +}; + +export type DashboardsPlanDashboardRenameResponse = (RenamePlan); + export type DevicesStartPairingData = { requestBody?: (PairHello | null); }; @@ -2095,6 +2182,13 @@ export type FlowsDuplicateFlowData = { export type FlowsDuplicateFlowResponse = (FlowDetail); +export type FlowsPlanFlowRenameData = { + flowId: string; + requestBody: RenamePlanRequest; +}; + +export type FlowsPlanFlowRenameResponse = (RenamePlan); + export type FlowsReadNodeSourceData = { flowId: string; nodeId: string;