Runs: a flow taken from its inputs to its outputs, once
A cascade has no end worth recording; a run does. Parameters go in, the graph executes until it drains, and the result is kept — which is what an ML experiment is and what a CI-style job is, so both are one entity. Each run gets a state backend namespaced to itself, so two runs of one flow cannot overwrite each other's messages; that is a constructor argument rather than a change to the pipeline, because every key the engine keeps already goes through the state backend. Its record is written by the driver thread rather than folded off the event bus, which drops what it cannot keep up with. Its own Redis stream wakes an engine up, and from the claim onwards the database row is the truth: redelivering hours of training because an acknowledgement was late is not recovery, so a stale lease is what marks a run whose engine died. Flows gain mode: batch, which are built and validated but never activated, and nodes gain a device label for the worker that must run them. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD
This commit is contained in:
Claude Fable 5
@@ -0,0 +1,126 @@
|
||||
"""Add run, run_node, run_metric and run_artifact
|
||||
|
||||
Runs are the batch shape of the engine: a flow taken from its inputs to its
|
||||
outputs once, with parameters that identify it and a result worth keeping.
|
||||
Deliberately its own tables rather than columns on flow_run — a cascade row is
|
||||
rolled up and pruned on a retention window, and an experiment must not be.
|
||||
|
||||
Revision ID: a3f1c07b52d9
|
||||
Revises: 087c44e16304
|
||||
Create Date: 2026-08-18 09:12:44.108312
|
||||
|
||||
"""
|
||||
import sqlalchemy as sa
|
||||
import sqlmodel.sql.sqltypes
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = 'a3f1c07b52d9'
|
||||
down_revision = '087c44e16304'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
op.create_table(
|
||||
'run',
|
||||
sa.Column('id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.Column('flow_version', sa.Integer(), nullable=False),
|
||||
sa.Column('commit', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('params', sa.JSON(), nullable=True),
|
||||
sa.Column(
|
||||
'params_digest',
|
||||
sqlmodel.sql.sqltypes.AutoString(length=64),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column('seed', sa.Integer(), nullable=True),
|
||||
sa.Column('group_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
|
||||
sa.Column('parent_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=True),
|
||||
sa.Column('cause', sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False),
|
||||
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
|
||||
sa.Column(
|
||||
'status_reason',
|
||||
sqlmodel.sql.sqltypes.AutoString(length=1024),
|
||||
nullable=False,
|
||||
),
|
||||
sa.Column('labels', sa.JSON(), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
|
||||
sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column('duration_ms', sa.Float(), nullable=False),
|
||||
sa.Column('result', sa.JSON(), nullable=True),
|
||||
sa.Column('engine', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('lease_at', sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column('actor', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
op.create_index(op.f('ix_run_flow'), 'run', ['flow'], unique=False)
|
||||
op.create_index(op.f('ix_run_status'), 'run', ['status'], unique=False)
|
||||
op.create_index(op.f('ix_run_group_id'), 'run', ['group_id'], unique=False)
|
||||
op.create_index(op.f('ix_run_created_at'), 'run', ['created_at'], unique=False)
|
||||
op.create_index(
|
||||
op.f('ix_run_params_digest'), 'run', ['params_digest'], unique=False
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
'run_node',
|
||||
sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(length=16), nullable=False),
|
||||
sa.Column('attempt', sa.Integer(), nullable=False),
|
||||
sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column('duration_ms', sa.Float(), nullable=False),
|
||||
sa.Column('worker', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('error', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column('logs', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
|
||||
sa.Column(
|
||||
'cache_key', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False
|
||||
),
|
||||
sa.PrimaryKeyConstraint('run_id', 'node'),
|
||||
)
|
||||
op.create_index(
|
||||
op.f('ix_run_node_cache_key'), 'run_node', ['cache_key'], unique=False
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
'run_metric',
|
||||
sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False),
|
||||
sa.Column('step', sa.Integer(), nullable=False),
|
||||
sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.Column('ts', sa.Float(), nullable=False),
|
||||
sa.Column('value', sa.Float(), nullable=False),
|
||||
sa.PrimaryKeyConstraint('run_id', 'name', 'step'),
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
'run_artifact',
|
||||
sa.Column('run_id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False),
|
||||
sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
|
||||
sa.Column('digest', sqlmodel.sql.sqltypes.AutoString(length=71), nullable=False),
|
||||
sa.Column('size', sa.BigInteger(), nullable=False),
|
||||
sa.Column(
|
||||
'media_type', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=False
|
||||
),
|
||||
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
|
||||
sa.PrimaryKeyConstraint('run_id', 'name'),
|
||||
)
|
||||
op.create_index(
|
||||
op.f('ix_run_artifact_digest'), 'run_artifact', ['digest'], unique=False
|
||||
)
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_index(op.f('ix_run_artifact_digest'), table_name='run_artifact')
|
||||
op.drop_table('run_artifact')
|
||||
op.drop_table('run_metric')
|
||||
op.drop_index(op.f('ix_run_node_cache_key'), table_name='run_node')
|
||||
op.drop_table('run_node')
|
||||
op.drop_index(op.f('ix_run_params_digest'), table_name='run')
|
||||
op.drop_index(op.f('ix_run_created_at'), table_name='run')
|
||||
op.drop_index(op.f('ix_run_group_id'), table_name='run')
|
||||
op.drop_index(op.f('ix_run_status'), table_name='run')
|
||||
op.drop_index(op.f('ix_run_flow'), table_name='run')
|
||||
op.drop_table('run')
|
||||
@@ -10,6 +10,7 @@ from app.api.routes import (
|
||||
oauth,
|
||||
observability,
|
||||
private,
|
||||
runs,
|
||||
secrets,
|
||||
users,
|
||||
utils,
|
||||
@@ -27,6 +28,7 @@ api_router.include_router(dashboards.router)
|
||||
api_router.include_router(messages.router)
|
||||
api_router.include_router(modules.router)
|
||||
api_router.include_router(observability.router)
|
||||
api_router.include_router(runs.router)
|
||||
# Always mounted so the generated SDK stays the same shape; the endpoints
|
||||
# themselves refuse to work unless MCP is switched on.
|
||||
api_router.include_router(oauth.router)
|
||||
|
||||
@@ -0,0 +1,282 @@
|
||||
"""Runs over the API: submit one, watch it, read what it made.
|
||||
|
||||
Submitting returns immediately with a queued run — a training run is measured
|
||||
in hours, so nothing here waits for one. The way to follow a run is to poll it
|
||||
or to listen on the flow socket, which carries its start and finish.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlmodel import col, select
|
||||
|
||||
from app.api.deps import CurrentUser, SessionDep, get_current_user
|
||||
from app.flow.runs import RunRejected, RunService, new_run_id
|
||||
from app.flow.store import FlowNotFound
|
||||
from app.models import Run, RunArtifact, RunMetric, RunNode
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/runs", tags=["runs"], dependencies=[Depends(get_current_user)]
|
||||
)
|
||||
|
||||
#: A sweep bigger than this is almost always a mistake in a loop.
|
||||
MAX_SWEEP = 1000
|
||||
|
||||
|
||||
class RunCreate(BaseModel):
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
seed: int | None = None
|
||||
#: Run the unpublished draft instead of what is published.
|
||||
draft: bool = False
|
||||
|
||||
|
||||
class SweepEntry(BaseModel):
|
||||
params: dict[str, Any] = Field(default_factory=dict)
|
||||
seed: int | None = None
|
||||
|
||||
|
||||
class SweepCreate(BaseModel):
|
||||
runs: list[SweepEntry] = Field(default_factory=list)
|
||||
draft: bool = False
|
||||
|
||||
|
||||
class RunNodeRow(BaseModel):
|
||||
node: str
|
||||
status: str
|
||||
attempt: int
|
||||
duration_ms: float
|
||||
worker: str
|
||||
error: str
|
||||
logs: str
|
||||
|
||||
|
||||
class ArtifactRow(BaseModel):
|
||||
name: str
|
||||
node: str
|
||||
digest: str
|
||||
size: int
|
||||
media_type: str
|
||||
|
||||
|
||||
class RunRow(BaseModel):
|
||||
"""A run without its result, which is the part that can be large."""
|
||||
|
||||
id: str
|
||||
flow: str
|
||||
status: str
|
||||
status_reason: str
|
||||
cause: str
|
||||
params: dict[str, Any]
|
||||
params_digest: str
|
||||
seed: int | None
|
||||
group_id: str | None
|
||||
labels: list[str]
|
||||
created_at: Any
|
||||
started_at: Any = None
|
||||
finished_at: Any = None
|
||||
duration_ms: float
|
||||
actor: str
|
||||
|
||||
|
||||
class RunDetail(RunRow):
|
||||
result: dict[str, Any] = Field(default_factory=dict)
|
||||
commit: str = ""
|
||||
flow_version: int = 1
|
||||
nodes: list[RunNodeRow] = Field(default_factory=list)
|
||||
artifacts: list[ArtifactRow] = Field(default_factory=list)
|
||||
|
||||
|
||||
class MetricPoint(BaseModel):
|
||||
step: int
|
||||
ts: float
|
||||
value: float
|
||||
|
||||
|
||||
class MetricSeries(BaseModel):
|
||||
"""The shape a chart widget already draws, so comparing runs is a binding."""
|
||||
|
||||
label: str
|
||||
points: list[list[float]] = Field(default_factory=list)
|
||||
|
||||
|
||||
class SeriesAnswer(BaseModel):
|
||||
metric: str
|
||||
lines: list[MetricSeries] = Field(default_factory=list)
|
||||
|
||||
|
||||
def _service(request: Request) -> RunService:
|
||||
service: RunService | None = getattr(request.app.state, "run_service", None)
|
||||
if service is None:
|
||||
raise HTTPException(status_code=503, detail="Runs are not available")
|
||||
return service
|
||||
|
||||
|
||||
@router.post("/flows/{name}", response_model=RunRow, status_code=202)
|
||||
async def create_run(
|
||||
name: str, body: RunCreate, request: Request, user: CurrentUser
|
||||
) -> Any:
|
||||
"""Queue one run of a flow."""
|
||||
service = _service(request)
|
||||
try:
|
||||
return await run_in_threadpool(
|
||||
service.submit,
|
||||
name,
|
||||
params=body.params,
|
||||
seed=body.seed,
|
||||
cause="api",
|
||||
actor=user.email,
|
||||
draft=body.draft,
|
||||
)
|
||||
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.post("/flows/{name}/sweep", response_model=list[RunRow], status_code=202)
|
||||
async def create_sweep(
|
||||
name: str, body: SweepCreate, request: Request, user: CurrentUser
|
||||
) -> Any:
|
||||
"""Queue many runs of one flow under a shared group.
|
||||
|
||||
An ensemble is this with the same parameters and different seeds; a grid
|
||||
search is this with the parameters spread out. Either way the caller
|
||||
builds the list — the engine does not own a sweep grammar.
|
||||
"""
|
||||
if not body.runs:
|
||||
raise HTTPException(status_code=422, detail="A sweep needs at least one run")
|
||||
if len(body.runs) > MAX_SWEEP:
|
||||
raise HTTPException(
|
||||
status_code=422, detail=f"A sweep is capped at {MAX_SWEEP} runs"
|
||||
)
|
||||
service = _service(request)
|
||||
group = new_run_id()
|
||||
|
||||
def submit_all() -> list[Run]:
|
||||
return [
|
||||
service.submit(
|
||||
name,
|
||||
params=entry.params,
|
||||
seed=entry.seed,
|
||||
group_id=group,
|
||||
cause="sweep",
|
||||
actor=user.email,
|
||||
draft=body.draft,
|
||||
)
|
||||
for entry in body.runs
|
||||
]
|
||||
|
||||
try:
|
||||
return await run_in_threadpool(submit_all)
|
||||
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,
|
||||
flow: str | None = None,
|
||||
status: str | None = None,
|
||||
group: str | None = None,
|
||||
digest: str | None = None,
|
||||
limit: int = 50,
|
||||
) -> Any:
|
||||
"""Runs, newest first. The queryable table an experiment log needs."""
|
||||
statement = select(Run).order_by(col(Run.created_at).desc())
|
||||
if flow:
|
||||
statement = statement.where(col(Run.flow) == flow)
|
||||
if status:
|
||||
statement = statement.where(col(Run.status) == status)
|
||||
if group:
|
||||
statement = statement.where(col(Run.group_id) == group)
|
||||
if digest:
|
||||
statement = statement.where(col(Run.params_digest) == digest)
|
||||
return list(session.exec(statement.limit(min(limit, 500))))
|
||||
|
||||
|
||||
@router.get("/{run_id}", response_model=RunDetail)
|
||||
def read_run(run_id: str, session: SessionDep) -> Any:
|
||||
"""One run in full: what it was asked, what each node did, what it made."""
|
||||
run = session.get(Run, run_id)
|
||||
if run is None:
|
||||
raise HTTPException(status_code=404, detail="No such run")
|
||||
nodes = session.exec(select(RunNode).where(col(RunNode.run_id) == run_id)).all()
|
||||
artifacts = session.exec(
|
||||
select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
|
||||
).all()
|
||||
detail = RunDetail.model_validate(run, from_attributes=True)
|
||||
detail.nodes = [RunNodeRow.model_validate(n, from_attributes=True) for n in nodes]
|
||||
detail.artifacts = [
|
||||
ArtifactRow.model_validate(a, from_attributes=True) for a in artifacts
|
||||
]
|
||||
return detail
|
||||
|
||||
|
||||
@router.post("/{run_id}/cancel", response_model=RunRow)
|
||||
async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
|
||||
"""Stop a run. One already past its last node is left as it finished."""
|
||||
run = session.get(Run, run_id)
|
||||
if run is None:
|
||||
raise HTTPException(status_code=404, detail="No such run")
|
||||
service = _service(request)
|
||||
await run_in_threadpool(service.cancel, run_id)
|
||||
session.refresh(run)
|
||||
return run
|
||||
|
||||
|
||||
@router.get("/{run_id}/metrics", response_model=list[MetricPoint])
|
||||
def read_metrics(run_id: str, session: SessionDep, name: str, stride: int = 1) -> Any:
|
||||
"""One metric's series, in step order.
|
||||
|
||||
``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
|
||||
is 3000 points nobody can see.
|
||||
"""
|
||||
statement = (
|
||||
select(RunMetric)
|
||||
.where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == name)
|
||||
.order_by(col(RunMetric.step))
|
||||
)
|
||||
rows = list(session.exec(statement))
|
||||
if stride > 1:
|
||||
rows = rows[:: max(1, stride)]
|
||||
return rows
|
||||
|
||||
|
||||
@router.get("/series/compare", response_model=SeriesAnswer)
|
||||
def compare_metric(session: SessionDep, ids: str, metric: str) -> Any:
|
||||
"""One metric across several runs, as the chart widget's series shape.
|
||||
|
||||
This is the comparison view: it answers in the same shape a flow answers a
|
||||
chart's query with, so putting three training curves beside each other is
|
||||
a widget binding rather than a screen of its own.
|
||||
"""
|
||||
run_ids = [part for part in ids.split(",") if part]
|
||||
if not run_ids:
|
||||
raise HTTPException(status_code=422, detail="Name at least one run")
|
||||
runs = {
|
||||
run.id: run
|
||||
for run in session.exec(select(Run).where(col(Run.id).in_(run_ids))).all()
|
||||
}
|
||||
lines: list[MetricSeries] = []
|
||||
for run_id in run_ids:
|
||||
run = runs.get(run_id)
|
||||
if run is None:
|
||||
continue
|
||||
rows = session.exec(
|
||||
select(RunMetric)
|
||||
.where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == metric)
|
||||
.order_by(col(RunMetric.step))
|
||||
).all()
|
||||
label = run_id
|
||||
if run.seed is not None:
|
||||
label = f"{run_id} (seed {run.seed})"
|
||||
lines.append(
|
||||
MetricSeries(
|
||||
label=label, points=[[float(row.step), row.value] for row in rows]
|
||||
)
|
||||
)
|
||||
return SeriesAnswer(metric=metric, lines=lines)
|
||||
@@ -11,6 +11,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import logging
|
||||
import traceback
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any, cast
|
||||
@@ -40,7 +41,7 @@ from app.flow.nodes import (
|
||||
SwitchNode,
|
||||
TriggerNode,
|
||||
)
|
||||
from app.flow.pipeline import Pipeline, ValidationIssue, ValueSource
|
||||
from app.flow.pipeline import NodeOutcome, Pipeline, ValidationIssue, ValueSource
|
||||
from app.flow.schemas import (
|
||||
BrainEdge,
|
||||
BrainGraph,
|
||||
@@ -262,6 +263,8 @@ class FlowController:
|
||||
self.loaded: dict[str, LoadedNode] = {}
|
||||
self.issues: list[ValidationIssue] = []
|
||||
self.disabled: set[str] = set()
|
||||
#: Flows that only run when a run asks them to.
|
||||
self.batch: set[str] = set()
|
||||
self.supervisor = Supervisor(events)
|
||||
self.history_limits: dict[str, int] = {}
|
||||
self._lock = asyncio.Lock()
|
||||
@@ -306,6 +309,10 @@ class FlowController:
|
||||
for flow in published
|
||||
if not self.store.read_enabled(flow.name)
|
||||
}
|
||||
# A batch flow is built — the canvas draws it and validation covers
|
||||
# it — but never activated: it runs when a run asks it to, and a
|
||||
# subscription of its own would be a second way in.
|
||||
self.batch = {flow.name for flow in published if flow.mode == "batch"}
|
||||
# Off the loop: building a python node asks its worker to compile,
|
||||
# which waits for a free slot — and a busy node holds one for as
|
||||
# long as its timeout. On the loop that stalls every request, the
|
||||
@@ -373,8 +380,9 @@ class FlowController:
|
||||
if node is None:
|
||||
continue
|
||||
# A stopped flow gets no subscriptions, schedules or webhooks —
|
||||
# that is what stopping it means.
|
||||
if entry.flow in self.disabled:
|
||||
# that is what stopping it means. Nor does a batch flow, which has
|
||||
# no outside to listen to.
|
||||
if entry.flow in self.disabled or entry.flow in self.batch:
|
||||
continue
|
||||
node.supervisor = self.supervisor
|
||||
try:
|
||||
@@ -834,6 +842,33 @@ class FlowController:
|
||||
return
|
||||
self.pipeline.run(inputs or {}, nodes=self.pipeline.flow_nodes(flow))
|
||||
|
||||
def build_run_pipeline(
|
||||
self,
|
||||
flow: FlowDef,
|
||||
state: StateBackend,
|
||||
draft: bool = False,
|
||||
observer: Callable[[NodeOutcome], None] | None = None,
|
||||
) -> Pipeline:
|
||||
"""Build one flow as a pipeline of its own, for a single run.
|
||||
|
||||
Nothing here is activated: a run executes the graph from its inputs
|
||||
rather than waiting to be told something, so subscriptions, schedules
|
||||
and webhooks would only be a second copy of what the live pipeline
|
||||
already holds. The state is the run's, which is what keeps two runs of
|
||||
one flow from overwriting each other's messages.
|
||||
"""
|
||||
nodes, _loaded, initial_values, _inputs = self._build_flows([(flow, draft)])
|
||||
pipeline = Pipeline(
|
||||
nodes=nodes,
|
||||
state=state,
|
||||
events=self.events,
|
||||
max_workers=self.max_workers,
|
||||
initial_values=initial_values,
|
||||
observer=observer,
|
||||
)
|
||||
pipeline.history_limits = self.history_limits
|
||||
return pipeline
|
||||
|
||||
def run_preview(self, flow_name: str, inputs: dict[str, Any] | None = None) -> None:
|
||||
"""Run a flow's unpublished draft once, as the editor shows it.
|
||||
|
||||
|
||||
@@ -13,7 +13,7 @@ import threading
|
||||
import time
|
||||
import uuid
|
||||
from collections import deque
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Callable, Iterator
|
||||
from concurrent.futures import Future, ThreadPoolExecutor, wait
|
||||
from contextlib import contextmanager
|
||||
from typing import Any, Literal
|
||||
@@ -76,6 +76,22 @@ def node_source(node: Node) -> ValueSource:
|
||||
return ValueSource(kind="node", id=node.id, label=node.local_id)
|
||||
|
||||
|
||||
class NodeOutcome(BaseModel):
|
||||
"""How one node execution went.
|
||||
|
||||
Handed to whoever is watching a particular pipeline rather than published:
|
||||
a run has to record every node it ran, and the event bus drops what it
|
||||
cannot keep up with.
|
||||
"""
|
||||
|
||||
node: str
|
||||
ok: bool
|
||||
duration_ms: float = 0.0
|
||||
outputs: int = 0
|
||||
error: str = ""
|
||||
logs: str = ""
|
||||
|
||||
|
||||
class Pipeline:
|
||||
"""Directed graph of nodes with automatic dependency resolution."""
|
||||
|
||||
@@ -96,6 +112,7 @@ class Pipeline:
|
||||
"_queue",
|
||||
"_node_pool",
|
||||
"history_limits",
|
||||
"observer",
|
||||
)
|
||||
|
||||
def __init__(
|
||||
@@ -108,6 +125,7 @@ class Pipeline:
|
||||
disabled_flows: set[str] | None = None,
|
||||
work_queue: WorkQueue | None = None,
|
||||
node_pool: ThreadPoolExecutor | None = None,
|
||||
observer: Callable[[NodeOutcome], None] | None = None,
|
||||
) -> None:
|
||||
self._nodes = nodes or []
|
||||
# Stopped flows are stored and survive a restart; paused ones are a
|
||||
@@ -127,6 +145,8 @@ class Pipeline:
|
||||
self._queue = work_queue
|
||||
# A pool owned by the execution service, so a wave does not build one.
|
||||
self._node_pool = node_pool
|
||||
# Set by a run, which needs every node it executed written down.
|
||||
self.observer = observer
|
||||
# How deep to keep each message's series; a chart asking for more
|
||||
# than the default puts its message in here. Swapped, never mutated.
|
||||
self.history_limits: dict[str, int] = {}
|
||||
@@ -665,6 +685,7 @@ class Pipeline:
|
||||
}
|
||||
)
|
||||
|
||||
duration_ms = round((time.perf_counter() - started) * 1000, 2)
|
||||
self._publish(
|
||||
{
|
||||
"type": "node_executed",
|
||||
@@ -673,17 +694,44 @@ class Pipeline:
|
||||
# A node that returns nothing ran but published nothing,
|
||||
# which is a different thing to show than one that emitted.
|
||||
"outputs": len(result or {}),
|
||||
"duration_ms": round((time.perf_counter() - started) * 1000, 2),
|
||||
"duration_ms": duration_ms,
|
||||
"run": entry_id,
|
||||
"ts": time.time(),
|
||||
}
|
||||
)
|
||||
self._observe(
|
||||
NodeOutcome(
|
||||
node=node.id,
|
||||
ok=True,
|
||||
duration_ms=duration_ms,
|
||||
outputs=len(result or {}),
|
||||
logs=collected.text,
|
||||
)
|
||||
)
|
||||
return result
|
||||
except Exception as exc:
|
||||
# One failing node must not take the rest of the graph down.
|
||||
self.publish_error(node, exc, collected, entry_id)
|
||||
error = self.publish_error(node, exc, collected, entry_id)
|
||||
self._observe(
|
||||
NodeOutcome(
|
||||
node=node.id,
|
||||
ok=False,
|
||||
duration_ms=round((time.perf_counter() - started) * 1000, 2),
|
||||
error=error,
|
||||
logs=collected.text,
|
||||
)
|
||||
)
|
||||
return None
|
||||
|
||||
def _observe(self, outcome: NodeOutcome) -> None:
|
||||
"""Tell the run watching this pipeline, if there is one."""
|
||||
if self.observer is None:
|
||||
return
|
||||
try:
|
||||
self.observer(outcome)
|
||||
except Exception:
|
||||
logger.exception("Run observer failed for '%s'", outcome.node)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Running, stopped, paused
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
@@ -40,7 +40,8 @@ class WorkItem:
|
||||
"""One unit of journaled work.
|
||||
|
||||
:param kind: ``cascade`` replays a node's outputs and runs what is
|
||||
downstream; ``flush`` lets out what a node's rate limits held back.
|
||||
downstream; ``flush`` lets out what a node's rate limits held back;
|
||||
``run`` is a whole batch run, and carries only its id.
|
||||
:param node: The node the item is about — the source for a cascade, the
|
||||
target for a node item.
|
||||
:param flow: The flow that node belongs to, so gating needs no lookup.
|
||||
@@ -68,6 +69,9 @@ class WorkItem:
|
||||
entry_id: str = ""
|
||||
deliveries: int = 1
|
||||
enqueued_at: float = field(default_factory=time.time)
|
||||
#: The run this item is, on a ``run`` item. Everything else about a run is
|
||||
#: in its database row, so the journal only has to name it.
|
||||
run_id: str = ""
|
||||
|
||||
def to_fields(self) -> dict[str, str]:
|
||||
return {
|
||||
@@ -80,6 +84,7 @@ class WorkItem:
|
||||
"guard_key": self.guard_key,
|
||||
"guard_value": self.guard_value,
|
||||
"enqueued_at": str(self.enqueued_at),
|
||||
"run_id": self.run_id,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
@@ -98,6 +103,7 @@ class WorkItem:
|
||||
entry_id=entry_id,
|
||||
deliveries=deliveries,
|
||||
enqueued_at=float(fields.get("enqueued_at") or time.time()),
|
||||
run_id=fields.get("run_id", ""),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,516 @@
|
||||
"""Runs: a flow taken from its inputs to its outputs, once.
|
||||
|
||||
A cascade is what an always-on flow does when a value arrives — it has no
|
||||
beginning and no end worth recording. A *run* is the other shape the same
|
||||
engine can take: parameters go in, the graph executes until it drains, and
|
||||
what it produced is kept. That is what an ML experiment is, and what a
|
||||
CI-style job is, so both are this one entity.
|
||||
|
||||
Three things make a run different from a cascade, and each is deliberate:
|
||||
|
||||
* **Its own state.** Message names are global keys, so two runs of one flow
|
||||
sharing the engine's state would overwrite each other's values. A run gets a
|
||||
state backend namespaced to itself, which is a constructor argument rather
|
||||
than a change to the pipeline — every key the engine keeps, versions and
|
||||
node memory included, already goes through that backend.
|
||||
* **Its own record.** The event bus drops what it cannot keep up with, which
|
||||
is right for a live canvas and wrong for a result. The driver writes the
|
||||
run's rows itself, from the thread that is running it.
|
||||
* **Its own durability.** The queue wakes an engine up; from the moment a run
|
||||
starts, its database row is the truth. Redelivering hours of training
|
||||
because an acknowledgement was late is not recovery, so a run is
|
||||
acknowledged as soon as it is claimed and a stale lease — not an unacked
|
||||
stream entry — is what marks a run its engine died in the middle of.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from sqlalchemy import update
|
||||
from sqlmodel import Session, col, select
|
||||
|
||||
from app.core.db import engine as db_engine
|
||||
from app.flow.messages import qualify
|
||||
from app.flow.pipeline import NodeOutcome, Pipeline
|
||||
from app.flow.queue import WorkItem, WorkQueue
|
||||
from app.flow.schemas import FlowDef
|
||||
from app.flow.state import MemoryState, StateBackend
|
||||
from app.models import Run, RunNode
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from app.flow.controller import FlowController
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: How long a finished run's state is kept before Redis drops it. Long enough
|
||||
#: to look at what a failed run left behind, short enough not to accumulate.
|
||||
RUN_STATE_TTL = 24 * 3600
|
||||
#: How often a running run says it is still alive.
|
||||
LEASE_INTERVAL_S = 20.0
|
||||
#: A lease older than this belongs to an engine that is not coming back.
|
||||
LEASE_STALE_S = 90.0
|
||||
#: How often stale leases are looked for.
|
||||
SWEEP_INTERVAL_S = 30.0
|
||||
#: Runs driven at once. Node bodies are bounded by the worker pool anyway;
|
||||
#: this only bounds how many graphs are in flight.
|
||||
MAX_PARALLEL = 4
|
||||
CLAIM_COUNT = 4
|
||||
CLAIM_BLOCK_MS = 1000
|
||||
ERROR_CAP = 2000
|
||||
LOG_CAP = 8000
|
||||
|
||||
#: Where a run's state lives, so it can never collide with the engine's own.
|
||||
RUN_NAMESPACE = "run"
|
||||
|
||||
|
||||
class RunRejected(ValueError):
|
||||
"""The run cannot be made: bad parameters, or a flow that cannot batch."""
|
||||
|
||||
|
||||
def new_run_id() -> str:
|
||||
"""Time-ordered, so the newest runs sort last without reading a column."""
|
||||
return f"{int(time.time() * 1000):013d}-{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
def digest_of(params: dict[str, Any], seed: int | None) -> str:
|
||||
"""What identifies a run's inputs: same parameters, same digest."""
|
||||
canonical = json.dumps(
|
||||
{"params": params, "seed": seed}, sort_keys=True, separators=(",", ":")
|
||||
)
|
||||
return hashlib.sha256(canonical.encode()).hexdigest()
|
||||
|
||||
|
||||
def batch_issues(flow: FlowDef) -> list[str]:
|
||||
"""Why this flow cannot be run as a batch, if it cannot.
|
||||
|
||||
Only one thing genuinely breaks: a port with a discretization interval
|
||||
holds values back for a timer to release, and a run has no timer — the
|
||||
engine would drop them instead. A delay node is fine; without a queue to
|
||||
defer into it simply sleeps, which in a run is what was asked for.
|
||||
"""
|
||||
issues: list[str] = []
|
||||
for node in flow.nodes:
|
||||
for spec in list(node.requires) + list(node.provides):
|
||||
if spec.interval > 0:
|
||||
issues.append(
|
||||
f"Node '{node.id}' rate-limits '{spec.port or spec.name}'. "
|
||||
"A run has no timer to release what that holds back, so "
|
||||
"the value would be dropped. Remove the interval to run "
|
||||
"this flow as a batch."
|
||||
)
|
||||
return issues
|
||||
|
||||
|
||||
def required_labels(flow: FlowDef) -> list[str]:
|
||||
"""Worker labels this flow's nodes ask for."""
|
||||
return sorted(
|
||||
{node.device for node in flow.nodes if node.device and node.device.strip()}
|
||||
)
|
||||
|
||||
|
||||
def seed_values(flow: FlowDef, params: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Turn a run's parameters into the messages the flow starts from."""
|
||||
specs = {declared.spec.name: declared.spec for declared in flow.inputs}
|
||||
values: dict[str, Any] = {}
|
||||
for key, value in params.items():
|
||||
spec = specs.get(key)
|
||||
if spec is None:
|
||||
known = ", ".join(sorted(specs)) or "none"
|
||||
raise RunRejected(
|
||||
f"'{key}' is not an input of flow '{flow.name}' (it declares: {known})"
|
||||
)
|
||||
try:
|
||||
spec.check(value)
|
||||
except TypeError as exc:
|
||||
raise RunRejected(f"Parameter '{key}': {exc}") from exc
|
||||
values[qualify(flow.name, spec.name)] = value
|
||||
return values
|
||||
|
||||
|
||||
def collect_result(flow: FlowDef, state: StateBackend) -> dict[str, Any]:
|
||||
"""What the run produced, keyed by message name without the flow prefix."""
|
||||
prefix = f"{flow.name}."
|
||||
if flow.outputs:
|
||||
names = [qualify(flow.name, name) for name in flow.outputs]
|
||||
else:
|
||||
# Everything the flow ended up holding. The engine's own bookkeeping is
|
||||
# keyed by `__thing__:message`, so it never starts with the flow name.
|
||||
names = sorted(key for key in state.keys() if key.startswith(prefix))
|
||||
result: dict[str, Any] = {}
|
||||
for name in names:
|
||||
if name in state:
|
||||
result[name[len(prefix) :] if name.startswith(prefix) else name] = state[
|
||||
name
|
||||
]
|
||||
return result
|
||||
|
||||
|
||||
class RunService:
|
||||
"""Accepts runs, drives them, and writes down what they did."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
controller: FlowController,
|
||||
queue: WorkQueue,
|
||||
state_factory: Callable[[str], StateBackend] | None = None,
|
||||
parallel: int = MAX_PARALLEL,
|
||||
) -> None:
|
||||
self.controller = controller
|
||||
self.queue = queue
|
||||
# Without one, a run gets a private in-memory state — which is exactly
|
||||
# the isolation it wants, minus surviving the process.
|
||||
self._state_factory = state_factory or (lambda _ns: MemoryState())
|
||||
self.engine_name = f"{socket.gethostname()}-{os.getpid()}"[:64]
|
||||
self._pool = ThreadPoolExecutor(max_workers=parallel, thread_name_prefix="run")
|
||||
self._stop = threading.Event()
|
||||
self._consumer: threading.Thread | None = None
|
||||
self._keeper: threading.Thread | None = None
|
||||
# Runs this process is driving, and the pipeline each is running, so a
|
||||
# cancel has something to hold on to.
|
||||
self._active: dict[str, Pipeline] = {}
|
||||
self._cancelled: set[str] = set()
|
||||
self._lock = threading.Lock()
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Lifecycle
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def start(self) -> None:
|
||||
if self._consumer is not None:
|
||||
return
|
||||
self._consumer = threading.Thread(
|
||||
target=self._consume, name="run-consumer", daemon=True
|
||||
)
|
||||
self._consumer.start()
|
||||
self._keeper = threading.Thread(
|
||||
target=self._keep_leases, name="run-leases", daemon=True
|
||||
)
|
||||
self._keeper.start()
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop.set()
|
||||
for thread in (self._consumer, self._keeper):
|
||||
if thread is not None:
|
||||
thread.join(timeout=5)
|
||||
self._consumer = None
|
||||
self._keeper = None
|
||||
self._pool.shutdown(wait=False)
|
||||
self.queue.close()
|
||||
|
||||
def alive(self) -> bool:
|
||||
return self._consumer is not None and self._consumer.is_alive()
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Accepting work
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def submit(
|
||||
self,
|
||||
flow_name: str,
|
||||
params: dict[str, Any] | None = None,
|
||||
seed: int | None = None,
|
||||
group_id: str | None = None,
|
||||
cause: str = "api",
|
||||
actor: str = "",
|
||||
draft: bool = False,
|
||||
) -> Run:
|
||||
"""Journal a run and wake an engine up for it. Never blocks on it."""
|
||||
flow = self.controller.store.read_flow(flow_name, draft=draft)
|
||||
issues = batch_issues(flow)
|
||||
if issues:
|
||||
raise RunRejected(" ".join(issues))
|
||||
params = params or {}
|
||||
# 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)
|
||||
|
||||
run = Run(
|
||||
id=new_run_id(),
|
||||
flow=flow.name,
|
||||
flow_version=flow.version,
|
||||
commit=self.controller.store.head(),
|
||||
params=params,
|
||||
params_digest=digest_of(params, seed),
|
||||
seed=seed,
|
||||
group_id=group_id,
|
||||
cause=cause,
|
||||
status="queued",
|
||||
labels=required_labels(flow),
|
||||
created_at=datetime.now(timezone.utc),
|
||||
actor=actor,
|
||||
)
|
||||
with Session(db_engine) as session:
|
||||
session.add(run)
|
||||
session.commit()
|
||||
session.refresh(run)
|
||||
|
||||
self.queue.add(WorkItem(kind="run", node="", flow=flow.name, run_id=run.id))
|
||||
return run
|
||||
|
||||
def cancel(self, run_id: str) -> bool:
|
||||
"""Stop a run: nothing further is scheduled once its nodes return.
|
||||
|
||||
A node already executing is left to finish. Killing one needs the
|
||||
worker pool to know which run it belongs to, which is what the metric
|
||||
stream adds next; until then a cancel that arrives mid-node is a stop
|
||||
rather than an interruption.
|
||||
"""
|
||||
with self._lock:
|
||||
pipeline = self._active.get(run_id)
|
||||
if pipeline is None:
|
||||
# Not running here — if it is still queued, refusing to start
|
||||
# is all the cancelling it needs.
|
||||
cancelled = self._finish_queued(run_id)
|
||||
if cancelled:
|
||||
self._cancelled.add(run_id)
|
||||
return cancelled
|
||||
self._cancelled.add(run_id)
|
||||
# Reusing the pause gate: a gated node is never submitted, so the graph
|
||||
# drains instead of going further.
|
||||
pipeline.pause(self._flow_of(run_id) or "")
|
||||
return True
|
||||
|
||||
def _flow_of(self, run_id: str) -> str | None:
|
||||
with Session(db_engine) as session:
|
||||
run = session.get(Run, run_id)
|
||||
return run.flow if run else None
|
||||
|
||||
def _finish_queued(self, run_id: str) -> bool:
|
||||
with Session(db_engine) as session:
|
||||
result = session.exec(
|
||||
update(Run)
|
||||
.where(col(Run.id) == run_id, col(Run.status) == "queued")
|
||||
.values(
|
||||
status="cancelled",
|
||||
finished_at=datetime.now(timezone.utc),
|
||||
status_reason="Cancelled before it started",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
return bool(result.rowcount)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Threads
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _consume(self) -> None:
|
||||
failures = 0
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
items = self.queue.claim(CLAIM_COUNT, CLAIM_BLOCK_MS)
|
||||
failures = 0
|
||||
except Exception as exc:
|
||||
failures += 1
|
||||
logger.error("Could not claim runs: %s", exc)
|
||||
self._stop.wait(min(30.0, 2.0**failures))
|
||||
continue
|
||||
for item in items:
|
||||
# Acknowledged before it runs: from here on the row is the
|
||||
# record, and a lease that stops moving is what says otherwise.
|
||||
self.queue.ack(item)
|
||||
if not item.run_id:
|
||||
continue
|
||||
try:
|
||||
self._pool.submit(self._drive, item.run_id)
|
||||
except RuntimeError:
|
||||
logger.warning("Run %s not started: shutting down", item.run_id)
|
||||
|
||||
def _keep_leases(self) -> None:
|
||||
"""Say the local runs are alive, and clean up after engines that died."""
|
||||
last_sweep = 0.0
|
||||
while not self._stop.is_set():
|
||||
self._stop.wait(LEASE_INTERVAL_S)
|
||||
if self._stop.is_set():
|
||||
break
|
||||
with self._lock:
|
||||
mine = list(self._active)
|
||||
now = datetime.now(timezone.utc)
|
||||
try:
|
||||
if mine:
|
||||
with Session(db_engine) as session:
|
||||
session.exec(
|
||||
update(Run)
|
||||
.where(col(Run.id).in_(mine))
|
||||
.values(lease_at=now)
|
||||
)
|
||||
session.commit()
|
||||
if time.monotonic() - last_sweep >= SWEEP_INTERVAL_S:
|
||||
last_sweep = time.monotonic()
|
||||
self._sweep_abandoned(now)
|
||||
except Exception:
|
||||
logger.exception("Could not refresh run leases")
|
||||
|
||||
def _sweep_abandoned(self, now: datetime) -> None:
|
||||
cutoff = now - timedelta(seconds=LEASE_STALE_S)
|
||||
with Session(db_engine) as session:
|
||||
result = session.exec(
|
||||
update(Run)
|
||||
.where(
|
||||
col(Run.status) == "running",
|
||||
col(Run.lease_at) < cutoff,
|
||||
)
|
||||
.values(
|
||||
status="abandoned",
|
||||
finished_at=now,
|
||||
status_reason="The engine running it stopped reporting",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
if result.rowcount:
|
||||
logger.warning("Marked %d run(s) abandoned", result.rowcount)
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Driving one run
|
||||
# -------------------------------------------------------------------------
|
||||
|
||||
def _claim(self, run_id: str) -> Run | None:
|
||||
"""Take the run, or leave it: whoever moves it out of `queued` owns it.
|
||||
|
||||
The compare-and-swap is what makes a redelivered item harmless — the
|
||||
second engine to arrive updates nothing and walks away.
|
||||
"""
|
||||
now = datetime.now(timezone.utc)
|
||||
with Session(db_engine) as session:
|
||||
result = session.exec(
|
||||
update(Run)
|
||||
.where(col(Run.id) == run_id, col(Run.status) == "queued")
|
||||
.values(
|
||||
status="running",
|
||||
started_at=now,
|
||||
lease_at=now,
|
||||
engine=self.engine_name,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
if not result.rowcount:
|
||||
return None
|
||||
return session.exec(select(Run).where(col(Run.id) == run_id)).first()
|
||||
|
||||
def _drive(self, run_id: str) -> None:
|
||||
run = self._claim(run_id)
|
||||
if run is None:
|
||||
return
|
||||
started = time.perf_counter()
|
||||
status = "ok"
|
||||
reason = ""
|
||||
result: dict[str, Any] = {}
|
||||
state: StateBackend | None = None
|
||||
errors = 0
|
||||
|
||||
def observe(outcome: NodeOutcome) -> None:
|
||||
nonlocal errors
|
||||
if not outcome.ok:
|
||||
errors += 1
|
||||
self._record_node(run_id, outcome)
|
||||
|
||||
try:
|
||||
flow = self.controller.store.read_flow(run.flow)
|
||||
state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}")
|
||||
pipeline = self.controller.build_run_pipeline(
|
||||
flow, state=state, observer=observe
|
||||
)
|
||||
with self._lock:
|
||||
self._active[run_id] = pipeline
|
||||
if run_id in self._cancelled:
|
||||
pipeline.pause(flow.name)
|
||||
self._publish(run, "run_started")
|
||||
pipeline.run(seed_values(flow, run.params))
|
||||
result = collect_result(flow, state)
|
||||
with self._lock:
|
||||
cancelled = run_id in self._cancelled
|
||||
if cancelled:
|
||||
status = "cancelled"
|
||||
reason = "Cancelled while running"
|
||||
elif errors:
|
||||
status = "error"
|
||||
reason = f"{errors} node(s) failed"
|
||||
except Exception as exc:
|
||||
logger.exception("Run %s failed", run_id)
|
||||
status = "error"
|
||||
reason = f"{type(exc).__name__}: {exc}"[:1024]
|
||||
finally:
|
||||
with self._lock:
|
||||
self._active.pop(run_id, None)
|
||||
self._cancelled.discard(run_id)
|
||||
duration = round((time.perf_counter() - started) * 1000, 2)
|
||||
self._finish(run_id, status, reason, result, duration)
|
||||
run.status = status
|
||||
self._publish(run, "run_finished")
|
||||
# Its values were only ever this run's; nothing reads them once it
|
||||
# has a result. On Redis the namespace would expire anyway.
|
||||
if state is not None and status != "error":
|
||||
try:
|
||||
state.clear()
|
||||
except Exception:
|
||||
logger.warning("Could not clear state of run %s", run_id)
|
||||
|
||||
def _record_node(self, run_id: str, outcome: NodeOutcome) -> None:
|
||||
row = RunNode(
|
||||
run_id=run_id,
|
||||
node=outcome.node[:255],
|
||||
status="ok" if outcome.ok else "error",
|
||||
started_at=datetime.now(timezone.utc),
|
||||
duration_ms=outcome.duration_ms,
|
||||
error=outcome.error[:ERROR_CAP],
|
||||
logs=outcome.logs[:LOG_CAP],
|
||||
)
|
||||
try:
|
||||
with Session(db_engine) as session:
|
||||
session.merge(row)
|
||||
session.commit()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Could not record node '%s' of run %s", outcome.node, run_id
|
||||
)
|
||||
|
||||
def _finish(
|
||||
self,
|
||||
run_id: str,
|
||||
status: str,
|
||||
reason: str,
|
||||
result: dict[str, Any],
|
||||
duration_ms: float,
|
||||
) -> None:
|
||||
try:
|
||||
with Session(db_engine) as session:
|
||||
session.exec(
|
||||
update(Run)
|
||||
.where(col(Run.id) == run_id)
|
||||
.values(
|
||||
status=status,
|
||||
status_reason=reason,
|
||||
result=result,
|
||||
duration_ms=duration_ms,
|
||||
finished_at=datetime.now(timezone.utc),
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
except Exception:
|
||||
logger.exception("Could not close run %s", run_id)
|
||||
|
||||
def _publish(self, run: Run, kind: str) -> None:
|
||||
events = self.controller.events
|
||||
if events is None:
|
||||
return
|
||||
events.publish(
|
||||
{
|
||||
"type": kind,
|
||||
"flow": run.flow,
|
||||
"run": run.id,
|
||||
"status": run.status,
|
||||
"group": run.group_id or "",
|
||||
"ts": time.time(),
|
||||
}
|
||||
)
|
||||
@@ -48,7 +48,23 @@ class NodeDef(BaseModel):
|
||||
"Seconds this node's code may run before it is stopped. This "
|
||||
"covers the first call's imports, which can be much slower than "
|
||||
"the body. Above 60 the engine may deliver its work again while "
|
||||
"it is still running."
|
||||
"it is still running — in a batch run, which never redelivers, "
|
||||
"it is an idle timeout instead: silence this long is a kill."
|
||||
),
|
||||
)
|
||||
device: str | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Label of the worker this node's code must run on, such as 'gpu'. "
|
||||
"Empty means the engine's own workers. A run needing a label no "
|
||||
"attached worker carries waits rather than failing."
|
||||
),
|
||||
)
|
||||
device_policy: Literal["require", "prefer"] = Field(
|
||||
default="require",
|
||||
description=(
|
||||
"What to do when no worker carries `device`: wait for one, or run "
|
||||
"locally anyway."
|
||||
),
|
||||
)
|
||||
|
||||
@@ -73,6 +89,22 @@ class FlowDef(BaseModel):
|
||||
nodes: list[NodeDef] = Field(default_factory=list)
|
||||
inputs: list[FlowInput] = Field(default_factory=list)
|
||||
version: int = 1
|
||||
mode: Literal["live", "batch"] = Field(
|
||||
default="live",
|
||||
description=(
|
||||
"A live flow reacts to what arrives: its subscriptions, schedules "
|
||||
"and webhooks run until it is stopped. A batch flow only runs when "
|
||||
"a run asks it to, from its inputs to its outputs, and is never "
|
||||
"activated."
|
||||
),
|
||||
)
|
||||
outputs: list[str] = Field(
|
||||
default_factory=list,
|
||||
description=(
|
||||
"Messages a batch run reports as its result, unqualified. Empty "
|
||||
"means every message the flow ends up holding."
|
||||
),
|
||||
)
|
||||
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
|
||||
@@ -160,6 +160,11 @@ class FlowStore:
|
||||
def _lib_file(self, name: str) -> Path:
|
||||
return self.root / LIB_DIR / f"{name}.py"
|
||||
|
||||
def head(self) -> str:
|
||||
"""The commit the store is at, so a result can name the code it ran."""
|
||||
result = self._git("rev-parse", "HEAD")
|
||||
return result.stdout.strip() if result.returncode == 0 else ""
|
||||
|
||||
# -------------------------------------------------------------------------
|
||||
# Module requirements
|
||||
#
|
||||
|
||||
+32
-2
@@ -25,6 +25,7 @@ from app.flow.nodes.http import close_shared_client
|
||||
from app.flow.pipeline import ValueSource
|
||||
from app.flow.plugins import load_plugins
|
||||
from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
|
||||
from app.flow.runs import RUN_STATE_TTL, RunService
|
||||
from app.flow.secrets import init_secrets
|
||||
from app.flow.state import MemoryState, RedisState, StateBackend
|
||||
from app.flow.store import FlowStore
|
||||
@@ -46,13 +47,31 @@ def _state_backend() -> StateBackend:
|
||||
return MemoryState()
|
||||
|
||||
|
||||
def _work_queue() -> WorkQueue:
|
||||
def _work_queue(namespace: str = "pipeline") -> WorkQueue:
|
||||
"""Redis makes queued work survive the process; memory does not pretend to."""
|
||||
if settings.REDIS_HOST:
|
||||
return RedisWorkQueue(host=settings.REDIS_HOST, port=settings.REDIS_PORT)
|
||||
return RedisWorkQueue(
|
||||
host=settings.REDIS_HOST, port=settings.REDIS_PORT, namespace=namespace
|
||||
)
|
||||
return MemoryWorkQueue()
|
||||
|
||||
|
||||
def _run_state(namespace: str) -> StateBackend:
|
||||
"""A state backend of a run's own, which is what isolates it.
|
||||
|
||||
It expires: a finished run's messages are read out into its result, and
|
||||
what is left is only worth keeping while someone might look at it.
|
||||
"""
|
||||
if settings.REDIS_HOST:
|
||||
return RedisState(
|
||||
host=settings.REDIS_HOST,
|
||||
port=settings.REDIS_PORT,
|
||||
namespace=namespace,
|
||||
ttl=RUN_STATE_TTL,
|
||||
)
|
||||
return MemoryState()
|
||||
|
||||
|
||||
def _mcp_sessions() -> AbstractAsyncContextManager[None]:
|
||||
"""The MCP session manager's run scope, or nothing when MCP is off."""
|
||||
if not settings.MCP_ENABLED:
|
||||
@@ -110,6 +129,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
# Charts need a deeper series than the default; tell the engine
|
||||
# before it starts recording.
|
||||
controller.set_history_limits(dashboards.history_requirements())
|
||||
# Runs read from a stream of their own: a burst of sweep runs must not
|
||||
# stand between the automations and their work, and a run that takes an
|
||||
# hour must not be judged by the cascade reaper's timings.
|
||||
run_service = RunService(
|
||||
controller=controller,
|
||||
queue=_work_queue("run"),
|
||||
state_factory=_run_state,
|
||||
)
|
||||
app.state.run_service = run_service
|
||||
watchdog = LoopWatchdog(event_bus)
|
||||
app.state.watchdog = watchdog
|
||||
watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog")
|
||||
@@ -118,6 +146,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
MetricsCollector(event_bus).run(), name="metrics-collector"
|
||||
)
|
||||
await controller.start()
|
||||
run_service.start()
|
||||
try:
|
||||
# A mounted sub-app gets no lifespan of its own, so the MCP session
|
||||
# manager is entered here; without it every /mcp request fails.
|
||||
@@ -127,6 +156,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
||||
watchdog_task.cancel()
|
||||
alerts_task.cancel()
|
||||
metrics_task.cancel()
|
||||
await run_in_threadpool(run_service.stop)
|
||||
await controller.stop()
|
||||
pool.stop()
|
||||
close_shared_client()
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from pydantic import EmailStr
|
||||
from sqlalchemy import JSON, Column, DateTime
|
||||
@@ -279,3 +280,131 @@ class FlowRun(SQLModel, table=True):
|
||||
errors: int = 0
|
||||
duration_ms: float = 0.0
|
||||
deliveries: int = 1
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Runs
|
||||
#
|
||||
# A cascade above is what an always-on flow does when a value arrives. A run is
|
||||
# the other shape: a batch flow taken from its inputs to its outputs once, with
|
||||
# parameters that identify it and a result worth keeping. An experiment is a
|
||||
# run, and so is a CI-style job — same entity, different caller.
|
||||
#
|
||||
# Kept apart from the observability tables on purpose: those are rolled up and
|
||||
# pruned on a retention window, and an experiment nobody wants deleted must not
|
||||
# share that fate.
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Run(SQLModel, table=True):
|
||||
"""One finite execution of a flow, with what it was asked and what it made."""
|
||||
|
||||
__tablename__ = "run"
|
||||
|
||||
id: str = Field(primary_key=True, max_length=64)
|
||||
flow: str = Field(index=True, max_length=255)
|
||||
#: The flow document this ran, and the commit it was read at, so a result
|
||||
#: can be traced back to the code that produced it.
|
||||
flow_version: int = 1
|
||||
commit: str = Field(default="", max_length=64)
|
||||
params: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
|
||||
#: 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)
|
||||
seed: int | None = None
|
||||
#: Runs submitted together — a sweep, an ensemble.
|
||||
group_id: str | None = Field(default=None, index=True, max_length=64)
|
||||
#: The run this one was made from, on a retry.
|
||||
parent_id: str | None = Field(default=None, max_length=64)
|
||||
#: api, hook, sweep or cli.
|
||||
cause: str = Field(default="api", max_length=32)
|
||||
#: queued, running, ok, error, cancelled or abandoned.
|
||||
status: str = Field(default="queued", index=True, max_length=16)
|
||||
#: Why it is where it is: what it waits for, or what went wrong.
|
||||
status_reason: str = Field(default="", max_length=1024)
|
||||
#: Worker labels its nodes need, so a run with nowhere to go can say so.
|
||||
labels: list[str] = Field(sa_column=Column(JSON), default_factory=list)
|
||||
created_at: datetime = Field(
|
||||
index=True,
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
started_at: datetime | None = Field(
|
||||
default=None,
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
finished_at: datetime | None = Field(
|
||||
default=None,
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
duration_ms: float = 0.0
|
||||
#: The flow's declared outputs once it finished.
|
||||
result: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
|
||||
#: Which engine holds it, and when it last said so. A run whose lease has
|
||||
#: gone stale is one whose engine died mid-run.
|
||||
engine: str = Field(default="", max_length=64)
|
||||
lease_at: datetime | None = Field(
|
||||
default=None,
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
actor: str = Field(default="", max_length=255)
|
||||
|
||||
|
||||
class RunNode(SQLModel, table=True):
|
||||
"""What one node did inside a run: the per-stage view of it."""
|
||||
|
||||
__tablename__ = "run_node"
|
||||
|
||||
run_id: str = Field(primary_key=True, max_length=64)
|
||||
node: str = Field(primary_key=True, max_length=255)
|
||||
#: ok, error, skipped, cached or cancelled.
|
||||
status: str = Field(default="ok", max_length=16)
|
||||
attempt: int = 1
|
||||
started_at: datetime | None = Field(
|
||||
default=None,
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
duration_ms: float = 0.0
|
||||
#: Which worker ran it: "local", or a remote worker's name.
|
||||
worker: str = Field(default="local", max_length=64)
|
||||
error: str = ""
|
||||
logs: str = ""
|
||||
#: Everything this node's output depends on, hashed. Recorded from the
|
||||
#: start so that skipping a stage whose inputs have not changed is later a
|
||||
#: lookup rather than a migration.
|
||||
cache_key: str = Field(default="", index=True, max_length=64)
|
||||
|
||||
|
||||
class RunMetric(SQLModel, table=True):
|
||||
"""One number a run reported, at one step.
|
||||
|
||||
Written by the run's own driver rather than folded off the event bus: the
|
||||
bus drops what it cannot keep up with, which is the right trade for a live
|
||||
canvas and the wrong one for a training curve.
|
||||
"""
|
||||
|
||||
__tablename__ = "run_metric"
|
||||
|
||||
run_id: str = Field(primary_key=True, max_length=64)
|
||||
name: str = Field(primary_key=True, max_length=128)
|
||||
#: -1 for a value with no step of its own — a final score.
|
||||
step: int = Field(primary_key=True)
|
||||
node: str = Field(default="", max_length=255)
|
||||
ts: float = 0.0
|
||||
value: float = 0.0
|
||||
|
||||
|
||||
class RunArtifact(SQLModel, table=True):
|
||||
"""A file a run produced, addressed by the hash of its content."""
|
||||
|
||||
__tablename__ = "run_artifact"
|
||||
|
||||
run_id: str = Field(primary_key=True, max_length=64)
|
||||
name: str = Field(primary_key=True, max_length=255)
|
||||
node: str = Field(default="", max_length=255)
|
||||
digest: str = Field(default="", index=True, max_length=71)
|
||||
size: int = 0
|
||||
media_type: str = Field(default="application/octet-stream", max_length=128)
|
||||
created_at: datetime = Field(
|
||||
default_factory=lambda: datetime.now(timezone.utc),
|
||||
sa_type=DateTime(timezone=True), # type: ignore
|
||||
)
|
||||
|
||||
@@ -0,0 +1,170 @@
|
||||
"""What a run is made of: isolation, the per-node record, and what it reports.
|
||||
|
||||
The service itself needs a database, so what is checked here is the part that
|
||||
decides whether a run is correct — that two runs of one flow cannot see each
|
||||
other's messages, that every node executed is reported once, and that the
|
||||
parameters a caller sends are refused before anything runs if they are wrong.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from app.flow.messages import DType, MessageSpec
|
||||
from app.flow.nodes import Node
|
||||
from app.flow.pipeline import Pipeline
|
||||
from app.flow.runs import (
|
||||
RunRejected,
|
||||
batch_issues,
|
||||
collect_result,
|
||||
digest_of,
|
||||
required_labels,
|
||||
seed_values,
|
||||
)
|
||||
from app.flow.schemas import FlowDef, FlowInput, NodeDef
|
||||
from app.flow.state import MemoryState
|
||||
|
||||
|
||||
def spec(name: str, dtype: DType = DType.FLOAT, **kwargs) -> MessageSpec:
|
||||
return MessageSpec(name=name, dtype=dtype, **kwargs)
|
||||
|
||||
|
||||
def make_node(node_id: str, flow: str, f, requires=(), provides=()) -> Node:
|
||||
node = Node(f=f, requires=list(requires), provides=list(provides), name=node_id)
|
||||
node.assign_flow(flow, node_id)
|
||||
return node
|
||||
|
||||
|
||||
def double_flow() -> FlowDef:
|
||||
"""A flow with one input, one node and one declared output."""
|
||||
return FlowDef(
|
||||
name="study",
|
||||
mode="batch",
|
||||
inputs=[FlowInput(spec=spec("lr"), initial=0.1)],
|
||||
outputs=["loss"],
|
||||
nodes=[
|
||||
NodeDef(
|
||||
id="train",
|
||||
requires=[spec("lr")],
|
||||
provides=[spec("loss")],
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def build(flow: FlowDef, state: MemoryState, observer=None) -> Pipeline:
|
||||
"""The pipeline a run drives, without the controller that normally builds it."""
|
||||
node = make_node(
|
||||
"train",
|
||||
flow.name,
|
||||
lambda lr, params: {"loss": lr * 2},
|
||||
requires=[spec("lr")],
|
||||
provides=[spec("loss")],
|
||||
)
|
||||
return Pipeline(nodes=[node], state=state, observer=observer)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Isolation — the reason a run has a state backend of its own
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_two_runs_of_one_flow_do_not_see_each_other():
|
||||
flow = double_flow()
|
||||
first, second = MemoryState(), MemoryState()
|
||||
|
||||
build(flow, first).run(seed_values(flow, {"lr": 0.5}))
|
||||
build(flow, second).run(seed_values(flow, {"lr": 4.0}))
|
||||
|
||||
assert collect_result(flow, first) == {"loss": 1.0}
|
||||
assert collect_result(flow, second) == {"loss": 8.0}
|
||||
|
||||
|
||||
def test_result_falls_back_to_everything_the_flow_holds():
|
||||
flow = double_flow()
|
||||
flow.outputs = []
|
||||
state = MemoryState()
|
||||
|
||||
build(flow, state).run(seed_values(flow, {"lr": 1.0}))
|
||||
|
||||
# The input is part of what the flow ended up holding; the engine's own
|
||||
# bookkeeping keys are not.
|
||||
assert collect_result(flow, state) == {"lr": 1.0, "loss": 2.0}
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# The record — every node a run executed, reported once
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_observer_sees_every_node_once():
|
||||
flow = double_flow()
|
||||
seen = []
|
||||
|
||||
build(flow, MemoryState(), observer=seen.append).run(seed_values(flow, {"lr": 1.0}))
|
||||
|
||||
assert [(o.node, o.ok) for o in seen] == [("study.train", True)]
|
||||
assert seen[0].outputs == 1
|
||||
|
||||
|
||||
def test_observer_reports_a_failing_node_with_its_error():
|
||||
seen = []
|
||||
|
||||
def boom(params):
|
||||
raise ValueError("no convergence")
|
||||
|
||||
node = make_node("train", "study", boom, provides=[spec("loss")])
|
||||
Pipeline(nodes=[node], state=MemoryState(), observer=seen.append).run()
|
||||
|
||||
assert len(seen) == 1
|
||||
assert not seen[0].ok
|
||||
assert "no convergence" in seen[0].error
|
||||
|
||||
|
||||
def test_a_failing_observer_does_not_take_the_node_down():
|
||||
def refuse(_outcome):
|
||||
raise RuntimeError("the database is gone")
|
||||
|
||||
node = make_node(
|
||||
"train", "study", lambda params: {"loss": 1.0}, provides=[spec("loss")]
|
||||
)
|
||||
pipeline = Pipeline(nodes=[node], state=MemoryState(), observer=refuse)
|
||||
pipeline.run()
|
||||
|
||||
assert pipeline.state["study.loss"] == 1.0
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# What a caller may ask for
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parameters_must_be_declared_inputs():
|
||||
flow = double_flow()
|
||||
with pytest.raises(RunRejected, match="not an input"):
|
||||
seed_values(flow, {"learning_rate": 0.1})
|
||||
|
||||
|
||||
def test_parameters_are_type_checked_before_anything_runs():
|
||||
flow = double_flow()
|
||||
with pytest.raises(RunRejected, match="lr"):
|
||||
seed_values(flow, {"lr": "fast"})
|
||||
|
||||
|
||||
def test_a_rate_limited_port_cannot_be_run_as_a_batch():
|
||||
flow = double_flow()
|
||||
flow.nodes[0].provides = [spec("loss", interval=30)]
|
||||
# Without a queue there is no timer to release what an interval holds, so
|
||||
# the value would be dropped rather than delayed.
|
||||
assert batch_issues(flow)
|
||||
assert not batch_issues(double_flow())
|
||||
|
||||
|
||||
def test_the_digest_identifies_the_inputs_not_their_order():
|
||||
assert digest_of({"a": 1, "b": 2}, 3) == digest_of({"b": 2, "a": 1}, 3)
|
||||
assert digest_of({"a": 1}, 3) != digest_of({"a": 1}, 4)
|
||||
|
||||
|
||||
def test_labels_come_from_the_nodes_that_ask_for_a_device():
|
||||
flow = double_flow()
|
||||
assert required_labels(flow) == []
|
||||
flow.nodes[0].device = "gpu"
|
||||
assert required_labels(flow) == ["gpu"]
|
||||
Reference in New Issue
Block a user