diff --git a/backend/fluksio/ai/agent.py b/backend/fluksio/ai/agent.py index 2f01450..8b5e59f 100644 --- a/backend/fluksio/ai/agent.py +++ b/backend/fluksio/ai/agent.py @@ -34,9 +34,13 @@ operate flows through the tools; they are the same API the editor uses, so \ anything you save appears in the editor at once. How to work: +- A flow has an id and a key, its `name`, which also prefixes its messages. \ +Tools take either; write id: or key: when a tool says the name could \ +mean two flows. - Read before you write: get_flow before save_flow, and pass back the version \ you read. On a version conflict, read again and reapply the change. -- save_flow writes a draft, and validate_flow tells you whether it is sound. \ +- create_flow makes a new flow and save_flow changes one; both write a draft, \ +and validate_flow tells you whether it is sound. \ Publish only when the user asks for the change to go live; otherwise leave \ the draft for review in the editor and say so. - list_node_types gives the node types and their parameters; get_node_source \ @@ -45,7 +49,7 @@ shows how a Python node is written before you write one. use lists, `inline code` and fenced code blocks where they help, and no \ tables, images or HTML. -The flow document `save_flow` takes, as a JSON schema: +The flow document `create_flow` and `save_flow` take, as a JSON schema: """ # ponytail: a tool's answer is replayed on every later round of the @@ -57,7 +61,10 @@ _RESULT_LIMIT = 20_000 def system_prompt(flow: str | None) -> str: prompt = _SYSTEM + json.dumps(FlowDef.model_json_schema()) if flow: - prompt += f"\n\nThe user is looking at the flow named {flow!r}." + prompt += ( + f"\n\nThe user is looking at the flow with the id {flow!r}. Name it " + f"to the tools as id:{flow}." + ) return prompt diff --git a/backend/fluksio/api/routes/ai.py b/backend/fluksio/api/routes/ai.py index cc520ef..7ad3f7e 100644 --- a/backend/fluksio/api/routes/ai.py +++ b/backend/fluksio/api/routes/ai.py @@ -103,7 +103,7 @@ class ControlNode(BaseModel): class TurnBody(BaseModel): message: str = Field(min_length=1, max_length=20_000) - #: The flow open in the editor, when the chat is opened from one. + #: The id of the flow open in the editor, when the chat is opened from one. flow: str | None = Field(default=None, pattern=NAME_PATTERN.pattern) diff --git a/backend/fluksio/mcp/server.py b/backend/fluksio/mcp/server.py index c7a03d5..3b855d5 100644 --- a/backend/fluksio/mcp/server.py +++ b/backend/fluksio/mcp/server.py @@ -8,19 +8,24 @@ means an agent cannot do anything a person could not do in the browser. The caller's token rides along on every hop, so the API sees the agent's own identity rather than some service account. + +A tool takes a flow by its id or its key, resolved by `fluksio.sdk.resolve` +from the flows the caller can list, and calls the API with the id. """ from __future__ import annotations import logging from contextvars import ContextVar -from typing import Any +from typing import Annotated, Any import httpx from mcp.server.context import CallNext, HandlerResult, ServerRequestContext from mcp.server.mcpserver import MCPServer +from pydantic import Field from fluksio.sdk.client import seg as _seg +from fluksio.sdk.resolve import Resolved, Unresolved, resolve logger = logging.getLogger(__name__) @@ -126,6 +131,80 @@ async def _call(method: str, path: str, **kwargs: Any) -> Any: return response.json() +async def _resolve( + flow: str, *, mutation: bool = False, history: bool = False +) -> Resolved | dict[str, Any]: + """The flow *flow* names, or the error to hand back. + + Read from what the caller may list, with the caller's token, so nothing + is resolved that the API would not show them. ``history`` counts deleted + flows too, under the key their runs last had. + """ + listing = await _call("GET", "/flows/") + if "error" in listing: + return dict(listing) + flows = [(str(row["id"]), str(row["name"])) for row in listing.get("data") or []] + if history: + overview = await _call("GET", "/runs/overview") + if isinstance(overview, dict): + return overview + flows += [ + (str(row["flow"]), str(row.get("flow_name") or row["flow"])) + for row in overview + ] + try: + return resolve(flow, flows, mutation=mutation) + except Unresolved as exc: + return {"error": str(exc)} + + +def _warned(answer: Any, found: Resolved) -> Any: + """*answer*, saying which flow was taken when the name could mean two.""" + if not found.warning: + return answer + if isinstance(answer, dict): + return {**answer, "warning": found.warning} + return {"result": answer, "warning": found.warning} + + +async def _on_flow( + flow: str, method: str, tail: str = "", *, mutation: bool, **kwargs: Any +) -> Any: + """One call to ``/flows/`` for the flow *flow* names. + + ``mutation`` is whether the call changes anything, which a name two flows + answer to is refused for rather than guessed. + """ + found = await _resolve(flow, mutation=mutation) + if not isinstance(found, Resolved): + return found + answer = await _call(method, f"/flows/{_seg(found.id)}{tail}", **kwargs) + return _warned(answer, found) + + +async def _narrowed(flow: str | None, path: str, params: dict[str, Any]) -> Any: + """A history read at *path*, narrowed to the flow *flow* names when it + names one, a deleted one included.""" + if not flow: + return await _call("GET", path, params=params) + found = await _resolve(flow, history=True) + if not isinstance(found, Resolved): + return found + params["flow"] = found.id + return _warned(await _call("GET", path, params=params), found) + + +#: Every tool's *flow* argument, described where an agent reads the schema. +FlowArg = Annotated[ + str, + Field( + description=( + "The flow's id or its key. Write id: or key: when one flow's " + "id is another's key." + ) + ), +] + # ----------------------------------------------------------------------------- # Reading # ----------------------------------------------------------------------------- @@ -138,13 +217,14 @@ async def list_flows() -> Any: @mcp.tool() -async def get_flow(name: str) -> Any: +async def get_flow(flow: FlowArg) -> Any: """One flow: its definition, the state of its nodes, and its problems. The definition is the working copy — unpublished edits included — which is - what to base a change on. + what to base a change on. It carries the flow's ``id`` and its key, + ``name``. """ - return await _call("GET", f"/flows/{_seg(name)}") + return await _on_flow(flow, "GET", mutation=False) @mcp.tool() @@ -154,21 +234,21 @@ async def list_node_types() -> Any: @mcp.tool() -async def get_node_source(name: str, node_id: str) -> Any: +async def get_node_source(flow: FlowArg, node_id: str) -> Any: """The Python source of one node.""" - return await _call("GET", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/source") + return await _on_flow(flow, "GET", f"/nodes/{_seg(node_id)}/source", mutation=False) @mcp.tool() -async def get_flow_state(name: str) -> Any: +async def get_flow_state(flow: FlowArg) -> Any: """The last value seen on every message of a flow.""" - return await _call("GET", f"/flows/{_seg(name)}/state") + return await _on_flow(flow, "GET", "/state", mutation=False) @mcp.tool() -async def get_message_history(name: str, message: str) -> Any: +async def get_message_history(flow: FlowArg, message: str) -> Any: """Recent numeric values of one message, oldest first.""" - return await _call("GET", f"/flows/{_seg(name)}/history/{_seg(message)}") + return await _on_flow(flow, "GET", f"/history/{_seg(message)}", mutation=False) @mcp.tool() @@ -216,21 +296,36 @@ async def get_graph() -> Any: @mcp.tool() -async def save_flow(name: str, definition: dict[str, Any]) -> Any: - """Save a flow as an unpublished draft. +async def create_flow(definition: dict[str, Any]) -> Any: + """Create a new flow, as an unpublished draft. + + ``definition`` is a flow document like the one `get_flow` returns under + ``definition``. Its ``name`` is the new flow's key, which no other flow + may hold. The server gives the flow its id, so an ``id`` or ``version`` in + the document is ignored; the answer carries both. Change an existing flow + with `save_flow` instead. + """ + document = {k: v for k, v in definition.items() if k not in ("id", "version")} + return await _call("POST", "/flows/", json=document) + + +@mcp.tool() +async def save_flow(flow: FlowArg, definition: dict[str, Any]) -> Any: + """Save changes to an existing flow as an unpublished draft. ``definition`` is a whole flow document — the shape `get_flow` returns under ``definition``. It must carry the ``version`` that was read, so an edit someone else made in between is refused rather than overwritten; on a - conflict, read the flow again and reapply the change. + conflict, read the flow again and reapply the change. Its ``name``, the + key, cannot change here. A new flow is `create_flow`. Nothing here reaches the engine until `publish_flow`. """ - return await _call("PUT", f"/flows/{_seg(name)}", json=definition) + return await _on_flow(flow, "PUT", mutation=True, json=definition) @mcp.tool() -async def save_node_source(name: str, node_id: str, code: str) -> Any: +async def save_node_source(flow: FlowArg, node_id: str, code: str) -> Any: """Save a node's Python source and report whether it compiles. A node defines ``process(...)``, taking one argument per input port and @@ -238,35 +333,39 @@ async def save_node_source(name: str, node_id: str, code: str) -> Any: or dropped there adds or drops the input port or setting in the draft, which moves its version on: read the flow again before saving it. """ - return await _call( - "PUT", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/source", json={"code": code} + return await _on_flow( + flow, + "PUT", + f"/nodes/{_seg(node_id)}/source", + mutation=True, + json={"code": code}, ) @mcp.tool() -async def publish_flow(name: str, version: int) -> Any: +async def publish_flow(flow: FlowArg, version: int) -> Any: """Deploy a flow's unpublished changes: this is what puts them live.""" - return await _call( - "POST", f"/flows/{_seg(name)}/publish", json={"version": version} + return await _on_flow( + flow, "POST", "/publish", mutation=True, json={"version": version} ) @mcp.tool() -async def discard_draft(name: str) -> Any: +async def discard_draft(flow: FlowArg) -> Any: """Throw unpublished changes away and go back to what is running.""" - return await _call("POST", f"/flows/{_seg(name)}/discard-draft") + return await _on_flow(flow, "POST", "/discard-draft", mutation=True) @mcp.tool() -async def delete_flow(name: str) -> Any: +async def delete_flow(flow: FlowArg) -> Any: """Delete a flow and the code of its nodes.""" - return await _call("DELETE", f"/flows/{_seg(name)}") + return await _on_flow(flow, "DELETE", mutation=True) @mcp.tool() -async def validate_flow(name: str) -> Any: +async def validate_flow(flow: FlowArg) -> Any: """What would keep this flow from running: loops, unconnected inputs.""" - return await _call("POST", f"/flows/{_seg(name)}/validate") + return await _on_flow(flow, "POST", "/validate", mutation=False) # ----------------------------------------------------------------------------- @@ -275,57 +374,59 @@ async def validate_flow(name: str) -> Any: @mcp.tool() -async def run_flow(name: str, inputs: dict[str, Any] | None = None) -> Any: +async def run_flow(flow: FlowArg, inputs: dict[str, Any] | None = None) -> Any: """Run every node of a flow once and return the resulting state. With unpublished changes this runs the draft, so a change can be tried before it is published. """ - return await _call( - "POST", f"/flows/{_seg(name)}/run", json={"inputs": inputs or {}} + return await _on_flow( + flow, "POST", "/run", mutation=True, json={"inputs": inputs or {}} ) @mcp.tool() async def trigger_node( - name: str, node_id: str, values: dict[str, Any] | None = None + flow: FlowArg, node_id: str, values: dict[str, Any] | None = None ) -> Any: """Feed values into one node and run everything downstream of it.""" - return await _call( + return await _on_flow( + flow, "POST", - f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/trigger", + f"/nodes/{_seg(node_id)}/trigger", + mutation=True, json={"values": values or {}}, ) @mcp.tool() -async def start_flow(name: str) -> Any: +async def start_flow(flow: FlowArg) -> Any: """Let the engine run this flow: subscriptions, schedules and webhooks.""" - return await _call("POST", f"/flows/{_seg(name)}/start") + return await _on_flow(flow, "POST", "/start", mutation=True) @mcp.tool() -async def stop_flow(name: str) -> Any: +async def stop_flow(flow: FlowArg) -> Any: """Take a flow off the engine. Nothing of it stays subscribed or scheduled.""" - return await _call("POST", f"/flows/{_seg(name)}/stop") + return await _on_flow(flow, "POST", "/stop", mutation=True) @mcp.tool() -async def pause_flow(name: str) -> Any: +async def pause_flow(flow: FlowArg) -> Any: """Hold a flow's nodes while its incoming values keep arriving.""" - return await _call("POST", f"/flows/{_seg(name)}/pause") + return await _on_flow(flow, "POST", "/pause", mutation=True) @mcp.tool() -async def resume_flow(name: str) -> Any: +async def resume_flow(flow: FlowArg) -> Any: """Let a paused flow carry on, running whatever was held back.""" - return await _call("POST", f"/flows/{_seg(name)}/resume") + return await _on_flow(flow, "POST", "/resume", mutation=True) @mcp.tool() -async def cancel_node(name: str, node_id: str) -> Any: +async def cancel_node(flow: FlowArg, node_id: str) -> Any: """Stop a node's code while it is running. Nothing to stop is not an error.""" - return await _call("POST", f"/flows/{_seg(name)}/nodes/{_seg(node_id)}/cancel") + return await _on_flow(flow, "POST", f"/nodes/{_seg(node_id)}/cancel", mutation=True) # ----------------------------------------------------------------------------- @@ -360,36 +461,45 @@ async def get_health() -> Any: return await _call("GET", "/observability/summary") +#: The optional flow a history question is narrowed to. +FlowFilter = Annotated[ + str | None, + Field( + description=( + "Only this flow: its id or its key, or the key it had before it was " + "deleted. Write id: or key: when the name could mean two." + ) + ), +] + + @mcp.tool() async def get_metrics( - flow: str | None = None, node: str | None = None, hours: int = 24 + flow: FlowFilter = None, node: str | None = None, hours: int = 24 ) -> Any: - """Executions, errors and timings per minute over the last few hours.""" + """Executions, errors and timings per minute over the last few hours. + + ``node`` is ``.``. + """ params: dict[str, Any] = {"hours": hours} - if flow: - params["flow"] = flow if node: params["node"] = node - return await _call("GET", "/observability/timeseries", params=params) + return await _narrowed(flow, "/observability/timeseries", params) @mcp.tool() -async def list_failures(flow: str | None = None, limit: int = 50) -> Any: +async def list_failures(flow: FlowFilter = None, limit: int = 50) -> Any: """Recent failures with their tracebacks, newest first.""" params: dict[str, Any] = {"kind": "failure", "limit": limit} - if flow: - params["flow"] = flow - return await _call("GET", "/observability/events", params=params) + return await _narrowed(flow, "/observability/events", params) @mcp.tool() -async def list_runs(flow: str | None = None, limit: int = 50) -> Any: +async def list_runs(flow: FlowFilter = None, limit: int = 50) -> Any: """Recent cascades: what triggered them, how long they took, how they ended. A page of rows plus the total number matching, which says whether the limit cut anything off. """ params: dict[str, Any] = {"limit": limit} - if flow: - params["flow"] = flow - return await _call("GET", "/observability/runs", params=params) + return await _narrowed(flow, "/observability/runs", params) diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index bdffc07..c41fb5e 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -776,18 +776,23 @@ class Flow: def submit( self, *, seed: int | None = None, client: Any = None, **params: Any ) -> Any: - """Start a run of this flow and hand back a handle to it.""" + """Start a run of this flow and hand back a handle to it. + + By its key, which is all a declaration knows, and as a key: an id + that happens to be spelled the same is another flow's. + """ from fluksio.sdk.client import Client - return (client or Client()).submit(self.name, params, seed=seed) + return (client or Client()).submit(f"key:{self.name}", params, seed=seed) def runs(self, limit: int = 20, client: Any = None) -> list[dict[str, Any]]: - """This flow's runs, newest first.""" + """This flow's runs, newest first: the flow holding its key now, not + one deleted before it that held the key too.""" from fluksio.sdk.client import Client - rows: list[dict[str, Any]] = (client or Client()).runs( - flow=self.name, limit=limit - ) + engine = client or Client() + flow_id = engine.resolve_flow(f"key:{self.name}").id + rows: list[dict[str, Any]] = engine.runs(flow=f"id:{flow_id}", limit=limit) return rows def __repr__(self) -> str: diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index dd64ee2..f59015d 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -43,6 +43,7 @@ from fluksio.sdk.client import ( sync, ) from fluksio.sdk.declare import ImportFailed, declare, declare_module +from fluksio.sdk.resolve import Unresolved __all__ = ["add_parsers", "discover"] @@ -72,6 +73,21 @@ def _fail(message: str) -> int: NO_ENGINE = "Is one running? `fluksio serve` starts one." +@contextmanager +def _plain_warnings() -> Iterator[None]: + """What the SDK warns about, one line each on stderr. + + A terminal wants the sentence, not where in the library it was raised. + """ + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + try: + yield + finally: + for warning in caught: + print(f"fluksio: warning: {warning.message}", file=sys.stderr) + + def _unreachable(exc: Exception, note: str = NO_ENGINE) -> int: """The engine did not answer. Say so as a sentence, not a traceback.""" return _fail( @@ -387,15 +403,19 @@ def cmd_sync(args: argparse.Namespace) -> int: return _unreachable(exc, "Nothing was published; sync again when it is back.") for report in reports: + # The id too when it is not the key, since that is what a URL holds. + name = ( + report.flow if report.id == report.flow else f"{report.flow} ({report.id})" + ) if report.unchanged: - _say(f" {report.flow}: unchanged") + _say(f" {name}: unchanged") continue what = "created" if report.created else "updated" detail = ", ".join(report.changed) # Nothing to publish is not the same as left as a draft, and saying # the second when it was the first reads as work not finished. state = "published" if report.published else "draft" if report.drafted else "" - _say(f" {report.flow}: {what} ({detail})" + (f" — {state}" if state else "")) + _say(f" {name}: {what} ({detail})" + (f" — {state}" if state else "")) if any(report.forgot_code for report in reports): engine = _engine_version(client) _say( @@ -533,7 +553,13 @@ def _declaration(client: Client, flow: str) -> tuple[str, str, Path] | None: """ if not flow: return None - stored = client.get_flow(flow) or {} + try: + # As the run that follows will: a name two flows answer to is refused + # there, and not guessed at here. + flow_id = client.resolve_flow(flow, mutation=True).id + except Unresolved: + return None + stored = client.get_flow(flow_id) or {} origin = (stored.get("definition") or {}).get("origin") or {} declared_in = str(origin.get("declared_in") or "") module = str(origin.get("module") or "") @@ -716,7 +742,9 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: with _client_for(args) as client: if not args.no_sync: _sync_first(client, args.sync, args.flow) - stored = client.get_flow(args.flow) + # After the sync, which may be what creates it. + flow_id = client.resolve_flow(args.flow, mutation=True).id + stored = client.get_flow(flow_id) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") definition = stored.get("definition") or {} @@ -732,7 +760,7 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int: ): params = _ask_params(definition) handle = client.submit( - args.flow, + f"id:{flow_id}", params, seed=args.seed, no_cache=args.no_cache, @@ -938,7 +966,7 @@ def _status_screen(client: Client) -> Any: state, "red" if state == "error" else "yellow" ), ), - Text(str(row.get("flow", "")), style="dim"), + Text(str(row.get("flow_name") or row.get("flow", "")), style="dim"), Text(_dur(row.get("duration_ms")), style="dim"), Text( _ago(row.get("finished_at") or row.get("created_at")), style="dim" @@ -952,7 +980,9 @@ def _status_screen(client: Client) -> Any: broken.add_column("") for event in failures: where = " ".join( - str(event.get(key, "")) for key in ("flow", "node") if event.get(key) + str(event.get(key, "")) + for key in ("flow_name", "node") + if event.get(key) ) broken.add_row( Text(f" {where}", style="red"), @@ -1076,7 +1106,8 @@ def _stamp(row: dict[str, Any]) -> str: def cmd_runs(args: argparse.Namespace) -> int: try: with _client_for(args, retries=0) as client: - rows = client.runs(flow=args.flow, limit=args.limit) + with _plain_warnings(): + rows = client.runs(flow=args.flow, limit=args.limit) declared = _declared(client, {str(row["flow"]) for row in rows}) except (SyncError, ApiError) as exc: return _fail(str(exc)) @@ -1100,8 +1131,9 @@ def cmd_runs(args: argparse.Namespace) -> int: params = json.dumps(shown) if len(params) > room: params = params[: room - 3] + "..." + flow = row.get("flow_name") or row["flow"] _say( - f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " + f"{row['id']} {_status(row['status'], 9)} {flow:<16} " f"{_dur(row['duration_ms']):>8} {_ago(row.get('created_at')):>9} " f"{_stamp(row):<22} {params}" ) @@ -1269,11 +1301,12 @@ def cmd_sweep(args: argparse.Namespace) -> int: with _client_for(args) as client: if not args.no_sync: _sync_first(client, args.sync, args.flow) - stored = client.get_flow(args.flow) + flow_id = client.resolve_flow(args.flow, mutation=True).id + stored = client.get_flow(flow_id) if stored is None: return _fail(f"no flow '{args.flow}' on that engine") entries = _grid(stored.get("definition") or {}, args.param, args.seed) - handles = client.sweep(args.flow, entries, no_cache=args.no_cache) + handles = client.sweep(f"id:{flow_id}", entries, no_cache=args.no_cache) for handle, entry in zip(handles, entries, strict=True): _say(f"{handle.id} queued {json.dumps(entry['params'])}") if not wait: @@ -1401,13 +1434,10 @@ def _export( try: with _client_for(args, retries=0) as client: try: - if getattr(args, "list_names", False): - return _list_names(client, args) - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") + with _plain_warnings(): + if getattr(args, "list_names", False): + return _list_names(client, args) rows = fetch(client) - for warning in caught: - print(f"fluksio: warning: {warning.message}", file=sys.stderr) except ApiError as exc: if exc.status != 404: raise diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 9a607ad..83589e9 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -18,7 +18,9 @@ from pathlib import Path from typing import Any from urllib.parse import quote +from fluksio import __version__ from fluksio.sdk import MARKER, Declared, Flow, SyncError +from fluksio.sdk.resolve import Resolved, resolve __all__ = [ "Client", @@ -62,6 +64,15 @@ WAIT_TOLERANCE = 5 #: Runs per `POST /runs/import`, which is what the engine caps one request at. IMPORT_PAGE = 200 +#: What to do about an engine from before flow ids. It lists flows without an +#: id and has no route to create one by, so this client cannot tell which flow +#: a name means, nor make one. +OLD_ENGINE = ( + "this engine is older than flow ids, and this client " + f"({__version__}) needs them. Upgrade the engine first: " + "`pip install -U fluksio` where it runs" +) + #: What a project-local instance is called, beside `.venv` and `.git`. DATA_DIR_NAME = ".fluksio" @@ -251,13 +262,49 @@ class Client: """The listing of the flow holding the key *key*, or ``None``.""" result = self._call("GET", "/flows/", params={"name": key}) found = list(result.get("data") or []) + if found and "id" not in found[0]: + # It ignored the filter too, so this is every flow, not the one. + raise SyncError(OLD_ENGINE) return found[0] if found else None def create_flow(self, document: dict[str, Any]) -> dict[str, Any]: """A new flow, under the id the engine issues for it.""" - result: dict[str, Any] = self._call("POST", "/flows/", json=document) + try: + result: dict[str, Any] = self._call("POST", "/flows/", json=document) + except ApiError as exc: + if exc.status in (404, 405): + raise SyncError(OLD_ENGINE) from exc + raise return result + def resolve_flow( + self, flow: str, *, mutation: bool = False, history: bool = False + ) -> Resolved: + """The flow *flow* names, by its id or its key (`fluksio.sdk.resolve`). + + ``history`` counts deleted flows too, under the key their runs last + had, for a question about the past. A name two flows answer to is + refused when ``mutation`` says the call changes something, and comes + back with a warning otherwise, for the caller to show. + """ + rows = self.flows() + if any("id" not in row for row in rows): + raise SyncError(OLD_ENGINE) + flows = [(str(row["id"]), str(row["name"])) for row in rows] + if history: + flows += [ + (str(row["flow"]), str(row.get("flow_name") or row["flow"])) + for row in self._call("GET", "/runs/overview") + ] + return resolve(flow, flows, mutation=mutation) + + def _flow_id(self, flow: str, **how: bool) -> str: + """The id behind *flow*, warning when it could have meant another.""" + found = self.resolve_flow(flow, **how) + if found.warning: + warnings.warn(found.warning, stacklevel=3) + return found.id + def put_flow(self, document: dict[str, Any]) -> dict[str, Any]: """Save over the flow *document* names by its id, or by its key for one whose id is its key.""" @@ -338,14 +385,15 @@ class Client: ) -> list[dict[str, Any]]: """What went wrong, or who changed what. Newest first. - Engine-wide, and narrowed by flow, by run or by time. ``run`` takes a - run id and answers what the engine recorded during it. For the failure - that ended a run, :attr:`RunHandle.failures` is still the shorter road: - it reads the run's own node rows, which carry the traceback too. + Engine-wide, and narrowed by flow (its id or key), by run or by time. + ``run`` takes a run id and answers what the engine recorded during it. + For the failure that ended a run, :attr:`RunHandle.failures` is still + the shorter road: it reads the run's own node rows, which carry the + traceback too. """ query: dict[str, Any] = {"kind": kind, "limit": limit} if flow: - query["flow"] = flow + query["flow"] = self._flow_id(flow, history=True) if run: query["run"] = run for name, value in (("since", since), ("until", until)): @@ -366,8 +414,9 @@ class Client: no_cache: bool = False, cause: str = "sdk", ) -> RunHandle: - """Queue a run. ``cause`` is what the history records it as coming - from — a script is the default, `fluksio run` says so itself. + """Queue a run of the flow *flow*, its id or key. ``cause`` is what + the history records it as coming from — a script is the default, + `fluksio run` says so itself. Safe to retry: the key is minted here, once per call, so an attempt that timed out on the way back is answered with the run it made rather @@ -375,7 +424,7 @@ class Client: """ row = self._call( "POST", - f"/runs/flows/{seg(flow)}", + f"/runs/flows/{seg(self._flow_id(flow, mutation=True))}", idempotent=True, json={ "params": params or {}, @@ -401,7 +450,7 @@ class Client: keyed = [{**entry, "idempotency_key": uuid.uuid4().hex} for entry in entries] rows = self._call( "POST", - f"/runs/flows/{seg(flow)}/sweep", + f"/runs/flows/{seg(self._flow_id(flow, mutation=True))}/sweep", idempotent=True, json={"runs": keyed, "no_cache": no_cache}, ) @@ -410,10 +459,17 @@ class Client: def import_runs(self, entries: list[dict[str, Any]]) -> list[dict[str, Any]]: """Record runs executed somewhere else, a page at a time. - Each entry is the shape `POST /runs/import` takes. Safe to send again: - one whose `external_id` is already on the engine answers - `created: False` and adds nothing. + Each entry is the shape `POST /runs/import` takes, but its `flow` may + be the key as well as the id. Safe to send again: one whose + `external_id` is already on the engine answers `created: False` and + adds nothing. """ + named = {str(entry["flow"]) for entry in entries if entry.get("flow")} + ids = {flow: self._flow_id(flow, mutation=True) for flow in named} + entries = [ + {**entry, "flow": ids[str(entry["flow"])]} if entry.get("flow") else entry + for entry in entries + ] answers: list[dict[str, Any]] = [] for start in range(0, len(entries), IMPORT_PAGE): answers.extend( @@ -431,9 +487,10 @@ class Client: return result def runs(self, flow: str = "", limit: int = 20, **filters: Any) -> Any: + """Runs, newest first, of every flow or of *flow*, its id or key.""" query = {"limit": limit, **filters} if flow: - query["flow"] = flow + query["flow"] = self._flow_id(flow, history=True) return self._call("GET", "/runs", params=query) def metrics(self, run_id: str, name: str = "", stride: int = 1) -> Any: @@ -456,7 +513,7 @@ class Client: if named: query["ids"] = ",".join(named) if flow: - query["flow"] = flow + query["flow"] = self._flow_id(flow, history=True) return self._call("GET", "/runs/metrics/names", params=query) def compare(self, ids: Iterable[str], metric: str, x: str = "") -> Any: @@ -482,7 +539,8 @@ class Client: """Runs' series as long rows: run, name, step, ts, value. ``pd.DataFrame(client.export_metrics(flow="train"))`` is every loss - curve of that flow with the run id on each row. ``name`` keeps the + curve of that flow with the run id on each row. ``flow`` is its id or + key. ``name`` keeps the metrics it lists, ``stride`` thins each curve, and the selection is narrowed the way :meth:`runs` is — ``status=``, ``group=``, ``since=``, ``until=``. @@ -530,7 +588,7 @@ class Client: and that is a warning, since the rows alone look complete. """ if flow: - query["flow"] = flow + query["flow"] = self._flow_id(flow, history=True) named = ",".join(ids) if named: query["ids"] = named @@ -728,6 +786,9 @@ class SyncReport: def __init__(self, flow: str) -> None: self.flow = flow + #: The id this engine holds the flow under. Another engine's differs: + #: the key is what instances share. + self.id = "" self.created = False self.changed: list[str] = [] self.published = False @@ -835,30 +896,41 @@ def _sync_one( force: bool, ) -> SyncReport: report = SyncReport(target.name) - # The declared key is what an instance shares; the id is this engine's, - # and every request after the lookup names the flow by it. - found = client.find_flow(target.name) - flow_id = str(found["id"]) if found else "" - stored = client.get_flow(flow_id) if flow_id else None + document = target.document(origin) + # The declared key is what instances share; the id is this engine's, and + # every request after the lookup names the flow by it. Looked up a second + # time only when another sync created the flow in between, and then its + # flow is checked as any stored one is before it is written over. + raced = False version = 1 # What the drift check already read, so the write loop below does not ask # the engine for every node's body a second time. known: dict[str, str] = {} - if stored is None: + while True: + found = client.find_flow(target.name) + flow_id = str(found["id"]) if found else "" + stored = client.get_flow(flow_id) if flow_id else None + if stored is not None: + definition = stored.get("definition") or {} + version = int(definition.get("version") or 1) + if not force: + known = _refuse_on_drift(client, target, flow_id, definition) + saved = client.put_flow(document | {"id": flow_id, "version": version}) + break + try: + saved = client.create_flow(document) + except Conflict: + # The key was free a moment ago. Anything else the engine refuses + # a create for is refused again, and that one is raised. + if raced: + raise + raced = True + continue report.created = True - else: - definition = stored.get("definition") or {} - version = int(definition.get("version") or 1) - if not force: - known = _refuse_on_drift(client, target, definition) - - document = target.document(origin) + flow_id = str((saved.get("definition") or {}).get("id") or "") + break + report.id = flow_id moved = _moved_code(stored, document) - if stored is None: - saved = client.create_flow(document) - flow_id = str((saved.get("definition") or {}).get("id") or target.name) - else: - saved = client.put_flow(document | {"id": flow_id, "version": version}) # An engine older than the per-node digest parses the document without # those fields and stores it without them, silently: no error, no 404, and # a cache still keyed on the whole repository. @@ -930,7 +1002,7 @@ def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list def _refuse_on_drift( - client: Client, target: Flow | Declared, definition: dict[str, Any] + client: Client, target: Flow | Declared, flow_id: str, definition: dict[str, Any] ) -> dict[str, str]: """Stop before overwriting work that was done somewhere else. @@ -956,9 +1028,7 @@ def _refuse_on_drift( for stored_node in definition.get("nodes") or []: node_id = str(stored_node.get("id")) try: - entry = client.get_source_entry( - str(definition.get("id") or target.name), node_id - ) + entry = client.get_source_entry(flow_id, node_id) except ApiError: continue if entry.get("missing"): diff --git a/backend/fluksio/sdk/resolve.py b/backend/fluksio/sdk/resolve.py new file mode 100644 index 0000000..f9f010d --- /dev/null +++ b/backend/fluksio/sdk/resolve.py @@ -0,0 +1,87 @@ +"""Which flow an argument names: its id or its key, id first. + +The API takes ids and only ids. People type keys, and scripts written before +ids existed pass keys, so the CLI, the SDK and the MCP tools answer the same +question here, with the same words. + +Pure: the caller fetches what it may see with its own credentials and hands +it over as ``(id, key)`` pairs. The SDK carries no engine code, so this lives +with it, and the server's MCP tools import it from here. +""" + +from __future__ import annotations + +import difflib +from collections.abc import Iterable +from dataclasses import dataclass + +from fluksio.sdk import SyncError + +__all__ = ["Resolved", "Unresolved", "resolve"] + +#: The explicit forms. A colon is legal in neither an id nor a key, so they +#: cannot clash with a real value. +ID = "id:" +KEY = "key:" + + +class Unresolved(SyncError): + """Nothing, or more than one thing, answers to the argument.""" + + +@dataclass(frozen=True) +class Resolved: + id: str + #: Empty for an explicit id nobody lists any more. + key: str + #: Set when the argument was one flow's id and another's key, and the + #: id was taken. Read-only callers show it; a change never gets this far. + warning: str = "" + + +def resolve( + value: str, flows: Iterable[tuple[str, str]], *, mutation: bool = False +) -> Resolved: + """The flow *value* names among *flows*. + + ``id:`` is taken as said, listed or not: history outlives its flow, + and the engine answers for an id nobody holds. ``key:`` is the flow + holding that key. A plain string is an id if one has it, else a key. + + A plain string that is one flow's id and another's key picks the id with + a warning, or is refused when *mutation* says the call changes something. + A key several flows answer to, which history can hold, is always refused. + """ + keys: dict[str, str] = {} + for flow_id, key in flows: + keys.setdefault(flow_id, key) + if value.startswith(ID): + named = value[len(ID) :] + return Resolved(named, keys.get(named, "")) + explicit = value.startswith(KEY) + key = value[len(KEY) :] if explicit else value + holders = [flow_id for flow_id, held in keys.items() if held == key] + if not explicit and value in keys: + others = [flow_id for flow_id in holders if flow_id != value] + if not others: + return Resolved(value, keys[value]) + both = ( + f"'{value}' is the id of '{keys[value]}' and the key of " + f"'{', '.join(others)}'." + ) + if mutation: + raise Unresolved(f"{both} Pass id:{value} or key:{value}.") + return Resolved( + value, keys[value], f"{both} Using the id. Pass key:{value} for the other." + ) + if len(holders) == 1: + return Resolved(holders[0], key) + if holders: + raise Unresolved( + f"Several flows have had the key '{key}': {', '.join(holders)}. " + "Pass id:." + ) + close = difflib.get_close_matches(key, sorted(set(keys.values())), n=3) + hint = f" Closest keys: {', '.join(close)}." if close else "" + what = "key" if explicit else "id or key" + raise Unresolved(f"No flow with the {what} '{key}'.{hint}") diff --git a/backend/fluksio/tui/app.py b/backend/fluksio/tui/app.py index e3ea5ed..7b3d20e 100644 --- a/backend/fluksio/tui/app.py +++ b/backend/fluksio/tui/app.py @@ -657,7 +657,7 @@ class ServeApp(App[int]): "·" if run_id in self.picked else " ", run_id[-8:], str(row.get("status", "")), - str(row.get("flow", "")), + str(row.get("flow_name") or row.get("flow", "")), _dur(row.get("duration_ms")), params.ljust(room) if len(params) <= room else params[: room - 1] + "…", key=run_id, diff --git a/backend/fluksio/tui/compare.py b/backend/fluksio/tui/compare.py index 638dc31..660f531 100644 --- a/backend/fluksio/tui/compare.py +++ b/backend/fluksio/tui/compare.py @@ -163,7 +163,7 @@ class CompareScreen(Screen[None]): def show_table(self, rows: list[dict[str, Any]], names: list[str]) -> None: self.names = names - flow = str(rows[0].get("flow", "")) if rows else "" + flow = str(rows[0].get("flow_name") or rows[0].get("flow", "")) if rows else "" drawn = ( f" {MAX_SERIES} of {len(self.ids)} drawn" if len(self.ids) > MAX_SERIES diff --git a/backend/tests/api/routes/test_sync.py b/backend/tests/api/routes/test_sync.py index 3cee2d1..c1eba84 100644 --- a/backend/tests/api/routes/test_sync.py +++ b/backend/tests/api/routes/test_sync.py @@ -213,6 +213,70 @@ def test_sync_refuses_a_flow_it_did_not_create(api, flows): theirs.name = "train" +def test_two_first_syncs_of_one_key_end_with_one_flow(api, flows): + """The second looked before the first had created the flow. + + Its create is refused for the key, so it looks again and updates the + first one's flow, checked first as any stored flow is: a canvas edit made + in between is refused rather than overwritten. + """ + + class Late(Client): + looked = False + + def find_flow(self, key): + if self.looked: + return super().find_flow(key) + # What it saw before the other sync created the flow. + self.looked = True + return None + + target = flows["train"] + target.name = "raced" + try: + first = sync([target], api, origin=ORIGIN)[0] + second = sync([target], Late(http=api.http, token=api.token), origin=ORIGIN)[0] + + assert first.created and not second.created + assert second.id == first.id + assert [f["id"] for f in api.flows() if f["name"] == "raced"] == [first.id] + + api.put_source(first.id, "fit", "def process(dataset, lr):\n return {}\n") + with pytest.raises(SyncError, match="edited on the canvas"): + sync([target], Late(http=api.http, token=api.token), origin=ORIGIN) + finally: + target.name = "train" + found = api.find_flow("raced") + if found: + api.http.delete(f"{PREFIX}/flows/{found['id']}") + + +def test_a_synced_flow_is_addressed_by_its_key_under_an_issued_id(api, flows): + """The id is the engine's own, opaque; the client takes the key and + sends the id, and the history answers with both.""" + report = sync([flows["train"]], api, origin=ORIGIN)[0] + train = flow_id(api, "train") + assert report.id == train + assert train != "train" and train.startswith("f_") + + [made] = api.import_runs( + [ + { + "flow": "train", + "external_id": "elsewhere:1", + "params": {"lr": 0.05}, + "created_at": "2026-03-01T10:00:00Z", + "result": {"score": 0.9}, + } + ] + ) + rows = api.runs(flow="train") + assert made["id"] in {row["id"] for row in rows} + assert {(row["flow"], row["flow_name"]) for row in rows} == {(train, "train")} + exported = api.export_runs(flow="train") + assert {(row["flow"], row["flow_name"]) for row in exported} == {(train, "train")} + + def test_use_stores_the_rewiring_it_was_given(api, flows): sync([flows["finetune"]], api, origin=ORIGIN) diff --git a/backend/tests/mcp/test_mcp_http.py b/backend/tests/mcp/test_mcp_http.py index 87b71cf..bb13e16 100644 --- a/backend/tests/mcp/test_mcp_http.py +++ b/backend/tests/mcp/test_mcp_http.py @@ -264,6 +264,13 @@ def test_an_agent_cannot_write_code_its_person_may_not( ).json() assert me["is_superuser"] is False token = mcp_token(uuid.UUID(me["id"]), agent) + # One they may create: the tool only reaches the route for a listed flow. + flows = f"{settings.API_V1_STR}/flows" + made = client.post( + f"{flows}/", headers=normal_user_token_headers, json={"name": "agent_gate"} + ) + assert made.status_code == 201, made.text + flow_id = made.json()["definition"]["id"] async def block(http: httpx.AsyncClient) -> httpx.Response: return await http.post( @@ -272,11 +279,14 @@ def test_an_agent_cannot_write_code_its_person_may_not( token, "tools/call", name="save_node_source", - arguments={"name": "any", "node_id": "fn", "code": "import os"}, + arguments={"flow": "agent_gate", "node_id": "fn", "code": "import os"}, ), ) - answer = json.loads(over_mcp(block).json()["result"]["content"][0]["text"]) + try: + answer = json.loads(over_mcp(block).json()["result"]["content"][0]["text"]) + finally: + client.delete(f"{flows}/{flow_id}", headers=normal_user_token_headers) assert answer == {"error": CODE_IS_SUPERUSERS, "status": 403} diff --git a/backend/tests/mcp/test_mcp_tools.py b/backend/tests/mcp/test_mcp_tools.py index 6f6bf48..ddbf86a 100644 --- a/backend/tests/mcp/test_mcp_tools.py +++ b/backend/tests/mcp/test_mcp_tools.py @@ -6,38 +6,80 @@ import httpx from fluksio.mcp import server +#: What the engine lists: the grandfathered `heating`, now keyed `warmth`, and +#: the later flow that took its old key. +LISTED = [{"id": "heating", "name": "warmth"}, {"id": "f_heat0000", "name": "heating"}] -def test_a_name_cannot_climb_out_of_its_route() -> None: - """httpx resolves ``..``, so ``save_flow("../ai/config")`` was ``PUT /ai/config``. - A prompt-injected assistant acting for a superuser could repoint the - provider that way while the stored key stayed. - """ +def _calls(*tools) -> list[str]: + """What the tools sent past the flow listing, run against a recording + engine.""" seen: list[str] = [] def record(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/flows/" and request.method == "GET": + return httpx.Response(200, json={"data": LISTED, "count": len(LISTED)}) seen.append(f"{request.method} {request.url.raw_path.decode()}") return httpx.Response(200, json={}) - async def calls() -> None: + async def calls() -> list: previous = server._client mock = httpx.AsyncClient( transport=httpx.MockTransport(record), base_url="http://test" ) server.set_client(mock) try: - await server.save_flow("../ai/config", {}) - await server.delete_flow("..") - await server.get_node_source("..", "../source") + return [await tool() for tool in tools] finally: server.set_client(previous) await mock.aclose() - asyncio.run(calls()) + answers = asyncio.run(calls()) + return seen + [answer for answer in answers if answer != {}] + + +def test_a_name_cannot_climb_out_of_its_route() -> None: + """httpx resolves ``..``, so ``save_flow("../ai/config")`` was ``PUT /ai/config``. + + A prompt-injected assistant acting for a superuser could repoint the + provider that way while the stored key stayed. A flow is now only ever a + listed one, and what follows it is still one segment. + """ + seen = _calls( + lambda: server.save_flow("../ai/config", {}), + lambda: server.get_node_source("f_heat0000", "../source"), + ) + assert seen == [ - "PUT /api/v1/flows/%2E%2E%2Fai%2Fconfig", - "DELETE /api/v1/flows/%2E%2E", - "GET /api/v1/flows/%2E%2E/nodes/%2E%2E%2Fsource/source", + "GET /api/v1/flows/f_heat0000/nodes/%2E%2E%2Fsource/source", + {"error": "No flow with the id or key '../ai/config'."}, + ] + + +def test_a_flow_argument_goes_through_the_resolver() -> None: + """Id first; a read of a name two flows answer to says so, a change of + it is refused, and the explicit forms say which.""" + seen = _calls( + lambda: server.get_flow("heating"), + lambda: server.delete_flow("heating"), + lambda: server.delete_flow("key:heating"), + lambda: server.get_flow("warmth"), + ) + + warning = ( + "'heating' is the id of 'warmth' and the key of 'f_heat0000'. " + "Using the id. Pass key:heating for the other." + ) + refusal = ( + "'heating' is the id of 'warmth' and the key of 'f_heat0000'. " + "Pass id:heating or key:heating." + ) + assert seen == [ + "GET /api/v1/flows/heating", + "DELETE /api/v1/flows/f_heat0000", + "GET /api/v1/flows/heating", + {"warning": warning}, + {"error": refusal}, ] diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py index 8092f7b..aaf7c1a 100644 --- a/backend/tests/sdk/test_client.py +++ b/backend/tests/sdk/test_client.py @@ -3,7 +3,26 @@ import httpx import pytest +from fluksio.sdk import SyncError from fluksio.sdk.client import WAIT_TOLERANCE, ApiError, Client, RunHandle +from fluksio.sdk.resolve import Unresolved + +#: What the engine lists: `train` under an issued id, and the grandfathered +#: `heating`, now keyed `warmth`, whose old key a later flow took. +LISTED = [ + {"id": "f_train000", "name": "train"}, + {"id": "heating", "name": "warmth"}, + {"id": "f_heat0000", "name": "heating"}, +] + + +def listing(request): + """The engine's answer to the lookups a flow argument costs, else None.""" + if request.url.path == "/api/v1/flows/": + return httpx.Response(200, json={"data": LISTED, "count": len(LISTED)}) + if request.url.path == "/api/v1/runs/overview": + return httpx.Response(200, json=[]) + return None def a_client(handler, monkeypatch, **kwargs): @@ -65,6 +84,9 @@ def test_submit_carries_one_key_across_its_retries(monkeypatch): bodies = [] def handler(request): + if listed := listing(request): + return listed + assert request.url.path == "/api/v1/runs/flows/f_train000" bodies.append(httpx.Response(200, content=request.content).json()) if len(bodies) < 3: raise httpx.ConnectError("no route", request=request) @@ -81,6 +103,8 @@ def test_a_sweep_keys_every_entry(monkeypatch): seen = {} def handler(request): + if listed := listing(request): + return listed body = httpx.Response(200, content=request.content).json() seen["runs"] = body["runs"] return httpx.Response(202, json=[{"id": "r1"}, {"id": "r2"}]) @@ -137,6 +161,8 @@ def test_an_export_is_read_as_rows(monkeypatch): seen = {} def handler(request): + if listed := listing(request): + return listed seen.update(request.url.params) return httpx.Response( 200, @@ -149,7 +175,8 @@ def test_an_export_is_read_as_rows(monkeypatch): assert seen["format"] == "jsonl" assert seen["ids"] == "r1,r2" - assert seen["flow"] == "train" + # The key, resolved: the API takes the id and only the id. + assert seen["flow"] == "f_train000" assert seen["status"] == "ok" assert [row["step"] for row in rows] == [0, 1] @@ -201,3 +228,46 @@ def test_a_name_cannot_climb_out_of_its_route(monkeypatch): "/api/v1/flows/%2E%2E/nodes/%2E%2E%2Fx/source", "/api/v1/runs/%2E%2E", ] + + +def test_a_flow_argument_goes_through_the_resolver(monkeypatch): + """A read of a name two flows answer to warns; a run of it is refused.""" + asked = [] + + def handler(request): + if listed := listing(request): + return listed + asked.append( + f"{request.method} {request.url.path}?{request.url.query.decode()}" + ) + return httpx.Response(200, json=[]) + + client = a_client(handler, monkeypatch) + with pytest.warns(UserWarning, match="Pass key:heating for the other"): + client.runs(flow="heating") + with pytest.raises(Unresolved, match="Pass id:heating or key:heating"): + client.submit("heating") + client.runs(flow="key:heating") + + assert asked == [ + "GET /api/v1/runs?limit=20&flow=heating", + "GET /api/v1/runs?limit=20&flow=f_heat0000", + ] + + +def test_an_engine_from_before_ids_is_named_as_such(monkeypatch): + """It lists flows without ids and has no route to create one by.""" + + def handler(request): + if request.method == "POST": + return httpx.Response(405, json={"detail": "Method Not Allowed"}) + return httpx.Response(200, json={"data": [{"name": "train"}], "count": 1}) + + client = a_client(handler, monkeypatch) + for call in ( + lambda: client.runs(flow="train"), + lambda: client.find_flow("train"), + lambda: client.create_flow({"name": "train"}), + ): + with pytest.raises(SyncError, match="older than flow ids"): + call() diff --git a/backend/tests/sdk/test_declare.py b/backend/tests/sdk/test_declare.py index 9fc9119..6877b0c 100644 --- a/backend/tests/sdk/test_declare.py +++ b/backend/tests/sdk/test_declare.py @@ -8,6 +8,7 @@ import pytest from fluksio.sdk import FLOWS, SyncError, cli from fluksio.sdk.declare import ImportFailed, declare, declare_module +from fluksio.sdk.resolve import Resolved STUDY = """\ import os @@ -131,6 +132,9 @@ def test_a_failing_import_is_reported_by_sync_and_warned_about_by_run( assert "Traceback" in seen.err and "study.py" in seen.err class Engine: + def resolve_flow(self, flow, **how): + return Resolved(flow, flow) + def get_flow(self, name): return { "definition": {"origin": {"declared_in": "study.py", "module": "study"}} diff --git a/backend/tests/sdk/test_resolve.py b/backend/tests/sdk/test_resolve.py new file mode 100644 index 0000000..55936f1 --- /dev/null +++ b/backend/tests/sdk/test_resolve.py @@ -0,0 +1,74 @@ +"""Which flow an argument names: the one matrix every client relies on.""" + +import pytest + +from fluksio.sdk.resolve import Resolved, Unresolved, resolve + +#: `heating` is the grandfathered id of the flow now keyed `warmth`; a later +#: flow took the key `heating` under an issued id. +FLOWS = [ + ("heating", "warmth"), + ("f_k3q7m2xa", "heating"), + ("train", "train"), + ("f_aaaaaaaa", "lights"), +] + + +def test_an_id_alone_picks_its_flow(): + assert resolve("f_aaaaaaaa", FLOWS) == Resolved("f_aaaaaaaa", "lights") + + +def test_a_key_alone_picks_its_flow(): + assert resolve("lights", FLOWS) == Resolved("f_aaaaaaaa", "lights") + + +def test_an_id_that_is_its_own_key_is_no_ambiguity(): + assert resolve("train", FLOWS, mutation=True) == Resolved("train", "train") + + +def test_one_flows_id_and_anothers_key_reads_the_id_with_a_warning(): + found = resolve("heating", FLOWS) + + assert (found.id, found.key) == ("heating", "warmth") + assert found.warning == ( + "'heating' is the id of 'warmth' and the key of 'f_k3q7m2xa'. " + "Using the id. Pass key:heating for the other." + ) + + +def test_a_change_refuses_a_name_two_flows_answer_to(): + with pytest.raises(Unresolved, match="Pass id:heating or key:heating"): + resolve("heating", FLOWS, mutation=True) + + +def test_the_explicit_forms_say_which(): + assert resolve("id:heating", FLOWS, mutation=True) == Resolved("heating", "warmth") + assert resolve("key:heating", FLOWS, mutation=True) == Resolved( + "f_k3q7m2xa", "heating" + ) + + +def test_neither_names_the_closest_keys(): + with pytest.raises(Unresolved) as caught: + resolve("lihgts", FLOWS) + + assert str(caught.value) == ( + "No flow with the id or key 'lihgts'. Closest keys: lights." + ) + with pytest.raises(Unresolved, match="No flow with the key 'f_aaaaaaaa'"): + resolve("key:f_aaaaaaaa", FLOWS) + + +def test_an_explicit_id_outlives_its_flow(): + """History stays filed under a deleted flow's id, which nobody lists.""" + assert resolve("id:f_gone0000", FLOWS) == Resolved("f_gone0000", "") + + +def test_a_key_history_holds_twice_is_refused_not_guessed(): + """A deleted flow and the later one that took its key.""" + history = [("f_old00000", "train"), ("f_new00000", "train")] + + for value in ("train", "key:train"): + with pytest.raises(Unresolved, match="f_old00000, f_new00000"): + resolve(value, history) + assert resolve("id:f_old00000", history) == Resolved("f_old00000", "train") diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index 5416d93..e0e4861 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -7,6 +7,7 @@ from pathlib import Path import pytest from fluksio.cli import load_or_create_secret_key +from fluksio.sdk.resolve import Resolved def test_importing_the_cli_does_not_build_the_settings() -> None: @@ -242,6 +243,9 @@ def test_a_local_run_always_waits(monkeypatch) -> None: return self class FakeClient: + def resolve_flow(self, flow: str, **how: bool) -> Resolved: + return Resolved(flow, flow) + def get_flow(self, name: str) -> dict[str, object]: return {"definition": {"inputs": []}} @@ -264,7 +268,8 @@ def test_a_local_run_always_waits(monkeypatch) -> None: assert cli.cmd_run(args, []) == 0 # "cli" rather than the SDK's default: the history says which asked. assert submitted == { - "flow": "train", + # Resolved once, then named by id so the submit does not ask again. + "flow": "id:train", "no_cache": True, "waited": True, "cause": "cli", @@ -289,6 +294,9 @@ def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None: raise KeyboardInterrupt class FakeClient: + def resolve_flow(self, flow: str, **how: bool) -> Resolved: + return Resolved(flow, flow) + def get_flow(self, name: str) -> dict[str, object]: return {"definition": {"inputs": []}} @@ -429,6 +437,9 @@ def test_a_failed_run_prints_what_each_node_said(monkeypatch, capsys) -> None: return self class FakeClient: + def resolve_flow(self, flow: str, **how: bool) -> Resolved: + return Resolved(flow, flow) + def get_flow(self, name: str) -> dict[str, object]: return {"definition": {"inputs": []}} @@ -699,6 +710,9 @@ def test_run_imports_the_module_that_declared_the_flow(tmp_path, monkeypatch): class Engine: origin: dict = {} + def resolve_flow(self, flow, **how): + return Resolved(flow, flow) + def get_flow(self, name): return {"definition": {"origin": self.origin}} @@ -1196,3 +1210,50 @@ def test_the_dashboard_subscribes_where_the_browser_does() -> None: assert socket_url("http://127.0.0.1:8000") == "ws://127.0.0.1:8000/api/v1/flows/ws" # TLS on the one side is TLS on the other. assert socket_url("https://engine.example.com").startswith("wss://") + + +def test_a_flow_argument_goes_through_the_resolver(monkeypatch, capsys) -> None: + """A run of a name two flows answer to is refused; a listing warns.""" + from contextlib import contextmanager + + import httpx + + from fluksio.cli import _parser + from fluksio.sdk import cli + from fluksio.sdk.client import Client + + listed = [ + {"id": "heating", "name": "warmth"}, + {"id": "f_heat0000", "name": "heating"}, + ] + asked: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/flows/": + return httpx.Response(200, json={"data": listed, "count": 2}) + if request.url.path == "/api/v1/runs/overview": + return httpx.Response(200, json=[]) + asked.append(f"{request.url.path}?{request.url.query.decode()}") + return httpx.Response(200, json=[]) + + @contextmanager + def engine(): + http = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://e") + yield Client(http=http, token="t") + + monkeypatch.setattr(cli, "_engine_client", engine) + parser = _parser() + + run = parser.parse_args(["run", "heating", "--local", "--no-sync"]) + assert cli.cmd_run(run, []) == 1 + assert "Pass id:heating or key:heating" in capsys.readouterr().err + assert asked == [] + + assert ( + cli.cmd_runs(parser.parse_args(["runs", "--flow", "heating", "--local"])) == 0 + ) + assert ( + "fluksio: warning: 'heating' is the id of 'warmth' and the key of " + "'f_heat0000'. Using the id. Pass key:heating for the other.\n" + ) in capsys.readouterr().err + assert asked == ["/api/v1/runs?limit=20&flow=heating"] diff --git a/docs/code/agents.md b/docs/code/agents.md index 32dfaeb..f57e5cf 100644 --- a/docs/code/agents.md +++ b/docs/code/agents.md @@ -41,13 +41,20 @@ page. | Inspect values | `get_flow_state`, `get_message_history` | | Inspect the instance | `list_secrets` (names only), `get_modules`, `get_health`, `get_metrics`, `list_failures`, `list_runs` (live cascades) | | Inspect devices | `list_devices`, `get_device` | -| Edit flows | `save_flow`, `save_node_source`, `validate_flow`, `publish_flow`, `discard_draft`, `delete_flow` | +| Edit flows | `create_flow`, `save_flow`, `save_node_source`, `validate_flow`, `publish_flow`, `discard_draft`, `delete_flow` | | Operate flows | `run_flow`, `trigger_node`, `start_flow`, `stop_flow`, `pause_flow`, `resume_flow`, `cancel_node` | | Install node packages | `apply_modules` | Tools call the same API as the browser. Drafts, validation, version checks, and superuser restrictions apply. Secrets cannot be read back. +A tool's `flow` argument is the flow's id or its key, resolved from the flows +the agent's user can list, as the [CLI does](cli.md#naming-a-flow). When a +name could mean two flows, a reading tool uses the id and adds a `warning` to +its answer, and a tool that changes something refuses until the agent writes +`id:` or `key:`. `create_flow` makes a new flow, under an id the server +issues. `save_flow` only changes an existing one and cannot change its key. + A useful workflow: ask the agent to inspect the flow, save a draft, and validate it. Review the draft before asking it to publish. The agent has a publish tool, so make your intent explicit. diff --git a/docs/code/cli.md b/docs/code/cli.md index aea82a0..3652281 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -153,6 +153,24 @@ is a fallback. `FLUKSIO_URL` and `FLUKSIO_TOKEN` override saved values; command options `--url` and `--token` override both. +A client needs an engine at least as new as itself. An engine from before +flow ids cannot create flows the way this client does, and the client says +so and asks you to upgrade the engine. + +### Naming a flow + +Wherever a command takes a flow, pass its id or its key. A flow created +before ids existed has its key as its id; a new one has an id like +`f_k3q7m2xa`. The id is tried first, then the key, and an unknown name +lists the closest keys. + +When a name is one flow's id and another flow's key, write `id:heating` or +`key:heating` to say which. A command that only reads uses the id and warns +on stderr. A command that changes something, such as `run` or `sweep`, +refuses until you choose. `runs` and `export` also find deleted flows by the +key they last had, and refuse a key several of them have had. `id:` always +works, even for a deleted flow's history. + ### `fluksio login` ```sh @@ -198,7 +216,12 @@ inside the project's repository. Sync records source digests for [stage caching](../concepts/runs.md#stage-caching) and refreshes local workers, so edited imports are picked up. An unchanged -flow reports `unchanged`. See the +flow reports `unchanged`. + +Sync finds a flow by its declared key. The first sync creates it, and the +engine gives it an id, which sync prints beside the key. Another engine gives +the same flow another id, so the key is what your code and every instance +share. See the [data science guide](../getting-started/data-science.md) for a complete example. ### `fluksio run` @@ -275,7 +298,8 @@ instance without a served engine. fluksio runs --flow train --limit 10 ``` -Lists recent runs, newest first. Omit `--flow` for all flows. +Lists recent runs, newest first. Omit `--flow` for all flows. The flow +column shows its key. `--local` reads the local instance. ### `fluksio cancel` @@ -342,6 +366,8 @@ select columns, including dotted paths such as `final_metrics.train_loss`. Both support `--flow`, repeated `--run ID`, `--group`, `--status`, `--since`, `--until` and `--local`. The run ID joins results to metrics. +Rows name the flow by its id in `flow` and its key in `flow_name`, and +metric names keep the key, as in `train.loss`. An export stops at the newest 10 000 runs, and when a selection matches more, the CLI and the Python client warn how many were left out. diff --git a/docs/code/workers.md b/docs/code/workers.md index 198b731..b11e67b 100644 --- a/docs/code/workers.md +++ b/docs/code/workers.md @@ -73,7 +73,7 @@ removed; pass credentials through [Secrets](../interface/operations.md#secrets). ## Send a node to it -Set these fields on a node through `PUT /api/v1/flows/{name}`: +Set these fields on a node through `PUT /api/v1/flows/{id}`: ```json { diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 1ee538d..4b474d6 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -218,7 +218,8 @@ keeps its runs whatever its key is later, and a new flow that takes the key of a deleted one starts without them. Metric names keep the key, as in `train.loss`. -Python returns the same rows directly: +Python returns the same rows directly. `flow` takes the flow's id or key, +as everywhere in the client: ```python from fluksio.sdk.client import Client @@ -249,8 +250,9 @@ Client().import_runs([{ }]) ``` -The flow must exist, and parameters are checked against its inputs. Over the -HTTP API an entry's `flow` is the flow's id. Repeated imports of the same +The flow must exist, and parameters are checked against its inputs. The +Python client takes the flow's id or key for `flow`. Over the HTTP API an +entry's `flow` is the flow's id. Repeated imports of the same `external_id` reuse the existing record. Imported runs have no code commit or digest because the engine did not execute them. Entries may also carry metric points and base64-encoded artifacts; see the