Record which inputs a run's caller passed, and mark defaults from it
`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).
This commit is contained in:
@@ -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")
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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({
|
||||
<dt className={cn(LABEL, "flex items-baseline gap-1.5")}>
|
||||
{name}
|
||||
{note && (
|
||||
<span className="rounded-full border border-border px-1.5 text-[10px]">
|
||||
<span className="rounded-full border border-border px-1.5 text-xs">
|
||||
{note}
|
||||
</span>
|
||||
)}
|
||||
|
||||
@@ -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 && (
|
||||
<ParamDiff rows={runs.filter((run) => selected.includes(run.id))} />
|
||||
)}
|
||||
{picked.length > 1 && <ParamDiff rows={picked} />}
|
||||
</section>
|
||||
</div>
|
||||
)
|
||||
@@ -359,7 +360,7 @@ export function FlowRail({
|
||||
>
|
||||
<span className="min-w-0 flex-1 truncate">{label}</span>
|
||||
{note && (
|
||||
<span className="rounded-full border border-border px-1.5 text-[10px]">
|
||||
<span className="rounded-full border border-border px-1.5 text-xs">
|
||||
{note}
|
||||
</span>
|
||||
)}
|
||||
@@ -784,16 +785,22 @@ function ParamDiff({ rows }: { rows: RunRow[] }) {
|
||||
)}
|
||||
{varying.map((key) => (
|
||||
<TableCell key={key}>
|
||||
{key in run.params ? (
|
||||
<ParamValue value={run.params[key]} />
|
||||
) : 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.
|
||||
<span className="flex items-baseline gap-1.5">
|
||||
<ParamValue value={declared.get(key)} />
|
||||
<span className="rounded-full border border-border px-1.5 text-[10px] text-muted-foreground">
|
||||
default
|
||||
</span>
|
||||
<ParamValue
|
||||
value={
|
||||
key in run.params
|
||||
? run.params[key]
|
||||
: declared.get(key)
|
||||
}
|
||||
/>
|
||||
{tookDefault(run, key) && (
|
||||
<span className="rounded-full border border-border px-1.5 text-xs text-muted-foreground">
|
||||
default
|
||||
</span>
|
||||
)}
|
||||
</span>
|
||||
) : (
|
||||
<span className="text-muted-foreground text-xs">unset</span>
|
||||
|
||||
@@ -321,6 +321,17 @@ export function varyingKeys(runs: { params: Record<string, unknown> }[]) {
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<string, unknown>; 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 "—"
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user