Add POST /runs/import for runs executed elsewhere

A history recorded before the move (MLflow experiments) had nowhere to
go, since runs were only made by running a flow. Each entry is stored as
it ended: parameters completed and checked like a submit, cause
"import", empty commit and code digest, metrics and artifacts written
as given, nothing queued or announced. Run.external_id (new migration)
makes a second import a no-op and is an export column so a row joins
back. The request is all or nothing and capped at 200 entries;
Client.import_runs pages a longer history.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QgG6TJw5qBZzaLghQSRVrq
This commit is contained in:
2026-09-16 17:01:07 +02:00
co-authored by Claude Opus 5
parent eeaa46d4ae
commit d9630f66f5
12 changed files with 848 additions and 32 deletions
@@ -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")
+95 -2
View File
@@ -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",
)
+173 -28
View File
@@ -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:
+7 -1
View File
@@ -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:<experiment>/<run>`. 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
+22
View File
@@ -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
+129
View File
@@ -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"
+29
View File
@@ -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
]