diff --git a/backend/app/api/main.py b/backend/app/api/main.py index 6c18c4b..79ff173 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -2,6 +2,7 @@ from fastapi import APIRouter from app.api.routes import ( alerts, + artifacts, dashboards, flows, login, @@ -29,6 +30,7 @@ api_router.include_router(messages.router) api_router.include_router(modules.router) api_router.include_router(observability.router) api_router.include_router(runs.router) +api_router.include_router(artifacts.router) # Always mounted so the generated SDK stays the same shape; the endpoints # themselves refuse to work unless MCP is switched on. api_router.include_router(oauth.router) diff --git a/backend/app/api/routes/artifacts.py b/backend/app/api/routes/artifacts.py new file mode 100644 index 0000000..44d393e --- /dev/null +++ b/backend/app/api/routes/artifacts.py @@ -0,0 +1,59 @@ +"""Artifacts over HTTP: the one way bytes get in and out of the store. + +A node on this host could reach the directory itself, but a node on a remote +worker cannot — and having one path rather than two is what keeps a flow's +code the same wherever it runs. +""" + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi.responses import StreamingResponse +from pydantic import BaseModel + +from app.api.deps import get_current_user +from app.flow.artifacts import ArtifactStore + +router = APIRouter( + prefix="/artifacts", tags=["artifacts"], dependencies=[Depends(get_current_user)] +) + + +class ArtifactRef(BaseModel): + digest: str + size: int + media_type: str + name: str = "" + + +def _store(request: Request) -> ArtifactStore: + store: ArtifactStore | None = getattr(request.app.state, "artifact_store", None) + if store is None: + raise HTTPException(status_code=503, detail="The artifact store is not ready") + return store + + +@router.put("", response_model=ArtifactRef) +async def put_artifact( + request: Request, + name: str = Query(default=""), + media_type: str = Query(default=""), +) -> Any: + """Store the request body and answer with the reference to it.""" + store = _store(request) + body = await request.body() + return store.put([body], name=name, media_type=media_type) + + +@router.get("/{digest}") +def get_artifact(digest: str, request: Request) -> Any: + """Stream one artifact back.""" + store = _store(request) + path = store.path(digest) + if path is None: + raise HTTPException(status_code=404, detail="No such artifact") + return StreamingResponse( + store.read(digest), + media_type="application/octet-stream", + headers={"Content-Length": str(path.stat().st_size)}, + ) diff --git a/backend/app/flow/artifacts.py b/backend/app/flow/artifacts.py new file mode 100644 index 0000000..61acedd --- /dev/null +++ b/backend/app/flow/artifacts.py @@ -0,0 +1,140 @@ +"""The artifact store: bytes a node produced, addressed by their content. + +A typed message carries JSON, which is what lets the same value pass through +Redis, the work queue and the worker protocol unchanged. A model checkpoint is +not that, so it does not travel as a message — it is written here and the +message carries a reference to it. + +Addressed by digest rather than by run, for three reasons. A sweep whose fifty +configs share one preprocessed input stores it once. A reference stays valid +however it is passed around, because it names content instead of a location. +And the digest is what a stage cache will compare, so building it in now is +what keeps that from being a change to the message contract later. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import logging +import os +import tempfile +from collections.abc import Iterable, Iterator +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +#: How much is read at a time when hashing or serving. +CHUNK = 1024 * 1024 +DIGEST_PREFIX = "sha256:" + + +def is_reference(value: Any) -> bool: + """Whether a message payload is an artifact reference.""" + return isinstance(value, dict) and str(value.get("digest", "")).startswith( + DIGEST_PREFIX + ) + + +def valid_digest(digest: str) -> bool: + """Guard for anything that reaches the filesystem from outside.""" + if not digest.startswith(DIGEST_PREFIX): + return False + body = digest[len(DIGEST_PREFIX) :] + return len(body) == 64 and all(c in "0123456789abcdef" for c in body) + + +class ArtifactStore: + """Content-addressed files under one directory.""" + + def __init__(self, root: Path) -> None: + self.root = root + self.root.mkdir(parents=True, exist_ok=True) + + def _path(self, digest: str) -> Path: + body = digest[len(DIGEST_PREFIX) :] + # Two levels, so a directory listing stays usable at a hundred thousand + # artifacts. + return self.root / body[:2] / body + + def put( + self, chunks: Iterable[bytes], name: str = "", media_type: str = "" + ) -> dict[str, Any]: + """Store a stream and return the reference to it. + + Written to a temporary file first and moved into place once the digest + is known, so a half-written artifact never has a name anyone can find. + A file already there is left alone: identical content is identical. + """ + digester = hashlib.sha256() + size = 0 + handle = tempfile.NamedTemporaryFile(dir=self.root, delete=False) + try: + with handle: + for chunk in chunks: + digester.update(chunk) + size += len(chunk) + handle.write(chunk) + digest = DIGEST_PREFIX + digester.hexdigest() + target = self._path(digest) + target.parent.mkdir(parents=True, exist_ok=True) + if target.exists(): + os.unlink(handle.name) + else: + # Same content from two nodes at once is one rename winning and + # the other replacing a byte-identical file. + os.replace(handle.name, target) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(handle.name) + raise + return { + "digest": digest, + "size": size, + "media_type": media_type or "application/octet-stream", + "name": name, + } + + def put_file(self, path: Path, media_type: str = "") -> dict[str, Any]: + with path.open("rb") as handle: + return self.put( + iter(lambda: handle.read(CHUNK), b""), + name=path.name, + media_type=media_type, + ) + + def path(self, digest: str) -> Path | None: + """Where the bytes are, or None if this store does not have them.""" + if not valid_digest(digest): + return None + target = self._path(digest) + return target if target.exists() else None + + def read(self, digest: str) -> Iterator[bytes]: + target = self.path(digest) + if target is None: + raise FileNotFoundError(digest) + with target.open("rb") as handle: + while chunk := handle.read(CHUNK): + yield chunk + + def collect(self, keep: set[str]) -> int: + """Delete what no run refers to any more. Returns how many went. + + The caller passes every digest still recorded; anything else in the + store was produced by a run that has since been pruned, or never got a + row at all because the run failed between writing and recording. + """ + removed = 0 + for entry in self.root.glob("*/*"): + if not entry.is_file(): + continue + if DIGEST_PREFIX + entry.name in keep: + continue + try: + entry.unlink() + removed += 1 + except OSError: + logger.warning("Could not remove artifact %s", entry.name) + return removed diff --git a/backend/app/flow/messages.py b/backend/app/flow/messages.py index 7deb0c3..f0da36c 100644 --- a/backend/app/flow/messages.py +++ b/backend/app/flow/messages.py @@ -25,8 +25,13 @@ class DType(str, Enum): knows what it is getting before anything runs, which is what lets the dashboard picker offer a message and refuse a wrong binding. - Binary payloads (tensors, images) will arrive later as explicitly declared - codec fields; until then everything on the wire is JSON. + Binary payloads — tensors, checkpoints, images — travel as ``artifact``: + the bytes go to the artifact store and the message carries a reference to + them. That keeps everything on the wire JSON, which is what the state + backend, the queue and the worker protocol all rely on, and it means a + thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be + needed for payloads too small to be worth a round trip, and nothing asks + for that yet. """ FLOAT = "float" @@ -42,6 +47,9 @@ class DType(str, Enum): RECORD = "record" #: Ordered items of one declared shape; see :attr:`MessageSpec.item`. LIST = "list" + #: A reference to stored bytes: + #: ``{"digest": "sha256:…", "size": int, "media_type": str, "name": str}``. + ARTIFACT = "artifact" _JSON_TYPES = (dict, list, str, int, float, bool, type(None)) @@ -92,6 +100,21 @@ def _is_series(value: Any) -> bool: ) +def _is_artifact(value: Any) -> bool: + """A reference to stored bytes, not the bytes themselves. + + The digest is what makes it one: it names content rather than a location, + so the same file produced twice is stored once and a reference stays valid + wherever the store is reachable from. + """ + return ( + isinstance(value, dict) + and isinstance(value.get("digest"), str) + and value["digest"].startswith("sha256:") + and isinstance(value.get("size"), int) + ) + + def _matches(dtype: DType, value: Any) -> bool: """Whether one value satisfies a scalar or record type.""" if dtype is DType.BOOL: @@ -104,6 +127,8 @@ def _matches(dtype: DType, value: Any) -> bool: return isinstance(value, str) if dtype is DType.RECORD: return _is_record(value) + if dtype is DType.ARTIFACT: + return _is_artifact(value) return isinstance(value, _JSON_TYPES) @@ -190,7 +215,7 @@ class MessageSpec(BaseModel): return str(value).lower() in ("true", "1", "yes", "on") if self.dtype is DType.STR: return value if isinstance(value, str) else json.dumps(value) - if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST): + if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST, DType.ARTIFACT): # A structured payload arriving as text is the same hint a numeric # one is; the shape itself is still checked afterwards. return json.loads(value) if isinstance(value, str) else value diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index c452d72..8eb96f0 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -21,6 +21,7 @@ from typing import Any, Literal from pydantic import BaseModel from app.flow import logs +from app.flow.artifacts import is_reference from app.flow.events import EventBus from app.flow.messages import flow_of from app.flow.nodes import Node @@ -90,6 +91,9 @@ class NodeOutcome(BaseModel): outputs: int = 0 error: str = "" logs: str = "" + #: Artifact references this node emitted, keyed by the message carrying + #: them — what a run records so a result can be opened later. + artifacts: dict[str, dict[str, Any]] = {} class Pipeline: @@ -706,6 +710,11 @@ class Pipeline: duration_ms=duration_ms, outputs=len(result or {}), logs=collected.text, + artifacts={ + name: value + for name, value in (result or {}).items() + if is_reference(value) + }, ) ) return result diff --git a/backend/app/flow/runs.py b/backend/app/flow/runs.py index e8b1b58..9b33040 100644 --- a/backend/app/flow/runs.py +++ b/backend/app/flow/runs.py @@ -49,7 +49,7 @@ from app.flow.pipeline import NodeOutcome, Pipeline from app.flow.queue import WorkItem, WorkQueue from app.flow.schemas import FlowDef from app.flow.state import MemoryState, StateBackend -from app.models import Run, RunMetric, RunNode +from app.models import Run, RunArtifact, RunMetric, RunNode logger = logging.getLogger(__name__) @@ -587,6 +587,19 @@ class RunService: try: with Session(db_engine) as session: session.merge(row) + for message, ref in outcome.artifacts.items(): + session.merge( + RunArtifact( + run_id=run_id, + name=message[:255], + node=outcome.node[:255], + digest=str(ref.get("digest") or "")[:71], + size=int(ref.get("size") or 0), + media_type=str( + ref.get("media_type") or "application/octet-stream" + )[:128], + ) + ) session.commit() except Exception: logger.exception( diff --git a/backend/app/flow/worker_main.py b/backend/app/flow/worker_main.py index ac47db7..cd4268b 100644 --- a/backend/app/flow/worker_main.py +++ b/backend/app/flow/worker_main.py @@ -36,8 +36,11 @@ import contextlib import hashlib import io import json +import tempfile import time import traceback +import urllib.parse +import urllib.request from collections.abc import Callable from types import ModuleType from typing import Any, cast @@ -46,6 +49,14 @@ from typing import Any, cast #: pipe or the reply. MAX_LOG = 16 * 1024 +#: Where the artifact store is, from this worker's point of view: a directory +#: when it shares the engine's filesystem, a URL when it does not. +ARTIFACT_DIR_ENV = "FLUKSIO_ARTIFACT_DIR" +ARTIFACT_URL_ENV = "FLUKSIO_ARTIFACT_URL" +ARTIFACT_TOKEN_ENV = "FLUKSIO_ARTIFACT_TOKEN" +ARTIFACT_CACHE_ENV = "FLUKSIO_ARTIFACT_CACHE" +ARTIFACT_TIMEOUT_S = 300 + #: The reply channel, opened by ``main``. Also what an event line goes down. _RPC: Any = None #: The call being served, so an event can say which one it belongs to. @@ -96,6 +107,119 @@ class _Reporter(ModuleType): } ) + def save_artifact( + self, + source: Any, + name: str = "", + media_type: str = "application/octet-stream", + ) -> dict[str, Any]: + """Store bytes or a file and return the reference to return onward. + + Return the reference from an ``artifact`` port: a checkpoint is far too + big to be a message, and the reference is what the next node opens. + """ + if isinstance(source, (bytes, bytearray)): + data = bytes(source) + name = name or "artifact.bin" + else: + path = str(source) + with open(path, "rb") as handle: + data = handle.read() + name = name or os.path.basename(path) + return _store_bytes(data, name, media_type) + + def load_artifact(self, ref: dict[str, Any]) -> str: + """Fetch an artifact and hand back a local path to read it from.""" + digest = str((ref or {}).get("digest") or "") + if not digest.startswith("sha256:"): + raise ValueError("not an artifact reference") + return _fetch(digest) + + +def _store_bytes(data: bytes, name: str, media_type: str) -> dict[str, Any]: + """Write to the artifact store, whichever end of it this worker can see. + + A worker in the engine's own container writes the file; one on another host + puts it over HTTP. Node code cannot tell the difference, which is the point + — the same flow runs either place. The hashing is repeated from + ``app.flow.artifacts`` rather than imported, because nothing of the app is + importable here. + """ + digest = "sha256:" + hashlib.sha256(data).hexdigest() + directory = os.environ.get(ARTIFACT_DIR_ENV) + if directory: + target = os.path.join(directory, digest[7:9], digest[7:]) + if not os.path.exists(target): + os.makedirs(os.path.dirname(target), exist_ok=True) + # Written beside the target and moved, so a reader never opens a + # half-written artifact. + handle, temporary = tempfile.mkstemp(dir=os.path.dirname(target)) + with os.fdopen(handle, "wb") as out: + out.write(data) + os.replace(temporary, target) + else: + _put_over_http(data, name, media_type) + return { + "digest": digest, + "size": len(data), + "media_type": media_type, + "name": name, + } + + +def _put_over_http(data: bytes, name: str, media_type: str) -> None: + base = os.environ.get(ARTIFACT_URL_ENV) + if not base: + raise RuntimeError( + "this worker has no artifact store — neither " + f"{ARTIFACT_DIR_ENV} nor {ARTIFACT_URL_ENV} is set" + ) + query = urllib.parse.urlencode({"name": name, "media_type": media_type}) + request = urllib.request.Request( + f"{base.rstrip('/')}?{query}", data=data, method="PUT" + ) + _authorize(request) + with urllib.request.urlopen(request, timeout=ARTIFACT_TIMEOUT_S): + pass + + +def _fetch(digest: str) -> str: + """A local path holding this artifact's bytes, downloading it if needed.""" + directory = os.environ.get(ARTIFACT_DIR_ENV) + if directory: + target = os.path.join(directory, digest[7:9], digest[7:]) + if not os.path.exists(target): + raise FileNotFoundError(digest) + return target + + # Cached by digest: content-addressing means a file once fetched is never + # stale, so a sweep pulls a shared input across the network once. + cache = os.environ.get(ARTIFACT_CACHE_ENV) or os.path.join( + tempfile.gettempdir(), "fluksio-artifacts" + ) + target = os.path.join(cache, digest[7:]) + if os.path.exists(target): + return target + base = os.environ.get(ARTIFACT_URL_ENV) + if not base: + raise RuntimeError(f"this worker cannot reach the artifact store: {digest}") + os.makedirs(cache, exist_ok=True) + request = urllib.request.Request(f"{base.rstrip('/')}/{digest}") + _authorize(request) + handle, temporary = tempfile.mkstemp(dir=cache) + with urllib.request.urlopen(request, timeout=ARTIFACT_TIMEOUT_S) as response: + with os.fdopen(handle, "wb") as out: + while chunk := response.read(1024 * 1024): + out.write(chunk) + os.replace(temporary, target) + return target + + +def _authorize(request: Any) -> None: + token = os.environ.get(ARTIFACT_TOKEN_ENV) + if token: + request.add_header("Authorization", f"Bearer {token}") + def _install_reporter() -> None: """Put ``fluksio`` on the import path of every node this worker runs.""" diff --git a/backend/app/flow/workers.py b/backend/app/flow/workers.py index 2d1887c..5ee4853 100644 --- a/backend/app/flow/workers.py +++ b/backend/app/flow/workers.py @@ -42,7 +42,7 @@ ENV_DENY_PREFIXES = ("POSTGRES_", "FIRST_SUPERUSER", "SENTRY_DSN") ENV_DENY_WORDS = ("PASSWORD", "SECRET", "TOKEN") -def worker_env() -> dict[str, str]: +def worker_env(extra: dict[str, str] | None = None) -> dict[str, str]: """The engine's environment with its credentials taken out. A denylist, so ``PATH``, ``HOME``, the locale and whatever else a venv @@ -50,13 +50,19 @@ def worker_env() -> dict[str, str]: user in the same container, with the same filesystem and the same network. It only means node code cannot read the deployment's secrets straight out of its own environment. + + ``extra`` is put back afterwards, deliberately: the artifact store's + address is something the worker has to be told, and one of its names would + otherwise be caught by the denylist for containing "TOKEN". """ - return { + env = { key: value for key, value in os.environ.items() if not key.startswith(ENV_DENY_PREFIXES) and not any(word in key.upper() for word in ENV_DENY_WORDS) } + env.update(extra or {}) + return env class RemoteError(Exception): @@ -99,7 +105,9 @@ def _remote_class(name: str) -> type[RemoteError]: class _Worker: """One subprocess, and the framing of one request/response over its pipes.""" - def __init__(self, python: str, generation: int) -> None: + def __init__( + self, python: str, generation: int, env: dict[str, str] | None = None + ) -> None: self.generation = generation self.cancelled = False # What a read took past the end of a line. A node reporting quickly @@ -112,7 +120,7 @@ class _Worker: stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True, - env=worker_env(), + env=worker_env(env), ) def alive(self) -> bool: @@ -176,11 +184,18 @@ class PythonWorkerPool: """ def __init__( - self, python: str, size: int = 4, events: EventBus | None = None + self, + python: str, + size: int = 4, + events: EventBus | None = None, + env: dict[str, str] | None = None, ) -> None: self.python = python self.size = size self.events = events + # Put into every worker's environment past the denylist — where the + # artifact store is, which node code needs and cannot guess. + self.env = env or {} self._idle: queue.Queue[_Worker | None] = queue.Queue() # Keyed by (run, node): a sweep has the same node executing in several # runs at once, and cancelling one of them must not kill the others. @@ -256,7 +271,7 @@ class PythonWorkerPool: if slot is not None: slot.kill() try: - return _Worker(self.python, self._generation) + return _Worker(self.python, self._generation, self.env) except Exception as exc: self._idle.put(None) raise RemoteError(f"worker unavailable: {exc}") from exc diff --git a/backend/app/main.py b/backend/app/main.py index 70fac28..6733f23 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -16,6 +16,7 @@ from app.core import security from app.core.config import settings from app.flow import logs, modules from app.flow.alerts import AlertManager +from app.flow.artifacts import ArtifactStore from app.flow.controller import FlowController from app.flow.dashboards import DashboardStore from app.flow.events import event_bus @@ -30,6 +31,7 @@ from app.flow.secrets import init_secrets from app.flow.state import MemoryState, RedisState, StateBackend from app.flow.store import FlowStore from app.flow.watchdog import LoopWatchdog +from app.flow.worker_main import ARTIFACT_DIR_ENV from app.flow.workers import PythonWorkerPool @@ -100,10 +102,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: store = FlowStore(settings.FLOWS_DIR) # The packages node code imports, before anything tries to import them. await run_in_threadpool(modules.reconcile, store) + # Beside the flows rather than in them: an artifact is what a run produced, + # not something anyone wrote, so it has no business in the git repository. + artifacts = ArtifactStore(settings.FLOWS_DIR.parent / "artifacts") + app.state.artifact_store = artifacts pool = PythonWorkerPool( python=modules.venv_python(), size=settings.FLOW_MAX_WORKERS, events=event_bus, + # A worker in this container writes to the store directly; a remote one + # is given a URL instead. Node code calls the same two functions. + env={ARTIFACT_DIR_ENV: str(artifacts.root)}, ) pool.start() app.state.worker_pool = pool diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index 8c30ef1..2687019 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -7,6 +7,9 @@ from collections.abc import Iterator import pytest +from app.flow.artifacts import ArtifactStore +from app.flow.messages import DType, MessageSpec +from app.flow.worker_main import ARTIFACT_DIR_ENV from app.flow.workers import NodeCancelled, NodeTimeout, PythonWorkerPool @@ -280,3 +283,45 @@ def test_cancelling_one_run_leaves_the_same_node_in_another_alone(pool): assert pool.cancel("demo.hold", run_id="run-a") is True started.wait(timeout=5) thread.join(timeout=5) + + +def test_a_node_saves_and_loads_an_artifact(tmp_path): + # Bytes never travel as a message: the node stores them and returns a + # reference, which the next node opens. + store = ArtifactStore(tmp_path / "artifacts") + pool = PythonWorkerPool( + python=sys.executable, size=1, env={ARTIFACT_DIR_ENV: str(store.root)} + ) + pool.start() + try: + ref = pool.run( + "demo", + "save", + "import fluksio\n" + "def process(params):\n" + " return {'weights': fluksio.save_artifact(b'x' * 2048, 'w.npz')}\n", + {}, + {}, + "demo.save", + timeout=10, + )["weights"] + + assert ref["size"] == 2048 + assert MessageSpec(name="weights", dtype=DType.ARTIFACT).check(ref) is None + assert store.path(ref["digest"]) is not None + + loaded = pool.run( + "demo", + "load", + "import fluksio\n" + "def process(weights, params):\n" + " with open(fluksio.load_artifact(weights), 'rb') as f:\n" + " return {'size': len(f.read())}\n", + {"weights": ref}, + {}, + "demo.load", + timeout=10, + ) + assert loaded == {"size": 2048} + finally: + pool.stop() diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 287afac..4ea7a3c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -57,6 +57,59 @@ export const ApplyResultSchema = { title: 'ApplyResult' } as const; +export const ArtifactRefSchema = { + properties: { + digest: { + type: 'string', + title: 'Digest' + }, + size: { + type: 'integer', + title: 'Size' + }, + media_type: { + type: 'string', + title: 'Media Type' + }, + name: { + type: 'string', + title: 'Name', + default: '' + } + }, + type: 'object', + required: ['digest', 'size', 'media_type'], + title: 'ArtifactRef' +} as const; + +export const ArtifactRowSchema = { + properties: { + name: { + type: 'string', + title: 'Name' + }, + node: { + type: 'string', + title: 'Node' + }, + digest: { + type: 'string', + title: 'Digest' + }, + size: { + type: 'integer', + title: 'Size' + }, + media_type: { + type: 'string', + title: 'Media Type' + } + }, + type: 'object', + required: ['name', 'node', 'digest', 'size', 'media_type'], + title: 'ArtifactRow' +} as const; + export const Body_login_login_access_tokenSchema = { properties: { grant_type: { @@ -320,7 +373,7 @@ export const ChannelSchema = { export const DTypeSchema = { type: 'string', - enum: ['float', 'int', 'str', 'bool', 'json', 'series', 'record', 'list'], + enum: ['float', 'int', 'str', 'bool', 'json', 'series', 'record', 'list', 'artifact'], title: 'DType', description: `Serializable payload types. @@ -329,8 +382,13 @@ are declared shapes rather than "some JSON": a widget or a downstream node knows what it is getting before anything runs, which is what lets the dashboard picker offer a message and refuse a wrong binding. -Binary payloads (tensors, images) will arrive later as explicitly declared -codec fields; until then everything on the wire is JSON.` +Binary payloads — tensors, checkpoints, images — travel as \`\`artifact\`\`: +the bytes go to the artifact store and the message carries a reference to +them. That keeps everything on the wire JSON, which is what the state +backend, the queue and the worker protocol all rely on, and it means a +thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be +needed for payloads too small to be worth a round trip, and nothing asks +for that yet.` } as const; export const DashboardDef_InputSchema = { @@ -642,6 +700,21 @@ export const FlowDef_InputSchema = { type: 'integer', title: 'Version', default: 1 + }, + mode: { + type: 'string', + enum: ['live', 'batch'], + title: 'Mode', + description: 'A live flow reacts to what arrives: its subscriptions, schedules and webhooks run until it is stopped. A batch flow only runs when a run asks it to, from its inputs to its outputs, and is never activated.', + default: 'live' + }, + outputs: { + items: { + type: 'string' + }, + type: 'array', + title: 'Outputs', + description: 'Messages a batch run reports as its result, unqualified. Empty means every message the flow ends up holding.' } }, type: 'object', @@ -679,6 +752,21 @@ export const FlowDef_OutputSchema = { type: 'integer', title: 'Version', default: 1 + }, + mode: { + type: 'string', + enum: ['live', 'batch'], + title: 'Mode', + description: 'A live flow reacts to what arrives: its subscriptions, schedules and webhooks run until it is stopped. A batch flow only runs when a run asks it to, from its inputs to its outputs, and is never activated.', + default: 'live' + }, + outputs: { + items: { + type: 'string' + }, + type: 'array', + title: 'Outputs', + description: 'Messages a batch run reports as its result, unqualified. Empty means every message the flow ends up holding.' } }, type: 'object', @@ -1220,6 +1308,49 @@ export const MessagesPublicSchema = { title: 'MessagesPublic' } as const; +export const MetricPointSchema = { + properties: { + step: { + type: 'integer', + title: 'Step' + }, + ts: { + type: 'number', + title: 'Ts' + }, + value: { + type: 'number', + title: 'Value' + } + }, + type: 'object', + required: ['step', 'ts', 'value'], + title: 'MetricPoint' +} as const; + +export const MetricSeriesSchema = { + properties: { + label: { + type: 'string', + title: 'Label' + }, + points: { + items: { + items: { + type: 'number' + }, + type: 'array' + }, + type: 'array', + title: 'Points' + } + }, + type: 'object', + required: ['label'], + title: 'MetricSeries', + description: 'The shape a chart widget already draws, so comparing runs is a binding.' +} as const; + export const ModulePackageSchema = { properties: { name: { @@ -1347,7 +1478,26 @@ export const NodeDef_InputSchema = { } ], title: 'Timeout', - description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running." + description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running — in a batch run, which never redelivers, it is an idle timeout instead: silence this long is a kill." + }, + device: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Device', + description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing." + }, + device_policy: { + type: 'string', + enum: ['require', 'prefer'], + title: 'Device Policy', + description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', + default: 'require' } }, type: 'object', @@ -1417,7 +1567,26 @@ export const NodeDef_OutputSchema = { } ], title: 'Timeout', - description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running." + description: "Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running — in a batch run, which never redelivers, it is an idle timeout instead: silence this long is a kill." + }, + device: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Device', + description: "Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing." + }, + device_policy: { + type: 'string', + enum: ['require', 'prefer'], + title: 'Device Policy', + description: 'What to do when no worker carries `device`: wait for one, or run locally anyway.', + default: 'require' } }, type: 'object', @@ -1893,6 +2062,182 @@ export const RuleSchema = { description: 'Which events go to which channels.' } as const; +export const RunCreateSchema = { + properties: { + params: { + additionalProperties: true, + type: 'object', + title: 'Params' + }, + seed: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Seed' + }, + draft: { + type: 'boolean', + title: 'Draft', + default: false + } + }, + type: 'object', + title: 'RunCreate' +} as const; + +export const RunDetailSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + flow: { + type: 'string', + title: 'Flow' + }, + status: { + type: 'string', + title: 'Status' + }, + status_reason: { + type: 'string', + title: 'Status Reason' + }, + cause: { + type: 'string', + title: 'Cause' + }, + params: { + additionalProperties: true, + type: 'object', + title: 'Params' + }, + params_digest: { + type: 'string', + title: 'Params Digest' + }, + seed: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Seed' + }, + group_id: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Group Id' + }, + labels: { + items: { + type: 'string' + }, + type: 'array', + title: 'Labels' + }, + created_at: { + title: 'Created At' + }, + started_at: { + title: 'Started At' + }, + finished_at: { + title: 'Finished At' + }, + duration_ms: { + type: 'number', + title: 'Duration Ms' + }, + actor: { + type: 'string', + title: 'Actor' + }, + result: { + additionalProperties: true, + type: 'object', + title: 'Result' + }, + commit: { + type: 'string', + title: 'Commit', + default: '' + }, + flow_version: { + type: 'integer', + title: 'Flow Version', + default: 1 + }, + nodes: { + items: { + '$ref': '#/components/schemas/RunNodeRow' + }, + type: 'array', + title: 'Nodes' + }, + artifacts: { + items: { + '$ref': '#/components/schemas/ArtifactRow' + }, + type: 'array', + title: 'Artifacts' + } + }, + type: 'object', + required: ['id', 'flow', 'status', 'status_reason', 'cause', 'params', 'params_digest', 'seed', 'group_id', 'labels', 'created_at', 'duration_ms', 'actor'], + title: 'RunDetail' +} as const; + +export const RunNodeRowSchema = { + properties: { + node: { + type: 'string', + title: 'Node' + }, + status: { + type: 'string', + title: 'Status' + }, + attempt: { + type: 'integer', + title: 'Attempt' + }, + duration_ms: { + type: 'number', + title: 'Duration Ms' + }, + worker: { + type: 'string', + title: 'Worker' + }, + error: { + type: 'string', + title: 'Error' + }, + logs: { + type: 'string', + title: 'Logs' + } + }, + type: 'object', + required: ['node', 'status', 'attempt', 'duration_ms', 'worker', 'error', 'logs'], + title: 'RunNodeRow' +} as const; + export const RunRequestSchema = { properties: { inputs: { @@ -1906,63 +2251,6 @@ export const RunRequestSchema = { title: 'RunRequest' } as const; -export const RunRowSchema = { - properties: { - id: { - type: 'string', - title: 'Id' - }, - flow: { - type: 'string', - title: 'Flow' - }, - source: { - type: 'string', - title: 'Source' - }, - status: { - type: 'string', - title: 'Status' - }, - started_at: { - type: 'string', - format: 'date-time', - title: 'Started At' - }, - finished_at: { - anyOf: [ - { - type: 'string', - format: 'date-time' - }, - { - type: 'null' - } - ], - title: 'Finished At' - }, - nodes: { - type: 'integer', - title: 'Nodes' - }, - errors: { - type: 'integer', - title: 'Errors' - }, - duration_ms: { - type: 'number', - title: 'Duration Ms' - }, - deliveries: { - type: 'integer', - title: 'Deliveries' - } - }, - type: 'object', - required: ['id', 'flow', 'source', 'status', 'started_at', 'nodes', 'errors', 'duration_ms', 'deliveries'], - title: 'RunRow' -} as const; - export const SecretNamesSchema = { properties: { data: { @@ -2044,6 +2332,25 @@ export const SectionDef_OutputSchema = { description: 'A grid of widgets under a heading.' } as const; +export const SeriesAnswerSchema = { + properties: { + metric: { + type: 'string', + title: 'Metric' + }, + lines: { + items: { + '$ref': '#/components/schemas/MetricSeries' + }, + type: 'array', + title: 'Lines' + } + }, + type: 'object', + required: ['metric'], + title: 'SeriesAnswer' +} as const; + export const SeriesPointSchema = { properties: { ts: { @@ -2092,6 +2399,48 @@ export const ShareRequestSchema = { title: 'ShareRequest' } as const; +export const SweepCreateSchema = { + properties: { + runs: { + items: { + '$ref': '#/components/schemas/SweepEntry' + }, + type: 'array', + title: 'Runs' + }, + draft: { + type: 'boolean', + title: 'Draft', + default: false + } + }, + type: 'object', + title: 'SweepCreate' +} as const; + +export const SweepEntrySchema = { + properties: { + params: { + additionalProperties: true, + type: 'object', + title: 'Params' + }, + seed: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Seed' + } + }, + type: 'object', + title: 'SweepEntry' +} as const; + export const TokenSchema = { properties: { access_token: { @@ -2635,6 +2984,147 @@ export const app__api__routes__messages__PublishRequestSchema = { title: 'PublishRequest' } as const; +export const app__api__routes__observability__RunRowSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + flow: { + type: 'string', + title: 'Flow' + }, + source: { + type: 'string', + title: 'Source' + }, + status: { + type: 'string', + title: 'Status' + }, + started_at: { + type: 'string', + format: 'date-time', + title: 'Started At' + }, + finished_at: { + anyOf: [ + { + type: 'string', + format: 'date-time' + }, + { + type: 'null' + } + ], + title: 'Finished At' + }, + nodes: { + type: 'integer', + title: 'Nodes' + }, + errors: { + type: 'integer', + title: 'Errors' + }, + duration_ms: { + type: 'number', + title: 'Duration Ms' + }, + deliveries: { + type: 'integer', + title: 'Deliveries' + } + }, + type: 'object', + required: ['id', 'flow', 'source', 'status', 'started_at', 'nodes', 'errors', 'duration_ms', 'deliveries'], + title: 'RunRow' +} as const; + +export const app__api__routes__runs__RunRowSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + flow: { + type: 'string', + title: 'Flow' + }, + status: { + type: 'string', + title: 'Status' + }, + status_reason: { + type: 'string', + title: 'Status Reason' + }, + cause: { + type: 'string', + title: 'Cause' + }, + params: { + additionalProperties: true, + type: 'object', + title: 'Params' + }, + params_digest: { + type: 'string', + title: 'Params Digest' + }, + seed: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Seed' + }, + group_id: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Group Id' + }, + labels: { + items: { + type: 'string' + }, + type: 'array', + title: 'Labels' + }, + created_at: { + title: 'Created At' + }, + started_at: { + title: 'Started At' + }, + finished_at: { + title: 'Finished At' + }, + duration_ms: { + type: 'number', + title: 'Duration Ms' + }, + actor: { + type: 'string', + title: 'Actor' + } + }, + type: 'object', + required: ['id', 'flow', 'status', 'status_reason', 'cause', 'params', 'params_digest', 'seed', 'group_id', 'labels', 'created_at', 'duration_ms', 'actor'], + title: 'RunRow', + description: 'A run without its result, which is the part that can be large.' +} as const; + export const app__flow__schemas__MessageValueSchema = { properties: { value: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 93174e4..27f62ab 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse } from './types.gen'; +import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsReadRunData, RunsReadRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse } from './types.gen'; export class AlertsService { /** @@ -61,6 +61,52 @@ export class AlertsService { } } +export class ArtifactsService { + /** + * Put Artifact + * Store the request body and answer with the reference to it. + * @param data The data for the request. + * @param data.name + * @param data.mediaType + * @returns ArtifactRef Successful Response + * @throws ApiError + */ + public static putArtifact(data: ArtifactsPutArtifactData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'PUT', + url: '/api/v1/artifacts', + query: { + name: data.name, + media_type: data.mediaType + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Get Artifact + * Stream one artifact back. + * @param data The data for the request. + * @param data.digest + * @returns unknown Successful Response + * @throws ApiError + */ + public static getArtifact(data: ArtifactsGetArtifactData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/artifacts/{digest}', + path: { + digest: data.digest + }, + errors: { + 422: 'Validation Error' + } + }); + } +} + export class DashboardsService { /** * Read Dashboards @@ -1210,7 +1256,7 @@ export class ObservabilityService { * @param data.since * @param data.until * @param data.limit - * @returns RunRow Successful Response + * @returns app__api__routes__observability__RunRow Successful Response * @throws ApiError */ public static readRuns(data: ObservabilityReadRunsData = {}): CancelablePromise { @@ -1306,6 +1352,188 @@ export class PrivateService { } } +export class RunsService { + /** + * Create Run + * Queue one run of a flow. + * @param data The data for the request. + * @param data.name + * @param data.requestBody + * @returns app__api__routes__runs__RunRow Successful Response + * @throws ApiError + */ + public static createRun(data: RunsCreateRunData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/runs/flows/{name}', + path: { + name: data.name + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Create Sweep + * Queue many runs of one flow under a shared group. + * + * An ensemble is this with the same parameters and different seeds; a grid + * search is this with the parameters spread out. Either way the caller + * builds the list — the engine does not own a sweep grammar. + * @param data The data for the request. + * @param data.name + * @param data.requestBody + * @returns app__api__routes__runs__RunRow Successful Response + * @throws ApiError + */ + public static createSweep(data: RunsCreateSweepData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/runs/flows/{name}/sweep', + path: { + name: data.name + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Runs + * Runs, newest first. The queryable table an experiment log needs. + * @param data The data for the request. + * @param data.flow + * @param data.status + * @param data.group + * @param data.digest + * @param data.limit + * @returns app__api__routes__runs__RunRow Successful Response + * @throws ApiError + */ + public static readRuns(data: RunsReadRunsData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/runs', + query: { + flow: data.flow, + status: data.status, + group: data.group, + digest: data.digest, + limit: data.limit + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Run + * One run in full: what it was asked, what each node did, what it made. + * @param data The data for the request. + * @param data.runId + * @returns RunDetail Successful Response + * @throws ApiError + */ + public static readRun(data: RunsReadRunData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/runs/{run_id}', + path: { + run_id: data.runId + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Cancel Run + * Stop a run. One already past its last node is left as it finished. + * @param data The data for the request. + * @param data.runId + * @returns app__api__routes__runs__RunRow Successful Response + * @throws ApiError + */ + public static cancelRun(data: RunsCancelRunData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/runs/{run_id}/cancel', + path: { + run_id: data.runId + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Metrics + * One metric's series, in step order. + * + * ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart + * is 3000 points nobody can see. + * @param data The data for the request. + * @param data.runId + * @param data.name + * @param data.stride + * @returns MetricPoint Successful Response + * @throws ApiError + */ + public static readMetrics(data: RunsReadMetricsData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/runs/{run_id}/metrics', + path: { + run_id: data.runId + }, + query: { + name: data.name, + stride: data.stride + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Compare Metric + * One metric across several runs, as the chart widget's series shape. + * + * This is the comparison view: it answers in the same shape a flow answers a + * chart's query with, so putting three training curves beside each other is + * a widget binding rather than a screen of its own. + * @param data The data for the request. + * @param data.ids + * @param data.metric + * @returns SeriesAnswer Successful Response + * @throws ApiError + */ + public static compareMetric(data: RunsCompareMetricData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/runs/series/compare', + query: { + ids: data.ids, + metric: data.metric + }, + errors: { + 422: 'Validation Error' + } + }); + } +} + export class SecretsService { /** * Read Secrets diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 7c421d8..d3457dc 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -39,6 +39,42 @@ export type app__api__routes__messages__PublishRequest = { source_detail?: string; }; +export type app__api__routes__observability__RunRow = { + id: string; + flow: string; + source: string; + status: string; + started_at: string; + finished_at?: (string | null); + nodes: number; + errors: number; + duration_ms: number; + deliveries: number; +}; + +/** + * A run without its result, which is the part that can be large. + */ +export type app__api__routes__runs__RunRow = { + id: string; + flow: string; + status: string; + status_reason: string; + cause: string; + params: { + [key: string]: unknown; + }; + params_digest: string; + seed: (number | null); + group_id: (string | null); + labels: Array<(string)>; + created_at: unknown; + started_at?: unknown; + finished_at?: unknown; + duration_ms: number; + actor: string; +}; + /** * The last payload seen on a message. */ @@ -59,6 +95,21 @@ export type ApplyResult = { output?: string; }; +export type ArtifactRef = { + digest: string; + size: number; + media_type: string; + name?: string; +}; + +export type ArtifactRow = { + name: string; + node: string; + digest: string; + size: number; + media_type: string; +}; + export type Body_login_login_access_token = { grant_type?: (string | null); username: string; @@ -190,10 +241,15 @@ export type DeadLetter = { * knows what it is getting before anything runs, which is what lets the * dashboard picker offer a message and refuse a wrong binding. * - * Binary payloads (tensors, images) will arrive later as explicitly declared - * codec fields; until then everything on the wire is JSON. + * Binary payloads — tensors, checkpoints, images — travel as ``artifact``: + * the bytes go to the artifact store and the message carries a reference to + * them. That keeps everything on the wire JSON, which is what the state + * backend, the queue and the worker protocol all rely on, and it means a + * thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be + * needed for payloads too small to be worth a round trip, and nothing asks + * for that yet. */ -export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'record' | 'list'; +export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'record' | 'list' | 'artifact'; /** * Something wired into this flow that is not a node in it. @@ -231,8 +287,21 @@ export type FlowDef_Input = { nodes?: Array; inputs?: Array; version?: number; + /** + * A live flow reacts to what arrives: its subscriptions, schedules and webhooks run until it is stopped. A batch flow only runs when a run asks it to, from its inputs to its outputs, and is never activated. + */ + mode?: 'live' | 'batch'; + /** + * Messages a batch run reports as its result, unqualified. Empty means every message the flow ends up holding. + */ + outputs?: Array<(string)>; }; +/** + * A live flow reacts to what arrives: its subscriptions, schedules and webhooks run until it is stopped. A batch flow only runs when a run asks it to, from its inputs to its outputs, and is never activated. + */ +export type mode = 'live' | 'batch'; + /** * One atomic flow. */ @@ -242,6 +311,14 @@ export type FlowDef_Output = { nodes?: Array; inputs?: Array; version?: number; + /** + * A live flow reacts to what arrives: its subscriptions, schedules and webhooks run until it is stopped. A batch flow only runs when a run asks it to, from its inputs to its outputs, and is never activated. + */ + mode?: 'live' | 'batch'; + /** + * Messages a batch run reports as its result, unqualified. Empty means every message the flow ends up holding. + */ + outputs?: Array<(string)>; }; /** @@ -422,6 +499,20 @@ export type MessagesPublic = { count: number; }; +export type MetricPoint = { + step: number; + ts: number; + value: number; +}; + +/** + * The shape a chart widget already draws, so comparing runs is a binding. + */ +export type MetricSeries = { + label: string; + points?: Array>; +}; + /** * One package installed in the venv node code runs on. */ @@ -464,11 +555,24 @@ export type NodeDef_Input = { provides?: Array; source_ref?: (string | null); /** - * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running. + * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running — in a batch run, which never redelivers, it is an idle timeout instead: silence this long is a kill. */ timeout?: (number | null); + /** + * Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing. + */ + device?: (string | null); + /** + * What to do when no worker carries `device`: wait for one, or run locally anyway. + */ + device_policy?: 'require' | 'prefer'; }; +/** + * What to do when no worker carries `device`: wait for one, or run locally anyway. + */ +export type device_policy = 'require' | 'prefer'; + /** * A node as stored: identity, configuration and ports. * @@ -487,9 +591,17 @@ export type NodeDef_Output = { provides?: Array; source_ref?: (string | null); /** - * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running. + * Seconds this node's code may run before it is stopped. This covers the first call's imports, which can be much slower than the body. Above 60 the engine may deliver its work again while it is still running — in a batch run, which never redelivers, it is an idle timeout instead: silence this long is a kill. */ timeout?: (number | null); + /** + * Label of the worker this node's code must run on, such as 'gpu'. Empty means the engine's own workers. A run needing a label no attached worker carries waits rather than failing. + */ + device?: (string | null); + /** + * What to do when no worker carries `device`: wait for one, or run locally anyway. + */ + device_policy?: 'require' | 'prefer'; }; /** @@ -624,25 +736,57 @@ export type Rule = { cooldown_s?: number; }; +export type RunCreate = { + params?: { + [key: string]: unknown; + }; + seed?: (number | null); + draft?: boolean; +}; + +export type RunDetail = { + id: string; + flow: string; + status: string; + status_reason: string; + cause: string; + params: { + [key: string]: unknown; + }; + params_digest: string; + seed: (number | null); + group_id: (string | null); + labels: Array<(string)>; + created_at: unknown; + started_at?: unknown; + finished_at?: unknown; + duration_ms: number; + actor: string; + result?: { + [key: string]: unknown; + }; + commit?: string; + flow_version?: number; + nodes?: Array; + artifacts?: Array; +}; + +export type RunNodeRow = { + node: string; + status: string; + attempt: number; + duration_ms: number; + worker: string; + error: string; + logs: string; +}; + export type RunRequest = { inputs?: { [key: string]: unknown; }; }; -export type RunRow = { - id: string; - flow: string; - source: string; - status: string; - started_at: string; - finished_at?: (string | null); - nodes: number; - errors: number; - duration_ms: number; - deliveries: number; -}; - export type SecretNames = { data: Array<(string)>; count: number; @@ -670,6 +814,11 @@ export type SectionDef_Output = { widgets?: Array; }; +export type SeriesAnswer = { + metric: string; + lines?: Array; +}; + export type SeriesPoint = { ts: number; executions: number; @@ -684,6 +833,18 @@ export type ShareRequest = { lib_name: string; }; +export type SweepCreate = { + runs?: Array; + draft?: boolean; +}; + +export type SweepEntry = { + params?: { + [key: string]: unknown; + }; + seed?: (number | null); +}; + export type Token = { access_token: string; token_type?: string; @@ -808,6 +969,19 @@ export type AlertsTestChannelData = { export type AlertsTestChannelResponse = (Message); +export type ArtifactsPutArtifactData = { + mediaType?: string; + name?: string; +}; + +export type ArtifactsPutArtifactResponse = (ArtifactRef); + +export type ArtifactsGetArtifactData = { + digest: string; +}; + +export type ArtifactsGetArtifactResponse = (unknown); + export type DashboardsReadDashboardsResponse = (DashboardsPublic); export type DashboardsReadDashboardData = { @@ -1117,7 +1291,7 @@ export type ObservabilityReadRunsData = { until?: (string | null); }; -export type ObservabilityReadRunsResponse = (Array); +export type ObservabilityReadRunsResponse = (Array); export type ObservabilityReadEventsData = { flow?: (string | null); @@ -1141,6 +1315,57 @@ export type PrivateCreateUserData = { export type PrivateCreateUserResponse = (UserPublic); +export type RunsCreateRunData = { + name: string; + requestBody: RunCreate; +}; + +export type RunsCreateRunResponse = (app__api__routes__runs__RunRow); + +export type RunsCreateSweepData = { + name: string; + requestBody: SweepCreate; +}; + +export type RunsCreateSweepResponse = (Array); + +export type RunsReadRunsData = { + digest?: (string | null); + flow?: (string | null); + group?: (string | null); + limit?: number; + status?: (string | null); +}; + +export type RunsReadRunsResponse = (Array); + +export type RunsReadRunData = { + runId: string; +}; + +export type RunsReadRunResponse = (RunDetail); + +export type RunsCancelRunData = { + runId: string; +}; + +export type RunsCancelRunResponse = (app__api__routes__runs__RunRow); + +export type RunsReadMetricsData = { + name: string; + runId: string; + stride?: number; +}; + +export type RunsReadMetricsResponse = (Array); + +export type RunsCompareMetricData = { + ids: string; + metric: string; +}; + +export type RunsCompareMetricResponse = (SeriesAnswer); + export type SecretsReadSecretsResponse = (SecretNames); export type SecretsSaveSecretData = { diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index 073a502..1ef83fb 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -59,6 +59,7 @@ const DTYPES: DType[] = [ "series", "record", "list", + "artifact", ] /** What a list may hold. One declared level: no list of lists. */ @@ -805,6 +806,9 @@ const PLACEHOLDER: Record = { series: '{"lines": []}', record: "{}", list: "[]", + // Bytes never travel as a message: the node stores them and returns what + // the next one opens. + artifact: 'fluksio.save_artifact(b"", "result.bin")', } const SCAFFOLD_DOC =