Skip to content

API Reference

The API reference is generated from the public Python modules.

Core

Fovux configuration loader.

Reads config.toml from FOVUX_HOME, applies environment variable overrides, and exposes a typed FovuxConfig object.

FovuxConfig

Bases: BaseModel

Root Fovux configuration.

Source code in src/fovux/config.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class FovuxConfig(BaseModel):
    """Root Fovux configuration."""

    version: str = "1.0"
    policy_mode: str = "developer"
    paths: PathsConfig = Field(default_factory=PathsConfig)
    training: TrainingConfig = Field(default_factory=TrainingConfig)
    inference: InferenceConfig = Field(default_factory=InferenceConfig)
    ui: UIConfig = Field(default_factory=UIConfig)
    telemetry: TelemetryConfig = Field(default_factory=TelemetryConfig)
    validation: ValidationConfig = Field(default_factory=ValidationConfig)

    @property
    def fovux_paths(self) -> FovuxPaths:
        """Return a FovuxPaths resolved from the current config."""
        env_home = os.environ.get("FOVUX_HOME")
        home = Path(env_home).expanduser().resolve() if env_home else Path(self.paths.home)
        return FovuxPaths(home, self.paths.model_dump(), load_config_file=False)

    @property
    def telemetry_enabled(self) -> bool:
        """Return effective telemetry setting, respecting FOVUX_NO_TELEMETRY."""
        if os.environ.get("FOVUX_NO_TELEMETRY", "").strip():
            return False
        return self.telemetry.enabled

fovux_paths property

Return a FovuxPaths resolved from the current config.

telemetry_enabled property

Return effective telemetry setting, respecting FOVUX_NO_TELEMETRY.

InferenceConfig

Bases: BaseModel

Default inference parameters.

Source code in src/fovux/config.py 
42
43
44
45
46
class InferenceConfig(BaseModel):
    """Default inference parameters."""

    default_conf: float = 0.25
    default_iou: float = 0.45

PathsConfig

Bases: BaseModel

Filesystem path configuration.

Source code in src/fovux/config.py 
24
25
26
27
28
29
30
31
class PathsConfig(BaseModel):
    """Filesystem path configuration."""

    home: str = "~/.fovux"
    runs: str = "runs"
    models: str = "models"
    cache: str = "cache"
    exports: str = "exports"

TelemetryConfig

Bases: BaseModel

Telemetry configuration. Off by default.

Source code in src/fovux/config.py 
55
56
57
58
59
class TelemetryConfig(BaseModel):
    """Telemetry configuration. Off by default."""

    enabled: bool = False
    endpoint: str = ""

TrainingConfig

Bases: BaseModel

Default training parameters.

Source code in src/fovux/config.py 
34
35
36
37
38
39
class TrainingConfig(BaseModel):
    """Default training parameters."""

    default_device: str = "auto"
    default_workers: int = 8
    default_patience: int = 50

UIConfig

Bases: BaseModel

UI preferences.

Source code in src/fovux/config.py 
49
50
51
52
class UIConfig(BaseModel):
    """UI preferences."""

    preferred_backend: str = "onnxruntime"

ValidationConfig

Bases: BaseModel

Filesystem safety limits for local processing.

Source code in src/fovux/config.py 
62
63
64
65
class ValidationConfig(BaseModel):
    """Filesystem safety limits for local processing."""

    max_file_size_mb: int = 100

clear_config_cache()

Clear the process-local configuration cache.

Source code in src/fovux/config.py 
131
132
133
def clear_config_cache() -> None:
    """Clear the process-local configuration cache."""
    _CONFIG_CACHE.clear()

load_config(config_path=None)

Load Fovux configuration from config.toml.

Priority: env vars > config.toml > defaults.

Parameters:

Name Type Description Default
config_path Path | None

Override path to config.toml. Defaults to FOVUX_HOME/config.toml.

 None

Returns:

Type Description
FovuxConfig

Validated FovuxConfig instance.
Source code in src/fovux/config.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def load_config(config_path: Path | None = None) -> FovuxConfig:
    """Load Fovux configuration from config.toml.

    Priority: env vars > config.toml > defaults.

    Args:
        config_path: Override path to config.toml. Defaults to FOVUX_HOME/config.toml.

    Returns:
        Validated FovuxConfig instance.
    """
    target_path = config_path or (get_fovux_home() / "config.toml")
    return _load_config_cached(str(target_path))

write_default_config(config_path)

Write a default config.toml to the given path.

Parameters:

Name Type Description Default
config_path Path

Destination path for config.toml.

 required
Source code in src/fovux/config.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def write_default_config(config_path: Path) -> None:
    """Write a default config.toml to the given path.

    Args:
        config_path: Destination path for config.toml.
    """
    config_path.parent.mkdir(parents=True, exist_ok=True)
    default: dict[str, Any] = {
        "fovux": {
            "version": "1.0",
            "policy_mode": "developer",
            "paths": {
                "home": "~/.fovux",
                "runs": "runs",
                "models": "models",
                "cache": "cache",
                "exports": "exports",
            },
            "training": {
                "default_device": "auto",
                "default_workers": 8,
                "default_patience": 50,
            },
            "inference": {
                "default_conf": 0.25,
                "default_iou": 0.45,
            },
            "ui": {
                "preferred_backend": "onnxruntime",
            },
            "telemetry": {
                "enabled": False,
                "endpoint": "",
            },
            "validation": {
                "max_file_size_mb": 100,
            },
        }
    }
    with config_path.open("wb") as f:
        tomli_w.dump(default, f)

Fovux exception hierarchy.

All exceptions raised by Fovux tools are subclasses of FovuxError. External library exceptions (ultralytics, cv2, onnx) must never bubble up raw.

FovuxCheckpointNotFoundError

Bases: FovuxEvalError

Checkpoint file does not exist.

Source code in src/fovux/core/errors.py 
124
125
126
127
128
129
130
131
132
133
134
class FovuxCheckpointNotFoundError(FovuxEvalError):
    """Checkpoint file does not exist."""

    code = "FOVUX_EVAL_001"

    def __init__(self, path: str) -> None:
        """Initialize with the missing checkpoint path."""
        super().__init__(
            f"Checkpoint not found: {path}",
            hint="Provide a valid .pt file path or a run_id with a best.pt.",
        )

__init__(path)

Initialize with the missing checkpoint path.

Source code in src/fovux/core/errors.py 
129
130
131
132
133
134
def __init__(self, path: str) -> None:
    """Initialize with the missing checkpoint path."""
    super().__init__(
        f"Checkpoint not found: {path}",
        hint="Provide a valid .pt file path or a run_id with a best.pt.",
    )

FovuxConfigError

Bases: FovuxError

Base class for configuration errors.

Source code in src/fovux/core/errors.py 
177
178
179
180
class FovuxConfigError(FovuxError):
    """Base class for configuration errors."""

    code = "FOVUX_CONFIG_000"

FovuxDatasetEmptyError

Bases: FovuxDatasetError

Dataset contains zero images.

Source code in src/fovux/core/errors.py 
68
69
70
71
72
73
74
75
76
77
78
class FovuxDatasetEmptyError(FovuxDatasetError):
    """Dataset contains zero images."""

    code = "FOVUX_DATASET_003"

    def __init__(self, path: str, message: str | None = None) -> None:
        """Initialize with the empty dataset path."""
        super().__init__(
            message or f"Dataset at {path} contains no images.",
            hint="Ensure the dataset root contains image files (jpg, png, bmp, webp).",
        )

__init__(path, message=None)

Initialize with the empty dataset path.

Source code in src/fovux/core/errors.py 
73
74
75
76
77
78
def __init__(self, path: str, message: str | None = None) -> None:
    """Initialize with the empty dataset path."""
    super().__init__(
        message or f"Dataset at {path} contains no images.",
        hint="Ensure the dataset root contains image files (jpg, png, bmp, webp).",
    )

FovuxDatasetError

Bases: FovuxError

Base class for dataset-related errors.

Source code in src/fovux/core/errors.py 
43
44
45
46
class FovuxDatasetError(FovuxError):
    """Base class for dataset-related errors."""

    code = "FOVUX_DATASET_000"

FovuxDatasetFormatError

Bases: FovuxDatasetError

Dataset format cannot be detected or is malformed.

Source code in src/fovux/core/errors.py 
62
63
64
65
class FovuxDatasetFormatError(FovuxDatasetError):
    """Dataset format cannot be detected or is malformed."""

    code = "FOVUX_DATASET_002"

FovuxDatasetNotFoundError

Bases: FovuxDatasetError

Dataset path does not exist.

Source code in src/fovux/core/errors.py 
49
50
51
52
53
54
55
56
57
58
59
class FovuxDatasetNotFoundError(FovuxDatasetError):
    """Dataset path does not exist."""

    code = "FOVUX_DATASET_001"

    def __init__(self, path: str) -> None:
        """Initialize with the missing path."""
        super().__init__(
            f"Dataset path not found: {path}",
            hint="Check that the path exists and is accessible.",
        )

__init__(path)

Initialize with the missing path.

Source code in src/fovux/core/errors.py 
54
55
56
57
58
59
def __init__(self, path: str) -> None:
    """Initialize with the missing path."""
    super().__init__(
        f"Dataset path not found: {path}",
        hint="Check that the path exists and is accessible.",
    )

FovuxError

Bases: Exception

Base exception for all Fovux errors.

Attributes:

Name Type Description
code str

Stable error code string (e.g. FOVUX_DATASET_001).
message

Human-readable description.
hint

Optional remediation hint for the user.
Source code in src/fovux/core/errors.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class FovuxError(Exception):
    """Base exception for all Fovux errors.

    Attributes:
        code: Stable error code string (e.g. FOVUX_DATASET_001).
        message: Human-readable description.
        hint: Optional remediation hint for the user.
    """

    code: str = "FOVUX_000"

    def __init__(self, message: str, hint: str | None = None) -> None:
        """Initialize FovuxError.

        Args:
            message: Human-readable error description.
            hint: Optional remediation hint.
        """
        super().__init__(message)
        self.message = message
        self.hint = hint

    def __str__(self) -> str:
        """Return formatted error string."""
        base = f"[{self.code}] {self.message}"
        if self.hint:
            return f"{base}\nHint: {self.hint}"
        return base

__init__(message, hint=None)

Initialize FovuxError.

Parameters:

Name Type Description Default
message str

Human-readable error description.

 required
hint str | None

Optional remediation hint.

 None
Source code in src/fovux/core/errors.py 
21
22
23
24
25
26
27
28
29
30
def __init__(self, message: str, hint: str | None = None) -> None:
    """Initialize FovuxError.

    Args:
        message: Human-readable error description.
        hint: Optional remediation hint.
    """
    super().__init__(message)
    self.message = message
    self.hint = hint

__str__()

Return formatted error string.

Source code in src/fovux/core/errors.py 
32
33
34
35
36
37
def __str__(self) -> str:
    """Return formatted error string."""
    base = f"[{self.code}] {self.message}"
    if self.hint:
        return f"{base}\nHint: {self.hint}"
    return base

FovuxEvalError

Bases: FovuxError

Base class for evaluation-related errors.

Source code in src/fovux/core/errors.py 
118
119
120
121
class FovuxEvalError(FovuxError):
    """Base class for evaluation-related errors."""

    code = "FOVUX_EVAL_000"

FovuxExportError

Bases: FovuxError

Base class for export-related errors.

Source code in src/fovux/core/errors.py 
140
141
142
143
class FovuxExportError(FovuxError):
    """Base class for export-related errors."""

    code = "FOVUX_EXPORT_000"

FovuxExportParityError

Bases: FovuxExportError

Roundtrip parity check failed after export.

Source code in src/fovux/core/errors.py 
146
147
148
149
class FovuxExportParityError(FovuxExportError):
    """Roundtrip parity check failed after export."""

    code = "FOVUX_EXPORT_001"

FovuxInferenceError

Bases: FovuxError

Base class for inference-related errors.

Source code in src/fovux/core/errors.py 
155
156
157
158
class FovuxInferenceError(FovuxError):
    """Base class for inference-related errors."""

    code = "FOVUX_INFER_000"

FovuxPathValidationError

Bases: FovuxConfigError

Raised when a filesystem path violates local safety checks.

Source code in src/fovux/core/errors.py 
183
184
185
186
187
188
189
190
191
192
193
class FovuxPathValidationError(FovuxConfigError):
    """Raised when a filesystem path violates local safety checks."""

    code = "FOVUX_CONFIG_001"

    def __init__(self, path: str, reason: str, hint: str | None = None) -> None:
        """Initialize the path validation error."""
        super().__init__(
            f"Path validation failed for {path}: {reason}",
            hint=hint or "Use a path inside the intended project or dataset root.",
        )

__init__(path, reason, hint=None)

Initialize the path validation error.

Source code in src/fovux/core/errors.py 
188
189
190
191
192
193
def __init__(self, path: str, reason: str, hint: str | None = None) -> None:
    """Initialize the path validation error."""
    super().__init__(
        f"Path validation failed for {path}: {reason}",
        hint=hint or "Use a path inside the intended project or dataset root.",
    )

FovuxRtspConnectionError

Bases: FovuxInferenceError

RTSP stream could not be opened.

Source code in src/fovux/core/errors.py 
161
162
163
164
165
166
167
168
169
170
171
class FovuxRtspConnectionError(FovuxInferenceError):
    """RTSP stream could not be opened."""

    code = "FOVUX_INFER_001"

    def __init__(self, url: str) -> None:
        """Initialize with the failing RTSP URL."""
        super().__init__(
            f"Could not open RTSP stream: {url}",
            hint="Verify the stream URL, network connectivity, and credentials.",
        )

__init__(url)

Initialize with the failing RTSP URL.

Source code in src/fovux/core/errors.py 
166
167
168
169
170
171
def __init__(self, url: str) -> None:
    """Initialize with the failing RTSP URL."""
    super().__init__(
        f"Could not open RTSP stream: {url}",
        hint="Verify the stream URL, network connectivity, and credentials.",
    )

FovuxTrainingAlreadyRunningError

Bases: FovuxTrainingError

Attempt to start training on an already-running run.

Source code in src/fovux/core/errors.py 
103
104
105
106
class FovuxTrainingAlreadyRunningError(FovuxTrainingError):
    """Attempt to start training on an already-running run."""

    code = "FOVUX_TRAIN_002"

FovuxTrainingError

Bases: FovuxError

Base class for training-related errors.

Source code in src/fovux/core/errors.py 
84
85
86
87
class FovuxTrainingError(FovuxError):
    """Base class for training-related errors."""

    code = "FOVUX_TRAIN_000"

FovuxTrainingRunNotFoundError

Bases: FovuxTrainingError

Run ID does not exist in the registry.

Source code in src/fovux/core/errors.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class FovuxTrainingRunNotFoundError(FovuxTrainingError):
    """Run ID does not exist in the registry."""

    code = "FOVUX_TRAIN_001"

    def __init__(self, run_id: str) -> None:
        """Initialize with the missing run_id."""
        super().__init__(
            f"Run not found: {run_id}",
            hint="Use `model_list` to see available runs.",
        )

__init__(run_id)

Initialize with the missing run_id.

Source code in src/fovux/core/errors.py 
 95
 96
 97
 98
 99
100
def __init__(self, run_id: str) -> None:
    """Initialize with the missing run_id."""
    super().__init__(
        f"Run not found: {run_id}",
        hint="Use `model_list` to see available runs.",
    )

FovuxTrainingSubprocessError

Bases: FovuxTrainingError

Training subprocess exited with non-zero code.

Source code in src/fovux/core/errors.py 
109
110
111
112
class FovuxTrainingSubprocessError(FovuxTrainingError):
    """Training subprocess exited with non-zero code."""

    code = "FOVUX_TRAIN_003"

Fovux home directory resolution and path helpers.

FovuxPaths

Typed container for all Fovux filesystem paths.

Attributes:

Name Type Description
home

Root Fovux data directory.
runs

Training run directories.
models

Pretrained / imported checkpoints.
cache

Perceptual hashes, thumbnails, dataset cache.
exports

Exported ONNX / TFLite / TensorRT artifacts.
datasets

Indexed dataset metadata (not raw images).
runs_db

SQLite runs index.
config_file

User-level config.toml.
Source code in src/fovux/core/paths.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class FovuxPaths:
    """Typed container for all Fovux filesystem paths.

    Attributes:
        home: Root Fovux data directory.
        runs: Training run directories.
        models: Pretrained / imported checkpoints.
        cache: Perceptual hashes, thumbnails, dataset cache.
        exports: Exported ONNX / TFLite / TensorRT artifacts.
        datasets: Indexed dataset metadata (not raw images).
        runs_db: SQLite runs index.
        config_file: User-level config.toml.
    """

    def __init__(
        self,
        home: Path,
        path_overrides: Mapping[str, str] | None = None,
        *,
        load_config_file: bool = True,
    ) -> None:
        """Initialize with a home directory."""
        self.home = home.expanduser().resolve()
        self.config_file = self.home / "config.toml"
        overrides = dict(path_overrides or {})
        if path_overrides is None and load_config_file:
            overrides.update(_load_path_overrides(self.config_file))
        self.runs = _resolve_child_path(self.home, overrides.get("runs", "runs"))
        self.models = _resolve_child_path(self.home, overrides.get("models", "models"))
        self.cache = _resolve_child_path(self.home, overrides.get("cache", "cache"))
        self.exports = _resolve_child_path(self.home, overrides.get("exports", "exports"))
        self.datasets = self.home / "datasets"
        self.runs_db = self.home / "runs.db"

    def run_dir(self, run_id: str) -> Path:
        """Return the directory for a specific run.

        Args:
            run_id: The run identifier string.

        Returns:
            Path to the run directory.
        """
        return self.runs / run_id

    def __repr__(self) -> str:
        """Return debug representation."""
        return f"FovuxPaths(home={self.home})"

__init__(home, path_overrides=None, *, load_config_file=True)

Initialize with a home directory.

Source code in src/fovux/core/paths.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def __init__(
    self,
    home: Path,
    path_overrides: Mapping[str, str] | None = None,
    *,
    load_config_file: bool = True,
) -> None:
    """Initialize with a home directory."""
    self.home = home.expanduser().resolve()
    self.config_file = self.home / "config.toml"
    overrides = dict(path_overrides or {})
    if path_overrides is None and load_config_file:
        overrides.update(_load_path_overrides(self.config_file))
    self.runs = _resolve_child_path(self.home, overrides.get("runs", "runs"))
    self.models = _resolve_child_path(self.home, overrides.get("models", "models"))
    self.cache = _resolve_child_path(self.home, overrides.get("cache", "cache"))
    self.exports = _resolve_child_path(self.home, overrides.get("exports", "exports"))
    self.datasets = self.home / "datasets"
    self.runs_db = self.home / "runs.db"

__repr__()

Return debug representation.

Source code in src/fovux/core/paths.py 
94
95
96
def __repr__(self) -> str:
    """Return debug representation."""
    return f"FovuxPaths(home={self.home})"

run_dir(run_id)

Return the directory for a specific run.

Parameters:

Name Type Description Default
run_id str

The run identifier string.

 required

Returns:

Type Description
Path

Path to the run directory.
Source code in src/fovux/core/paths.py 
83
84
85
86
87
88
89
90
91
92
def run_dir(self, run_id: str) -> Path:
    """Return the directory for a specific run.

    Args:
        run_id: The run identifier string.

    Returns:
        Path to the run directory.
    """
    return self.runs / run_id

ensure_fovux_dirs(home=None)

Create all required Fovux subdirectories and return a FovuxPaths instance.

Parameters:

Name Type Description Default
home Path | None

Override the home directory (defaults to get_fovux_home()).

 None

Returns:

Type Description
FovuxPaths

FovuxPaths with all directories created.
Source code in src/fovux/core/paths.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def ensure_fovux_dirs(home: Path | None = None) -> FovuxPaths:
    """Create all required Fovux subdirectories and return a FovuxPaths instance.

    Args:
        home: Override the home directory (defaults to get_fovux_home()).

    Returns:
        FovuxPaths with all directories created.
    """
    if home is None:
        home = get_fovux_home()
    paths = FovuxPaths(home)
    paths.runs.mkdir(parents=True, exist_ok=True)
    paths.models.mkdir(parents=True, exist_ok=True)
    paths.cache.mkdir(parents=True, exist_ok=True)
    paths.exports.mkdir(parents=True, exist_ok=True)
    paths.datasets.mkdir(parents=True, exist_ok=True)
    return paths

get_fovux_home()

Resolve the Fovux home directory.

Priority order: 1. FOVUX_HOME environment variable 2. ~/.fovux

Returns:

Type Description
Path

Absolute Path to the Fovux home directory.
Source code in src/fovux/core/paths.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def get_fovux_home() -> Path:
    """Resolve the Fovux home directory.

    Priority order:
    1. FOVUX_HOME environment variable
    2. ~/.fovux

    Returns:
        Absolute Path to the Fovux home directory.
    """
    env_home = os.environ.get("FOVUX_HOME")
    if env_home:
        return Path(env_home).expanduser().resolve()

    default_home = Path.home() / ".fovux"
    return default_home

SQLite-backed run registry.

Tracks all training runs: metadata, status, metrics pointers.

ArtifactRecord

Bases: Base

ORM model for artifacts (e.g. checkpoints, exports, datasets).

Source code in src/fovux/core/runs.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class ArtifactRecord(Base):
    """ORM model for artifacts (e.g. checkpoints, exports, datasets)."""

    __tablename__ = "artifacts"

    id = Column(String, primary_key=True)
    run_id = Column(String, nullable=True, index=True)
    type = Column(String, nullable=False)  # e.g., checkpoint, dataset, export
    path = Column(String, nullable=False)
    sha256 = Column(String, nullable=True)
    size = Column(Integer, nullable=True)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    extra_json = Column(Text, nullable=False, default="{}")

AuditEventRecord

Bases: Base

ORM model for security/system audit events.

Source code in src/fovux/core/runs.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
class AuditEventRecord(Base):
    """ORM model for security/system audit events."""

    __tablename__ = "audit_events"

    id = Column(Integer, primary_key=True, autoincrement=True)
    actor = Column(String, nullable=False)
    action = Column(String, nullable=False)
    entity_type = Column(String, nullable=False)
    entity_id = Column(String, nullable=False)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    details_json = Column(Text, nullable=False, default="{}")

Base

Bases: DeclarativeBase

SQLAlchemy declarative base.

Source code in src/fovux/core/runs.py 
38
39
class Base(DeclarativeBase):
    """SQLAlchemy declarative base."""

DatasetRecord

Bases: Base

ORM model for datasets.

Source code in src/fovux/core/runs.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class DatasetRecord(Base):
    """ORM model for datasets."""

    __tablename__ = "datasets"

    fingerprint = Column(String, primary_key=True)
    path = Column(String, nullable=False)
    class_map_json = Column(Text, nullable=False, default="{}")
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    extra_json = Column(Text, nullable=False, default="{}")

ExportRecord

Bases: Base

ORM model for model exports.

Source code in src/fovux/core/runs.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
class ExportRecord(Base):
    """ORM model for model exports."""

    __tablename__ = "exports"

    id = Column(String, primary_key=True)
    run_id = Column(String, nullable=True, index=True)
    source_checkpoint = Column(String, nullable=False)
    artifact_path = Column(String, nullable=False)
    format = Column(String, nullable=False)
    duration_s = Column(Float, nullable=True)
    validation_result_json = Column(Text, nullable=True)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

MetricRecord

Bases: Base

ORM model for step or epoch-level metrics.

Source code in src/fovux/core/runs.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
class MetricRecord(Base):
    """ORM model for step or epoch-level metrics."""

    __tablename__ = "metrics"

    id = Column(Integer, primary_key=True, autoincrement=True)
    run_id = Column(String, nullable=False, index=True)
    epoch = Column(Integer, nullable=False)
    metric_key = Column(String, nullable=False)
    metric_value = Column(Float, nullable=False)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

ModelRecord

Bases: Base

ORM model for registered or downloaded models.

Source code in src/fovux/core/runs.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class ModelRecord(Base):
    """ORM model for registered or downloaded models."""

    __tablename__ = "models"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    task = Column(String, nullable=False)
    path = Column(String, nullable=True)
    sha256 = Column(String, nullable=True)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

OperationEventRecord

Bases: Base

ORM model for operation lifecycle events (for SSE resume).

Source code in src/fovux/core/runs.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class OperationEventRecord(Base):
    """ORM model for operation lifecycle events (for SSE resume)."""

    __tablename__ = "operation_events"

    id = Column(Integer, primary_key=True, autoincrement=True)
    operation_id = Column(String, nullable=False, index=True)
    event_type = Column(String, nullable=False)  # status_change, progress, etc.
    data_json = Column(Text, nullable=False)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

OperationRecord

Bases: Base

ORM model for a background operation.

Source code in src/fovux/core/runs.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class OperationRecord(Base):
    """ORM model for a background operation."""

    __tablename__ = "operations"

    id = Column(String, primary_key=True)
    idempotency_key = Column(String, nullable=True, unique=True, index=True)
    tool = Column(String, nullable=False)
    arguments_json = Column(Text, nullable=False)
    status = Column(String, nullable=False, default="pending")
    progress = Column(Integer, nullable=True)
    result_json = Column(Text, nullable=True)
    error_type = Column(String, nullable=True)
    error = Column(Text, nullable=True)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    started_at = Column(DateTime, nullable=True)
    finished_at = Column(DateTime, nullable=True)
    run_id = Column(String, nullable=True)

ReviewQueueEntry

Bases: Base

ORM model for active learning review queue entries.

Source code in src/fovux/core/runs.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class ReviewQueueEntry(Base):
    """ORM model for active learning review queue entries."""

    __tablename__ = "review_queue"

    id: Any = Column(String, primary_key=True)
    image_path: Any = Column(String, nullable=False)
    dataset_path: Any = Column(String, nullable=False)
    score: Any = Column(Float, nullable=False)
    reason: Any = Column(String, nullable=False)
    status: Any = Column(String, nullable=False, default="pending")
    predictions_json: Any = Column(Text, nullable=False, default="[]")
    corrected_labels_json: Any = Column(Text, nullable=True)
    created_at: Any = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

RunEventRecord

Bases: Base

ORM model for run lifecycle events (status transition, artifact creation, etc.).

Source code in src/fovux/core/runs.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class RunEventRecord(Base):
    """ORM model for run lifecycle events (status transition, artifact creation, etc.)."""

    __tablename__ = "run_events"

    id = Column(Integer, primary_key=True, autoincrement=True)
    run_id = Column(String, nullable=True, index=True)
    event_type = Column(String, nullable=False)  # status_transition, artifact_created, audit, etc.
    from_status = Column(String, nullable=True)
    to_status = Column(String, nullable=True)
    message = Column(Text, nullable=True)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    extra_json = Column(Text, nullable=False, default="{}")

RunRecord

Bases: Base

ORM model for a training run row.

Source code in src/fovux/core/runs.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class RunRecord(Base):
    """ORM model for a training run row."""

    __tablename__ = "runs"

    id = Column(String, primary_key=True)
    status = Column(String, nullable=False, default="pending")
    model = Column(String, nullable=False)
    dataset_path = Column(String, nullable=False)
    task = Column(String, nullable=False, default="detect")
    epochs = Column(Integer, nullable=False)
    created_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )
    started_at = Column(DateTime, nullable=True)
    finished_at = Column(DateTime, nullable=True)
    pid = Column(Integer, nullable=True)
    run_path = Column(String, nullable=False)
    tags_json = Column(Text, nullable=False, default="[]")
    extra_json = Column(Text, nullable=False, default="{}")

    # Experiment intelligence / lineage tracking metadata
    dataset_fingerprint = Column(String, nullable=True)
    config_hash = Column(String, nullable=True)
    code_version = Column(String, nullable=True)
    env_summary = Column(Text, nullable=True)
    parent_run_id = Column(String, nullable=True)

RunRegistry

CRUD interface for the SQLite runs registry.

Parameters:

Name Type Description Default
db_path Path

Path to the SQLite database file.

 required
Source code in src/fovux/core/runs.py 
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
class RunRegistry:
    """CRUD interface for the SQLite runs registry.

    Args:
        db_path: Path to the SQLite database file.
    """

    def __init__(self, db_path: Path) -> None:
        """Initialize and create tables if needed."""
        self._engine = create_engine(
            f"sqlite:///{db_path}",
            echo=False,
            poolclass=NullPool,
            connect_args={"check_same_thread": False},
        )

        @sa_event.listens_for(self._engine, "connect")
        def _set_sqlite_pragmas(
            dbapi_conn: sqlite3.Connection,
            _connection_record: object,
        ) -> None:
            dbapi_conn.execute("PRAGMA journal_mode=WAL")
            dbapi_conn.execute("PRAGMA synchronous=NORMAL")
            dbapi_conn.execute("PRAGMA foreign_keys=ON")

        Base.metadata.create_all(self._engine)
        self._run_migrations()
        self._Session = sessionmaker(bind=self._engine, expire_on_commit=False)

    def _run_migrations(self) -> None:
        """Ensure schema migrations are executed, adding columns if needed."""
        with self._engine.begin() as conn:
            conn.execute(
                text(
                    "CREATE TABLE IF NOT EXISTS schema_migrations ("
                    "version INTEGER PRIMARY KEY,"
                    "applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP"
                    ")"
                )
            )
            row = conn.execute(text("SELECT MAX(version) FROM schema_migrations")).fetchone()
            current_version = row[0] if row and row[0] is not None else 0

        if current_version < 1:
            self._apply_migration_1()

    def _apply_migration_1(self) -> None:
        """Add columns to runs table for dataset/config tracking."""
        with self._engine.begin() as conn:
            res = conn.execute(text("PRAGMA table_info(runs)")).fetchall()
            existing_cols = {r[1] for r in res}

            new_cols = {
                "dataset_fingerprint": "TEXT",
                "config_hash": "TEXT",
                "code_version": "TEXT",
                "env_summary": "TEXT",
                "parent_run_id": "TEXT",
            }

            for col_name, col_type in new_cols.items():
                if col_name not in existing_cols:
                    conn.execute(text(f"ALTER TABLE runs ADD COLUMN {col_name} {col_type}"))

            conn.execute(
                text(
                    "INSERT OR IGNORE INTO schema_migrations (version, applied_at) "
                    "VALUES (1, :applied_at)"
                ),
                {"applied_at": datetime.now(UTC).replace(tzinfo=None)},
            )

    def close(self) -> None:
        """Dispose the SQLite engine and release pooled connections."""
        self._engine.dispose()

    def _auto_metadata(
        self,
        model: str,
        dataset_path: Path,
        task: str,
        epochs: int,
        extra: dict[str, Any] | None = None,
    ) -> tuple[str, str, str, str]:
        """Automatically calculate metadata fields if they are None."""
        import hashlib

        from fovux import __version__ as fovux_version

        # 1. Dataset Fingerprint
        try:
            from fovux.core.dataset_config import _find_yolo_yaml

            yaml_path = _find_yolo_yaml(dataset_path)
            content = yaml_path.read_bytes()
            dataset_fingerprint = hashlib.sha256(content).hexdigest()
        except Exception:
            dataset_fingerprint = hashlib.sha256(
                str(dataset_path.resolve()).encode("utf-8")
            ).hexdigest()

        # 2. Config Hash
        payload = {
            "model": model,
            "dataset_path": str(dataset_path.resolve()),
            "epochs": epochs,
            "task": task,
            "extra": extra or {},
        }
        encoded = json.dumps(payload, sort_keys=True).encode("utf-8")
        config_hash = hashlib.sha256(encoded).hexdigest()

        # 3. Code Version
        code_version = fovux_version

        # 4. Env Summary
        import platform
        import sys

        summary: dict[str, Any] = {
            "python_version": sys.version.split()[0],
            "platform": platform.platform(),
        }
        try:
            import torch  # type: ignore[import-not-found]

            summary["torch_version"] = torch.__version__
            summary["cuda_available"] = torch.cuda.is_available()
        except ImportError:
            summary["torch_version"] = "not_installed"
            summary["cuda_available"] = False
        env_summary = json.dumps(summary)

        return dataset_fingerprint, config_hash, code_version, env_summary

    def _register_lineage(
        self,
        session: Session,
        run_id: str,
        model: str,
        dataset_path: Path,
        task: str,
        dataset_fingerprint: str,
    ) -> None:
        """Internal helper to populate datasets and models tables for lineage tracking."""
        # 1. Dataset Registration
        try:
            from fovux.core.dataset_utils import read_yolo_data_yaml

            data = read_yolo_data_yaml(dataset_path)
            class_map = data.get("names", {})
        except Exception:
            class_map = {}

        db_dataset = session.get(DatasetRecord, dataset_fingerprint)
        if db_dataset is None:
            db_dataset = DatasetRecord(
                fingerprint=dataset_fingerprint,
                path=str(dataset_path.resolve()),
                class_map_json=json.dumps(class_map),
            )
            session.merge(db_dataset)

        # 2. Model Registration
        model_name = Path(model).name
        db_model = session.get(ModelRecord, model_name)
        if db_model is None:
            db_model = ModelRecord(
                id=model_name,
                name=model_name,
                task=task,
                path=model,
            )
            session.merge(db_model)

        # 3. Status Transition Event to Pending
        event = RunEventRecord(
            run_id=run_id,
            event_type="status_transition",
            from_status=None,
            to_status="pending",
            message="Run reserved and initialized in pending state",
        )
        session.add(event)

    def reserve_run_slot(
        self,
        run_id: str,
        run_path: Path,
        model: str,
        dataset_path: Path,
        task: str,
        epochs: int,
        max_concurrent_runs: int,
        tags: list[str] | None = None,
        extra: dict[str, Any] | None = None,
        dataset_fingerprint: str | None = None,
        config_hash: str | None = None,
        code_version: str | None = None,
        env_summary: str | None = None,
        parent_run_id: str | None = None,
    ) -> RunRecord:
        """Reserve a run slot atomically, locking the DB if concurrent limit is set."""
        auto_fp, auto_ch, auto_cv, auto_env = self._auto_metadata(
            model, dataset_path, task, epochs, extra
        )
        dataset_fingerprint = dataset_fingerprint or auto_fp
        config_hash = config_hash or auto_ch
        code_version = code_version or auto_cv
        env_summary = env_summary or auto_env

        with self._Session() as session:
            with session.begin():
                if max_concurrent_runs > 0:
                    active_count = (
                        session.query(RunRecord)
                        .filter(RunRecord.status.in_(["running", "pending"]))
                        .count()
                    )
                    if active_count >= max_concurrent_runs:
                        from fovux.core.errors import FovuxTrainingAlreadyRunningError

                        raise FovuxTrainingAlreadyRunningError(
                            f"Cannot start run '{run_id}': {active_count} "
                            f"concurrent training run(s) already active and "
                            f"max_concurrent_runs={max_concurrent_runs}."
                        )

                record = RunRecord(
                    id=run_id,
                    run_path=str(run_path),
                    model=model,
                    dataset_path=str(dataset_path),
                    task=task,
                    epochs=epochs,
                    status="pending",
                    created_at=datetime.now(UTC).replace(tzinfo=None),
                    tags_json=json.dumps(tags or []),
                    extra_json=json.dumps(extra or {}),
                    dataset_fingerprint=dataset_fingerprint,
                    config_hash=config_hash,
                    code_version=code_version,
                    env_summary=env_summary,
                    parent_run_id=parent_run_id,
                )
                session.add(record)
                self._register_lineage(
                    session, run_id, model, dataset_path, task, dataset_fingerprint
                )
            session.refresh(record)
            return record

    def create_run(
        self,
        run_id: str,
        run_path: Path,
        model: str,
        dataset_path: Path,
        task: str,
        epochs: int,
        tags: list[str] | None = None,
        extra: dict[str, Any] | None = None,
        dataset_fingerprint: str | None = None,
        config_hash: str | None = None,
        code_version: str | None = None,
        env_summary: str | None = None,
        parent_run_id: str | None = None,
    ) -> RunRecord:
        """Insert a new run record.

        Args:
            run_id: Unique run identifier.
            run_path: Path to the run directory.
            model: Model name or path.
            dataset_path: Path to the dataset.
            task: YOLO task (detect, segment, classify, pose, obb).
            epochs: Total training epochs.
            tags: Optional list of user tags.
            extra: Optional extra metadata dict.
            dataset_fingerprint: Optional fingerprint of dataset.
            config_hash: Optional hash of config.
            code_version: Optional code version.
            env_summary: Optional system environment summary.
            parent_run_id: Optional parent run/checkpoint reference.

        Returns:
            The newly created RunRecord.
        """
        auto_fp, auto_ch, auto_cv, auto_env = self._auto_metadata(
            model, dataset_path, task, epochs, extra
        )
        dataset_fingerprint = dataset_fingerprint or auto_fp
        config_hash = config_hash or auto_ch
        code_version = code_version or auto_cv
        env_summary = env_summary or auto_env

        with self._Session() as session:
            with session.begin():
                record = RunRecord(
                    id=run_id,
                    run_path=str(run_path),
                    model=model,
                    dataset_path=str(dataset_path),
                    task=task,
                    epochs=epochs,
                    status="pending",
                    created_at=datetime.now(UTC).replace(tzinfo=None),
                    tags_json=json.dumps(tags or []),
                    extra_json=json.dumps(extra or {}),
                    dataset_fingerprint=dataset_fingerprint,
                    config_hash=config_hash,
                    code_version=code_version,
                    env_summary=env_summary,
                    parent_run_id=parent_run_id,
                )
                session.add(record)
                self._register_lineage(
                    session, run_id, model, dataset_path, task, dataset_fingerprint
                )
            session.refresh(record)
            return record

    def get_run(self, run_id: str) -> RunRecord | None:
        """Fetch a run by ID.

        Args:
            run_id: The run identifier.

        Returns:
            RunRecord or None if not found.
        """
        with self._Session() as session:
            stmt = select(RunRecord).where(RunRecord.id == run_id)
            return session.execute(stmt).scalar_one_or_none()

    def update_status(
        self,
        run_id: str,
        status: RunStatus,
        pid: int | None = None,
    ) -> None:
        """Update run status and optionally pid.

        Args:
            run_id: The run identifier.
            status: New status value.
            pid: Process ID of the training subprocess (if applicable).
        """
        with self._Session() as session:
            stmt = select(RunRecord).where(RunRecord.id == run_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return

            current_status = str(record.status)
            if current_status != status:
                valid_targets = {
                    "pending": {"running", "complete", "failed", "stopped", "archived"},
                    "running": {"complete", "failed", "stopped", "archived"},
                    "complete": {"running", "archived"},
                    "failed": {"running", "archived"},
                    "stopped": {"running", "archived"},
                    "archived": {"pending", "running"},
                }
                if status not in valid_targets.get(current_status, set()):
                    raise ValueError(
                        f"Invalid run status transition from '{current_status}' to '{status}'"
                    )

                # Log status transition event
                event = RunEventRecord(
                    run_id=run_id,
                    event_type="status_transition",
                    from_status=current_status,
                    to_status=status,
                    message=f"Run status transitioned from '{current_status}' to '{status}'",
                )
                session.add(event)

                # Log audit event
                audit = AuditEventRecord(
                    actor="system",
                    action="status_transition",
                    entity_type="run",
                    entity_id=run_id,
                    details_json=json.dumps({"from": current_status, "to": status}),
                )
                session.add(audit)

            record.status = status  # type: ignore[assignment]
            if pid is not None:
                record.pid = pid  # type: ignore[assignment]
            if status == "running" and record.started_at is None:
                record.started_at = datetime.now(UTC).replace(tzinfo=None)
            if status in ("complete", "failed", "stopped", "archived"):
                record.finished_at = datetime.now(UTC).replace(tzinfo=None)  # type: ignore[assignment]
            session.commit()

    def list_runs(
        self,
        status: RunStatus | None = None,
        limit: int = 100,
        offset: int = 0,
    ) -> list[RunRecord]:
        """List runs, optionally filtered by status.

        Args:
            status: Filter by this status if provided.
            limit: Maximum number of results.
            offset: Number of results to skip before returning rows.

        Returns:
            List of RunRecord objects ordered by created_at desc.
        """
        with self._Session() as session:
            stmt = (
                select(RunRecord)
                .order_by(RunRecord.created_at.desc())
                .offset(max(offset, 0))
                .limit(max(limit, 1))
            )
            if status is not None:
                stmt = stmt.where(RunRecord.status == status)
            return list(session.execute(stmt).scalars().all())

    def delete_run(self, run_id: str) -> bool:
        """Delete a run record.

        Args:
            run_id: The run identifier.

        Returns:
            True if deleted, False if not found.
        """
        with self._Session() as session:
            stmt = select(RunRecord).where(RunRecord.id == run_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return False
            session.delete(record)
            session.commit()
            return True

    def update_tags(self, run_id: str, tags: list[str]) -> bool:
        """Replace a run's tag list."""
        with self._Session() as session:
            stmt = select(RunRecord).where(RunRecord.id == run_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return False
            record.tags_json = json.dumps(tags)  # type: ignore[assignment]
            session.commit()
            return True

    def update_extra(self, run_id: str, extra: dict[str, Any]) -> bool:
        """Merge extra metadata into a run record."""
        with self._Session() as session:
            stmt = select(RunRecord).where(RunRecord.id == run_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return False
            current = json.loads(str(record.extra_json or "{}"))
            if not isinstance(current, dict):
                current = {}
            current.update(extra)
            record.extra_json = json.dumps(current)  # type: ignore[assignment]
            session.commit()
            return True

    def create_operation(
        self,
        op_id: str,
        tool: str,
        arguments: dict[str, Any],
        idempotency_key: str | None = None,
    ) -> OperationRecord:
        """Insert a new operation record."""
        with self._Session() as session:
            record = OperationRecord(
                id=op_id,
                tool=tool,
                arguments_json=json.dumps(arguments),
                idempotency_key=idempotency_key,
                status="pending",
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.add(record)
            session.commit()
            session.refresh(record)
            return record

    def get_operation(self, op_id: str) -> OperationRecord | None:
        """Fetch an operation by ID."""
        with self._Session() as session:
            stmt = select(OperationRecord).where(OperationRecord.id == op_id)
            return session.execute(stmt).scalar_one_or_none()

    def get_operation_by_idempotency_key(self, idempotency_key: str) -> OperationRecord | None:
        """Fetch an operation by idempotency key."""
        with self._Session() as session:
            stmt = select(OperationRecord).where(OperationRecord.idempotency_key == idempotency_key)
            return session.execute(stmt).scalar_one_or_none()

    def update_operation_status(
        self,
        op_id: str,
        status: str,
        error_type: str | None = None,
        error: str | None = None,
        result: dict[str, Any] | None = None,
        run_id: str | None = None,
    ) -> None:
        """Update operation status, timestamps, results/errors."""
        with self._Session() as session:
            stmt = select(OperationRecord).where(OperationRecord.id == op_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return
            record.status = status  # type: ignore[assignment]
            if status == "running" and record.started_at is None:
                record.started_at = datetime.now(UTC).replace(tzinfo=None)
            elif status in ("succeeded", "failed", "cancelled"):
                record.finished_at = datetime.now(UTC).replace(tzinfo=None)  # type: ignore[assignment]
            if error_type is not None:
                record.error_type = error_type  # type: ignore[assignment]
            if error is not None:
                record.error = error  # type: ignore[assignment]
            if result is not None:
                record.result_json = json.dumps(result)  # type: ignore[assignment]
            if run_id is not None:
                record.run_id = run_id  # type: ignore[assignment]
            session.commit()

    def update_operation_progress(self, op_id: str, progress: int) -> None:
        """Update progress percentage of an operation."""
        with self._Session() as session:
            stmt = select(OperationRecord).where(OperationRecord.id == op_id)
            record = session.execute(stmt).scalar_one_or_none()
            if record is None:
                return
            record.progress = progress  # type: ignore[assignment]
            session.commit()

    def list_operations(self, limit: int = 100) -> list[OperationRecord]:
        """List operations ordered by created_at desc."""
        with self._Session() as session:
            stmt = select(OperationRecord).order_by(OperationRecord.created_at.desc()).limit(limit)
            return list(session.execute(stmt).scalars().all())

    def create_operation_event(
        self,
        op_id: str,
        event_type: str,
        data: dict[str, Any],
    ) -> OperationEventRecord:
        """Create and persist a lifecycle event for an operation."""
        with self._Session() as session:
            event_rec = OperationEventRecord(
                operation_id=op_id,
                event_type=event_type,
                data_json=json.dumps(data),
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.add(event_rec)
            session.commit()
            session.refresh(event_rec)
            return event_rec

    def list_operation_events(
        self,
        last_event_id: int | None = None,
        limit: int = 1000,
    ) -> list[OperationEventRecord]:
        """List operation events newer than last_event_id."""
        with self._Session() as session:
            stmt = select(OperationEventRecord).order_by(OperationEventRecord.id.asc()).limit(limit)
            if last_event_id is not None:
                stmt = stmt.where(OperationEventRecord.id > last_event_id)
            return list(session.execute(stmt).scalars().all())

    def add_artifact(
        self,
        artifact_id: str,
        run_id: str | None,
        artifact_type: str,
        path: Path,
        sha256: str | None = None,
        size: int | None = None,
        extra: dict[str, Any] | None = None,
    ) -> ArtifactRecord:
        """Register or merge a new artifact record."""
        import hashlib

        path_str = str(path.resolve())
        if path.exists() and path.is_file():
            if size is None:
                size = path.stat().st_size
            if sha256 is None:
                h = hashlib.sha256()
                try:
                    with path.open("rb") as f:
                        for chunk in iter(lambda: f.read(65536), b""):
                            h.update(chunk)
                    sha256 = h.hexdigest()
                except Exception:
                    sha256 = None

        with self._Session() as session:
            record = ArtifactRecord(
                id=artifact_id,
                run_id=run_id,
                type=artifact_type,
                path=path_str,
                sha256=sha256,
                size=size,
                extra_json=json.dumps(extra or {}),
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.merge(record)

            # Log audit event
            audit = AuditEventRecord(
                actor="system",
                action="add_artifact",
                entity_type="artifact",
                entity_id=artifact_id,
                details_json=json.dumps({"type": artifact_type, "path": path_str}),
            )
            session.add(audit)
            session.commit()
            return record

    def record_export(
        self,
        export_id: str,
        run_id: str | None,
        source_checkpoint: Path,
        artifact_path: Path,
        format: str,
        duration_s: float | None = None,
        validation_result: dict[str, Any] | None = None,
    ) -> ExportRecord:
        """Record a model export and associate it with a corresponding artifact."""
        with self._Session() as session:
            record = ExportRecord(
                id=export_id,
                run_id=run_id,
                source_checkpoint=str(source_checkpoint.resolve()),
                artifact_path=str(artifact_path.resolve()),
                format=format,
                duration_s=duration_s,
                validation_result_json=json.dumps(validation_result or {}),
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.add(record)
            session.commit()

        # Also register the export file as an artifact
        self.add_artifact(
            artifact_id=f"art_{export_id}",
            run_id=run_id,
            artifact_type="export",
            path=artifact_path,
            extra={"format": format, "duration_s": duration_s},
        )
        return record

    def list_run_events(self, run_id: str | None = None, limit: int = 1000) -> list[RunEventRecord]:
        """List run lifecycle and audit events."""
        with self._Session() as session:
            stmt = select(RunEventRecord).order_by(RunEventRecord.created_at.desc()).limit(limit)
            if run_id is not None:
                stmt = stmt.where(RunEventRecord.run_id == run_id)
            return list(session.execute(stmt).scalars().all())

    def log_audit_event(
        self,
        actor: str,
        action: str,
        entity_type: str,
        entity_id: str,
        details: dict[str, Any],
    ) -> AuditEventRecord:
        """Log an audit event to the database."""
        with self._Session() as session:
            record = AuditEventRecord(
                actor=actor,
                action=action,
                entity_type=entity_type,
                entity_id=entity_id,
                details_json=json.dumps(details),
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.add(record)
            session.commit()
            session.refresh(record)
            return record

    def list_audit_events(self, limit: int = 100, offset: int = 0) -> list[AuditEventRecord]:
        """List audit events ordered by created_at desc."""
        with self._Session() as session:
            stmt = (
                select(AuditEventRecord)
                .order_by(AuditEventRecord.created_at.desc())
                .offset(max(offset, 0))
                .limit(max(limit, 1))
            )
            return list(session.execute(stmt).scalars().all())

    def list_artifacts(self, run_id: str | None = None, limit: int = 1000) -> list[ArtifactRecord]:
        """List registered artifacts."""
        with self._Session() as session:
            stmt = select(ArtifactRecord).order_by(ArtifactRecord.created_at.desc()).limit(limit)
            if run_id is not None:
                stmt = stmt.where(ArtifactRecord.run_id == run_id)
            return list(session.execute(stmt).scalars().all())

    def get_dataset(self, fingerprint: str) -> DatasetRecord | None:
        """Fetch a dataset record by fingerprint."""
        with self._Session() as session:
            return session.get(DatasetRecord, fingerprint)

    def list_datasets(self, limit: int = 100) -> list[DatasetRecord]:
        """List registered datasets."""
        with self._Session() as session:
            stmt = select(DatasetRecord).order_by(DatasetRecord.created_at.desc()).limit(limit)
            return list(session.execute(stmt).scalars().all())

    def list_exports(self, run_id: str | None = None, limit: int = 100) -> list[ExportRecord]:
        """List recorded exports."""
        with self._Session() as session:
            stmt = select(ExportRecord).order_by(ExportRecord.created_at.desc()).limit(limit)
            if run_id is not None:
                stmt = stmt.where(ExportRecord.run_id == run_id)
            return list(session.execute(stmt).scalars().all())

    def add_metric(self, run_id: str, epoch: int, key: str, value: float) -> MetricRecord:
        """Record an epoch-level training metric."""
        with self._Session() as session:
            record = MetricRecord(
                run_id=run_id,
                epoch=epoch,
                metric_key=key,
                metric_value=value,
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.add(record)
            session.commit()
            return record

    def list_metrics(self, run_id: str, limit: int = 1000) -> list[MetricRecord]:
        """List metrics for a run."""
        with self._Session() as session:
            stmt = (
                select(MetricRecord)
                .where(MetricRecord.run_id == run_id)
                .order_by(MetricRecord.epoch.asc(), MetricRecord.created_at.asc())
                .limit(limit)
            )
            return list(session.execute(stmt).scalars().all())

    def add_review_queue_entry(
        self,
        entry_id: str,
        image_path: Path,
        dataset_path: Path,
        score: float,
        reason: str,
        predictions: list[dict[str, Any]],
    ) -> ReviewQueueEntry:
        """Add or update an active learning review queue entry."""
        with self._Session() as session:
            record = ReviewQueueEntry(
                id=entry_id,
                image_path=str(image_path.resolve()),
                dataset_path=str(dataset_path.resolve()),
                score=score,
                reason=reason,
                status="pending",
                predictions_json=json.dumps(predictions),
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
            session.merge(record)
            session.commit()
            return record

    def get_review_queue_entry(self, entry_id: str) -> ReviewQueueEntry | None:
        """Fetch a review queue entry by ID."""
        with self._Session() as session:
            return session.get(ReviewQueueEntry, entry_id)

    def list_review_queue_entries(
        self,
        dataset_path: Path | None = None,
        status: str = "pending",
        limit: int = 100,
    ) -> list[ReviewQueueEntry]:
        """List active learning review queue entries ordered by score desc."""
        with self._Session() as session:
            stmt = (
                select(ReviewQueueEntry)
                .where(ReviewQueueEntry.status == status)
                .order_by(ReviewQueueEntry.score.desc())
                .limit(limit)
            )
            if dataset_path is not None:
                stmt = stmt.where(ReviewQueueEntry.dataset_path == str(dataset_path.resolve()))
            return list(session.execute(stmt).scalars().all())

    def update_review_queue_status(
        self,
        entry_id: str,
        status: str,
        corrected_labels: list[dict[str, Any]] | None = None,
    ) -> bool:
        """Update review queue entry status and corrections."""
        with self._Session() as session:
            record = session.get(ReviewQueueEntry, entry_id)
            if record is None:
                return False
            record.status = status
            if corrected_labels is not None:
                record.corrected_labels_json = json.dumps(corrected_labels)
            session.commit()
            return True

__init__(db_path)

Initialize and create tables if needed.

Source code in src/fovux/core/runs.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def __init__(self, db_path: Path) -> None:
    """Initialize and create tables if needed."""
    self._engine = create_engine(
        f"sqlite:///{db_path}",
        echo=False,
        poolclass=NullPool,
        connect_args={"check_same_thread": False},
    )

    @sa_event.listens_for(self._engine, "connect")
    def _set_sqlite_pragmas(
        dbapi_conn: sqlite3.Connection,
        _connection_record: object,
    ) -> None:
        dbapi_conn.execute("PRAGMA journal_mode=WAL")
        dbapi_conn.execute("PRAGMA synchronous=NORMAL")
        dbapi_conn.execute("PRAGMA foreign_keys=ON")

    Base.metadata.create_all(self._engine)
    self._run_migrations()
    self._Session = sessionmaker(bind=self._engine, expire_on_commit=False)

add_artifact(artifact_id, run_id, artifact_type, path, sha256=None, size=None, extra=None)

Register or merge a new artifact record.

Source code in src/fovux/core/runs.py 
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
def add_artifact(
    self,
    artifact_id: str,
    run_id: str | None,
    artifact_type: str,
    path: Path,
    sha256: str | None = None,
    size: int | None = None,
    extra: dict[str, Any] | None = None,
) -> ArtifactRecord:
    """Register or merge a new artifact record."""
    import hashlib

    path_str = str(path.resolve())
    if path.exists() and path.is_file():
        if size is None:
            size = path.stat().st_size
        if sha256 is None:
            h = hashlib.sha256()
            try:
                with path.open("rb") as f:
                    for chunk in iter(lambda: f.read(65536), b""):
                        h.update(chunk)
                sha256 = h.hexdigest()
            except Exception:
                sha256 = None

    with self._Session() as session:
        record = ArtifactRecord(
            id=artifact_id,
            run_id=run_id,
            type=artifact_type,
            path=path_str,
            sha256=sha256,
            size=size,
            extra_json=json.dumps(extra or {}),
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.merge(record)

        # Log audit event
        audit = AuditEventRecord(
            actor="system",
            action="add_artifact",
            entity_type="artifact",
            entity_id=artifact_id,
            details_json=json.dumps({"type": artifact_type, "path": path_str}),
        )
        session.add(audit)
        session.commit()
        return record

add_metric(run_id, epoch, key, value)

Record an epoch-level training metric.

Source code in src/fovux/core/runs.py 
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
def add_metric(self, run_id: str, epoch: int, key: str, value: float) -> MetricRecord:
    """Record an epoch-level training metric."""
    with self._Session() as session:
        record = MetricRecord(
            run_id=run_id,
            epoch=epoch,
            metric_key=key,
            metric_value=value,
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.add(record)
        session.commit()
        return record

add_review_queue_entry(entry_id, image_path, dataset_path, score, reason, predictions)

Add or update an active learning review queue entry.

Source code in src/fovux/core/runs.py 
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
def add_review_queue_entry(
    self,
    entry_id: str,
    image_path: Path,
    dataset_path: Path,
    score: float,
    reason: str,
    predictions: list[dict[str, Any]],
) -> ReviewQueueEntry:
    """Add or update an active learning review queue entry."""
    with self._Session() as session:
        record = ReviewQueueEntry(
            id=entry_id,
            image_path=str(image_path.resolve()),
            dataset_path=str(dataset_path.resolve()),
            score=score,
            reason=reason,
            status="pending",
            predictions_json=json.dumps(predictions),
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.merge(record)
        session.commit()
        return record

close()

Dispose the SQLite engine and release pooled connections.

Source code in src/fovux/core/runs.py 
354
355
356
def close(self) -> None:
    """Dispose the SQLite engine and release pooled connections."""
    self._engine.dispose()

create_operation(op_id, tool, arguments, idempotency_key=None)

Insert a new operation record.

Source code in src/fovux/core/runs.py 
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
def create_operation(
    self,
    op_id: str,
    tool: str,
    arguments: dict[str, Any],
    idempotency_key: str | None = None,
) -> OperationRecord:
    """Insert a new operation record."""
    with self._Session() as session:
        record = OperationRecord(
            id=op_id,
            tool=tool,
            arguments_json=json.dumps(arguments),
            idempotency_key=idempotency_key,
            status="pending",
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.add(record)
        session.commit()
        session.refresh(record)
        return record

create_operation_event(op_id, event_type, data)

Create and persist a lifecycle event for an operation.

Source code in src/fovux/core/runs.py 
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
def create_operation_event(
    self,
    op_id: str,
    event_type: str,
    data: dict[str, Any],
) -> OperationEventRecord:
    """Create and persist a lifecycle event for an operation."""
    with self._Session() as session:
        event_rec = OperationEventRecord(
            operation_id=op_id,
            event_type=event_type,
            data_json=json.dumps(data),
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.add(event_rec)
        session.commit()
        session.refresh(event_rec)
        return event_rec

create_run(run_id, run_path, model, dataset_path, task, epochs, tags=None, extra=None, dataset_fingerprint=None, config_hash=None, code_version=None, env_summary=None, parent_run_id=None)

Insert a new run record.

Parameters:

Name Type Description Default
run_id str

Unique run identifier.

 required
run_path Path

Path to the run directory.

 required
model str

Model name or path.

 required
dataset_path Path

Path to the dataset.

 required
task str

YOLO task (detect, segment, classify, pose, obb).

 required
epochs int

Total training epochs.

 required
tags list[str] | None

Optional list of user tags.

 None
extra dict[str, Any] | None

Optional extra metadata dict.

 None
dataset_fingerprint str | None

Optional fingerprint of dataset.

 None
config_hash str | None

Optional hash of config.

 None
code_version str | None

Optional code version.

 None
env_summary str | None

Optional system environment summary.

 None
parent_run_id str | None

Optional parent run/checkpoint reference.

 None

Returns:

Type Description
RunRecord

The newly created RunRecord.
Source code in src/fovux/core/runs.py 
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
def create_run(
    self,
    run_id: str,
    run_path: Path,
    model: str,
    dataset_path: Path,
    task: str,
    epochs: int,
    tags: list[str] | None = None,
    extra: dict[str, Any] | None = None,
    dataset_fingerprint: str | None = None,
    config_hash: str | None = None,
    code_version: str | None = None,
    env_summary: str | None = None,
    parent_run_id: str | None = None,
) -> RunRecord:
    """Insert a new run record.

    Args:
        run_id: Unique run identifier.
        run_path: Path to the run directory.
        model: Model name or path.
        dataset_path: Path to the dataset.
        task: YOLO task (detect, segment, classify, pose, obb).
        epochs: Total training epochs.
        tags: Optional list of user tags.
        extra: Optional extra metadata dict.
        dataset_fingerprint: Optional fingerprint of dataset.
        config_hash: Optional hash of config.
        code_version: Optional code version.
        env_summary: Optional system environment summary.
        parent_run_id: Optional parent run/checkpoint reference.

    Returns:
        The newly created RunRecord.
    """
    auto_fp, auto_ch, auto_cv, auto_env = self._auto_metadata(
        model, dataset_path, task, epochs, extra
    )
    dataset_fingerprint = dataset_fingerprint or auto_fp
    config_hash = config_hash or auto_ch
    code_version = code_version or auto_cv
    env_summary = env_summary or auto_env

    with self._Session() as session:
        with session.begin():
            record = RunRecord(
                id=run_id,
                run_path=str(run_path),
                model=model,
                dataset_path=str(dataset_path),
                task=task,
                epochs=epochs,
                status="pending",
                created_at=datetime.now(UTC).replace(tzinfo=None),
                tags_json=json.dumps(tags or []),
                extra_json=json.dumps(extra or {}),
                dataset_fingerprint=dataset_fingerprint,
                config_hash=config_hash,
                code_version=code_version,
                env_summary=env_summary,
                parent_run_id=parent_run_id,
            )
            session.add(record)
            self._register_lineage(
                session, run_id, model, dataset_path, task, dataset_fingerprint
            )
        session.refresh(record)
        return record

delete_run(run_id)

Delete a run record.

Parameters:

Name Type Description Default
run_id str

The run identifier.

 required

Returns:

Type Description
bool

True if deleted, False if not found.
Source code in src/fovux/core/runs.py 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
def delete_run(self, run_id: str) -> bool:
    """Delete a run record.

    Args:
        run_id: The run identifier.

    Returns:
        True if deleted, False if not found.
    """
    with self._Session() as session:
        stmt = select(RunRecord).where(RunRecord.id == run_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return False
        session.delete(record)
        session.commit()
        return True

get_dataset(fingerprint)

Fetch a dataset record by fingerprint.

Source code in src/fovux/core/runs.py 
 999
1000
1001
1002
def get_dataset(self, fingerprint: str) -> DatasetRecord | None:
    """Fetch a dataset record by fingerprint."""
    with self._Session() as session:
        return session.get(DatasetRecord, fingerprint)

get_operation(op_id)

Fetch an operation by ID.

Source code in src/fovux/core/runs.py 
773
774
775
776
777
def get_operation(self, op_id: str) -> OperationRecord | None:
    """Fetch an operation by ID."""
    with self._Session() as session:
        stmt = select(OperationRecord).where(OperationRecord.id == op_id)
        return session.execute(stmt).scalar_one_or_none()

get_operation_by_idempotency_key(idempotency_key)

Fetch an operation by idempotency key.

Source code in src/fovux/core/runs.py 
779
780
781
782
783
def get_operation_by_idempotency_key(self, idempotency_key: str) -> OperationRecord | None:
    """Fetch an operation by idempotency key."""
    with self._Session() as session:
        stmt = select(OperationRecord).where(OperationRecord.idempotency_key == idempotency_key)
        return session.execute(stmt).scalar_one_or_none()

get_review_queue_entry(entry_id)

Fetch a review queue entry by ID.

Source code in src/fovux/core/runs.py 
1068
1069
1070
1071
def get_review_queue_entry(self, entry_id: str) -> ReviewQueueEntry | None:
    """Fetch a review queue entry by ID."""
    with self._Session() as session:
        return session.get(ReviewQueueEntry, entry_id)

get_run(run_id)

Fetch a run by ID.

Parameters:

Name Type Description Default
run_id str

The run identifier.

 required

Returns:

Type Description
RunRecord | None

RunRecord or None if not found.
Source code in src/fovux/core/runs.py 
604
605
606
607
608
609
610
611
612
613
614
615
def get_run(self, run_id: str) -> RunRecord | None:
    """Fetch a run by ID.

    Args:
        run_id: The run identifier.

    Returns:
        RunRecord or None if not found.
    """
    with self._Session() as session:
        stmt = select(RunRecord).where(RunRecord.id == run_id)
        return session.execute(stmt).scalar_one_or_none()

list_artifacts(run_id=None, limit=1000)

List registered artifacts.

Source code in src/fovux/core/runs.py 
991
992
993
994
995
996
997
def list_artifacts(self, run_id: str | None = None, limit: int = 1000) -> list[ArtifactRecord]:
    """List registered artifacts."""
    with self._Session() as session:
        stmt = select(ArtifactRecord).order_by(ArtifactRecord.created_at.desc()).limit(limit)
        if run_id is not None:
            stmt = stmt.where(ArtifactRecord.run_id == run_id)
        return list(session.execute(stmt).scalars().all())

list_audit_events(limit=100, offset=0)

List audit events ordered by created_at desc.

Source code in src/fovux/core/runs.py 
980
981
982
983
984
985
986
987
988
989
def list_audit_events(self, limit: int = 100, offset: int = 0) -> list[AuditEventRecord]:
    """List audit events ordered by created_at desc."""
    with self._Session() as session:
        stmt = (
            select(AuditEventRecord)
            .order_by(AuditEventRecord.created_at.desc())
            .offset(max(offset, 0))
            .limit(max(limit, 1))
        )
        return list(session.execute(stmt).scalars().all())

list_datasets(limit=100)

List registered datasets.

Source code in src/fovux/core/runs.py 
1004
1005
1006
1007
1008
def list_datasets(self, limit: int = 100) -> list[DatasetRecord]:
    """List registered datasets."""
    with self._Session() as session:
        stmt = select(DatasetRecord).order_by(DatasetRecord.created_at.desc()).limit(limit)
        return list(session.execute(stmt).scalars().all())

list_exports(run_id=None, limit=100)

List recorded exports.

Source code in src/fovux/core/runs.py 
1010
1011
1012
1013
1014
1015
1016
def list_exports(self, run_id: str | None = None, limit: int = 100) -> list[ExportRecord]:
    """List recorded exports."""
    with self._Session() as session:
        stmt = select(ExportRecord).order_by(ExportRecord.created_at.desc()).limit(limit)
        if run_id is not None:
            stmt = stmt.where(ExportRecord.run_id == run_id)
        return list(session.execute(stmt).scalars().all())

list_metrics(run_id, limit=1000)

List metrics for a run.

Source code in src/fovux/core/runs.py 
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
def list_metrics(self, run_id: str, limit: int = 1000) -> list[MetricRecord]:
    """List metrics for a run."""
    with self._Session() as session:
        stmt = (
            select(MetricRecord)
            .where(MetricRecord.run_id == run_id)
            .order_by(MetricRecord.epoch.asc(), MetricRecord.created_at.asc())
            .limit(limit)
        )
        return list(session.execute(stmt).scalars().all())

list_operation_events(last_event_id=None, limit=1000)

List operation events newer than last_event_id.

Source code in src/fovux/core/runs.py 
850
851
852
853
854
855
856
857
858
859
860
def list_operation_events(
    self,
    last_event_id: int | None = None,
    limit: int = 1000,
) -> list[OperationEventRecord]:
    """List operation events newer than last_event_id."""
    with self._Session() as session:
        stmt = select(OperationEventRecord).order_by(OperationEventRecord.id.asc()).limit(limit)
        if last_event_id is not None:
            stmt = stmt.where(OperationEventRecord.id > last_event_id)
        return list(session.execute(stmt).scalars().all())

list_operations(limit=100)

List operations ordered by created_at desc.

Source code in src/fovux/core/runs.py 
825
826
827
828
829
def list_operations(self, limit: int = 100) -> list[OperationRecord]:
    """List operations ordered by created_at desc."""
    with self._Session() as session:
        stmt = select(OperationRecord).order_by(OperationRecord.created_at.desc()).limit(limit)
        return list(session.execute(stmt).scalars().all())

list_review_queue_entries(dataset_path=None, status='pending', limit=100)

List active learning review queue entries ordered by score desc.

Source code in src/fovux/core/runs.py 
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
def list_review_queue_entries(
    self,
    dataset_path: Path | None = None,
    status: str = "pending",
    limit: int = 100,
) -> list[ReviewQueueEntry]:
    """List active learning review queue entries ordered by score desc."""
    with self._Session() as session:
        stmt = (
            select(ReviewQueueEntry)
            .where(ReviewQueueEntry.status == status)
            .order_by(ReviewQueueEntry.score.desc())
            .limit(limit)
        )
        if dataset_path is not None:
            stmt = stmt.where(ReviewQueueEntry.dataset_path == str(dataset_path.resolve()))
        return list(session.execute(stmt).scalars().all())

list_run_events(run_id=None, limit=1000)

List run lifecycle and audit events.

Source code in src/fovux/core/runs.py 
949
950
951
952
953
954
955
def list_run_events(self, run_id: str | None = None, limit: int = 1000) -> list[RunEventRecord]:
    """List run lifecycle and audit events."""
    with self._Session() as session:
        stmt = select(RunEventRecord).order_by(RunEventRecord.created_at.desc()).limit(limit)
        if run_id is not None:
            stmt = stmt.where(RunEventRecord.run_id == run_id)
        return list(session.execute(stmt).scalars().all())

list_runs(status=None, limit=100, offset=0)

List runs, optionally filtered by status.

Parameters:

Name Type Description Default
status RunStatus | None

Filter by this status if provided.

 None
limit int

Maximum number of results.

 100
offset int

Number of results to skip before returning rows.

 0

Returns:

Type Description
list[RunRecord]

List of RunRecord objects ordered by created_at desc.
Source code in src/fovux/core/runs.py 
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
def list_runs(
    self,
    status: RunStatus | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[RunRecord]:
    """List runs, optionally filtered by status.

    Args:
        status: Filter by this status if provided.
        limit: Maximum number of results.
        offset: Number of results to skip before returning rows.

    Returns:
        List of RunRecord objects ordered by created_at desc.
    """
    with self._Session() as session:
        stmt = (
            select(RunRecord)
            .order_by(RunRecord.created_at.desc())
            .offset(max(offset, 0))
            .limit(max(limit, 1))
        )
        if status is not None:
            stmt = stmt.where(RunRecord.status == status)
        return list(session.execute(stmt).scalars().all())

log_audit_event(actor, action, entity_type, entity_id, details)

Log an audit event to the database.

Source code in src/fovux/core/runs.py 
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
def log_audit_event(
    self,
    actor: str,
    action: str,
    entity_type: str,
    entity_id: str,
    details: dict[str, Any],
) -> AuditEventRecord:
    """Log an audit event to the database."""
    with self._Session() as session:
        record = AuditEventRecord(
            actor=actor,
            action=action,
            entity_type=entity_type,
            entity_id=entity_id,
            details_json=json.dumps(details),
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.add(record)
        session.commit()
        session.refresh(record)
        return record

record_export(export_id, run_id, source_checkpoint, artifact_path, format, duration_s=None, validation_result=None)

Record a model export and associate it with a corresponding artifact.

Source code in src/fovux/core/runs.py 
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
def record_export(
    self,
    export_id: str,
    run_id: str | None,
    source_checkpoint: Path,
    artifact_path: Path,
    format: str,
    duration_s: float | None = None,
    validation_result: dict[str, Any] | None = None,
) -> ExportRecord:
    """Record a model export and associate it with a corresponding artifact."""
    with self._Session() as session:
        record = ExportRecord(
            id=export_id,
            run_id=run_id,
            source_checkpoint=str(source_checkpoint.resolve()),
            artifact_path=str(artifact_path.resolve()),
            format=format,
            duration_s=duration_s,
            validation_result_json=json.dumps(validation_result or {}),
            created_at=datetime.now(UTC).replace(tzinfo=None),
        )
        session.add(record)
        session.commit()

    # Also register the export file as an artifact
    self.add_artifact(
        artifact_id=f"art_{export_id}",
        run_id=run_id,
        artifact_type="export",
        path=artifact_path,
        extra={"format": format, "duration_s": duration_s},
    )
    return record

reserve_run_slot(run_id, run_path, model, dataset_path, task, epochs, max_concurrent_runs, tags=None, extra=None, dataset_fingerprint=None, config_hash=None, code_version=None, env_summary=None, parent_run_id=None)

Reserve a run slot atomically, locking the DB if concurrent limit is set.

Source code in src/fovux/core/runs.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
def reserve_run_slot(
    self,
    run_id: str,
    run_path: Path,
    model: str,
    dataset_path: Path,
    task: str,
    epochs: int,
    max_concurrent_runs: int,
    tags: list[str] | None = None,
    extra: dict[str, Any] | None = None,
    dataset_fingerprint: str | None = None,
    config_hash: str | None = None,
    code_version: str | None = None,
    env_summary: str | None = None,
    parent_run_id: str | None = None,
) -> RunRecord:
    """Reserve a run slot atomically, locking the DB if concurrent limit is set."""
    auto_fp, auto_ch, auto_cv, auto_env = self._auto_metadata(
        model, dataset_path, task, epochs, extra
    )
    dataset_fingerprint = dataset_fingerprint or auto_fp
    config_hash = config_hash or auto_ch
    code_version = code_version or auto_cv
    env_summary = env_summary or auto_env

    with self._Session() as session:
        with session.begin():
            if max_concurrent_runs > 0:
                active_count = (
                    session.query(RunRecord)
                    .filter(RunRecord.status.in_(["running", "pending"]))
                    .count()
                )
                if active_count >= max_concurrent_runs:
                    from fovux.core.errors import FovuxTrainingAlreadyRunningError

                    raise FovuxTrainingAlreadyRunningError(
                        f"Cannot start run '{run_id}': {active_count} "
                        f"concurrent training run(s) already active and "
                        f"max_concurrent_runs={max_concurrent_runs}."
                    )

            record = RunRecord(
                id=run_id,
                run_path=str(run_path),
                model=model,
                dataset_path=str(dataset_path),
                task=task,
                epochs=epochs,
                status="pending",
                created_at=datetime.now(UTC).replace(tzinfo=None),
                tags_json=json.dumps(tags or []),
                extra_json=json.dumps(extra or {}),
                dataset_fingerprint=dataset_fingerprint,
                config_hash=config_hash,
                code_version=code_version,
                env_summary=env_summary,
                parent_run_id=parent_run_id,
            )
            session.add(record)
            self._register_lineage(
                session, run_id, model, dataset_path, task, dataset_fingerprint
            )
        session.refresh(record)
        return record

update_extra(run_id, extra)

Merge extra metadata into a run record.

Source code in src/fovux/core/runs.py 
736
737
738
739
740
741
742
743
744
745
746
747
748
749
def update_extra(self, run_id: str, extra: dict[str, Any]) -> bool:
    """Merge extra metadata into a run record."""
    with self._Session() as session:
        stmt = select(RunRecord).where(RunRecord.id == run_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return False
        current = json.loads(str(record.extra_json or "{}"))
        if not isinstance(current, dict):
            current = {}
        current.update(extra)
        record.extra_json = json.dumps(current)  # type: ignore[assignment]
        session.commit()
        return True

update_operation_progress(op_id, progress)

Update progress percentage of an operation.

Source code in src/fovux/core/runs.py 
815
816
817
818
819
820
821
822
823
def update_operation_progress(self, op_id: str, progress: int) -> None:
    """Update progress percentage of an operation."""
    with self._Session() as session:
        stmt = select(OperationRecord).where(OperationRecord.id == op_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return
        record.progress = progress  # type: ignore[assignment]
        session.commit()

update_operation_status(op_id, status, error_type=None, error=None, result=None, run_id=None)

Update operation status, timestamps, results/errors.

Source code in src/fovux/core/runs.py 
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
def update_operation_status(
    self,
    op_id: str,
    status: str,
    error_type: str | None = None,
    error: str | None = None,
    result: dict[str, Any] | None = None,
    run_id: str | None = None,
) -> None:
    """Update operation status, timestamps, results/errors."""
    with self._Session() as session:
        stmt = select(OperationRecord).where(OperationRecord.id == op_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return
        record.status = status  # type: ignore[assignment]
        if status == "running" and record.started_at is None:
            record.started_at = datetime.now(UTC).replace(tzinfo=None)
        elif status in ("succeeded", "failed", "cancelled"):
            record.finished_at = datetime.now(UTC).replace(tzinfo=None)  # type: ignore[assignment]
        if error_type is not None:
            record.error_type = error_type  # type: ignore[assignment]
        if error is not None:
            record.error = error  # type: ignore[assignment]
        if result is not None:
            record.result_json = json.dumps(result)  # type: ignore[assignment]
        if run_id is not None:
            record.run_id = run_id  # type: ignore[assignment]
        session.commit()

update_review_queue_status(entry_id, status, corrected_labels=None)

Update review queue entry status and corrections.

Source code in src/fovux/core/runs.py 
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
def update_review_queue_status(
    self,
    entry_id: str,
    status: str,
    corrected_labels: list[dict[str, Any]] | None = None,
) -> bool:
    """Update review queue entry status and corrections."""
    with self._Session() as session:
        record = session.get(ReviewQueueEntry, entry_id)
        if record is None:
            return False
        record.status = status
        if corrected_labels is not None:
            record.corrected_labels_json = json.dumps(corrected_labels)
        session.commit()
        return True

update_status(run_id, status, pid=None)

Update run status and optionally pid.

Parameters:

Name Type Description Default
run_id str

The run identifier.

 required
status RunStatus

New status value.

 required
pid int | None

Process ID of the training subprocess (if applicable).

 None
Source code in src/fovux/core/runs.py 
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
def update_status(
    self,
    run_id: str,
    status: RunStatus,
    pid: int | None = None,
) -> None:
    """Update run status and optionally pid.

    Args:
        run_id: The run identifier.
        status: New status value.
        pid: Process ID of the training subprocess (if applicable).
    """
    with self._Session() as session:
        stmt = select(RunRecord).where(RunRecord.id == run_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return

        current_status = str(record.status)
        if current_status != status:
            valid_targets = {
                "pending": {"running", "complete", "failed", "stopped", "archived"},
                "running": {"complete", "failed", "stopped", "archived"},
                "complete": {"running", "archived"},
                "failed": {"running", "archived"},
                "stopped": {"running", "archived"},
                "archived": {"pending", "running"},
            }
            if status not in valid_targets.get(current_status, set()):
                raise ValueError(
                    f"Invalid run status transition from '{current_status}' to '{status}'"
                )

            # Log status transition event
            event = RunEventRecord(
                run_id=run_id,
                event_type="status_transition",
                from_status=current_status,
                to_status=status,
                message=f"Run status transitioned from '{current_status}' to '{status}'",
            )
            session.add(event)

            # Log audit event
            audit = AuditEventRecord(
                actor="system",
                action="status_transition",
                entity_type="run",
                entity_id=run_id,
                details_json=json.dumps({"from": current_status, "to": status}),
            )
            session.add(audit)

        record.status = status  # type: ignore[assignment]
        if pid is not None:
            record.pid = pid  # type: ignore[assignment]
        if status == "running" and record.started_at is None:
            record.started_at = datetime.now(UTC).replace(tzinfo=None)
        if status in ("complete", "failed", "stopped", "archived"):
            record.finished_at = datetime.now(UTC).replace(tzinfo=None)  # type: ignore[assignment]
        session.commit()

update_tags(run_id, tags)

Replace a run's tag list.

Source code in src/fovux/core/runs.py 
725
726
727
728
729
730
731
732
733
734
def update_tags(self, run_id: str, tags: list[str]) -> bool:
    """Replace a run's tag list."""
    with self._Session() as session:
        stmt = select(RunRecord).where(RunRecord.id == run_id)
        record = session.execute(stmt).scalar_one_or_none()
        if record is None:
            return False
        record.tags_json = json.dumps(tags)  # type: ignore[assignment]
        session.commit()
        return True

SchemaMigrationRecord

Bases: Base

ORM model for schema migrations.

Source code in src/fovux/core/runs.py 
113
114
115
116
117
118
119
120
121
122
123
class SchemaMigrationRecord(Base):
    """ORM model for schema migrations."""

    __tablename__ = "schema_migrations"

    version = Column(Integer, primary_key=True)
    applied_at = Column(
        DateTime,
        nullable=False,
        default=lambda: datetime.now(UTC).replace(tzinfo=None),
    )

TagRecord

Bases: Base

ORM model for entity tags.

Source code in src/fovux/core/runs.py 
253
254
255
256
257
258
259
260
261
class TagRecord(Base):
    """ORM model for entity tags."""

    __tablename__ = "tags"

    id = Column(Integer, primary_key=True, autoincrement=True)
    entity_type = Column(String, nullable=False)  # run, artifact, dataset
    entity_id = Column(String, nullable=False)
    tag = Column(String, nullable=False)

close_registry(db_path=None)

Dispose cached registry engines for one database or all databases.

Source code in src/fovux/core/runs.py 
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
def close_registry(db_path: Path | None = None) -> None:
    """Dispose cached registry engines for one database or all databases."""
    if db_path is None:
        with _REGISTRIES_LOCK:
            registries = list(_REGISTRIES.values())
            _REGISTRIES.clear()
        for registry in registries:
            registry.close()
        return

    resolved = db_path.expanduser().resolve()
    with _REGISTRIES_LOCK:
        cached_registry = _REGISTRIES.pop(resolved) if resolved in _REGISTRIES else None
    if cached_registry is not None:
        cached_registry.close()

get_registry(db_path)

Return a process-local singleton registry for a database path.

Source code in src/fovux/core/runs.py 
1109
1110
1111
1112
1113
1114
1115
1116
1117
def get_registry(db_path: Path) -> RunRegistry:
    """Return a process-local singleton registry for a database path."""
    resolved = db_path.expanduser().resolve()
    with _REGISTRIES_LOCK:
        registry = _REGISTRIES.get(resolved)
        if registry is None:
            registry = RunRegistry(resolved)
            _REGISTRIES[resolved] = registry
        return registry

Structured logger backed by structlog for production-grade observability.

Respects

FOVUX_LOG_LEVEL — DEBUG | INFO | WARNING | ERROR (default: INFO) FOVUX_LOG_FORMAT — json | pretty (default: pretty) NO_COLOR — disable ANSI colours when set to any non-empty value

configure_logging(level=None, fmt=None)

Configure structured logging for the Fovux process.

Safe to call multiple times — subsequent calls reconfigure in-place.

Parameters:

Name Type Description Default
level str | None

Override for FOVUX_LOG_LEVEL.

 None
fmt str | None

Override for FOVUX_LOG_FORMAT ("json" or "pretty").

 None
Source code in src/fovux/core/logging.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def configure_logging(
    level: str | None = None,
    fmt: str | None = None,
) -> None:
    """Configure structured logging for the Fovux process.

    Safe to call multiple times — subsequent calls reconfigure in-place.

    Args:
        level: Override for FOVUX_LOG_LEVEL.
        fmt: Override for FOVUX_LOG_FORMAT (``"json"`` or ``"pretty"``).
    """
    log_level_name = (level or os.environ.get("FOVUX_LOG_LEVEL", "INFO")).upper()
    log_fmt = fmt or os.environ.get("FOVUX_LOG_FORMAT", "pretty")
    level_num = getattr(logging, log_level_name, logging.INFO)

    logging.basicConfig(
        level=level_num,
        stream=sys.stderr,
        format="%(message)s",
        force=True,
    )

    try:
        import structlog

        _configure_structlog(structlog, level_num, log_fmt)
    except ImportError:
        pass

get_logger(name)

Return a structured logger for the given module name.

Uses structlog when available, falls back to stdlib logging.Logger.

Parameters:

Name Type Description Default
name str

Typically __name__ of the calling module.

 required

Returns:

Type Description
_Logger

A logger compatible with log.info("event", key=value) style.
Source code in src/fovux/core/logging.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_logger(name: str) -> _Logger:
    """Return a structured logger for the given module name.

    Uses structlog when available, falls back to stdlib ``logging.Logger``.

    Args:
        name: Typically ``__name__`` of the calling module.

    Returns:
        A logger compatible with ``log.info("event", key=value)`` style.
    """
    try:
        import structlog

        return cast(_Logger, structlog.get_logger(name))
    except ImportError:
        return cast(_Logger, logging.getLogger(name))

HTTP route definitions for fovux-studio integration.

Endpoints

GET /runs — list all runs GET /runs/{run_id} — single run metadata GET /runs/{run_id}/stream — canonical SSE stream of metrics.jsonl lines GET /runs/{run_id}/metrics — compatibility SSE stream of metrics.jsonl lines POST /tools/{name} — proxy to MCP tool call

CreateOperationInput

Bases: BaseModel

Input parameters to create a background operation.

Source code in src/fovux/http/routes.py 
845
846
847
848
849
850
851
class CreateOperationInput(BaseModel):
    """Input parameters to create a background operation."""

    tool: str
    arguments: dict[str, Any] = {}
    idempotency_key: str | None = None
    challenge_id: str | None = None

RunsSearchInput

Bases: BaseModel

Input payload for run search filters.

Source code in src/fovux/http/routes.py 
158
159
160
161
162
163
164
165
class RunsSearchInput(BaseModel):
    """Input payload for run search filters."""

    query: str | None = None
    tags: list[str] = []
    status: list[str] = []
    min_map50: float | None = None
    limit: int = 50

cancel_operation_route(id, request) async

Request cancellation of a background operation.

Source code in src/fovux/http/routes.py 
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
@router.post("/operations/{id}/cancel")
async def cancel_operation_route(id: str, request: Request) -> JSONResponse:
    """Request cancellation of a background operation."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    record = registry.get_operation(id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Operation {id} not found.")

    if record.status in ("succeeded", "failed", "cancelled"):
        return JSONResponse(
            status_code=200,
            content=_operation_summary(record),
        )

    tasks = getattr(request.app.state, "active_operation_tasks", {})
    task = tasks.get(id)
    if task is not None:
        task.cancel()

    if record.run_id:
        from fovux.tools.train_stop import train_stop

        try:
            train_stop(run_id=str(record.run_id))
        except Exception:  # noqa: S110
            pass

    registry.update_operation_status(id, "cancelled")
    registry.create_operation_event(id, "status_change", {"status": "cancelled"})
    _notify_sse_listeners(request.app.state, id, "status_change", {"status": "cancelled"})

    record = registry.get_operation(id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Operation {id} not found.")
    return JSONResponse(
        status_code=200,
        content=_operation_summary(record),
    )

create_operation_route(request, body) async

Create a persistent background operation with an optional idempotency key.

Source code in src/fovux/http/routes.py 
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
@router.post("/operations")
async def create_operation_route(
    request: Request,
    body: CreateOperationInput,
) -> JSONResponse:
    """Create a persistent background operation with an optional idempotency key."""
    import uuid

    from fovux.core.auth import ALL_SCOPES, is_known_session_token, resolve_session_scopes
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry
    from fovux.http.challenge import prune_expired_challenges, verify_challenge
    from fovux.http.tool_proxy import check_scope, payload_hash, policy_for_tool

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    auth_header = request.headers.get("Authorization", "")
    raw_token = auth_header.removeprefix("Bearer ").strip()
    scopes = ALL_SCOPES
    if is_known_session_token(raw_token):
        scopes = resolve_session_scopes(raw_token)

    policy = policy_for_tool(body.tool)
    check_scope(policy, scopes)

    if policy.requires_confirmation:
        challenges = cast(dict[str, Any], request.app.state.challenges)
        prune_expired_challenges(challenges)
        challenge_id = body.challenge_id or body.arguments.get("challenge_id")
        if not isinstance(challenge_id, str) or not challenge_id.strip():
            raise HTTPException(
                status_code=403,
                detail={
                    "code": "FOVUX_HTTP_001",
                    "message": f"Tool '{body.tool}' requires a confirmation challenge.",
                    "hint": (
                        "Call POST /tools/{name}/challenge first, then include the challenge_id."
                    ),
                },
            )
        verify_challenge(
            challenges,
            challenge_id=challenge_id,
            tool_name=body.tool,
            args_hash=payload_hash(
                {k: v for k, v in body.arguments.items() if k != "challenge_id"}
            ),
        )

    if body.idempotency_key:
        existing = registry.get_operation_by_idempotency_key(body.idempotency_key)
        if existing is not None:
            return JSONResponse(
                status_code=200,
                content=_operation_summary(existing),
            )

    op_id = f"op_{uuid.uuid4().hex[:12]}"
    record = registry.create_operation(
        op_id=op_id,
        tool=body.tool,
        arguments=body.arguments,
        idempotency_key=body.idempotency_key,
    )

    semaphores = cast(dict[str, asyncio.Semaphore], request.app.state.tool_semaphores)
    semaphore = semaphores[body.tool]

    if not hasattr(request.app.state, "active_operation_tasks"):
        request.app.state.active_operation_tasks = {}

    task = asyncio.create_task(
        _run_operation_in_background(
            op_id=op_id,
            tool_name=body.tool,
            payload=body.arguments,
            semaphore=semaphore,
            app_state=request.app.state,
        )
    )
    request.app.state.active_operation_tasks[op_id] = task

    registry.create_operation_event(op_id, "status_change", {"status": "pending"})

    return JSONResponse(
        status_code=201,
        content=_operation_summary(record),
    )

get_dataset(fingerprint) async

Fetch single dataset record by fingerprint.

Source code in src/fovux/http/routes.py 
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
@router.get("/datasets/{fingerprint}")
async def get_dataset(fingerprint: str) -> JSONResponse:
    """Fetch single dataset record by fingerprint."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    d = registry.get_dataset(fingerprint)
    if d is None:
        raise HTTPException(status_code=404, detail=f"Dataset {fingerprint} not found.")
    return JSONResponse(
        {
            "fingerprint": d.fingerprint,
            "path": d.path,
            "class_map": json.loads(cast(str, d.class_map_json)) if d.class_map_json else {},
            "created_at": d.created_at.isoformat() if d.created_at else None,
        }
    )

get_operation_logs_route(id) async

Fetch or stream execution logs for a background operation.

Source code in src/fovux/http/routes.py 
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
@router.get("/operations/{id}/logs")
async def get_operation_logs_route(id: str) -> StreamingResponse:
    """Fetch or stream execution logs for a background operation."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    record = registry.get_operation(id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Operation {id} not found.")

    log_file = paths.home / "operations" / f"{id}.log"
    if record.run_id:
        run_log = paths.home / "runs" / record.run_id / "stdout.log"
        if run_log.exists():
            log_file = run_log

    async def log_generator() -> AsyncIterator[str]:
        for _ in range(20):
            if log_file.exists():
                break
            await asyncio.sleep(0.1)

        if not log_file.exists():
            yield "Log file not found.\n"
            return

        with open(log_file, encoding="utf-8", errors="replace") as f:
            while True:
                line = f.readline()
                if line:
                    yield line
                else:
                    op = registry.get_operation(id)
                    if op is None or op.status in ("succeeded", "failed", "cancelled"):
                        remaining = f.read()
                        if remaining:
                            yield remaining
                        break
                    await asyncio.sleep(0.5)

    return StreamingResponse(
        log_generator(),
        media_type="text/plain",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

get_operation_result_route(id) async

Fetch the final result of a background operation.

Source code in src/fovux/http/routes.py 
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
@router.get("/operations/{id}/result")
async def get_operation_result_route(id: str) -> JSONResponse:
    """Fetch the final result of a background operation."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    record = registry.get_operation(id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Operation {id} not found.")

    if record.status == "succeeded":
        res = {}
        if record.result_json:
            res = json.loads(str(record.result_json))
        return JSONResponse(res)
    elif record.status == "failed":
        return JSONResponse(
            status_code=500,
            content={
                "operation_id": record.id,
                "status": record.status,
                "error_type": record.error_type,
                "error": record.error,
            },
        )
    elif record.status == "cancelled":
        return JSONResponse(
            status_code=400,
            content={
                "operation_id": record.id,
                "status": record.status,
                "message": "Operation was cancelled.",
            },
        )
    else:
        return JSONResponse(
            status_code=202,
            content={
                "operation_id": record.id,
                "status": record.status,
                "message": "Operation is still running.",
            },
        )

get_operation_route(id) async

Get status and metadata for a background operation.

Source code in src/fovux/http/routes.py 
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
@router.get("/operations/{id}")
async def get_operation_route(id: str) -> JSONResponse:
    """Get status and metadata for a background operation."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    record = registry.get_operation(id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Operation {id} not found.")

    return JSONResponse(_operation_summary(record))

get_run(run_id) async

Get metadata for a single run.

Source code in src/fovux/http/routes.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@router.get("/runs/{run_id}")
async def get_run(run_id: str) -> JSONResponse:
    """Get metadata for a single run."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    record = registry.get_run(run_id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found.")
    run_path = Path(record.run_path)
    status_payload = _read_status_payload(run_path)
    status = str(status_payload.get("status") or record.status)
    current_epoch, best_map50 = read_metrics_summary(run_path)
    raw_tags = cast(str, record.tags_json or "[]")
    try:
        tags = json.loads(raw_tags)
    except Exception:
        tags = []
    return JSONResponse(
        {
            "id": record.id,
            "status": status,
            "model": record.model,
            "dataset_path": record.dataset_path,
            "task": record.task,
            "epochs": record.epochs,
            "pid": record.pid,
            "run_path": record.run_path,
            "current_epoch": current_epoch,
            "best_map50": best_map50,
            "created_at": record.created_at.isoformat() if record.created_at else None,
            "started_at": record.started_at.isoformat() if record.started_at else None,
            "finished_at": record.finished_at.isoformat() if record.finished_at else None,
            "tags": tags,
        }
    )

get_run_events(run_id) async

Fetch all lifecycle and audit events for a single run.

Source code in src/fovux/http/routes.py 
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
@router.get("/runs/{run_id}/events")
async def get_run_events(run_id: str) -> JSONResponse:
    """Fetch all lifecycle and audit events for a single run."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    record = registry.get_run(run_id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found.")

    events = registry.list_run_events(run_id)
    return JSONResponse(
        [
            {
                "id": ev.id,
                "event_type": ev.event_type,
                "from_status": ev.from_status,
                "to_status": ev.to_status,
                "message": ev.message,
                "created_at": ev.created_at.isoformat() if ev.created_at else None,
                "extra": json.loads(cast(str, ev.extra_json)) if ev.extra_json else None,
            }
            for ev in events
        ]
    )

get_run_lineage(run_id) async

Fetch experiment lineage information for a run.

Source code in src/fovux/http/routes.py 
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
@router.get("/runs/{run_id}/lineage")
async def get_run_lineage(run_id: str) -> JSONResponse:
    """Fetch experiment lineage information for a run."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    record = registry.get_run(run_id)
    if record is None:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found.")

    artifacts = registry.list_artifacts(run_id)
    exports = registry.list_exports(run_id)
    events = registry.list_run_events(run_id)

    return JSONResponse(
        {
            "run_id": record.id,
            "dataset_path": record.dataset_path,
            "dataset_fingerprint": record.dataset_fingerprint,
            "config_hash": record.config_hash,
            "code_version": record.code_version,
            "env_summary": (
                json.loads(cast(str, record.env_summary)) if record.env_summary else None
            ),
            "parent_run_id": record.parent_run_id,
            "artifacts": [
                {
                    "id": a.id,
                    "type": a.type,
                    "path": a.path,
                    "sha256": a.sha256,
                    "size": a.size,
                    "created_at": a.created_at.isoformat() if a.created_at else None,
                }
                for a in artifacts
            ],
            "exports": [
                {
                    "id": e.id,
                    "source_checkpoint": e.source_checkpoint,
                    "artifact_path": e.artifact_path,
                    "format": e.format,
                    "duration_s": e.duration_s,
                    "validation_result": json.loads(cast(str, e.validation_result_json))
                    if e.validation_result_json
                    else None,
                    "created_at": e.created_at.isoformat() if e.created_at else None,
                }
                for e in exports
            ],
            "events": [
                {
                    "id": ev.id,
                    "event_type": ev.event_type,
                    "from_status": ev.from_status,
                    "to_status": ev.to_status,
                    "message": ev.message,
                    "created_at": ev.created_at.isoformat() if ev.created_at else None,
                }
                for ev in events
            ],
        }
    )

health() async

Health check endpoint.

Source code in src/fovux/http/routes.py 
42
43
44
45
46
47
@router.get("/health")
async def health() -> dict[str, str]:
    """Health check endpoint."""
    from fovux import __version__

    return {"status": "ok", "version": __version__, "service": "fovux-mcp"}

list_datasets() async

List all registered datasets in the ledger.

Source code in src/fovux/http/routes.py 
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
@router.get("/datasets")
async def list_datasets() -> JSONResponse:
    """List all registered datasets in the ledger."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    datasets = registry.list_datasets()
    return JSONResponse(
        [
            {
                "fingerprint": d.fingerprint,
                "path": d.path,
                "class_map": json.loads(cast(str, d.class_map_json)) if d.class_map_json else {},
                "created_at": d.created_at.isoformat() if d.created_at else None,
            }
            for d in datasets
        ]
    )

list_exports() async

List all model exports recorded in the ledger.

Source code in src/fovux/http/routes.py 
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
@router.get("/exports")
async def list_exports() -> JSONResponse:
    """List all model exports recorded in the ledger."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    exports = registry.list_exports()
    return JSONResponse(
        [
            {
                "id": e.id,
                "run_id": e.run_id,
                "source_checkpoint": e.source_checkpoint,
                "artifact_path": e.artifact_path,
                "format": e.format,
                "duration_s": e.duration_s,
                "validation_result": json.loads(cast(str, e.validation_result_json))
                if e.validation_result_json
                else None,
                "created_at": e.created_at.isoformat() if e.created_at else None,
            }
            for e in exports
        ]
    )

list_runs() async

List all training runs.

Source code in src/fovux/http/routes.py 
50
51
52
53
54
55
56
57
58
59
60
@router.get("/runs")
async def list_runs() -> JSONResponse:
    """List all training runs."""
    from fovux.core.paths import ensure_fovux_dirs

    paths = ensure_fovux_dirs()
    from fovux.core.runs import get_registry

    registry = get_registry(paths.runs_db)
    records = registry.list_runs()
    return JSONResponse([_run_summary(record) for record in records])

prometheus_metrics(request) async

Expose a small Prometheus-compatible metrics snapshot when enabled.

Source code in src/fovux/http/routes.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@router.get("/metrics")
async def prometheus_metrics(request: Request) -> PlainTextResponse:
    """Expose a small Prometheus-compatible metrics snapshot when enabled."""
    if not bool(getattr(request.app.state, "metrics_enabled", False)):
        raise HTTPException(status_code=404, detail="Metrics endpoint is disabled.")

    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    records = registry.list_runs(limit=10000)
    active_runs = sum(1 for record in records if record.status == "running")
    total_runs = len(records)
    lines = [
        "# HELP fovux_active_runs Number of currently running Fovux training runs.",
        "# TYPE fovux_active_runs gauge",
        f"fovux_active_runs {active_runs}",
        "# HELP fovux_runs_total Number of runs tracked by the local registry.",
        "# TYPE fovux_runs_total gauge",
        f"fovux_runs_total {total_runs}",
    ]
    return PlainTextResponse("\n".join(lines) + "\n")

proxy_tool(request, name, payload=_EMPTY_PAYLOAD) async

Invoke a local Fovux tool through the HTTP transport.

Tools that require confirmation must include a valid challenge_id obtained from POST /tools/{name}/challenge.

Source code in src/fovux/http/routes.py 
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
@router.post("/tools/{name}")
async def proxy_tool(
    request: Request,
    name: str,
    payload: dict[str, object] = _EMPTY_PAYLOAD,
) -> JSONResponse:
    """Invoke a local Fovux tool through the HTTP transport.

    Tools that require confirmation must include a valid challenge_id
    obtained from POST /tools/{name}/challenge.
    """
    from fovux.core.auth import token_fingerprint
    from fovux.http.tool_proxy import (
        HttpToolPolicyError,
        invoke_tool,
        payload_hash,
        policy_for_tool,
    )

    logger = get_logger(__name__)
    origin = request.headers.get("origin")
    if origin is None and request.client is not None:
        origin = request.client.host
    actor = token_fingerprint(str(request.app.state.auth_token))
    args_hash = payload_hash(payload)
    operation_id = _tool_operation_id(name, args_hash)
    operation_key = f"{name}:{args_hash}"
    from fovux.http.challenge import prune_expired_challenges, verify_challenge

    started = time.monotonic()
    try:
        policy = policy_for_tool(name)
        if policy.requires_confirmation:
            challenges = cast(dict[str, Any], request.app.state.challenges)
            prune_expired_challenges(challenges)
            challenge_id = payload.get("challenge_id")
            if not isinstance(challenge_id, str) or not challenge_id.strip():
                raise HttpToolPolicyError(
                    f"Tool '{name}' requires a confirmation challenge.",
                    hint=(
                        "Call POST /tools/{name}/challenge first, then include the "
                        "returned challenge_id in the tool payload."
                    ),
                )
            challenge_payload = {k: v for k, v in payload.items() if k != "challenge_id"}
            verify_challenge(
                challenges,
                challenge_id=challenge_id,
                tool_name=name,
                args_hash=payload_hash(challenge_payload),
            )

        semaphores = cast(dict[str, asyncio.Semaphore], request.app.state.tool_semaphores)
        semaphore = semaphores[name]
        operations = cast(dict[str, asyncio.Future[Any]], request.app.state.tool_operations)
        operation_results = cast(
            dict[str, dict[str, object]],
            request.app.state.tool_operation_results,
        )
        _prune_tool_operation_results(operation_results)
        completed_operation = _pop_fresh_tool_operation_result(operation_results, operation_key)
        if completed_operation is not None:
            if completed_operation.get("status") == "succeeded":
                result = cast(dict[str, Any], completed_operation.get("result") or {})
                logger.info(
                    "http_tool_audit",
                    actor=actor,
                    origin=origin,
                    tool=name,
                    args_hash=args_hash,
                    status="success",
                    duration_ms=int((time.monotonic() - started) * 1000),
                    failure_class=None,
                )
                return JSONResponse(result)
            logger.warning(
                "http_tool_audit",
                actor=actor,
                origin=origin,
                tool=name,
                args_hash=args_hash,
                status="failed",
                duration_ms=int((time.monotonic() - started) * 1000),
                failure_class="background_operation_failed",
            )
            return JSONResponse(
                status_code=500,
                content={
                    "operation_id": completed_operation.get("operation_id", operation_id),
                    "status": completed_operation.get("status", "failed"),
                    "error_type": completed_operation.get("error_type"),
                    "error": completed_operation.get("error"),
                },
            )
        running_operation = operations.get(operation_key)
        if running_operation is not None and not running_operation.done():
            logger.info(
                "http_tool_audit",
                actor=actor,
                origin=origin,
                tool=name,
                args_hash=args_hash,
                status="accepted",
                duration_ms=int((time.monotonic() - started) * 1000),
                failure_class=None,
            )
            return JSONResponse(
                status_code=202,
                content={
                    "operation_id": operation_id,
                    "status": "running",
                    "message": "Tool execution is still running.",
                },
            )
        try:
            await asyncio.wait_for(semaphore.acquire(), timeout=0.01)
        except TimeoutError as exc:
            logger.warning(
                "http_tool_audit",
                actor=actor,
                origin=origin,
                tool=name,
                args_hash=args_hash,
                status="rejected",
                duration_ms=0,
                failure_class="concurrency_limit",
            )
            raise HTTPException(status_code=429, detail="Tool concurrency limit exceeded.") from exc
        release_deferred = False
        try:
            worker_task = asyncio.create_task(asyncio.to_thread(invoke_tool, name, payload))
            try:
                result = await asyncio.wait_for(
                    asyncio.shield(worker_task),
                    timeout=policy.timeout_seconds,
                )
            except TimeoutError:
                operations[operation_key] = worker_task
                worker_task.add_done_callback(
                    _remember_timed_out_tool_worker(
                        semaphore=semaphore,
                        operations=operations,
                        results=operation_results,
                        operation_key=operation_key,
                        operation_id=operation_id,
                    )
                )
                release_deferred = True
                logger.warning(
                    "http_tool_audit",
                    actor=actor,
                    origin=origin,
                    tool=name,
                    args_hash=args_hash,
                    status="accepted",
                    duration_ms=int((time.monotonic() - started) * 1000),
                    failure_class="background_operation",
                )
                return JSONResponse(
                    status_code=202,
                    content={
                        "operation_id": operation_id,
                        "status": "running",
                        "message": (
                            "Tool execution exceeded the request timeout and continues once."
                        ),
                    },
                )
        finally:
            if not release_deferred:
                semaphore.release()
    except TimeoutError as exc:
        logger.warning(
            "http_tool_audit",
            actor=actor,
            origin=origin,
            tool=name,
            args_hash=args_hash,
            status="failed",
            duration_ms=int((time.monotonic() - started) * 1000),
            failure_class="timeout",
        )
        raise HTTPException(status_code=504, detail="Tool execution timed out.") from exc
    except HttpToolPolicyError as exc:
        logger.warning(
            "http_tool_audit",
            actor=actor,
            origin=origin,
            tool=name,
            args_hash=args_hash,
            status="rejected",
            duration_ms=int((time.monotonic() - started) * 1000),
            failure_class="policy",
        )
        detail = ErrorDetail(code=exc.code, message=exc.message, hint=exc.hint)
        raise HTTPException(status_code=403, detail=detail.model_dump(mode="json")) from exc
    except ValidationError as exc:
        logger.warning(
            "http_tool_audit",
            actor=actor,
            origin=origin,
            tool=name,
            args_hash=args_hash,
            status="failed",
            duration_ms=int((time.monotonic() - started) * 1000),
            failure_class="validation_error",
        )
        detail = ErrorDetail(
            code="FOVUX_HTTP_002",
            message="Tool payload validation failed.",
            hint=str(exc),
        )
        raise HTTPException(status_code=422, detail=detail.model_dump(mode="json")) from exc
    except FovuxError as exc:
        logger.warning(
            "http_tool_audit",
            actor=actor,
            origin=origin,
            tool=name,
            args_hash=args_hash,
            status="failed",
            duration_ms=int((time.monotonic() - started) * 1000),
            failure_class=exc.code,
        )
        detail = ErrorDetail(code=exc.code, message=exc.message, hint=exc.hint)
        raise HTTPException(
            status_code=400,
            detail=detail.model_dump(mode="json"),
        ) from exc

    logger.info(
        "http_tool_audit",
        actor=actor,
        origin=origin,
        tool=name,
        args_hash=args_hash,
        status="success",
        duration_ms=int((time.monotonic() - started) * 1000),
        failure_class=None,
    )
    return JSONResponse(result)

request_challenge(request, name, payload=_EMPTY_PAYLOAD) async

Request a confirmation challenge for a risky tool call.

Returns a challenge_id that must be included when calling the tool via POST /tools/{name}. Read-only tools do not require challenges.

Source code in src/fovux/http/routes.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
@router.post("/tools/{name}/challenge")
async def request_challenge(
    request: Request,
    name: str,
    payload: dict[str, object] = _EMPTY_PAYLOAD,
) -> JSONResponse:
    """Request a confirmation challenge for a risky tool call.

    Returns a challenge_id that must be included when calling the tool
    via POST /tools/{name}. Read-only tools do not require challenges.
    """
    from fovux.http.challenge import create_challenge, prune_expired_challenges
    from fovux.http.tool_proxy import (
        HttpToolPolicyError,
        payload_hash,
        policy_for_tool,
    )

    try:
        policy = policy_for_tool(name)
    except HttpToolPolicyError as exc:
        raise HTTPException(
            status_code=403,
            detail={
                "code": exc.code,
                "message": exc.message,
                "hint": exc.hint,
            },
        ) from exc
    if not policy.requires_confirmation:
        raise HTTPException(
            status_code=403,
            detail={
                "code": "FOVUX_HTTP_003",
                "message": f"Tool '{name}' does not require a confirmation challenge.",
                "hint": "Read-only tools can be called directly without a challenge.",
            },
        )

    challenges = cast(dict[str, Any], request.app.state.challenges)
    prune_expired_challenges(challenges)
    args_hash = payload_hash(payload)

    record = create_challenge(
        tool_name=name,
        args_hash=args_hash,
        risk_level=policy.category,
    )
    challenges[record.challenge_id] = record

    return JSONResponse(
        status_code=201,
        content={
            "challenge_id": record.challenge_id,
            "tool": name,
            "risk_level": policy.category,
            "summary": {
                "name": name,
                "args_hash": args_hash,
                "params": {
                    str(k): v
                    for k, v in payload.items()
                    if str(k) not in ("confirm", "challenge_id")
                },
            },
            "expires_at": record.expires_at,
        },
    )

search_runs(body) async

Search runs by text, tags, status, and minimum mAP50.

Source code in src/fovux/http/routes.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
@router.post("/runs/search")
async def search_runs(body: RunsSearchInput) -> JSONResponse:
    """Search runs by text, tags, status, and minimum mAP50."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)
    records = registry.list_runs(limit=max(body.limit, 1) * 4)

    matched: list[dict[str, object]] = []
    lowered_query = body.query.lower() if body.query else None
    required_statuses = {status.lower() for status in body.status}
    required_tags = {tag.lower() for tag in body.tags}

    for record in records:
        raw_tags = cast(str, record.tags_json or "[]")
        record_tags = {str(tag).lower() for tag in json.loads(raw_tags)}
        haystack = " ".join(
            [
                str(record.id),
                str(record.model),
                str(record.dataset_path),
                str(record.task),
                " ".join(record_tags),
                str(record.extra_json or ""),
            ]
        ).lower()
        if lowered_query and lowered_query not in haystack:
            continue
        if required_statuses and str(record.status).lower() not in required_statuses:
            continue
        if required_tags and not required_tags.issubset(record_tags):
            continue
        _, best_map50 = read_metrics_summary(Path(record.run_path))
        if body.min_map50 is not None and (best_map50 is None or best_map50 < body.min_map50):
            continue
        matched.append(
            {
                "id": record.id,
                "status": record.status,
                "model": record.model,
                "dataset_path": record.dataset_path,
                "task": record.task,
                "epochs": record.epochs,
                "created_at": record.created_at.isoformat() if record.created_at else None,
                "best_map50": best_map50,
                "tags": sorted(record_tags),
            }
        )
        if len(matched) >= body.limit:
            break
    return JSONResponse(matched)

sse_events_route(request) async

Server-Sent Events (SSE) stream of all operations events with resume support.

Source code in src/fovux/http/routes.py 
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
@router.get("/events")
async def sse_events_route(request: Request) -> StreamingResponse:
    """Server-Sent Events (SSE) stream of all operations events with resume support."""
    from fovux.core.paths import ensure_fovux_dirs
    from fovux.core.runs import get_registry

    paths = ensure_fovux_dirs()
    registry = get_registry(paths.runs_db)

    last_event_id_str = request.headers.get("Last-Event-ID") or request.query_params.get(
        "last_event_id"
    )
    last_event_id = None
    if last_event_id_str:
        try:
            last_event_id = int(last_event_id_str)
        except ValueError:
            pass

    shutdown_event = cast(asyncio.Event, request.app.state.shutdown_event)

    async def event_generator() -> AsyncIterator[str]:
        yield "retry: 5000\n\n"

        if last_event_id is not None:
            hist_events = registry.list_operation_events(last_event_id=last_event_id)
            for event in hist_events:
                yield f"id: {event.id}\nevent: {event.event_type}\ndata: {event.data_json}\n\n"

        queue: asyncio.Queue[tuple[int, str, dict[str, Any]]] = asyncio.Queue()
        if not hasattr(request.app.state, "sse_listeners"):
            request.app.state.sse_listeners = []
        request.app.state.sse_listeners.append(queue)

        try:
            while not shutdown_event.is_set():
                if await request.is_disconnected():
                    break
                try:
                    ev_id, ev_type, ev_data = await asyncio.wait_for(queue.get(), timeout=15.0)
                    yield f"id: {ev_id}\nevent: {ev_type}\ndata: {json.dumps(ev_data)}\n\n"
                except TimeoutError:
                    yield ": keep-alive\n\n"
        finally:
            if (
                hasattr(request.app.state, "sse_listeners")
                and queue in request.app.state.sse_listeners
            ):
                request.app.state.sse_listeners.remove(queue)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

stream_run_metrics(run_id, request) async

Stream normalized metric rows for a run over server-sent events.

Source code in src/fovux/http/routes.py 
223
224
225
226
@router.get("/runs/{run_id}/stream")
async def stream_run_metrics(run_id: str, request: Request) -> StreamingResponse:
    """Stream normalized metric rows for a run over server-sent events."""
    return _stream_run_metrics_response(run_id, request)

stream_run_metrics_compat(run_id, request) async

Compatibility alias for the canonical run metric stream.

Source code in src/fovux/http/routes.py 
229
230
231
232
@router.get("/runs/{run_id}/metrics")
async def stream_run_metrics_compat(run_id: str, request: Request) -> StreamingResponse:
    """Compatibility alias for the canonical run metric stream."""
    return _stream_run_metrics_response(run_id, request)

Schemas

Pydantic schemas for dataset tools.

AutoFixItem

Bases: BaseModel

An auto-fix recommendation.

Source code in src/fovux/schemas/dataset.py 
59
60
61
62
63
64
class AutoFixItem(BaseModel):
    """An auto-fix recommendation."""

    action: str
    description: str
    estimated_impact: str

ClassStat

Bases: BaseModel

Statistics for a single class.

Source code in src/fovux/schemas/dataset.py 
18
19
20
21
22
23
24
class ClassStat(BaseModel):
    """Statistics for a single class."""

    name: str
    count: int
    pct: float
    avg_bbox_area: float | None = None

DatasetAugmentInput

Bases: BaseModel

Input for dataset_augment.

Source code in src/fovux/schemas/dataset.py 
208
209
210
211
212
213
214
class DatasetAugmentInput(BaseModel):
    """Input for dataset_augment."""

    dataset_path: Path
    techniques: list[AugmentationTechnique] = Field(default_factory=default_augmentation_techniques)
    multiplier: int = 3
    output_path: Path

DatasetAugmentOutput

Bases: BaseModel

Output from dataset_augment.

Source code in src/fovux/schemas/dataset.py 
217
218
219
220
221
222
223
224
225
class DatasetAugmentOutput(BaseModel):
    """Output from dataset_augment."""

    dataset_path: Path
    output_path: Path
    source_images: int
    generated_images: int
    techniques: list[str]
    manifest_path: Path

DatasetConvertInput

Bases: BaseModel

Input for dataset_convert tool.

Source code in src/fovux/schemas/dataset.py 
186
187
188
189
190
191
192
193
194
class DatasetConvertInput(BaseModel):
    """Input for dataset_convert tool."""

    source_path: Path
    source_format: Literal["yolo", "coco", "voc", "auto"] = "auto"
    target_format: Literal["yolo", "coco", "voc"]
    target_path: Path
    copy_images: bool = False
    class_map: dict[str, str] | None = None

DatasetConvertOutput

Bases: BaseModel

Output from dataset_convert tool.

Source code in src/fovux/schemas/dataset.py 
197
198
199
200
201
202
203
204
205
class DatasetConvertOutput(BaseModel):
    """Output from dataset_convert tool."""

    images_processed: int
    annotations_converted: int
    annotations_skipped: int
    skip_reasons: dict[str, int]
    target_path: Path
    conversion_duration_seconds: float

DatasetFindDuplicatesInput

Bases: BaseModel

Input for dataset_find_duplicates tool.

Source code in src/fovux/schemas/dataset.py 
144
145
146
147
148
149
150
class DatasetFindDuplicatesInput(BaseModel):
    """Input for dataset_find_duplicates tool."""

    dataset_path: Path
    algorithm: Literal["phash", "dhash", "whash", "avg"] = "phash"
    hamming_threshold: int = 5
    across_splits: bool = True

DatasetFindDuplicatesOutput

Bases: BaseModel

Output from dataset_find_duplicates tool.

Source code in src/fovux/schemas/dataset.py 
153
154
155
156
157
158
159
160
class DatasetFindDuplicatesOutput(BaseModel):
    """Output from dataset_find_duplicates tool."""

    total_images: int
    duplicate_groups: list[DuplicateGroup]
    total_duplicates: int
    duplicate_pct: float
    analysis_duration_seconds: float

DatasetInspectInput

Bases: BaseModel

Input for dataset_inspect tool.

Source code in src/fovux/schemas/dataset.py 
41
42
43
44
45
46
47
class DatasetInspectInput(BaseModel):
    """Input for dataset_inspect tool."""

    dataset_path: Path
    format: Literal["yolo", "coco", "voc", "auto"] = "auto"
    include_samples: bool = True
    max_images_analyzed: int = 10_000

DatasetInspectOutput

Bases: BaseModel

Output from dataset_inspect tool.

Source code in src/fovux/schemas/dataset.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class DatasetInspectOutput(BaseModel):
    """Output from dataset_inspect tool."""

    format_detected: str
    total_images: int
    total_annotations: int
    num_classes: int
    classes: list[ClassStat]
    image_size_distribution: SizeHistogram
    bbox_size_distribution: SizeHistogram
    bbox_size_buckets: dict[str, int] = Field(default_factory=dict)
    bbox_count_per_image: Histogram
    orphan_images: int
    missing_label_images: list[Path] = Field(default_factory=list)
    orphan_annotations: int
    class_balance_gini: float
    splits_detected: dict[str, int]
    warnings: list[str]
    sample_paths: list[Path]
    analysis_duration_seconds: float

    # Intelligence capabilities
    quality_score: float = 100.0
    label_anomalies: LabelAnomalySummary = Field(default_factory=LabelAnomalySummary)
    duplicate_groups_count: int = 0
    total_duplicates_found: int = 0
    leaked_images: list[LeakageIssue] = Field(default_factory=list)
    auto_fix_plan: list[AutoFixItem] = Field(default_factory=list)
    dataset_card: str = ""

DatasetSplitInput

Bases: BaseModel

Input for dataset_split tool.

Source code in src/fovux/schemas/dataset.py 
163
164
165
166
167
168
169
170
171
172
class DatasetSplitInput(BaseModel):
    """Input for dataset_split tool."""

    dataset_path: Path
    ratios: tuple[float, float, float] = (0.7, 0.2, 0.1)
    stratify_by_class: bool = True
    seed: int = 42
    output_format: Literal["yolo", "coco"] = "yolo"
    overwrite: bool = False
    output_path: Path | None = None

DatasetSplitOutput

Bases: BaseModel

Output from dataset_split tool.

Source code in src/fovux/schemas/dataset.py 
175
176
177
178
179
180
181
182
183
class DatasetSplitOutput(BaseModel):
    """Output from dataset_split tool."""

    train_count: int
    val_count: int
    test_count: int
    stratification_report: dict[str, dict[str, int]]
    output_path: Path
    manifest_path: Path

DatasetValidateInput

Bases: BaseModel

Input for dataset_validate tool.

Source code in src/fovux/schemas/dataset.py 
116
117
118
119
120
121
122
123
124
class DatasetValidateInput(BaseModel):
    """Input for dataset_validate tool."""

    dataset_path: Path
    format: Literal["yolo", "coco", "voc", "auto"] = "auto"
    check_image_readable: bool = True
    check_bbox_bounds: bool = True
    check_class_id_range: bool = True
    strict: bool = False

DatasetValidateOutput

Bases: BaseModel

Output from dataset_validate tool.

Source code in src/fovux/schemas/dataset.py 
127
128
129
130
131
132
133
134
class DatasetValidateOutput(BaseModel):
    """Output from dataset_validate tool."""

    valid: bool
    errors: list[ValidationIssue]
    warnings: list[ValidationIssue]
    summary: str
    remediation_script: str | None = None

DuplicateGroup

Bases: BaseModel

A group of duplicate or near-duplicate images.

Source code in src/fovux/schemas/dataset.py 
137
138
139
140
141
class DuplicateGroup(BaseModel):
    """A group of duplicate or near-duplicate images."""

    images: list[Path]
    hamming_distance: int

Histogram

Bases: BaseModel

Generic histogram.

Source code in src/fovux/schemas/dataset.py 
34
35
36
37
38
class Histogram(BaseModel):
    """Generic histogram."""

    buckets: list[str]
    counts: list[int]

LabelAnomalySummary

Bases: BaseModel

Summary of label anomaly checks.

Source code in src/fovux/schemas/dataset.py 
50
51
52
53
54
55
56
class LabelAnomalySummary(BaseModel):
    """Summary of label anomaly checks."""

    tiny_boxes: int = 0
    out_of_bounds: int = 0
    empty_labels: int = 0
    suspiciously_overlapping: int = 0

LeakageIssue

Bases: BaseModel

Details of a leaked image between splits.

Source code in src/fovux/schemas/dataset.py 
67
68
69
70
71
72
73
class LeakageIssue(BaseModel):
    """Details of a leaked image between splits."""

    train_image: str
    val_image: str | None = None
    test_image: str | None = None
    reason: str

SizeHistogram

Bases: BaseModel

Simple histogram for size distributions.

Source code in src/fovux/schemas/dataset.py 
27
28
29
30
31
class SizeHistogram(BaseModel):
    """Simple histogram for size distributions."""

    buckets: list[str]
    counts: list[int]

ValidationIssue

Bases: BaseModel

A single validation issue.

Source code in src/fovux/schemas/dataset.py 
107
108
109
110
111
112
113
class ValidationIssue(BaseModel):
    """A single validation issue."""

    file: str
    line: int | None = None
    severity: Literal["error", "warning"]
    message: str

default_augmentation_techniques()

Return the default deterministic augmentation list.

Source code in src/fovux/schemas/dataset.py 
13
14
15
def default_augmentation_techniques() -> list[AugmentationTechnique]:
    """Return the default deterministic augmentation list."""
    return ["flip_h"]

Pydantic schemas for training tools.

TrainPreflightInput

Bases: BaseModel

Input for train_preflight tool (shares validation with TrainStartInput).

Source code in src/fovux/schemas/training.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class TrainPreflightInput(BaseModel):
    """Input for train_preflight tool (shares validation with TrainStartInput)."""

    dataset_path: Path
    model: str = "yolov8n.pt"
    epochs: int = Field(default=100, gt=0)
    batch: int = Field(default=16, gt=0)
    imgsz: int = Field(default=640, gt=0)
    device: str = "auto"
    task: Literal["detect", "segment", "classify", "pose", "obb"] = "detect"
    name: RunId | None = None
    force: bool = False
    max_concurrent_runs: int = Field(default=1, ge=0)
    tags: list[str] = Field(default_factory=list)
    options: TrainingOptions = Field(default_factory=lambda: TrainingOptions())
    extra_args: dict[str, Any] = Field(default_factory=dict)

    # Resource budgets
    max_runtime_seconds: int | None = Field(default=None, gt=0)
    max_disk_usage_gb: float | None = Field(default=None, gt=0.0)
    device_policy: Literal["any", "gpu_only", "cpu_only"] = "any"

    @model_validator(mode="before")
    @classmethod
    def merge_extra_args(cls, data: Any) -> Any:  # noqa: ANN401
        """Merge extra_args field values into options."""
        if isinstance(data, dict):
            extra_args = data.get("extra_args") or {}
            options = data.setdefault("options", {})
            for k, v in extra_args.items():
                options.setdefault(k, v)
        return data

    @field_validator("device")
    @classmethod
    def validate_device(cls, v: str) -> str:
        """Validate device via TrainStartInput."""
        return TrainStartInput.validate_device(v)

    @field_validator("model")
    @classmethod
    def validate_model(cls, v: str) -> str:
        """Allow any model string for preflight diagnostics."""
        return v

    @model_validator(mode="after")
    def validate_device_policy(self) -> TrainPreflightInput:
        """Validate device policy via TrainStartInput rules."""
        if self.device_policy == "gpu_only":
            if self.device == "cpu":
                raise ValueError("Device cannot be 'cpu' when device_policy is 'gpu_only'")
        elif self.device_policy == "cpu_only":
            if self.device not in ("cpu", "auto"):
                raise ValueError(
                    f"Device cannot be '{self.device}' when device_policy is 'cpu_only'"
                )
        return self

merge_extra_args(data) classmethod

Merge extra_args field values into options.

Source code in src/fovux/schemas/training.py 
152
153
154
155
156
157
158
159
160
161
@model_validator(mode="before")
@classmethod
def merge_extra_args(cls, data: Any) -> Any:  # noqa: ANN401
    """Merge extra_args field values into options."""
    if isinstance(data, dict):
        extra_args = data.get("extra_args") or {}
        options = data.setdefault("options", {})
        for k, v in extra_args.items():
            options.setdefault(k, v)
    return data

validate_device(v) classmethod

Validate device via TrainStartInput.

Source code in src/fovux/schemas/training.py 
163
164
165
166
167
@field_validator("device")
@classmethod
def validate_device(cls, v: str) -> str:
    """Validate device via TrainStartInput."""
    return TrainStartInput.validate_device(v)

validate_device_policy()

Validate device policy via TrainStartInput rules.

Source code in src/fovux/schemas/training.py 
175
176
177
178
179
180
181
182
183
184
185
186
@model_validator(mode="after")
def validate_device_policy(self) -> TrainPreflightInput:
    """Validate device policy via TrainStartInput rules."""
    if self.device_policy == "gpu_only":
        if self.device == "cpu":
            raise ValueError("Device cannot be 'cpu' when device_policy is 'gpu_only'")
    elif self.device_policy == "cpu_only":
        if self.device not in ("cpu", "auto"):
            raise ValueError(
                f"Device cannot be '{self.device}' when device_policy is 'cpu_only'"
            )
    return self

validate_model(v) classmethod

Allow any model string for preflight diagnostics.

Source code in src/fovux/schemas/training.py 
169
170
171
172
173
@field_validator("model")
@classmethod
def validate_model(cls, v: str) -> str:
    """Allow any model string for preflight diagnostics."""
    return v

TrainPreflightOutput

Bases: BaseModel

Output for train_preflight tool.

Source code in src/fovux/schemas/training.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class TrainPreflightOutput(BaseModel):
    """Output for train_preflight tool."""

    dataset_valid: bool
    dataset_classes_count: int
    dataset_path: str
    model_valid: bool
    model_source: str
    device_available: bool
    resolved_device: str
    disk_space_valid: bool
    available_disk_space_gb: float
    output_path_valid: bool
    resolved_run_dir: str
    concurrency_valid: bool
    active_runs_count: int
    warnings: list[str] = Field(default_factory=list)

TrainResumeInput

Bases: BaseModel

Input for train_resume tool.

Source code in src/fovux/schemas/training.py 
250
251
252
253
254
class TrainResumeInput(BaseModel):
    """Input for train_resume tool."""

    run_id: RunId
    epochs: int | None = None

TrainResumeOutput

Bases: BaseModel

Output from train_resume tool.

Source code in src/fovux/schemas/training.py 
257
258
259
260
261
262
263
class TrainResumeOutput(BaseModel):
    """Output from train_resume tool."""

    run_id: str
    status: str
    pid: int | None
    run_path: Path

TrainStartInput

Bases: BaseModel

Input for train_start tool.

Source code in src/fovux/schemas/training.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class TrainStartInput(BaseModel):
    """Input for train_start tool."""

    dataset_path: Path
    model: str = "yolov8n.pt"
    epochs: int = Field(default=100, gt=0)
    batch: int = Field(default=16, gt=0)
    imgsz: int = Field(default=640, gt=0)
    device: str = "auto"
    task: Literal["detect", "segment", "classify", "pose", "obb"] = "detect"
    name: RunId | None = None
    force: bool = False
    max_concurrent_runs: int = Field(default=1, ge=0)
    tags: list[str] = Field(default_factory=list)
    options: TrainingOptions = Field(default_factory=lambda: TrainingOptions())
    extra_args: dict[str, Any] = Field(default_factory=dict)

    # Resource budgets
    max_runtime_seconds: int | None = Field(default=None, gt=0)
    max_disk_usage_gb: float | None = Field(default=None, gt=0.0)
    device_policy: Literal["any", "gpu_only", "cpu_only"] = "any"

    @model_validator(mode="before")
    @classmethod
    def merge_extra_args(cls, data: Any) -> Any:  # noqa: ANN401
        """Merge extra_args field values into options."""
        if isinstance(data, dict):
            extra_args = data.get("extra_args") or {}
            options = data.setdefault("options", {})
            for k, v in extra_args.items():
                options.setdefault(k, v)
        return data

    @field_validator("device")
    @classmethod
    def validate_device(cls, v: str) -> str:
        """Validate device is cpu, cuda, or specific GPU index."""
        v_lower = v.strip().lower()
        if v_lower in ("auto", "cpu", "gpu", "cuda"):
            return v_lower
        if v_lower.startswith(("cuda:", "gpu:")):
            parts = v_lower.split(":")
            if len(parts) == 2 and parts[1].isdigit():
                return v_lower
        raise ValueError(
            "device must be 'auto', 'cpu', 'cuda', 'gpu', or specific index like 'cuda:0'"
        )

    @field_validator("model")
    @classmethod
    def validate_model(cls, v: str) -> str:
        """Validate model source file extension format."""
        v = v.strip()
        if not v.endswith((".pt", ".yaml", ".yml")):
            raise ValueError("model source must end with .pt or .yaml/.yml")
        return v

    @model_validator(mode="after")
    def validate_device_policy(self) -> TrainStartInput:
        """Check compatibility between device and device_policy."""
        if self.device_policy == "gpu_only":
            if self.device == "cpu":
                raise ValueError("Device cannot be 'cpu' when device_policy is 'gpu_only'")
        elif self.device_policy == "cpu_only":
            if self.device not in ("cpu", "auto"):
                raise ValueError(
                    f"Device cannot be '{self.device}' when device_policy is 'cpu_only'"
                )
        return self

merge_extra_args(data) classmethod

Merge extra_args field values into options.

Source code in src/fovux/schemas/training.py 
81
82
83
84
85
86
87
88
89
90
@model_validator(mode="before")
@classmethod
def merge_extra_args(cls, data: Any) -> Any:  # noqa: ANN401
    """Merge extra_args field values into options."""
    if isinstance(data, dict):
        extra_args = data.get("extra_args") or {}
        options = data.setdefault("options", {})
        for k, v in extra_args.items():
            options.setdefault(k, v)
    return data

validate_device(v) classmethod

Validate device is cpu, cuda, or specific GPU index.

Source code in src/fovux/schemas/training.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@field_validator("device")
@classmethod
def validate_device(cls, v: str) -> str:
    """Validate device is cpu, cuda, or specific GPU index."""
    v_lower = v.strip().lower()
    if v_lower in ("auto", "cpu", "gpu", "cuda"):
        return v_lower
    if v_lower.startswith(("cuda:", "gpu:")):
        parts = v_lower.split(":")
        if len(parts) == 2 and parts[1].isdigit():
            return v_lower
    raise ValueError(
        "device must be 'auto', 'cpu', 'cuda', 'gpu', or specific index like 'cuda:0'"
    )

validate_device_policy()

Check compatibility between device and device_policy.

Source code in src/fovux/schemas/training.py 
116
117
118
119
120
121
122
123
124
125
126
127
@model_validator(mode="after")
def validate_device_policy(self) -> TrainStartInput:
    """Check compatibility between device and device_policy."""
    if self.device_policy == "gpu_only":
        if self.device == "cpu":
            raise ValueError("Device cannot be 'cpu' when device_policy is 'gpu_only'")
    elif self.device_policy == "cpu_only":
        if self.device not in ("cpu", "auto"):
            raise ValueError(
                f"Device cannot be '{self.device}' when device_policy is 'cpu_only'"
            )
    return self

validate_model(v) classmethod

Validate model source file extension format.

Source code in src/fovux/schemas/training.py 
107
108
109
110
111
112
113
114
@field_validator("model")
@classmethod
def validate_model(cls, v: str) -> str:
    """Validate model source file extension format."""
    v = v.strip()
    if not v.endswith((".pt", ".yaml", ".yml")):
        raise ValueError("model source must end with .pt or .yaml/.yml")
    return v

TrainStartOutput

Bases: BaseModel

Output from train_start tool.

Source code in src/fovux/schemas/training.py 
208
209
210
211
212
213
214
class TrainStartOutput(BaseModel):
    """Output from train_start tool."""

    run_id: RunId
    status: str
    pid: int | None
    run_path: Path

TrainStatusInput

Bases: BaseModel

Input for train_status tool.

Source code in src/fovux/schemas/training.py 
217
218
219
220
class TrainStatusInput(BaseModel):
    """Input for train_status tool."""

    run_id: RunId

TrainStatusOutput

Bases: BaseModel

Output from train_status tool.

Source code in src/fovux/schemas/training.py 
223
224
225
226
227
228
229
230
231
232
class TrainStatusOutput(BaseModel):
    """Output from train_status tool."""

    run_id: RunId
    status: str
    pid: int | None
    elapsed_seconds: float | None
    current_epoch: int | None
    best_map50: float | None
    run_path: Path

TrainStopInput

Bases: BaseModel

Input for train_stop tool.

Source code in src/fovux/schemas/training.py 
235
236
237
238
239
class TrainStopInput(BaseModel):
    """Input for train_stop tool."""

    run_id: RunId
    force: bool = False

TrainStopOutput

Bases: BaseModel

Output from train_stop tool.

Source code in src/fovux/schemas/training.py 
242
243
244
245
246
247
class TrainStopOutput(BaseModel):
    """Output from train_stop tool."""

    run_id: str
    status: str
    message: str

TrainingOptions

Bases: BaseModel

Supported training options for YOLO.

Source code in src/fovux/schemas/training.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class TrainingOptions(BaseModel):
    """Supported training options for YOLO."""

    model_config = {
        "extra": "forbid",
    }

    optimizer: Literal["SGD", "Adam", "AdamW", "RMSProp", "auto"] = "auto"
    lr0: float = Field(default=0.01, gt=0.0)
    lrf: float = Field(default=0.01, ge=0.0)
    momentum: float = Field(default=0.937, ge=0.0, le=1.0)
    weight_decay: float = Field(default=0.0005, ge=0.0)
    warmup_epochs: float = Field(default=3.0, ge=0.0)
    warmup_momentum: float = Field(default=0.8, ge=0.0, le=1.0)
    warmup_bias_lr: float = Field(default=0.1, ge=0.0)
    box: float = Field(default=7.5, ge=0.0)
    cls: float = Field(default=0.5, ge=0.0)
    dfl: float = Field(default=1.5, ge=0.0)
    pose: float = Field(default=12.0, ge=0.0)
    kobj: float = Field(default=1.0, ge=0.0)
    label_smoothing: float = Field(default=0.0, ge=0.0, le=1.0)
    nbs: int = Field(default=64, gt=0)
    overlap_mask: bool = True
    mask_ratio: int = Field(default=4, gt=0)
    dropout: float = Field(default=0.0, ge=0.0, le=1.0)
    val: bool = True
    save: bool = True
    save_period: int = Field(default=-1, ge=-1)
    cache: Literal["ram", "disk", "auto", ""] = ""
    workers: int = Field(default=8, ge=0, le=128)
    pretrained: bool = True
    seed: int = Field(default=0, ge=0)
    deterministic: bool = True
    single_cls: bool = False
    rect: bool = False
    cos_lr: bool = False
    close_mosaic: int = Field(default=10, ge=0)
    amp: bool = True
    fraction: float = Field(default=1.0, gt=0.0, le=1.0)
    freeze: int | list[int] | None = None
    patience: int | None = Field(default=None, ge=0)
    teacher_checkpoint: str | None = Field(default=None)
    distillation_temperature: float | None = Field(default=None, gt=0.0)
    distillation_alpha: float | None = Field(default=None, ge=0.0, le=1.0)

Pydantic schemas for evaluation tools.

CheckpointComparison

Bases: BaseModel

Comparison row for one checkpoint.

Source code in src/fovux/schemas/eval.py 
132
133
134
135
136
137
138
139
140
class CheckpointComparison(BaseModel):
    """Comparison row for one checkpoint."""

    checkpoint: str
    map50: float
    map50_95: float
    precision: float
    recall: float
    eval_duration_seconds: float

ConfusionEntry

Bases: BaseModel

A single confusion matrix entry.

Source code in src/fovux/schemas/eval.py 
77
78
79
80
81
82
class ConfusionEntry(BaseModel):
    """A single confusion matrix entry."""

    true_class: str
    predicted_class: str
    count: int

ErrorSample

Bases: BaseModel

A high-error image sample.

Source code in src/fovux/schemas/eval.py 
85
86
87
88
89
90
91
92
class ErrorSample(BaseModel):
    """A high-error image sample."""

    image_path: Path
    true_class: str
    predicted_class: str
    confidence: float
    iou: float | None = None

EvalCompareInput

Bases: BaseModel

Input for eval_compare tool.

Source code in src/fovux/schemas/eval.py 
119
120
121
122
123
124
125
126
127
128
129
class EvalCompareInput(BaseModel):
    """Input for eval_compare tool."""

    checkpoints: list[str]
    dataset_path: Path
    split: str = "val"
    batch: int = 16
    imgsz: int = 640
    device: str = "auto"
    conf: float = 0.25
    iou: float = 0.45

EvalCompareOutput

Bases: BaseModel

Output from eval_compare tool.

Source code in src/fovux/schemas/eval.py 
143
144
145
146
147
148
149
150
class EvalCompareOutput(BaseModel):
    """Output from eval_compare tool."""

    dataset_path: Path
    split: str
    results: list[CheckpointComparison]
    best_map50: str
    best_map50_95: str

EvalErrorAnalysisInput

Bases: BaseModel

Input for eval_error_analysis tool.

Source code in src/fovux/schemas/eval.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
class EvalErrorAnalysisInput(BaseModel):
    """Input for eval_error_analysis tool."""

    checkpoint: str
    dataset_path: Path
    split: str = "val"
    top_n: int = 10
    imgsz: int = 640
    device: str = "auto"
    conf: float = 0.25
    iou: float = 0.45

EvalErrorAnalysisOutput

Bases: BaseModel

Output from eval_error_analysis tool.

Source code in src/fovux/schemas/eval.py 
108
109
110
111
112
113
114
115
116
class EvalErrorAnalysisOutput(BaseModel):
    """Output from eval_error_analysis tool."""

    checkpoint: str
    confusion_matrix: list[ConfusionEntry]
    top_errors: list[ErrorSample]
    false_positive_count: int
    false_negative_count: int
    eval_duration_seconds: float

EvalPerClassInput

Bases: BaseModel

Input for eval_per_class tool.

Source code in src/fovux/schemas/eval.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
class EvalPerClassInput(BaseModel):
    """Input for eval_per_class tool."""

    checkpoint: str
    dataset_path: Path
    split: str = "val"
    batch: int = 16
    imgsz: int = 640
    device: str = "auto"
    conf: float = 0.25
    iou: float = 0.45
    sort_by: Literal["map50", "map50_95", "precision", "recall", "class_name"] = "map50"
    ascending: bool = True

EvalPerClassOutput

Bases: BaseModel

Output from eval_per_class tool.

Source code in src/fovux/schemas/eval.py 
68
69
70
71
72
73
74
class EvalPerClassOutput(BaseModel):
    """Output from eval_per_class tool."""

    checkpoint: str
    per_class: list[PerClassStat]
    worst_classes: list[PerClassStat]
    eval_duration_seconds: float

EvalRunInput

Bases: BaseModel

Input for eval_run tool.

Source code in src/fovux/schemas/eval.py 
11
12
13
14
15
16
17
18
19
20
21
22
class EvalRunInput(BaseModel):
    """Input for eval_run tool."""

    checkpoint: str
    dataset_path: Path
    split: str = "val"
    batch: int = 16
    imgsz: int = 640
    device: str = "auto"
    conf: float = 0.25
    iou: float = 0.45
    task: Literal["detect", "segment", "classify", "pose", "obb"] = "detect"

EvalRunOutput

Bases: BaseModel

Output from eval_run tool.

Source code in src/fovux/schemas/eval.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
class EvalRunOutput(BaseModel):
    """Output from eval_run tool."""

    checkpoint: str
    dataset_path: Path
    split: str
    map50: float
    map50_95: float
    precision: float
    recall: float
    per_class: list[PerClassStat]
    eval_duration_seconds: float
    results_dir: Path | None = None

PerClassStat

Bases: BaseModel

Per-class evaluation statistics.

Source code in src/fovux/schemas/eval.py 
25
26
27
28
29
30
31
32
33
34
35
class PerClassStat(BaseModel):
    """Per-class evaluation statistics."""

    class_id: int
    class_name: str
    images: int
    instances: int
    precision: float
    recall: float
    map50: float
    map50_95: float

Pydantic schemas for export and quantization tools.

ExportOnnxInput

Bases: BaseModel

Input for export_onnx tool.

Source code in src/fovux/schemas/export.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class ExportOnnxInput(BaseModel):
    """Input for export_onnx tool."""

    checkpoint: str
    output_path: Path | None = None
    imgsz: int = 640
    opset: int = 17
    dynamic: bool = False
    simplify: bool = True
    half: bool = False
    nms: bool = False
    device: str = "auto"
    parity_check: bool = True
    parity_tolerance: float = 1e-3

ExportOnnxOutput

Bases: BaseModel

Output from export_onnx tool.

Source code in src/fovux/schemas/export.py 
26
27
28
29
30
31
32
33
34
35
36
37
class ExportOnnxOutput(BaseModel):
    """Output from export_onnx tool."""

    checkpoint: str
    onnx_path: Path
    output_path: Path
    export_duration_seconds: float
    parity_passed: bool | None
    parity_max_diff: float | None
    file_size_mb: float
    opset: int
    model_size_bytes: int

ExportTfliteInput

Bases: BaseModel

Input for export_tflite tool.

Source code in src/fovux/schemas/export.py 
40
41
42
43
44
45
46
47
48
class ExportTfliteInput(BaseModel):
    """Input for export_tflite tool."""

    checkpoint: str
    output_path: Path | None = None
    imgsz: int = 640
    half: bool = False
    int8: bool = False
    device: str = "auto"

ExportTfliteOutput

Bases: BaseModel

Output from export_tflite tool.

Source code in src/fovux/schemas/export.py 
51
52
53
54
55
56
57
58
59
class ExportTfliteOutput(BaseModel):
    """Output from export_tflite tool."""

    checkpoint: str
    tflite_path: Path
    output_path: Path
    export_duration_seconds: float
    file_size_mb: float
    model_size_bytes: int

QuantizeInt8Input

Bases: BaseModel

Input for quantize_int8 tool.

Source code in src/fovux/schemas/export.py 
62
63
64
65
66
67
68
69
class QuantizeInt8Input(BaseModel):
    """Input for quantize_int8 tool."""

    checkpoint: str
    calibration_dataset: Path
    output_path: Path | None = None
    imgsz: int = 640
    device: str = "auto"

QuantizeInt8Output

Bases: BaseModel

Output from quantize_int8 tool.

Source code in src/fovux/schemas/export.py 
72
73
74
75
76
77
78
79
class QuantizeInt8Output(BaseModel):
    """Output from quantize_int8 tool."""

    checkpoint: str
    quantized_path: Path
    quantize_duration_seconds: float
    model_size_bytes: int
    size_reduction_pct: float

QuantizeReportInput

Bases: BaseModel

Input for quantize_report tool.

Source code in src/fovux/schemas/export.py 
82
83
84
85
86
87
88
89
90
91
92
class QuantizeReportInput(BaseModel):
    """Input for quantize_report tool."""

    original_checkpoint: str
    quantized_checkpoint: str
    dataset_path: Path
    split: str = "val"
    imgsz: int = 640
    device: str = "auto"
    max_map50_drop: float = 0.01
    strict: bool = False

QuantizeReportOutput

Bases: BaseModel

Output from quantize_report tool.

Source code in src/fovux/schemas/export.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class QuantizeReportOutput(BaseModel):
    """Output from quantize_report tool."""

    original_checkpoint: str
    quantized_checkpoint: str
    original_map50: float
    quantized_map50: float
    map50_delta: float
    verdict: str
    original_size_bytes: int
    quantized_size_bytes: int
    size_reduction_pct: float
    report_duration_seconds: float

Pydantic schemas for inference and latency benchmark tools.

ActiveLearningQueueItem

Bases: BaseModel

A single review queue entry.

Source code in src/fovux/schemas/inference.py 
265
266
267
268
269
270
271
272
273
274
275
276
class ActiveLearningQueueItem(BaseModel):
    """A single review queue entry."""

    id: str
    image_path: Path
    dataset_path: Path
    score: float
    reason: str
    status: str
    predictions: list[Detection] = Field(default_factory=list)
    corrected_labels: list[Detection] | None = None
    created_at: datetime | None = None

ActiveLearningQueueListInput

Bases: BaseModel

Input for active_learning_queue_list.

Source code in src/fovux/schemas/inference.py 
301
302
303
304
305
306
class ActiveLearningQueueListInput(BaseModel):
    """Input for active_learning_queue_list."""

    dataset_path: Path | None = None
    status: Literal["pending", "reviewed", "skipped"] = "pending"
    limit: int = 100

ActiveLearningQueueListOutput

Bases: BaseModel

Output for active_learning_queue_list.

Source code in src/fovux/schemas/inference.py 
309
310
311
312
class ActiveLearningQueueListOutput(BaseModel):
    """Output for active_learning_queue_list."""

    queue_entries: list[ActiveLearningQueueItem] = Field(default_factory=list)

ActiveLearningQueueRankInput

Bases: BaseModel

Input for active_learning_queue_rank.

Source code in src/fovux/schemas/inference.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
class ActiveLearningQueueRankInput(BaseModel):
    """Input for active_learning_queue_rank."""

    checkpoint: str
    unlabeled_pool: Path
    dataset_path: Path
    strategy: Literal["entropy", "margin", "least_confident", "diversity", "error_likelihood"] = (
        "entropy"
    )
    limit: int = 50
    imgsz: int = 640
    conf: float = 0.25
    device: str = "auto"

ActiveLearningQueueRankOutput

Bases: BaseModel

Output for active_learning_queue_rank.

Source code in src/fovux/schemas/inference.py 
294
295
296
297
298
class ActiveLearningQueueRankOutput(BaseModel):
    """Output for active_learning_queue_rank."""

    ranked_count: int
    queue_entries: list[ActiveLearningQueueItem] = Field(default_factory=list)

ActiveLearningQueueSubmitInput

Bases: BaseModel

Input for active_learning_queue_submit.

Source code in src/fovux/schemas/inference.py 
315
316
317
318
319
320
class ActiveLearningQueueSubmitInput(BaseModel):
    """Input for active_learning_queue_submit."""

    entry_id: str
    corrected_labels: list[Detection] = Field(default_factory=list)
    dataset_split: Literal["train", "val", "test"] = "train"

ActiveLearningQueueSubmitOutput

Bases: BaseModel

Output for active_learning_queue_submit.

Source code in src/fovux/schemas/inference.py 
323
324
325
326
327
328
329
class ActiveLearningQueueSubmitOutput(BaseModel):
    """Output for active_learning_queue_submit."""

    entry_id: str
    status: str
    copied_image_path: Path
    written_label_path: Path

ActiveLearningSelectInput

Bases: BaseModel

Input for active_learning_select.

Source code in src/fovux/schemas/inference.py 
193
194
195
196
197
198
199
200
201
202
class ActiveLearningSelectInput(BaseModel):
    """Input for active_learning_select."""

    checkpoint: str
    unlabeled_pool: Path
    strategy: Literal["entropy", "margin", "least_confident"] = "entropy"
    budget: int = 100
    imgsz: int = 640
    conf: float = 0.25
    device: str = "auto"

ActiveLearningSelectOutput

Bases: BaseModel

Output from active_learning_select.

Source code in src/fovux/schemas/inference.py 
205
206
207
208
209
210
211
class ActiveLearningSelectOutput(BaseModel):
    """Output from active_learning_select."""

    checkpoint: str
    strategy: str
    budget: int
    selected: list[dict[str, object]] = Field(default_factory=list)

BatchDetectionSummary

Bases: BaseModel

Per-image summary returned by infer_batch.

Source code in src/fovux/schemas/inference.py 
 96
 97
 98
 99
100
101
102
class BatchDetectionSummary(BaseModel):
    """Per-image summary returned by infer_batch."""

    image_path: Path
    detection_count: int
    detections_by_class: dict[str, int] = Field(default_factory=dict)
    output_path: Path | None = None

BenchmarkLatencyInput

Bases: BaseModel

Input for benchmark_latency.

Source code in src/fovux/schemas/inference.py 
83
84
85
86
87
88
89
90
91
92
93
class BenchmarkLatencyInput(BaseModel):
    """Input for benchmark_latency."""

    model_path: Path
    backend: Literal["onnxruntime", "tflite", "tensorrt", "pytorch"] = "onnxruntime"
    device: str = "auto"
    imgsz: int = 640
    batch_size: int = 1
    num_warmup: int = 10
    num_iterations: int = 100
    threads: int = 4

BenchmarkLatencyOutput

Bases: BaseModel

Output for benchmark_latency.

Source code in src/fovux/schemas/inference.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
class BenchmarkLatencyOutput(BaseModel):
    """Output for benchmark_latency."""

    backend: str
    device: str
    num_iterations: int
    latency_p50_ms: float
    latency_p95_ms: float
    latency_p99_ms: float
    latency_mean_ms: float
    latency_std_ms: float
    throughput_fps: float
    peak_memory_mb: float

Detection

Bases: BaseModel

A single model detection.

Source code in src/fovux/schemas/inference.py 
12
13
14
15
16
17
18
class Detection(BaseModel):
    """A single model detection."""

    class_id: int
    class_name: str
    confidence: float
    bbox_xyxy: list[float] = Field(default_factory=list)

DistillModelInput

Bases: BaseModel

Input for distill_model.

Source code in src/fovux/schemas/inference.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
class DistillModelInput(BaseModel):
    """Input for distill_model."""

    teacher_checkpoint: str
    student_model: str = "yolov8n.pt"
    dataset_path: Path
    temperature: float = 4.0
    alpha: float = 0.7
    epochs: int = 100
    batch: int = 16
    imgsz: int = 640
    device: str = "auto"
    name: str | None = None

DistillModelOutput

Bases: BaseModel

Output from distill_model.

Source code in src/fovux/schemas/inference.py 
229
230
231
232
233
234
235
236
237
class DistillModelOutput(BaseModel):
    """Output from distill_model."""

    run_id: str
    status: str
    pid: int | None
    run_path: Path
    teacher_checkpoint: str
    student_model: str

InferBatchInput

Bases: BaseModel

Input for infer_batch.

Source code in src/fovux/schemas/inference.py 
105
106
107
108
109
110
111
112
113
114
115
116
class InferBatchInput(BaseModel):
    """Input for infer_batch."""

    checkpoint: str
    input_dir: Path
    output_dir: Path | None = None
    imgsz: int = 640
    conf: float = 0.25
    save_annotated: bool = True
    export_format: Literal["json", "csv", "yolo_labels"] = "json"
    device: str = "auto"
    batch_size: int = 32

InferBatchOutput

Bases: BaseModel

Output for infer_batch.

Source code in src/fovux/schemas/inference.py 
119
120
121
122
123
124
125
126
127
128
129
130
class InferBatchOutput(BaseModel):
    """Output for infer_batch."""

    checkpoint: str
    input_dir: Path
    output_dir: Path | None = None
    export_format: str
    processed_images: int
    detection_count: int
    manifest_path: Path
    annotated_dir: Path | None = None
    preview: list[BatchDetectionSummary] = Field(default_factory=list)

InferEnsembleInput

Bases: BaseModel

Input for infer_ensemble.

Source code in src/fovux/schemas/inference.py 
148
149
150
151
152
153
154
155
156
157
class InferEnsembleInput(BaseModel):
    """Input for infer_ensemble."""

    checkpoints: list[str]
    image_path: Path
    fusion_method: Literal["wbf", "nms", "soft-nms"] = "wbf"
    weights: list[float] | None = None
    imgsz: int = 640
    conf: float = 0.25
    device: str = "auto"

InferEnsembleOutput

Bases: BaseModel

Output from infer_ensemble.

Source code in src/fovux/schemas/inference.py 
160
161
162
163
164
165
166
167
class InferEnsembleOutput(BaseModel):
    """Output from infer_ensemble."""

    checkpoints: list[str]
    image_path: Path
    fusion_method: str
    detections: list[dict[str, object]] = Field(default_factory=list)
    detection_count: int = 0

InferImageInput

Bases: BaseModel

Input for infer_image.

Source code in src/fovux/schemas/inference.py 
21
22
23
24
25
26
27
28
29
30
31
class InferImageInput(BaseModel):
    """Input for infer_image."""

    checkpoint: str
    image_path: Path
    imgsz: int = 640
    conf: float = 0.25
    iou: float = 0.45
    device: str = "auto"
    save_image: bool = False
    output_path: Path | None = None

InferImageOutput

Bases: BaseModel

Output for infer_image.

Source code in src/fovux/schemas/inference.py 
34
35
36
37
38
39
40
41
42
43
class InferImageOutput(BaseModel):
    """Output for infer_image."""

    checkpoint: str
    image_path: Path
    detections: list[Detection] = Field(default_factory=list)
    detection_count: int = 0
    detections_by_class: dict[str, int] = Field(default_factory=dict)
    inference_duration_seconds: float
    output_path: Path | None = None

InferRtspInput

Bases: BaseModel

Input for infer_rtsp.

Source code in src/fovux/schemas/inference.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class InferRtspInput(BaseModel):
    """Input for infer_rtsp."""

    checkpoint: str
    rtsp_url: str
    duration_seconds: int = 30
    imgsz: int = 640
    conf: float = 0.25
    save_video: bool = False
    output_path: Path | None = None
    frame_skip: int = 0
    device: str = "auto"
    max_reconnect_attempts: int = 10

    @model_validator(mode="after")
    def _validate_output(self) -> InferRtspInput:
        if self.save_video and self.output_path is None:
            raise ValueError("output_path is required when save_video=True")
        return self

InferRtspOutput

Bases: BaseModel

Output for infer_rtsp.

Source code in src/fovux/schemas/inference.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class InferRtspOutput(BaseModel):
    """Output for infer_rtsp."""

    frames_processed: int
    frames_skipped: int
    dropped_frames: int
    avg_fps: float
    detection_count: int
    detections_by_class: dict[str, int] = Field(default_factory=dict)
    connection_status: str
    reconnect_attempts: int = 0
    output_fps: float = 0.0
    duration_actual_seconds: float
    output_path: Path | None = None

ModelCompareVisualInput

Bases: BaseModel

Input for model_compare_visual.

Source code in src/fovux/schemas/inference.py 
170
171
172
173
174
175
176
177
178
179
class ModelCompareVisualInput(BaseModel):
    """Input for model_compare_visual."""

    checkpoint_a: str
    checkpoint_b: str
    image_path: Path
    output_path: Path | None = None
    imgsz: int = 640
    conf: float = 0.25
    device: str = "auto"

ModelCompareVisualOutput

Bases: BaseModel

Output from model_compare_visual.

Source code in src/fovux/schemas/inference.py 
182
183
184
185
186
187
188
189
190
class ModelCompareVisualOutput(BaseModel):
    """Output from model_compare_visual."""

    checkpoint_a: str
    checkpoint_b: str
    image_path: Path
    output_path: Path
    detections_a: int
    detections_b: int

SyncToMlflowOutput

Bases: BaseModel

Output from sync_to_mlflow.

Source code in src/fovux/schemas/inference.py 
256
257
258
259
260
261
262
class SyncToMlflowOutput(BaseModel):
    """Output from sync_to_mlflow."""

    run_id: str
    tracking_uri: str
    metrics_logged: int
    params_logged: int

TrainAdjustInput

Bases: BaseModel

Input for train_adjust.

Source code in src/fovux/schemas/inference.py 
240
241
242
243
244
245
class TrainAdjustInput(BaseModel):
    """Input for train_adjust."""

    run_id: str
    learning_rate: float | None = None
    mosaic: bool | None = None

TrainAdjustOutput

Bases: BaseModel

Output from train_adjust.

Source code in src/fovux/schemas/inference.py 
248
249
250
251
252
253
class TrainAdjustOutput(BaseModel):
    """Output from train_adjust."""

    run_id: str
    control_path: Path
    applied: dict[str, object]

Pydantic schemas for model and run management tools.

DemoInitInput

Bases: BaseModel

Input for demo_init.

Source code in src/fovux/schemas/management.py 
171
172
173
174
class DemoInitInput(BaseModel):
    """Input for demo_init."""

    target_path: str

DemoInitOutput

Bases: BaseModel

Output for demo_init.

Source code in src/fovux/schemas/management.py 
177
178
179
180
181
182
183
184
class DemoInitOutput(BaseModel):
    """Output for demo_init."""

    dataset_path: Path
    run_id: str
    run_path: Path
    model_path: Path
    export_path: Path

DeploymentAdviseInput

Bases: BaseModel

Input for deployment_advise.

Source code in src/fovux/schemas/management.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class DeploymentAdviseInput(BaseModel):
    """Input for deployment_advise."""

    model_path: str
    target_profile: Literal[
        "cpu_server",
        "nvidia_gpu_tensorrt",
        "jetson",
        "raspberry_pi",
        "android_tflite",
        "browser_wasm",
    ]
    dataset_path: str | None = None
    imgsz: int = 640

DeploymentAdviseOutput

Bases: BaseModel

Output for deployment_advise.

Source code in src/fovux/schemas/management.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class DeploymentAdviseOutput(BaseModel):
    """Output for deployment_advise."""

    target_profile: str
    model_path: str
    format: str
    model_size_mb: float
    compatibility_preflight: dict[str, Any]
    quantization_recommendation: str
    readiness_score: int
    parity_check: dict[str, Any]
    benchmark_results: dict[str, Any]
    risk_warnings: list[str]
    runtime_snippets: dict[str, str]
    report_path: Path

ModelArtifact

Bases: BaseModel

Metadata for a tracked model artifact.

Source code in src/fovux/schemas/management.py 
14
15
16
17
18
19
20
21
22
23
24
25
class ModelArtifact(BaseModel):
    """Metadata for a tracked model artifact."""

    name: str
    path: Path
    source: Literal["runs", "models"]
    format: str
    size_mb: float
    task: str | None = None
    run_id: str | None = None
    status: str | None = None
    modified_at: datetime | None = None

ModelListInput

Bases: BaseModel

Input for model_list.

Source code in src/fovux/schemas/management.py 
37
38
39
40
41
class ModelListInput(BaseModel):
    """Input for model_list."""

    offset: int = 0
    limit: int = 50

ModelListOutput

Bases: BaseModel

Output for model_list.

Source code in src/fovux/schemas/management.py 
28
29
30
31
32
33
34
class ModelListOutput(BaseModel):
    """Output for model_list."""

    models: list[ModelArtifact] = Field(default_factory=list)
    total: int = 0
    offset: int = 0
    limit: int = 50

RunArchiveInput

Bases: BaseModel

Input for run_archive.

Source code in src/fovux/schemas/management.py 
120
121
122
123
124
125
class RunArchiveInput(BaseModel):
    """Input for run_archive."""

    run_id: RunId
    delete_original: bool = True
    dry_run: bool = False

RunArchiveOutput

Bases: BaseModel

Output from run_archive.

Source code in src/fovux/schemas/management.py 
128
129
130
131
132
133
134
135
class RunArchiveOutput(BaseModel):
    """Output from run_archive."""

    run_id: str
    archive_path: Path
    archived_files: int
    deleted_original: bool
    dry_run: bool = False

RunCompareInput

Bases: BaseModel

Input for run_compare.

Source code in src/fovux/schemas/management.py 
66
67
68
69
70
class RunCompareInput(BaseModel):
    """Input for run_compare."""

    run_ids: list[RunId] = Field(default_factory=list)
    output_path: Path | None = None

RunCompareOutput

Bases: BaseModel

Output for run_compare.

Source code in src/fovux/schemas/management.py 
73
74
75
76
77
78
79
80
81
82
83
class RunCompareOutput(BaseModel):
    """Output for run_compare."""

    compared_runs: list[RunMetricSummary] = Field(default_factory=list)
    best_run_id: str | None = None
    report_path: Path
    chart_path: Path
    config_diffs: dict[str, dict[str, Any]] = Field(default_factory=dict)
    pareto_frontier_run_ids: list[str] = Field(default_factory=list)
    model_cards: dict[str, str] = Field(default_factory=dict)
    suggested_next_experiment: str = ""

RunDeleteInput

Bases: BaseModel

Input for run_delete.

Source code in src/fovux/schemas/management.py 
86
87
88
89
90
91
92
class RunDeleteInput(BaseModel):
    """Input for run_delete."""

    run_id: RunId
    delete_files: bool = True
    force: bool = False
    dry_run: bool = False

RunDeleteOutput

Bases: BaseModel

Output for run_delete.

Source code in src/fovux/schemas/management.py 
 95
 96
 97
 98
 99
100
101
102
103
class RunDeleteOutput(BaseModel):
    """Output for run_delete."""

    run_id: str
    deleted_registry: bool
    deleted_files: bool
    dry_run: bool = False
    run_path: str | None = None
    affected_files_count: int = 0

RunMetricSummary

Bases: BaseModel

Comparable run summary with experiment intelligence metrics.

Source code in src/fovux/schemas/management.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class RunMetricSummary(BaseModel):
    """Comparable run summary with experiment intelligence metrics."""

    run_id: str
    status: str
    model: str
    epochs: int
    current_epoch: int | None = None
    best_map50: float | None = None
    best_map50_95: float | None = None
    precision: float | None = None
    recall: float | None = None
    latency_ms: float | None = None
    model_size_mb: float | None = None
    config: dict[str, Any] = Field(default_factory=dict)
    dataset_fingerprint: str | None = None
    export_target: str | None = None
    pareto_optimal: bool = False
    promotion_state: Literal["draft", "candidate", "approved", "deployed"] = "draft"
    run_path: Path

RunTagInput

Bases: BaseModel

Input for run_tag.

Source code in src/fovux/schemas/management.py 
106
107
108
109
110
class RunTagInput(BaseModel):
    """Input for run_tag."""

    run_id: RunId
    tags: list[str] = Field(default_factory=list)

RunTagOutput

Bases: BaseModel

Output for run_tag.

Source code in src/fovux/schemas/management.py 
113
114
115
116
117
class RunTagOutput(BaseModel):
    """Output for run_tag."""

    run_id: str
    tags: list[str]