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

Record which inputs a run's caller passed, and mark defaults from it

Browse Source 
`Run.passed` (migration d5a3c8e1f246) keeps the names the caller chose, so
the run detail and the comparison table can mark a value as the flow's own
again; rows from before it fall back to the names their params lack. Retries
hand on their parent's list, imports record their entry's.

Also: a cancel of a queued run no longer leaves its id in `_cancelled`, a
linked comparison no longer crashes before the list loads, and the pre-commit
failures are fixed (stale client, `text-[10px]`, worker formatting).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ
This commit is contained in:
stroblme
2026-10-05 15:01:33 +02:00
co-authored by Claude Opus 5.5
parent 84e054cdd1
commit 8e655c842f
12 changed files with 134 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/alembic/versions/d5a3c8e1f246_run_passed.py
+30
View File
@@ -0,0 +1,30 @@
"""run.passed
A run's params hold every input it started from, declared values included, so
the record could no longer tell an input left at the flow's default from one
the caller chose. The names the caller passed go beside them.
Revision ID: d5a3c8e1f246
Revises: c4e9a2f7b318
Create Date: 2026-10-05
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "d5a3c8e1f246"
down_revision = "c4e9a2f7b318"
branch_labels = None
depends_on = None
def upgrade():
# Nullable: a run submitted before this did not record it, and null is that
# rather than a run that was passed nothing.
op.add_column("run", sa.Column("passed", sa.JSON(), nullable=True))
def downgrade():
op.drop_column("run", "passed")
backend/fluksio/api/routes/runs.py
+3
View File
@@ -169,6 +169,9 @@ class RunRow(BaseModel):
status_reason: str
cause: str
params: dict[str, Any]
#: The names in `params` the caller chose; any other took the flow's
#: declared value. Null for a run from before this was recorded.
passed: list[str] | None = None
params_digest: str
#: The user repository's commit, for a flow declared in code with the
#: decorators. Empty for one drawn on the canvas, where `commit` is the
backend/fluksio/flow/runs.py
+25 -11
View File
@@ -882,8 +882,13 @@ class RunService:
no_cache: bool = False,
idempotency_key: str | None = None,
parent_id: str | None = None,
passed: list[str] | None = None,
) -> Run:
"""Journal a run and wake an engine up for it. Never blocks on it."""
"""Journal a run and wake an engine up for it. Never blocks on it.
``passed`` overrides which of ``params`` count as the caller's own, for
a retry handing on its parent's whole params.
"""
# Before anything else, including reading the flow: a caller retrying a
# submit it never got an answer for is owed the run that answer was
# about, whatever the flow says now.
@@ -895,7 +900,7 @@ class RunService:
issues = batch_issues(flow, self.controller.store, draft=draft)
if issues:
raise RunRejected(" ".join(issues))
params, seed = self._complete(flow, params, seed)
params, seed, named = self._complete(flow, params, seed)
run = Run(
id=new_run_id(),
@@ -905,6 +910,7 @@ class RunService:
origin_commit=origin_commit(flow),
code_digest=run_digest(flow),
params=params,
passed=named if passed is None else passed,
params_digest=digest_of(params, seed),
seed=seed,
group_id=group_id,
@@ -939,13 +945,15 @@ class RunService:
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."""
) -> tuple[dict[str, Any], int | None, list[str]]:
"""The parameters and seed a run of ``flow`` starts from, checked, and
the names among them the caller chose."""
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)
passed = set(params)
# 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
@@ -958,6 +966,7 @@ class RunService:
# `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
passed.add("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
@@ -969,7 +978,7 @@ class RunService:
# 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
return params, seed, sorted(passed)
def _by_key(self, idempotency_key: str) -> Run | None:
"""The run a key already made, if it made one."""
@@ -1012,7 +1021,9 @@ class RunService:
flows[name] = self.controller.store.read_flow(name)
flow = flows[name]
try:
params, seed = self._complete(flow, entry["params"], entry["seed"])
params, seed, passed = 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"]
@@ -1021,6 +1032,7 @@ class RunService:
flow=flow.name,
flow_version=flow.version,
params=params,
passed=passed,
params_digest=digest_of(params, seed),
seed=seed,
group_id=entry["group_id"],
@@ -1108,11 +1120,10 @@ class RunService:
with self._lock:
if run_id not in self._active:
# Not running here — if it is still queued, refusing to start
# is all the cancelling it needs.
cancelled = self._finish_unstarted(run_id, "queued")
if cancelled:
self._cancelled.add(run_id)
return cancelled
# is all the cancelling it needs: the driver's claim finds the
# row no longer queued, so it needs no mark (and would never
# discard one).
return self._finish_unstarted(run_id, "queued")
self._cancelled.add(run_id)
pipeline = self._active[run_id]
if pipeline is None:
@@ -1165,6 +1176,9 @@ class RunService:
no_cache=source.no_cache,
draft=source.draft,
parent_id=source.id,
# What its parent's caller chose. A parent from before that was
# recorded falls back to the names its params carry.
passed=source.passed,
)
def _flow_of(self, run_id: str) -> str | None:
backend/fluksio/models.py
+3
View File
@@ -356,6 +356,9 @@ class Run(SQLModel, table=True):
#: repository, or for a flow drawn on the canvas.
code_digest: str = Field(default="", max_length=64)
params: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
#: The names in `params` the caller chose; the rest are the flow's declared
#: values. Null for a run recorded before this was kept.
passed: list[str] | None = Field(sa_column=Column(JSON), default=None)
#: sha256 over params and seed. Two runs of the same thing share it, which
#: is what makes "have I already run this?" a lookup.
params_digest: str = Field(default="", index=True, max_length=64)
backend/tests/api/routes/test_runs.py
+2
View File
@@ -454,6 +454,8 @@ def test_a_run_records_the_inputs_it_actually_starts_from():
spelled = service.submit("study", {"lr": 0.01, "epochs": 5})
made.append(spelled.id)
assert spelled.params_digest == defaulted.params_digest
# Only what the caller chose says which value was the flow's own.
assert (defaulted.passed, spelled.passed) == (["epochs"], ["epochs", "lr"])
finally:
with Session(db_engine) as session:
for run in session.exec(select(Run).where(col(Run.id).in_(made))).all():