A run says whether the dashboard, the CLI or a script asked for it
`POST /runs/flows/{name}` hardcoded `cause: "api"`, so every row in the
history claimed the same origin. The body now carries an optional `cause`,
closed to the values the column knows — the dashboard sends nothing and stays
"api", `fluksio run` says "cli", and the SDK client says "sdk".
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013Gf7WaExcJ9bs3kfJXB3nK
This commit is contained in:
@@ -6,7 +6,7 @@ or to listen on the flow socket, which carries its start and finish.
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from typing import Any
|
from typing import Any, Literal
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||||
from fastapi.concurrency import run_in_threadpool
|
from fastapi.concurrency import run_in_threadpool
|
||||||
@@ -35,6 +35,12 @@ def elapsed_ms(since: datetime) -> float:
|
|||||||
return round((datetime.now(UTC) - start).total_seconds() * 1000, 2)
|
return round((datetime.now(UTC) - start).total_seconds() * 1000, 2)
|
||||||
|
|
||||||
|
|
||||||
|
#: 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
|
||||||
|
#: column is only worth a table row if it means the same thing every time.
|
||||||
|
RunCause = Literal["api", "cli", "sdk"]
|
||||||
|
|
||||||
|
|
||||||
class RunCreate(BaseModel):
|
class RunCreate(BaseModel):
|
||||||
params: dict[str, Any] = Field(default_factory=dict)
|
params: dict[str, Any] = Field(default_factory=dict)
|
||||||
seed: int | None = None
|
seed: int | None = None
|
||||||
@@ -42,6 +48,8 @@ class RunCreate(BaseModel):
|
|||||||
draft: bool = False
|
draft: bool = False
|
||||||
#: Execute every node, whatever an earlier run already worked out.
|
#: Execute every node, whatever an earlier run already worked out.
|
||||||
no_cache: bool = False
|
no_cache: bool = False
|
||||||
|
#: Who is asking. The dashboard leaves it, and is the "api" default.
|
||||||
|
cause: RunCause = "api"
|
||||||
|
|
||||||
|
|
||||||
class SweepEntry(BaseModel):
|
class SweepEntry(BaseModel):
|
||||||
@@ -172,7 +180,7 @@ async def create_run(
|
|||||||
name,
|
name,
|
||||||
params=body.params,
|
params=body.params,
|
||||||
seed=body.seed,
|
seed=body.seed,
|
||||||
cause="api",
|
cause=body.cause,
|
||||||
actor=user.email,
|
actor=user.email,
|
||||||
draft=body.draft,
|
draft=body.draft,
|
||||||
no_cache=body.no_cache,
|
no_cache=body.no_cache,
|
||||||
|
|||||||
@@ -340,7 +340,7 @@ class Run(SQLModel, table=True):
|
|||||||
group_id: str | None = Field(default=None, index=True, max_length=64)
|
group_id: str | None = Field(default=None, index=True, max_length=64)
|
||||||
#: The run this one was made from, on a retry.
|
#: The run this one was made from, on a retry.
|
||||||
parent_id: str | None = Field(default=None, max_length=64)
|
parent_id: str | None = Field(default=None, max_length=64)
|
||||||
#: api, hook, sweep or cli.
|
#: Where it was asked for: api, cli, sdk, hook or sweep.
|
||||||
cause: str = Field(default="api", max_length=32)
|
cause: str = Field(default="api", max_length=32)
|
||||||
#: Re-execute every node, whatever the stage cache holds for it.
|
#: Re-execute every node, whatever the stage cache holds for it.
|
||||||
no_cache: bool = False
|
no_cache: bool = False
|
||||||
|
|||||||
@@ -459,7 +459,11 @@ def cmd_run(args: argparse.Namespace, rest: list[str]) -> int:
|
|||||||
):
|
):
|
||||||
params = _ask_params(definition)
|
params = _ask_params(definition)
|
||||||
handle = client.submit(
|
handle = client.submit(
|
||||||
args.flow, params, seed=args.seed, no_cache=args.no_cache
|
args.flow,
|
||||||
|
params,
|
||||||
|
seed=args.seed,
|
||||||
|
no_cache=args.no_cache,
|
||||||
|
cause="cli",
|
||||||
)
|
)
|
||||||
_say(f"{handle.id} queued {json.dumps(params)}")
|
_say(f"{handle.id} queued {json.dumps(params)}")
|
||||||
if not wait:
|
if not wait:
|
||||||
|
|||||||
@@ -226,11 +226,19 @@ class Client:
|
|||||||
params: dict[str, Any] | None = None,
|
params: dict[str, Any] | None = None,
|
||||||
seed: int | None = None,
|
seed: int | None = None,
|
||||||
no_cache: bool = False,
|
no_cache: bool = False,
|
||||||
|
cause: str = "sdk",
|
||||||
) -> RunHandle:
|
) -> RunHandle:
|
||||||
|
"""Queue a run. ``cause`` is what the history records it as coming
|
||||||
|
from — a script is the default, `fluksio run` says so itself."""
|
||||||
row = self._call(
|
row = self._call(
|
||||||
"POST",
|
"POST",
|
||||||
f"/runs/flows/{flow}",
|
f"/runs/flows/{flow}",
|
||||||
json={"params": params or {}, "seed": seed, "no_cache": no_cache},
|
json={
|
||||||
|
"params": params or {},
|
||||||
|
"seed": seed,
|
||||||
|
"no_cache": no_cache,
|
||||||
|
"cause": cause,
|
||||||
|
},
|
||||||
)
|
)
|
||||||
return RunHandle(self, row["id"], row)
|
return RunHandle(self, row["id"], row)
|
||||||
|
|
||||||
|
|||||||
@@ -311,6 +311,44 @@ def test_a_running_run_reports_how_long_it_has_been_going(
|
|||||||
assert rows[0]["duration_ms"] >= 30_000
|
assert rows[0]["duration_ms"] >= 30_000
|
||||||
|
|
||||||
|
|
||||||
|
def test_a_run_records_which_caller_asked_for_it(
|
||||||
|
client, superuser_token_headers, monkeypatch
|
||||||
|
):
|
||||||
|
"""The dashboard, the CLI and the SDK are told apart by what they send.
|
||||||
|
|
||||||
|
Client-supplied, so the vocabulary is closed: a column nobody can write
|
||||||
|
free text into is one a table can group by.
|
||||||
|
"""
|
||||||
|
seen: dict[str, object] = {}
|
||||||
|
|
||||||
|
class Recorder:
|
||||||
|
def submit(self, name, **kwargs):
|
||||||
|
seen.update(kwargs)
|
||||||
|
return Run(
|
||||||
|
id="cause-1",
|
||||||
|
flow=name,
|
||||||
|
cause=str(kwargs["cause"]),
|
||||||
|
created_at=datetime.now(UTC),
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr(client.app.state, "run_service", Recorder())
|
||||||
|
url = f"{settings.API_V1_STR}/runs/flows/demo"
|
||||||
|
|
||||||
|
answer = client.post(url, headers=superuser_token_headers, json={"cause": "cli"})
|
||||||
|
assert answer.status_code == 202
|
||||||
|
assert (seen["cause"], answer.json()["cause"]) == ("cli", "cli")
|
||||||
|
|
||||||
|
# Nothing said still means the dashboard, which is the only caller that
|
||||||
|
# does not name itself.
|
||||||
|
client.post(url, headers=superuser_token_headers, json={})
|
||||||
|
assert seen["cause"] == "api"
|
||||||
|
|
||||||
|
refused = client.post(
|
||||||
|
url, headers=superuser_token_headers, json={"cause": "somewhere else"}
|
||||||
|
)
|
||||||
|
assert refused.status_code == 422
|
||||||
|
|
||||||
|
|
||||||
def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers):
|
def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers):
|
||||||
"""`/overview` is declared before `/{run_id}`, which would swallow it."""
|
"""`/overview` is declared before `/{run_id}`, which would swallow it."""
|
||||||
answer = client.get(
|
answer = client.get(
|
||||||
|
|||||||
@@ -200,9 +200,10 @@ def test_a_local_run_always_waits(monkeypatch) -> None:
|
|||||||
def get_flow(self, name: str) -> dict[str, object]:
|
def get_flow(self, name: str) -> dict[str, object]:
|
||||||
return {"definition": {"inputs": []}}
|
return {"definition": {"inputs": []}}
|
||||||
|
|
||||||
def submit(self, flow, params, seed=None, no_cache=False):
|
def submit(self, flow, params, seed=None, no_cache=False, cause="sdk"):
|
||||||
submitted["flow"] = flow
|
submitted["flow"] = flow
|
||||||
submitted["no_cache"] = no_cache
|
submitted["no_cache"] = no_cache
|
||||||
|
submitted["cause"] = cause
|
||||||
return FakeHandle()
|
return FakeHandle()
|
||||||
|
|
||||||
def run(self, run_id: str) -> dict[str, object]:
|
def run(self, run_id: str) -> dict[str, object]:
|
||||||
@@ -216,7 +217,13 @@ def test_a_local_run_always_waits(monkeypatch) -> None:
|
|||||||
|
|
||||||
args = _parser().parse_args(["run", "train", "--local", "--no-sync", "--no-cache"])
|
args = _parser().parse_args(["run", "train", "--local", "--no-sync", "--no-cache"])
|
||||||
assert cli.cmd_run(args, []) == 0
|
assert cli.cmd_run(args, []) == 0
|
||||||
assert submitted == {"flow": "train", "no_cache": True, "waited": True}
|
# "cli" rather than the SDK's default: the history says which asked.
|
||||||
|
assert submitted == {
|
||||||
|
"flow": "train",
|
||||||
|
"no_cache": True,
|
||||||
|
"waited": True,
|
||||||
|
"cause": "cli",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None:
|
def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None:
|
||||||
@@ -240,7 +247,7 @@ def test_ctrl_c_while_waiting_cancels_the_run(monkeypatch) -> None:
|
|||||||
def get_flow(self, name: str) -> dict[str, object]:
|
def get_flow(self, name: str) -> dict[str, object]:
|
||||||
return {"definition": {"inputs": []}}
|
return {"definition": {"inputs": []}}
|
||||||
|
|
||||||
def submit(self, flow, params, seed=None, no_cache=False):
|
def submit(self, flow, params, seed=None, no_cache=False, cause="sdk"):
|
||||||
return FakeHandle()
|
return FakeHandle()
|
||||||
|
|
||||||
def cancel(self, run_id: str) -> None:
|
def cancel(self, run_id: str) -> None:
|
||||||
|
|||||||
+1
-1
@@ -102,7 +102,7 @@ published to. Flows own the namespace; everything else is a client of it.
|
|||||||
|
|
||||||
| Method | Path | What |
|
| Method | Path | What |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `POST` | `/runs/flows/{name}` | queue one run — `{"params": {...}, "seed": 7, "draft": false, "no_cache": false}` |
|
| `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/flows/{name}/sweep` | queue many, sharing a `group_id` |
|
||||||
| `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?limit=`, `?offset=` |
|
| `GET` | `/runs` | the queryable history: `?flow=`, `?status=`, `?group=`, `?digest=`, `?limit=`, `?offset=` |
|
||||||
| `GET` | `/runs/overview` | one row per flow that has runs, with how many are running or queued |
|
| `GET` | `/runs/overview` | one row per flow that has runs, with how many are running or queued |
|
||||||
|
|||||||
@@ -2384,6 +2384,12 @@ export const RunCreateSchema = {
|
|||||||
type: 'boolean',
|
type: 'boolean',
|
||||||
title: 'No Cache',
|
title: 'No Cache',
|
||||||
default: false
|
default: false
|
||||||
|
},
|
||||||
|
cause: {
|
||||||
|
type: 'string',
|
||||||
|
enum: ['api', 'cli', 'sdk'],
|
||||||
|
title: 'Cause',
|
||||||
|
default: 'api'
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
type: 'object',
|
type: 'object',
|
||||||
|
|||||||
@@ -865,8 +865,11 @@ export type RunCreate = {
|
|||||||
seed?: (number | null);
|
seed?: (number | null);
|
||||||
draft?: boolean;
|
draft?: boolean;
|
||||||
no_cache?: boolean;
|
no_cache?: boolean;
|
||||||
|
cause?: 'api' | 'cli' | 'sdk';
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export type cause = 'api' | 'cli' | 'sdk';
|
||||||
|
|
||||||
export type RunDetail = {
|
export type RunDetail = {
|
||||||
id: string;
|
id: string;
|
||||||
flow: string;
|
flow: string;
|
||||||
|
|||||||
Reference in New Issue
Block a user