Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Clients take a flow's id or key: one resolver in the SDK, id first, explicit id:/key:, a warning for reads and a refusal for changes when a name means two flows, deleted flows found by their last key; sync creates by POST, retries a create race through the drift checks and reports the id; MCP gets create_flow and resolves every flow argument with the caller's token; displays show flow_name; an engine from before ids is named as such

Browse Source
This commit is contained in:
stroblme
2026-11-02 19:39:14 +01:00
parent fd637e0f1e
commit 53916d869b
20 changed files with 816 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/ai/agent.py
+10 -3
View File
@@ -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:<x> or key:<x> 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
backend/fluksio/api/routes/ai.py
+1 -1
View File
@@ -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)
backend/fluksio/mcp/server.py
+165 -55
View File
@@ -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/<id><tail>`` 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:<x> or key:<x> 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:<x> or key:<x> 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 ``<flow key>.<node id>``.
"""
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)
backend/fluksio/sdk/__init__.py
+11 -6
View File
@@ -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:
backend/fluksio/sdk/cli.py
+47 -17
View File
@@ -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
backend/fluksio/sdk/client.py
+109 -39
View File
@@ -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"):
backend/fluksio/sdk/resolve.py
+87
View File
@@ -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:<x>`` is taken as said, listed or not: history outlives its flow,
and the engine answers for an id nobody holds. ``key:<x>`` 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:<one of them>."
)
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}")
backend/fluksio/tui/app.py
+1 -1
View File
@@ -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,
backend/fluksio/tui/compare.py
+1 -1
View File
@@ -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
backend/tests/api/routes/test_sync.py
+64
View File
@@ -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)
backend/tests/mcp/test_mcp_http.py
+12 -2
View File
@@ -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}
backend/tests/mcp/test_mcp_tools.py
+55 -13
View File
@@ -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},
]
backend/tests/sdk/test_client.py
+71 -1
View File
@@ -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()
backend/tests/sdk/test_declare.py
+4
View File
@@ -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"}}
backend/tests/sdk/test_resolve.py
+74
View File
@@ -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")
backend/tests/test_cli.py
+62 -1
View File
@@ -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"]
docs/code/agents.md
+8 -1
View File
@@ -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:<x>` or `key:<x>`. `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.
docs/code/cli.md
+28 -2
View File
@@ -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.
docs/code/workers.md
+1 -1
View File
@@ -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
{
docs/concepts/runs.md
+5 -3
View File
@@ -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