diff --git a/backend/fluksio/alembic/versions/d5a3c8e1f246_run_passed.py b/backend/fluksio/alembic/versions/d5a3c8e1f246_run_passed.py new file mode 100644 index 0000000..560e1d1 --- /dev/null +++ b/backend/fluksio/alembic/versions/d5a3c8e1f246_run_passed.py @@ -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") diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index 02c5fed..e1c1993 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -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 diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 2525d2a..1082af5 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -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: diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index b550f7f..8fc118a 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -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) diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index dcf4fc8..35bd238 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -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(): diff --git a/frontend/src/client/core/OpenAPI.ts b/frontend/src/client/core/OpenAPI.ts index 979d72b..556cdf0 100644 --- a/frontend/src/client/core/OpenAPI.ts +++ b/frontend/src/client/core/OpenAPI.ts @@ -48,7 +48,7 @@ export const OpenAPI: OpenAPIConfig = { PASSWORD: undefined, TOKEN: undefined, USERNAME: undefined, - VERSION: '0.1.5+dev', + VERSION: '0.1.6+dev', WITH_CREDENTIALS: false, interceptors: { request: new Interceptors(), diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index bfcec84..209099c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -3249,6 +3249,20 @@ export const RunDetailSchema = { type: 'object', title: 'Params' }, + passed: { + anyOf: [ + { + items: { + type: 'string' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Passed' + }, params_digest: { type: 'string', title: 'Params Digest' @@ -4604,6 +4618,20 @@ export const fluksio__api__routes__runs__RunRowSchema = { type: 'object', title: 'Params' }, + passed: { + anyOf: [ + { + items: { + type: 'string' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Passed' + }, params_digest: { type: 'string', title: 'Params Digest' diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 744a3e3..af5136f 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -506,6 +506,7 @@ export type fluksio__api__routes__runs__RunRow = { params: { [key: string]: unknown; }; + passed?: (Array<(string)> | null); params_digest: string; origin_commit?: string; code_digest?: string; @@ -1105,6 +1106,7 @@ export type RunDetail = { params: { [key: string]: unknown; }; + passed?: (Array<(string)> | null); params_digest: string; origin_commit?: string; code_digest?: string; diff --git a/frontend/src/components/Runs/RunDetail.tsx b/frontend/src/components/Runs/RunDetail.tsx index 5c626e9..a1a3d27 100644 --- a/frontend/src/components/Runs/RunDetail.tsx +++ b/frontend/src/components/Runs/RunDetail.tsx @@ -30,6 +30,7 @@ import { runQueryOptions, shortCommit, shortId, + tookDefault, useCancelRun, useFlowInputs, useRetryRun, @@ -64,11 +65,15 @@ export function RunDetail({ id }: { id: string }) { ...[...declared.keys()].map((name) => ({ name, value: name in run.params ? run.params[name] : declared.get(name), - fromFlow: !(name in run.params), + fromFlow: tookDefault(run, name), })), ...Object.keys(run.params) .filter((name) => !declared.has(name)) - .map((name) => ({ name, value: run.params[name], fromFlow: false })), + .map((name) => ({ + name, + value: run.params[name], + fromFlow: tookDefault(run, name), + })), ] return ( @@ -246,7 +251,7 @@ export function Entry({
{name} {note && ( - + {note} )} diff --git a/frontend/src/components/Runs/RunsScreen.tsx b/frontend/src/components/Runs/RunsScreen.tsx index 019af8d..53cb78e 100644 --- a/frontend/src/components/Runs/RunsScreen.tsx +++ b/frontend/src/components/Runs/RunsScreen.tsx @@ -62,6 +62,7 @@ import { STEP_AXIS, shortCommit, shortId, + tookDefault, useFlowInputs, varyingKeys, } from "./queries" @@ -96,6 +97,8 @@ export function RunsScreen({ const runs: RunRow[] = data?.pages.flat() ?? [] const selected = search.compare ? search.compare.split(",") : [] + // Only those the list has read: a linked comparison renders before it has. + const picked = runs.filter((run) => selected.includes(run.id)) const put = (next: string[]) => update({ @@ -305,9 +308,7 @@ export function RunsScreen({ {/* Below the chart, and only once there are two runs to hold apart. The table above keeps its height and the page scrolls instead: what differs is worth reading in full. */} - {selected.length > 1 && ( - selected.includes(run.id))} /> - )} + {picked.length > 1 && } ) @@ -359,7 +360,7 @@ export function FlowRail({ > {label} {note && ( - + {note} )} @@ -784,16 +785,22 @@ function ParamDiff({ rows }: { rows: RunRow[] }) { )} {varying.map((key) => ( - {key in run.params ? ( - - ) : declared.has(key) ? ( - // Not "empty": this run was never given the parameter, so - // what it ran with is what the flow declares. + {key in run.params || declared.has(key) ? ( + // Not "empty": a run never given the parameter ran with + // what the flow declares, and says so. - - - default - + + {tookDefault(run, key) && ( + + default + + )} ) : ( unset diff --git a/frontend/src/components/Runs/queries.ts b/frontend/src/components/Runs/queries.ts index 82e5e87..d679ba4 100644 --- a/frontend/src/components/Runs/queries.ts +++ b/frontend/src/components/Runs/queries.ts @@ -321,6 +321,17 @@ export function varyingKeys(runs: { params: Record }[]) { }) } +/** + * Whether a run took `name` from its flow rather than from its caller. A run + * from before `passed` was recorded can only say so for a name its params lack. + */ +export function tookDefault( + run: { params: Record; passed?: string[] | null }, + name: string, +): boolean { + return run.passed ? !run.passed.includes(name) : !(name in run.params) +} + /** A parameter value, as narrow as it can be written. */ export function paramText(value: unknown): string { if (value === null || value === undefined) return "—" diff --git a/worker/fluksio_worker/worker_main.py b/worker/fluksio_worker/worker_main.py index 3cccbcc..362f0ad 100644 --- a/worker/fluksio_worker/worker_main.py +++ b/worker/fluksio_worker/worker_main.py @@ -139,9 +139,7 @@ class _Reporter(ModuleType): bytes(source), name or "artifact.bin", media_type, volatile ) path = str(source) - return _store_file( - path, name or os.path.basename(path), media_type, volatile - ) + return _store_file(path, name or os.path.basename(path), media_type, volatile) def load_artifact(self, ref: dict[str, Any]) -> str: """Fetch an artifact and hand back a local path to read it from."""