diff --git a/backend/fluksio/alembic/versions/c4e9a2f7b318_run_external_id.py b/backend/fluksio/alembic/versions/c4e9a2f7b318_run_external_id.py new file mode 100644 index 0000000..29b1849 --- /dev/null +++ b/backend/fluksio/alembic/versions/c4e9a2f7b318_run_external_id.py @@ -0,0 +1,38 @@ +"""run.external_id + +Runs were only ever made by running a flow, so a history recorded somewhere +else before the move had nowhere to go. An imported run keeps the id it had +there, and the unique index is what makes importing it twice a no-op. + +Revision ID: c4e9a2f7b318 +Revises: a8f6f79e40be +Create Date: 2026-09-16 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "c4e9a2f7b318" +down_revision = "a8f6f79e40be" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column( + "run", + sa.Column( + "external_id", + sqlmodel.sql.sqltypes.AutoString(length=255), + nullable=True, + ), + ) + op.create_index("ix_run_external_id", "run", ["external_id"], unique=True) + + +def downgrade(): + op.drop_index("ix_run_external_id", table_name="run") + op.drop_column("run", "external_id") diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 03ebe12..02c5fed 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -23,6 +23,7 @@ from sqlalchemy import select as sa_select from sqlmodel import Session, col, select from fluksio.api.deps import CurrentUser, SessionDep, get_current_user +from fluksio.core.config import settings from fluksio.core.db import writing from fluksio.flow.events import event_bus from fluksio.flow.messages import requalify @@ -37,6 +38,9 @@ router = APIRouter( #: A sweep bigger than this is almost always a mistake in a loop. MAX_SWEEP = 1000 +#: Runs per import request. A history is thousands; the client pages it. +MAX_IMPORT = 200 + def elapsed_ms(since: datetime) -> float: """Milliseconds since an instant the columns stored as UTC.""" @@ -49,8 +53,8 @@ def _aware(when: datetime) -> datetime: return when if when.tzinfo else when.replace(tzinfo=UTC) -#: Where a caller may say a run came from. "sweep" is not here because the -#: sweep route writes it itself, and neither is a value a client made up: the +#: Where a caller may say a run came from. "sweep" and "import" are not here +#: because their routes write them, and neither is a value a client made up: the #: column is only worth a table row if it means the same thing every time. RunCause = Literal["api", "cli", "sdk"] @@ -83,6 +87,51 @@ class SweepCreate(BaseModel): no_cache: bool = False +class ImportMetric(BaseModel): + #: Flow-qualified, the way the engine names what a node reports. + name: str = Field(min_length=1, max_length=128) + step: int = -1 + ts: float = 0.0 + value: float + + +class ImportArtifact(BaseModel): + name: str = Field(min_length=1, max_length=255) + filename: str | None = Field(default=None, max_length=255) + media_type: str = Field(default="application/octet-stream", max_length=128) + content_base64: str + + +class ImportEntry(BaseModel): + """One run executed somewhere else, as it ended there.""" + + flow: str + #: What it was called where it ran. Importing it again is a no-op. + external_id: str = Field(min_length=1, max_length=255) + params: dict[str, Any] = Field(default_factory=dict) + seed: int | None = None + status: Literal["ok", "error", "cancelled"] = "ok" + created_at: datetime + started_at: datetime | None = None + finished_at: datetime | None = None + result: dict[str, Any] = Field(default_factory=dict) + metrics: list[ImportMetric] = Field(default_factory=list) + artifacts: list[ImportArtifact] = Field(default_factory=list) + labels: list[str] = Field(default_factory=list) + group_id: str | None = Field(default=None, max_length=64) + + +class ImportCreate(BaseModel): + runs: list[ImportEntry] = Field(default_factory=list) + + +class Imported(BaseModel): + external_id: str + id: str + #: False when an earlier import already made it. + created: bool + + class RunNodeRow(BaseModel): node: str status: str @@ -132,6 +181,9 @@ class RunRow(BaseModel): #: The flow store's own commit. Short, unlike `result`, so the list #: carries it: "what code produced this" is a question asked of a table. commit: str = "" + #: What it was called where it ran, for a run imported rather than + #: executed here. + external_id: str | None = None seed: int | None group_id: str | None #: Whether it ran the working copy rather than what is published. @@ -265,6 +317,45 @@ async def create_sweep( raise HTTPException(status_code=422, detail=str(exc)) from exc +@router.post("/import", response_model=list[Imported], status_code=202) +async def import_runs(body: ImportCreate, request: Request, user: CurrentUser) -> Any: + """Record runs that were executed somewhere else: a history from before. + + Stored as they ended, with `cause` "import" and no commit or code digest, + since this engine never saw the code. The whole request is checked before + anything is written, and an `external_id` already here answers with the + run it made rather than a second one. + """ + if not body.runs: + raise HTTPException(status_code=422, detail="An import needs at least one run") + if len(body.runs) > MAX_IMPORT: + raise HTTPException( + status_code=422, detail=f"An import is capped at {MAX_IMPORT} runs" + ) + cap = settings.MAX_ARTIFACT_BYTES + # The same limit `PUT /artifacts` holds, estimated from the encoded length: + # otherwise any account could fill the data volume this way instead. + if cap and any( + len(artifact.content_base64) * 3 // 4 > cap + for entry in body.runs + for artifact in entry.artifacts + ): + raise HTTPException( + status_code=413, detail=f"An artifact may be at most {cap} bytes" + ) + service = _service(request) + try: + return await run_in_threadpool( + service.import_runs, + [entry.model_dump() for entry in body.runs], + user.email, + ) + except FlowNotFound as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except RunRejected as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + @router.get("", response_model=list[RunRow]) def read_runs( session: SessionDep, @@ -373,6 +464,8 @@ RUN_COLUMNS = ( #: Set on a run made by retrying another, so a sweep that was completed #: rather than repeated says which row replaced which. "parent_id", + #: What an imported run was called where it ran, so the row joins back. + "external_id", "code_digest", "origin_commit", ) diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index e80b26d..f290d5c 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -25,6 +25,8 @@ Three things make a run different from a cascade, and each is deliberate: from __future__ import annotations +import base64 +import binascii import copy import hashlib import json @@ -103,6 +105,17 @@ def new_run_id() -> str: return f"{int(time.time() * 1000):013d}-{uuid.uuid4().hex[:8]}" +def _span_ms(started: datetime | None, finished: datetime | None) -> float: + """Milliseconds between two instants, or 0 without both. Naive is UTC.""" + if started is None or finished is None: + return 0.0 + utc = [ + when if when.tzinfo else when.replace(tzinfo=UTC) + for when in (started, finished) + ] + return round((utc[1] - utc[0]).total_seconds() * 1000, 2) + + def digest_of(params: dict[str, Any], seed: int | None) -> str: """What identifies a run's inputs: same parameters, same digest.""" canonical = json.dumps( @@ -881,34 +894,7 @@ class RunService: issues = batch_issues(flow, self.controller.store, draft=draft) if issues: raise RunRejected(" ".join(issues)) - params = params or {} - # Resolved before it is stored, so what the run records is the same - # reference a python caller would have passed and every later reader — - # the digest, the cache, the run detail — sees one spelling. - params = resolve_references(flow, params, self._artifacts) - # What the run actually starts from, not only what was passed: an input - # left out takes its declared value, and a row that records `{}` cannot - # say which. Folded literally — an initial is a value from the - # definition, never a reference to resolve. - declared = { - one.spec.name: one.initial for one in flow.inputs if one.initial is not None - } - # The run's own seed fills an input of that name, outranking what the - # flow declares and outranked by one passed as a parameter — the order - # `seed_values` applies, moved to where the record is written. - if seed is not None and any(one.spec.name == "seed" for one in flow.inputs): - declared["seed"] = seed - params = {**declared, **params} - # And back the other way, so the run-level column holds the seed the - # run actually used however it arrived. Otherwise `--seed 1` fills one - # column and a declared seed the other, and that is the single field an - # export still has to coalesce. - resolved_seed = params.get("seed") - if isinstance(resolved_seed, int) and not isinstance(resolved_seed, bool): - seed = resolved_seed - # Checked here rather than in the driver: a caller who mistyped a - # parameter should be told now, not by a run that fails in a minute. - seed_values(flow, params, seed) + params, seed = self._complete(flow, params, seed) run = Run( id=new_run_id(), @@ -950,6 +936,40 @@ class RunService: self.queue.add(WorkItem(kind="run", node="", flow=flow.name, run_id=run.id)) return run + def _complete( + self, flow: FlowDef, params: dict[str, Any] | None, seed: int | None + ) -> tuple[dict[str, Any], int | None]: + """The parameters and seed a run of ``flow`` starts from, checked.""" + params = params or {} + # Resolved before it is stored, so what the run records is the same + # reference a python caller would have passed and every later reader — + # the digest, the cache, the run detail — sees one spelling. + params = resolve_references(flow, params, self._artifacts) + # What the run actually starts from, not only what was passed: an input + # left out takes its declared value, and a row that records `{}` cannot + # say which. Folded literally — an initial is a value from the + # definition, never a reference to resolve. + declared = { + one.spec.name: one.initial for one in flow.inputs if one.initial is not None + } + # The run's own seed fills an input of that name, outranking what the + # flow declares and outranked by one passed as a parameter — the order + # `seed_values` applies, moved to where the record is written. + if seed is not None and any(one.spec.name == "seed" for one in flow.inputs): + declared["seed"] = seed + params = {**declared, **params} + # And back the other way, so the run-level column holds the seed the + # run actually used however it arrived. Otherwise `--seed 1` fills one + # column and a declared seed the other, and that is the single field an + # export still has to coalesce. + resolved_seed = params.get("seed") + if isinstance(resolved_seed, int) and not isinstance(resolved_seed, bool): + seed = resolved_seed + # Checked here rather than in the driver: a caller who mistyped a + # parameter should be told now, not by a run that fails in a minute. + seed_values(flow, params, seed) + return params, seed + def _by_key(self, idempotency_key: str) -> Run | None: """The run a key already made, if it made one.""" with Session(db_engine) as session: @@ -957,6 +977,131 @@ class RunService: select(Run).where(col(Run.idempotency_key) == idempotency_key) ).first() + def import_runs( + self, entries: list[dict[str, Any]], actor: str = "" + ) -> list[dict[str, Any]]: + """Record runs that were executed somewhere else, as they ended there. + + Parameters are completed and checked the way `submit` does it, so an + imported row reads like any other. The provenance columns stay empty: + an imported run ran code this engine never saw, and borrowing today's + commit would say otherwise. Nothing is queued and nothing announced — + the runs are over, and a `run_finished` per row would page whatever + alerts on a failure. + + All or nothing: every entry is checked before a row is written. One + whose `external_id` is already here answers with that run instead. + """ + answers: list[dict[str, Any]] = [] + rows: list[Run | RunMetric | RunArtifact] = [] + made: dict[str, str] = {} + flows: dict[str, FlowDef] = {} + for entry in entries: + external_id = str(entry["external_id"]) + existing = made.get(external_id) or getattr( + self._by_external(external_id), "id", None + ) + if existing: + answers.append( + {"external_id": external_id, "id": existing, "created": False} + ) + continue + name = str(entry["flow"]) + if name not in flows: + flows[name] = self.controller.store.read_flow(name) + flow = flows[name] + try: + params, seed = self._complete(flow, entry["params"], entry["seed"]) + except RunRejected as exc: + raise RunRejected(f"{external_id}: {exc}") from exc + started, finished = entry["started_at"], entry["finished_at"] + run = Run( + id=new_run_id(), + flow=flow.name, + flow_version=flow.version, + params=params, + params_digest=digest_of(params, seed), + seed=seed, + group_id=entry["group_id"], + cause="import", + status=entry["status"], + labels=entry["labels"], + created_at=entry["created_at"], + started_at=started, + finished_at=finished, + duration_ms=_span_ms(started, finished), + result=entry["result"], + actor=actor, + external_id=external_id, + ) + rows.append(run) + # Keyed as the table is, so a step reported twice is the later one + # rather than a batch refused over it. + points: dict[tuple[str, int], RunMetric] = {} + for metric in entry["metrics"]: + if not math.isfinite(metric["value"]): + # Refused on the way in by the ports, and not a value the + # column can hold. + continue + point = RunMetric( + run_id=run.id, + name=metric["name"], + step=metric["step"], + ts=metric["ts"], + value=metric["value"], + ) + points[(point.name, point.step)] = point + rows.extend(points.values()) + for artifact in entry["artifacts"]: + rows.append(self._imported_artifact(run.id, external_id, artifact)) + made[external_id] = run.id + answers.append({"external_id": external_id, "id": run.id, "created": True}) + if rows: + with Session(db_engine) as session: + session.add_all(rows) + try: + session.commit() + except IntegrityError as exc: + raise RunRejected( + "Some of these runs were imported meanwhile; send them again" + ) from exc + return answers + + def _imported_artifact( + self, run_id: str, external_id: str, artifact: dict[str, Any] + ) -> RunArtifact: + """Put an imported file's bytes in the store, and the row naming them. + + Stored before the batch is known to be good. A batch refused after + this leaves bytes nothing names, which `sweep_artifacts` collects. + """ + if self._artifacts is None: + raise RunRejected("This engine has no artifact store to import into") + try: + data = base64.b64decode(artifact["content_base64"], validate=True) + except binascii.Error as exc: + raise RunRejected( + f"{external_id}: artifact '{artifact['name']}' is not base64" + ) from exc + ref = self._artifacts.put( + [data], name=artifact["filename"] or "", media_type=artifact["media_type"] + ) + return RunArtifact( + run_id=run_id, + name=artifact["name"], + filename=artifact["filename"], + digest=ref["digest"], + size=ref["size"], + media_type=ref["media_type"], + ) + + def _by_external(self, external_id: str) -> Run | None: + """The run an earlier import made under this id, if there is one.""" + with Session(db_engine) as session: + return session.exec( + select(Run).where(col(Run.external_id) == external_id) + ).first() + def cancel(self, run_id: str) -> bool: """Stop a run: kill what it is executing, schedule nothing further.""" with self._lock: diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index e36649c..b550f7f 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -370,7 +370,13 @@ class Run(SQLModel, table=True): idempotency_key: str | None = Field( default=None, unique=True, index=True, max_length=64 ) - #: Where it was asked for: api, cli, sdk, hook or sweep. + #: What the run was called where it actually ran, for one imported rather + #: than executed here — `mlflow:/`. Unique where it is + #: set, so importing the same history twice adds nothing the second time. + external_id: str | None = Field( + default=None, unique=True, index=True, max_length=255 + ) + #: Where it was asked for: api, cli, sdk, hook, sweep or import. cause: str = Field(default="api", max_length=32) #: Re-execute every node, whatever the stage cache holds for it. no_cache: bool = False diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 77cd89c..db7ad9d 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -56,6 +56,9 @@ RETRY_STATUS = frozenset({502, 503, 504}) #: blip. WAIT_TOLERANCE = 5 +#: Runs per `POST /runs/import`, which is what the engine caps one request at. +IMPORT_PAGE = 200 + #: What a project-local instance is called, beside `.venv` and `.git`. DATA_DIR_NAME = ".fluksio" @@ -374,6 +377,25 @@ class Client: ) return [RunHandle(self, row["id"], row) for row in rows] + 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. + """ + answers: list[dict[str, Any]] = [] + for start in range(0, len(entries), IMPORT_PAGE): + answers.extend( + self._call( + "POST", + "/runs/import", + idempotent=True, + json={"runs": entries[start : start + IMPORT_PAGE]}, + ) + ) + return answers + def run(self, run_id: str) -> dict[str, Any]: result: dict[str, Any] = self._call("GET", f"/runs/{run_id}") return result diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 4f2f6e8..8d0a246 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -4,6 +4,7 @@ The pipeline half — what a hit restores and what a key is made of — is in `tests/flow/test_runs.py`, which runs without one. """ +import base64 import csv import json from datetime import UTC, datetime, timedelta @@ -494,6 +495,134 @@ def test_the_seed_is_recorded_the_same_way_however_it_arrived(): session.commit() +def test_a_run_made_elsewhere_is_imported_as_it_ended(tmp_path): + """History from before the move, completed like a submit and borrowing + nothing from today: no commit, no code digest, nothing queued.""" + flow = FlowDef( + name="history", + mode="batch", + inputs=[ + FlowInput(spec=MessageSpec(name="lr", dtype=DType.FLOAT), initial=0.01), + FlowInput(spec=MessageSpec(name="epochs", dtype=DType.INT)), + ], + ) + store = ArtifactStore(tmp_path) + queue = _Collect() + service = RunService(controller=_OneFlow(flow), queue=queue, artifacts=store) + started = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) + entry = { + "flow": "history", + "external_id": "mlflow:1/abc", + "params": {"epochs": 3}, + "seed": 1001, + "status": "ok", + "created_at": started, + "started_at": started, + "finished_at": started + timedelta(seconds=90), + "result": {"final": {"mse": 7.0}}, + "metrics": [ + {"name": "history.mse", "step": 0, "ts": 1.0, "value": 9.0}, + {"name": "history.mse", "step": 1, "ts": 2.0, "value": float("nan")}, + {"name": "history.mse", "step": 1, "ts": 3.0, "value": 7.0}, + ], + "artifacts": [ + { + "name": "model", + "filename": "model.txt", + "media_type": "text/plain", + "content_base64": base64.b64encode(b"weights").decode(), + } + ], + "labels": ["imported"], + "group_id": "mlflow:1", + } + made: list[str] = [] + try: + [answer] = service.import_runs([entry], actor="me@example.com") + made.append(answer["id"]) + assert answer["created"] is True + with Session(db_engine) as session: + run = session.get(Run, answer["id"]) + assert (run.cause, run.status, run.external_id) == ( + "import", + "ok", + "mlflow:1/abc", + ) + # The declared default is folded in, as a submit would have. + assert run.params == {"lr": 0.01, "epochs": 3} + assert run.params_digest + assert run.duration_ms == 90_000 + assert (run.commit, run.origin_commit, run.code_digest) == ("", "", "") + points = session.exec( + select(RunMetric).where(col(RunMetric.run_id) == run.id) + ).all() + assert sorted((point.step, point.value) for point in points) == [ + (0, 9.0), + (1, 7.0), + ] + [artifact] = session.exec( + select(RunArtifact).where(col(RunArtifact.run_id) == run.id) + ).all() + assert artifact.filename == "model.txt" + assert b"".join(store.read(artifact.digest)) == b"weights" + assert queue.items == [] + + # The same history again adds nothing, and says so. + assert service.import_runs([entry]) == [{**answer, "created": False}] + + # One refused entry takes the good one beside it down with it. + good = {**entry, "external_id": "mlflow:1/def", "artifacts": []} + bad = {**good, "external_id": "mlflow:1/ghi", "params": {"epochs": "three"}} + with pytest.raises(RunRejected, match="mlflow:1/ghi"): + service.import_runs([good, bad]) + assert service._by_external("mlflow:1/def") is None + finally: + with Session(db_engine) as session: + for table in (RunMetric, RunArtifact, Run): + column = Run.id if table is Run else table.run_id + for row in session.exec(select(table).where(col(column).in_(made))): + session.delete(row) + session.commit() + + +def test_an_import_answers_what_it_added(client, superuser_token_headers, monkeypatch): + """So a client can report what it actually added, and pages the rest.""" + seen: dict[str, object] = {} + + class Recorder: + def import_runs(self, entries, actor=""): + seen.update(entries=entries, actor=actor) + return [ + { + "external_id": one["external_id"], + "id": f"run-{index}", + "created": True, + } + for index, one in enumerate(entries) + ] + + monkeypatch.setattr(client.app.state, "run_service", Recorder()) + url = f"{settings.API_V1_STR}/runs/import" + entry = { + "flow": "history", + "external_id": "mlflow:1/abc", + "created_at": "2026-01-02T03:04:05Z", + } + + answer = client.post(url, headers=superuser_token_headers, json={"runs": [entry]}) + assert answer.status_code == 202 + assert answer.json() == [ + {"external_id": "mlflow:1/abc", "id": "run-0", "created": True} + ] + assert seen["entries"][0]["status"] == "ok" + assert seen["actor"] == settings.FIRST_SUPERUSER + + too_many = client.post( + url, headers=superuser_token_headers, json={"runs": [entry] * 201} + ) + assert too_many.status_code == 422 + + DRAFT_SOURCE = "def process():\n return {'answer': 'draft'}\n" PUBLISHED_SOURCE = "def process():\n return {'answer': 'published'}\n" diff --git a/backend/tests/sdk/test_client.py b/backend/tests/sdk/test_client.py index 70c1ab9..bab6c18 100644 --- a/backend/tests/sdk/test_client.py +++ b/backend/tests/sdk/test_client.py @@ -152,3 +152,32 @@ def test_an_export_is_read_as_rows(monkeypatch): assert seen["flow"] == "train" assert seen["status"] == "ok" assert [row["step"] for row in rows] == [0, 1] + + +def test_a_history_is_imported_a_page_at_a_time(monkeypatch): + """The engine caps a request; the caller hands over the whole history.""" + import json + + from fluksio.sdk.client import IMPORT_PAGE + + pages = [] + + def handler(request): + runs = json.loads(request.content)["runs"] + pages.append(len(runs)) + return httpx.Response( + 202, + json=[ + {"external_id": one["external_id"], "id": "r", "created": True} + for one in runs + ], + ) + + client = a_client(handler, monkeypatch) + entries = [{"external_id": f"mlflow:1/{n}"} for n in range(IMPORT_PAGE + 1)] + answers = client.import_runs(entries) + + assert pages == [IMPORT_PAGE, 1] + assert [one["external_id"] for one in answers] == [ + e["external_id"] for e in entries + ] diff --git a/docs/code/api.md b/docs/code/api.md index c2fbac9..fff708e 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -103,6 +103,7 @@ published to. Flows own the namespace; everything else is a client of it. |---|---|---| | `POST` | `/runs/flows/{name}` | queue one run — `{"params": {...}, "seed": 7, "draft": false, "no_cache": false}`. `"cause"` says where it came from — `api` (the default), `cli` or `sdk` | | `POST` | `/runs/flows/{name}/sweep` | queue many, sharing a `group_id` | +| `POST` | `/runs/import` | record runs executed somewhere else, up to 200 per request. Answers `[{"external_id", "id", "created"}]` | | `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?since=`, `?before=`, `?limit=`, `?offset=` | | `DELETE` | `/runs?flow=` | every run of one flow, and everything hanging off them. `flow` is required. Refused with 409 while one of them is still going | | `GET` | `/runs/overview` | one row per flow that has runs, with how many are running or queued | @@ -120,6 +121,16 @@ Submitting answers immediately with a `queued` run. Wrong parameters (an undeclared name, a value of the wrong type) come back as a 422 naming the problem, before anything executes. +`/runs/import` takes `{"runs": [...]}`. Each entry carries `flow`, +`external_id`, `params`, `seed`, `status` (`ok`, `error` or `cancelled`), +`created_at`, `started_at`, `finished_at`, `result`, `metrics` (`name`, +`step`, `ts`, `value`), `artifacts` (`name`, `filename`, `media_type`, +`content_base64`), `labels` and `group_id`. The parameters are completed and +checked as a submit's are, the run is stored with the cause `import` and an +empty commit and code digest, and nothing is queued. The request is checked +whole before anything is written, and an `external_id` already imported +answers with that run and `"created": false`. + `?digest=` filters by the hash of the parameters, which is how you find "every run that used exactly this configuration". diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index fb43e21..1221df0 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -383,6 +383,39 @@ path (`metric.final_metrics.train_loss`) and `metrics="final_metrics.train_loss" selects one. Inputs work the same way, and are compared leaf by leaf, so two configurations differing in one field give that field as the axis. +### Bringing in runs made elsewhere + +Runs are made by running a flow, so a history recorded before the move (an +MLflow experiment, say) has to be imported instead. Each entry names the flow, +the id the run had where it ran, and what it ended with: + +```python +Client().import_runs([ + { + "flow": "train", + "external_id": "mlflow:study-1/3f2a…", + "params": {"lr": 0.01}, + "seed": 1001, + "status": "ok", + "created_at": "2026-03-01T10:00:00Z", + "started_at": "2026-03-01T10:00:01Z", + "finished_at": "2026-03-01T10:40:12Z", + "result": {"final_metrics": {"train_loss": 0.08}}, + "metrics": [{"name": "train.train_loss", "step": 0, "ts": 1772359201.0, "value": 2.3}], + "artifacts": [{"name": "weights", "filename": "model.txt", + "media_type": "text/plain", "content_base64": "…"}], + "labels": ["imported"], + "group_id": "mlflow:study-1", + }, +]) +``` + +The parameters are completed with the flow's declared values and checked the +way a submit checks them. The run is stored with the cause `import` and no +commit or code digest, since this engine never saw the code it ran. Importing +the same `external_id` again adds nothing, and `fluksio export runs` carries +it as a column, so an exported row joins back to where it came from. + ### A dashboard, read against runs A run records values under the same names a dashboard binds to (a run of diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index f846aa5..ed7548c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -1456,6 +1456,212 @@ export const HistoryPointSchema = { description: 'One numeric value a message carried, and when.' } as const; +export const ImportArtifactSchema = { + properties: { + name: { + type: 'string', + maxLength: 255, + minLength: 1, + title: 'Name' + }, + filename: { + anyOf: [ + { + type: 'string', + maxLength: 255 + }, + { + type: 'null' + } + ], + title: 'Filename' + }, + media_type: { + type: 'string', + maxLength: 128, + title: 'Media Type', + default: 'application/octet-stream' + }, + content_base64: { + type: 'string', + title: 'Content Base64' + } + }, + type: 'object', + required: ['name', 'content_base64'], + title: 'ImportArtifact' +} as const; + +export const ImportCreateSchema = { + properties: { + runs: { + items: { + '$ref': '#/components/schemas/ImportEntry' + }, + type: 'array', + title: 'Runs' + } + }, + type: 'object', + title: 'ImportCreate' +} as const; + +export const ImportEntrySchema = { + properties: { + flow: { + type: 'string', + title: 'Flow' + }, + external_id: { + type: 'string', + maxLength: 255, + minLength: 1, + title: 'External Id' + }, + params: { + additionalProperties: true, + type: 'object', + title: 'Params' + }, + seed: { + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], + title: 'Seed' + }, + status: { + type: 'string', + enum: ['ok', 'error', 'cancelled'], + title: 'Status', + default: 'ok' + }, + created_at: { + type: 'string', + format: 'date-time', + title: 'Created At' + }, + started_at: { + anyOf: [ + { + type: 'string', + format: 'date-time' + }, + { + type: 'null' + } + ], + title: 'Started At' + }, + finished_at: { + anyOf: [ + { + type: 'string', + format: 'date-time' + }, + { + type: 'null' + } + ], + title: 'Finished At' + }, + result: { + additionalProperties: true, + type: 'object', + title: 'Result' + }, + metrics: { + items: { + '$ref': '#/components/schemas/ImportMetric' + }, + type: 'array', + title: 'Metrics' + }, + artifacts: { + items: { + '$ref': '#/components/schemas/ImportArtifact' + }, + type: 'array', + title: 'Artifacts' + }, + labels: { + items: { + type: 'string' + }, + type: 'array', + title: 'Labels' + }, + group_id: { + anyOf: [ + { + type: 'string', + maxLength: 64 + }, + { + type: 'null' + } + ], + title: 'Group Id' + } + }, + type: 'object', + required: ['flow', 'external_id', 'created_at'], + title: 'ImportEntry', + description: 'One run executed somewhere else, as it ended there.' +} as const; + +export const ImportMetricSchema = { + properties: { + name: { + type: 'string', + maxLength: 128, + minLength: 1, + title: 'Name' + }, + step: { + type: 'integer', + title: 'Step', + default: -1 + }, + ts: { + type: 'number', + title: 'Ts', + default: 0 + }, + value: { + type: 'number', + title: 'Value' + } + }, + type: 'object', + required: ['name', 'value'], + title: 'ImportMetric' +} as const; + +export const ImportedSchema = { + properties: { + external_id: { + type: 'string', + title: 'External Id' + }, + id: { + type: 'string', + title: 'Id' + }, + created: { + type: 'boolean', + title: 'Created' + } + }, + type: 'object', + required: ['external_id', 'id', 'created'], + title: 'Imported' +} as const; + export const LibraryNodeSchema = { properties: { name: { @@ -2879,6 +3085,17 @@ export const RunDetailSchema = { title: 'Commit', default: '' }, + external_id: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'External Id' + }, seed: { anyOf: [ { @@ -4197,6 +4414,17 @@ export const fluksio__api__routes__runs__RunRowSchema = { title: 'Commit', default: '' }, + external_id: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'External Id' + }, seed: { anyOf: [ { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index e8b496c..4428496 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, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, 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, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, 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, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AlertsService { /** @@ -1942,6 +1942,31 @@ export class RunsService { }); } + /** + * Import Runs + * Record runs that were executed somewhere else: a history from before. + * + * Stored as they ended, with `cause` "import" and no commit or code digest, + * since this engine never saw the code. The whole request is checked before + * anything is written, and an `external_id` already here answers with the + * run it made rather than a second one. + * @param data The data for the request. + * @param data.requestBody + * @returns Imported Successful Response + * @throws ApiError + */ + public static importRuns(data: RunsImportRunsData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/runs/import', + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Runs * Runs, newest first. The queryable table an experiment log needs. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index a10b029..bf54e3e 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -475,6 +475,7 @@ export type fluksio__api__routes__runs__RunRow = { origin_commit?: string; code_digest?: string; commit?: string; + external_id?: (string | null); seed: (number | null); group_id: (string | null); draft?: boolean; @@ -524,6 +525,55 @@ export type HTTPValidationError = { detail?: Array; }; +export type ImportArtifact = { + name: string; + filename?: (string | null); + media_type?: string; + content_base64: string; +}; + +export type ImportCreate = { + runs?: Array; +}; + +export type Imported = { + external_id: string; + id: string; + created: boolean; +}; + +/** + * One run executed somewhere else, as it ended there. + */ +export type ImportEntry = { + flow: string; + external_id: string; + params?: { + [key: string]: unknown; + }; + seed?: (number | null); + status?: 'ok' | 'error' | 'cancelled'; + created_at: string; + started_at?: (string | null); + finished_at?: (string | null); + result?: { + [key: string]: unknown; + }; + metrics?: Array; + artifacts?: Array; + labels?: Array<(string)>; + group_id?: (string | null); +}; + +export type status = 'ok' | 'error' | 'cancelled'; + +export type ImportMetric = { + name: string; + step?: number; + ts?: number; + value: number; +}; + /** * A node source shared across flows, and who is using it. */ @@ -1024,6 +1074,7 @@ export type RunDetail = { origin_commit?: string; code_digest?: string; commit?: string; + external_id?: (string | null); seed: (number | null); group_id: (string | null); draft?: boolean; @@ -1839,6 +1890,12 @@ export type RunsCreateSweepData = { export type RunsCreateSweepResponse = (Array); +export type RunsImportRunsData = { + requestBody: ImportCreate; +}; + +export type RunsImportRunsResponse = (Array); + export type RunsReadRunsData = { before?: (string | null); digest?: (string | null);