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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Stable compatibility facade for the SQLite-backed run registry.

ArtifactRecord

Bases: Base

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

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    extra_json = Column(Text, nullable=False, default="{}")

AuditEventRecord

Bases: Base

ORM model for security/system audit events.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    details_json = Column(Text, nullable=False, default="{}")

Base

Bases: DeclarativeBase

SQLAlchemy declarative base.

Source code in src/fovux/core/run_registry/models.py

class Base(DeclarativeBase):
    """SQLAlchemy declarative base."""

DatasetRecord

Bases: Base

ORM model for datasets.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    extra_json = Column(Text, nullable=False, default="{}")

ExportRecord

Bases: Base

ORM model for model exports.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

MetricRecord

Bases: Base

ORM model for step or epoch-level metrics.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

ModelRecord

Bases: Base

ORM model for registered or downloaded models.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

OperationEventRecord

Bases: Base

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

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

OperationRecord

Bases: Base

ORM model for a background operation.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    started_at = Column(UtcDateTime, nullable=True)
    finished_at = Column(UtcDateTime, 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/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

RunEventRecord

Bases: Base

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

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    extra_json = Column(Text, nullable=False, default="{}")

RunRecord

Bases: Base

ORM model for a training run row.

Source code in src/fovux/core/run_registry/models.py

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(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )
    started_at = Column(UtcDateTime, nullable=True)
    finished_at = Column(UtcDateTime, 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 compatibility facade 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/run_registry/facade.py

class RunRegistry:
    """CRUD compatibility facade for the SQLite runs registry.

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

    def __init__(self, db_path: Path) -> None:
        """Compose focused repositories over one SQLite database boundary."""
        self._database = RegistryDatabase(db_path)
        self._engine: Engine = self._database.engine
        self._Session: sessionmaker[Session] = self._database.session_factory
        self._metadata = RunMetadataProvider()
        self._events = EventStore(self._Session)
        self._catalog = CatalogRepository(self._Session, self._metadata)
        self._runs = RunRepository(
            self._Session,
            metadata_provider=self._metadata,
            event_store=self._events,
            catalog_repository=self._catalog,
        )
        self._operations = OperationRepository(self._Session)
        self._artifacts = ArtifactRepository(
            self._Session,
            metadata_provider=self._metadata,
            event_store=self._events,
        )

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

    # Compatibility boundary: callers rely on the complete historical signature.
    def reserve_run_slot(
        self,  # NOSONAR(S107)
        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."""
        request = RunCreateRequest(
            run_id=run_id,
            run_path=run_path,
            model=model,
            dataset_path=dataset_path,
            task=task,
            epochs=epochs,
            tags=tags,
            extra=extra,
            dataset_fingerprint=dataset_fingerprint,
            config_hash=config_hash,
            code_version=code_version,
            env_summary=env_summary,
            parent_run_id=parent_run_id,
        )
        return self._runs.reserve_run_slot(request, max_concurrent_runs)

    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."""
        request = RunCreateRequest(
            run_id=run_id,
            run_path=run_path,
            model=model,
            dataset_path=dataset_path,
            task=task,
            epochs=epochs,
            tags=tags,
            extra=extra,
            dataset_fingerprint=dataset_fingerprint,
            config_hash=config_hash,
            code_version=code_version,
            env_summary=env_summary,
            parent_run_id=parent_run_id,
        )
        return self._runs.create_run(request)

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

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

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

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

    def update_tags(self, run_id: str, tags: list[str]) -> bool:
        """Replace a run's tag list."""
        return self._runs.update_tags(run_id, tags)

    def update_extra(self, run_id: str, extra: dict[str, Any]) -> bool:
        """Merge extra metadata into a run record."""
        return self._runs.update_extra(run_id, extra)

    def create_operation(
        self,
        op_id: str,
        tool: str,
        arguments: dict[str, Any],
        idempotency_key: str | None = None,
    ) -> OperationRecord:
        """Insert a new operation record."""
        return self._operations.create_operation(
            op_id,
            tool,
            arguments,
            idempotency_key,
        )

    def get_operation(self, op_id: str) -> OperationRecord | None:
        """Fetch an operation by ID."""
        return self._operations.get_operation(op_id)

    def get_operation_by_idempotency_key(
        self,
        idempotency_key: str,
    ) -> OperationRecord | None:
        """Fetch an operation by idempotency key."""
        return self._operations.get_operation_by_idempotency_key(idempotency_key)

    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 lifecycle fields."""
        self._operations.update_operation_status(
            op_id,
            status,
            error_type,
            error,
            result,
            run_id,
        )

    def update_operation_progress(self, op_id: str, progress: int) -> None:
        """Update operation progress percentage."""
        self._operations.update_operation_progress(op_id, progress)

    def list_operations(self, limit: int = 100) -> list[OperationRecord]:
        """List operations ordered by creation time descending."""
        return self._operations.list_operations(limit)

    def create_operation_event(
        self,
        op_id: str,
        event_type: str,
        data: dict[str, Any],
    ) -> OperationEventRecord:
        """Create a durable lifecycle event for an operation."""
        return self._events.create_operation_event(op_id, event_type, data)

    def list_operation_events(
        self,
        last_event_id: int | None = None,
        limit: int = 1000,
    ) -> list[OperationEventRecord]:
        """List operation events newer than an optional event ID."""
        return self._events.list_operation_events(last_event_id, limit)

    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 an artifact record."""
        return self._artifacts.add_artifact(
            artifact_id,
            run_id,
            artifact_type,
            path,
            sha256,
            size,
            extra,
        )

    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 its associated artifact."""
        return self._artifacts.record_export(
            export_id,
            run_id,
            source_checkpoint,
            artifact_path,
            format,
            duration_s,
            validation_result,
        )

    def list_run_events(
        self,
        run_id: str | None = None,
        limit: int = 1000,
    ) -> list[RunEventRecord]:
        """List run lifecycle and audit events."""
        return self._events.list_run_events(run_id, limit)

    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."""
        return self._events.log_audit_event(
            actor,
            action,
            entity_type,
            entity_id,
            details,
        )

    def list_audit_events(
        self,
        limit: int = 100,
        offset: int = 0,
    ) -> list[AuditEventRecord]:
        """List audit events ordered by creation time descending."""
        return self._events.list_audit_events(limit, offset)

    def list_artifacts(
        self,
        run_id: str | None = None,
        limit: int = 1000,
    ) -> list[ArtifactRecord]:
        """List registered artifacts."""
        return self._artifacts.list_artifacts(run_id, limit)

    def get_dataset(self, fingerprint: str) -> DatasetRecord | None:
        """Fetch a dataset record by fingerprint."""
        return self._catalog.get_dataset(fingerprint)

    def list_datasets(self, limit: int = 100) -> list[DatasetRecord]:
        """List registered datasets."""
        return self._catalog.list_datasets(limit)

    def list_exports(
        self,
        run_id: str | None = None,
        limit: int = 100,
    ) -> list[ExportRecord]:
        """List recorded exports."""
        return self._artifacts.list_exports(run_id, limit)

    def add_metric(
        self,
        run_id: str,
        epoch: int,
        key: str,
        value: float,
    ) -> MetricRecord:
        """Record an epoch-level training metric."""
        return self._catalog.add_metric(run_id, epoch, key, value)

    def list_metrics(
        self,
        run_id: str,
        limit: int = 1000,
    ) -> list[MetricRecord]:
        """List metrics for a run."""
        return self._catalog.list_metrics(run_id, limit)

    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."""
        return self._catalog.add_review_queue_entry(
            entry_id,
            image_path,
            dataset_path,
            score,
            reason,
            predictions,
        )

    def get_review_queue_entry(self, entry_id: str) -> ReviewQueueEntry | None:
        """Fetch a review queue entry by ID."""
        return self._catalog.get_review_queue_entry(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."""
        return self._catalog.list_review_queue_entries(dataset_path, status, limit)

    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."""
        return self._catalog.update_review_queue_status(
            entry_id,
            status,
            corrected_labels,
        )

__init__(db_path)

Compose focused repositories over one SQLite database boundary.

Source code in src/fovux/core/run_registry/facade.py

def __init__(self, db_path: Path) -> None:
    """Compose focused repositories over one SQLite database boundary."""
    self._database = RegistryDatabase(db_path)
    self._engine: Engine = self._database.engine
    self._Session: sessionmaker[Session] = self._database.session_factory
    self._metadata = RunMetadataProvider()
    self._events = EventStore(self._Session)
    self._catalog = CatalogRepository(self._Session, self._metadata)
    self._runs = RunRepository(
        self._Session,
        metadata_provider=self._metadata,
        event_store=self._events,
        catalog_repository=self._catalog,
    )
    self._operations = OperationRepository(self._Session)
    self._artifacts = ArtifactRepository(
        self._Session,
        metadata_provider=self._metadata,
        event_store=self._events,
    )

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

Register or merge an artifact record.

Source code in src/fovux/core/run_registry/facade.py

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 an artifact record."""
    return self._artifacts.add_artifact(
        artifact_id,
        run_id,
        artifact_type,
        path,
        sha256,
        size,
        extra,
    )

add_metric(run_id, epoch, key, value)

Record an epoch-level training metric.

Source code in src/fovux/core/run_registry/facade.py

def add_metric(
    self,
    run_id: str,
    epoch: int,
    key: str,
    value: float,
) -> MetricRecord:
    """Record an epoch-level training metric."""
    return self._catalog.add_metric(run_id, epoch, key, value)

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/run_registry/facade.py

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."""
    return self._catalog.add_review_queue_entry(
        entry_id,
        image_path,
        dataset_path,
        score,
        reason,
        predictions,
    )

close()

Dispose the SQLite engine and release pooled connections.

Source code in src/fovux/core/run_registry/facade.py

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

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

Insert a new operation record.

Source code in src/fovux/core/run_registry/facade.py

def create_operation(
    self,
    op_id: str,
    tool: str,
    arguments: dict[str, Any],
    idempotency_key: str | None = None,
) -> OperationRecord:
    """Insert a new operation record."""
    return self._operations.create_operation(
        op_id,
        tool,
        arguments,
        idempotency_key,
    )

create_operation_event(op_id, event_type, data)

Create a durable lifecycle event for an operation.

Source code in src/fovux/core/run_registry/facade.py

def create_operation_event(
    self,
    op_id: str,
    event_type: str,
    data: dict[str, Any],
) -> OperationEventRecord:
    """Create a durable lifecycle event for an operation."""
    return self._events.create_operation_event(op_id, event_type, data)

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.

Source code in src/fovux/core/run_registry/facade.py

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."""
    request = RunCreateRequest(
        run_id=run_id,
        run_path=run_path,
        model=model,
        dataset_path=dataset_path,
        task=task,
        epochs=epochs,
        tags=tags,
        extra=extra,
        dataset_fingerprint=dataset_fingerprint,
        config_hash=config_hash,
        code_version=code_version,
        env_summary=env_summary,
        parent_run_id=parent_run_id,
    )
    return self._runs.create_run(request)

delete_run(run_id)

Delete a run record.

Source code in src/fovux/core/run_registry/facade.py

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

get_dataset(fingerprint)

Fetch a dataset record by fingerprint.

Source code in src/fovux/core/run_registry/facade.py

def get_dataset(self, fingerprint: str) -> DatasetRecord | None:
    """Fetch a dataset record by fingerprint."""
    return self._catalog.get_dataset(fingerprint)

get_operation(op_id)

Fetch an operation by ID.

Source code in src/fovux/core/run_registry/facade.py

def get_operation(self, op_id: str) -> OperationRecord | None:
    """Fetch an operation by ID."""
    return self._operations.get_operation(op_id)

get_operation_by_idempotency_key(idempotency_key)

Fetch an operation by idempotency key.

Source code in src/fovux/core/run_registry/facade.py

def get_operation_by_idempotency_key(
    self,
    idempotency_key: str,
) -> OperationRecord | None:
    """Fetch an operation by idempotency key."""
    return self._operations.get_operation_by_idempotency_key(idempotency_key)

get_review_queue_entry(entry_id)

Fetch a review queue entry by ID.

Source code in src/fovux/core/run_registry/facade.py

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

get_run(run_id)

Fetch a run by ID.

Source code in src/fovux/core/run_registry/facade.py

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

list_artifacts(run_id=None, limit=1000)

List registered artifacts.

Source code in src/fovux/core/run_registry/facade.py

def list_artifacts(
    self,
    run_id: str | None = None,
    limit: int = 1000,
) -> list[ArtifactRecord]:
    """List registered artifacts."""
    return self._artifacts.list_artifacts(run_id, limit)

list_audit_events(limit=100, offset=0)

List audit events ordered by creation time descending.

Source code in src/fovux/core/run_registry/facade.py

def list_audit_events(
    self,
    limit: int = 100,
    offset: int = 0,
) -> list[AuditEventRecord]:
    """List audit events ordered by creation time descending."""
    return self._events.list_audit_events(limit, offset)

list_datasets(limit=100)

List registered datasets.

Source code in src/fovux/core/run_registry/facade.py

def list_datasets(self, limit: int = 100) -> list[DatasetRecord]:
    """List registered datasets."""
    return self._catalog.list_datasets(limit)

list_exports(run_id=None, limit=100)

List recorded exports.

Source code in src/fovux/core/run_registry/facade.py

def list_exports(
    self,
    run_id: str | None = None,
    limit: int = 100,
) -> list[ExportRecord]:
    """List recorded exports."""
    return self._artifacts.list_exports(run_id, limit)

list_metrics(run_id, limit=1000)

List metrics for a run.

Source code in src/fovux/core/run_registry/facade.py

def list_metrics(
    self,
    run_id: str,
    limit: int = 1000,
) -> list[MetricRecord]:
    """List metrics for a run."""
    return self._catalog.list_metrics(run_id, limit)

list_operation_events(last_event_id=None, limit=1000)

List operation events newer than an optional event ID.

Source code in src/fovux/core/run_registry/facade.py

def list_operation_events(
    self,
    last_event_id: int | None = None,
    limit: int = 1000,
) -> list[OperationEventRecord]:
    """List operation events newer than an optional event ID."""
    return self._events.list_operation_events(last_event_id, limit)

list_operations(limit=100)

List operations ordered by creation time descending.

Source code in src/fovux/core/run_registry/facade.py

def list_operations(self, limit: int = 100) -> list[OperationRecord]:
    """List operations ordered by creation time descending."""
    return self._operations.list_operations(limit)

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

List active-learning review queue entries.

Source code in src/fovux/core/run_registry/facade.py

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."""
    return self._catalog.list_review_queue_entries(dataset_path, status, limit)

list_run_events(run_id=None, limit=1000)

List run lifecycle and audit events.

Source code in src/fovux/core/run_registry/facade.py

def list_run_events(
    self,
    run_id: str | None = None,
    limit: int = 1000,
) -> list[RunEventRecord]:
    """List run lifecycle and audit events."""
    return self._events.list_run_events(run_id, limit)

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

List runs, optionally filtered by status.

Source code in src/fovux/core/run_registry/facade.py

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

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

Log an audit event to the database.

Source code in src/fovux/core/run_registry/facade.py

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."""
    return self._events.log_audit_event(
        actor,
        action,
        entity_type,
        entity_id,
        details,
    )

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

Record a model export and its associated artifact.

Source code in src/fovux/core/run_registry/facade.py

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 its associated artifact."""
    return self._artifacts.record_export(
        export_id,
        run_id,
        source_checkpoint,
        artifact_path,
        format,
        duration_s,
        validation_result,
    )

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.

Source code in src/fovux/core/run_registry/facade.py

def reserve_run_slot(
    self,  # NOSONAR(S107)
    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."""
    request = RunCreateRequest(
        run_id=run_id,
        run_path=run_path,
        model=model,
        dataset_path=dataset_path,
        task=task,
        epochs=epochs,
        tags=tags,
        extra=extra,
        dataset_fingerprint=dataset_fingerprint,
        config_hash=config_hash,
        code_version=code_version,
        env_summary=env_summary,
        parent_run_id=parent_run_id,
    )
    return self._runs.reserve_run_slot(request, max_concurrent_runs)

update_extra(run_id, extra)

Merge extra metadata into a run record.

Source code in src/fovux/core/run_registry/facade.py

def update_extra(self, run_id: str, extra: dict[str, Any]) -> bool:
    """Merge extra metadata into a run record."""
    return self._runs.update_extra(run_id, extra)

update_operation_progress(op_id, progress)

Update operation progress percentage.

Source code in src/fovux/core/run_registry/facade.py

def update_operation_progress(self, op_id: str, progress: int) -> None:
    """Update operation progress percentage."""
    self._operations.update_operation_progress(op_id, progress)

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

Update operation lifecycle fields.

Source code in src/fovux/core/run_registry/facade.py

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 lifecycle fields."""
    self._operations.update_operation_status(
        op_id,
        status,
        error_type,
        error,
        result,
        run_id,
    )

update_review_queue_status(entry_id, status, corrected_labels=None)

Update review queue entry status and corrections.

Source code in src/fovux/core/run_registry/facade.py

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."""
    return self._catalog.update_review_queue_status(
        entry_id,
        status,
        corrected_labels,
    )

update_status(run_id, status, pid=None)

Update run status and optional process ID.

Source code in src/fovux/core/run_registry/facade.py

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

update_tags(run_id, tags)

Replace a run's tag list.

Source code in src/fovux/core/run_registry/facade.py

def update_tags(self, run_id: str, tags: list[str]) -> bool:
    """Replace a run's tag list."""
    return self._runs.update_tags(run_id, tags)

SchemaMigrationRecord

Bases: Base

ORM model for schema migrations.

Source code in src/fovux/core/run_registry/models.py

class SchemaMigrationRecord(Base):
    """ORM model for schema migrations."""

    __tablename__ = "schema_migrations"

    version = Column(Integer, primary_key=True)
    applied_at = Column(
        UtcDateTime,
        nullable=False,
        default=_utcnow_naive,
    )

TagRecord

Bases: Base

ORM model for entity tags.

Source code in src/fovux/core/run_registry/models.py

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)

UtcDateTime

Bases: TypeDecorator[datetime]

Store datetimes as ISO-8601 text to avoid sqlite3 default adapters.

Source code in src/fovux/core/run_registry/models.py

class UtcDateTime(TypeDecorator[datetime]):
    """Store datetimes as ISO-8601 text to avoid sqlite3 default adapters."""

    impl = String
    cache_ok = True

    def process_bind_param(self, value: datetime | None, _dialect: object) -> str | None:
        """Serialize a Python datetime before binding it to SQLite."""
        if value is None:
            return None
        return _serialize_datetime(value)

    def process_result_value(
        self, value: str | datetime | None, _dialect: object
    ) -> datetime | None:
        """Deserialize a stored SQLite value into a Python datetime."""
        if value is None:
            return None
        return _deserialize_datetime(value)

process_bind_param(value, _dialect)

Serialize a Python datetime before binding it to SQLite.

Source code in src/fovux/core/run_registry/models.py

def process_bind_param(self, value: datetime | None, _dialect: object) -> str | None:
    """Serialize a Python datetime before binding it to SQLite."""
    if value is None:
        return None
    return _serialize_datetime(value)

process_result_value(value, _dialect)

Deserialize a stored SQLite value into a Python datetime.

Source code in src/fovux/core/run_registry/models.py

def process_result_value(
    self, value: str | datetime | None, _dialect: object
) -> datetime | None:
    """Deserialize a stored SQLite value into a Python datetime."""
    if value is None:
        return None
    return _deserialize_datetime(value)

close_registry(db_path=None)

Dispose cached registry engines for one database or all databases.

Source code in src/fovux/core/runs.py

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

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

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

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))

Domain router aggregation and compatibility exports for the local HTTP API.

build_http_router()

Assemble every domain router exactly once.

Source code in src/fovux/http/routes/__init__.py

def build_http_router() -> APIRouter:
    """Assemble every domain router exactly once."""
    root = APIRouter()
    for domain_router in (
        health_router,
        runs_router,
        tools_router,
        operations_router,
        lineage_router,
        resources_router,
    ):
        root.include_router(domain_router)
    return root

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

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))

sse_events_route(request) async

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

Source code in src/fovux/http/routes/operations.py

@router.get("/events")
async def sse_events_route(request: Request) -> StreamingResponse:
    """Server-Sent Events (SSE) stream of all operations events with resume support."""
    services = _services(request)
    raw_last_id = request.headers.get("Last-Event-ID") or request.query_params.get("last_event_id")
    try:
        last_event_id = int(raw_last_id) if raw_last_id else None
    except ValueError:
        last_event_id = None
    shutdown_event = cast(asyncio.Event, request.app.state.shutdown_event)
    stream = services.operations.event_stream(
        services.operation_runtime,
        last_event_id=last_event_id,
        disconnect_check=request.is_disconnected,
        shutdown_event=shutdown_event,
    )
    return StreamingResponse(
        stream,
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Schemas

Pydantic schemas for dataset tools.

AutoFixItem

Bases: BaseModel

An auto-fix recommendation.

Source code in src/fovux/schemas/dataset.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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)
    preflight_approval_reason: str | None = None

    # 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

@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

@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

@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

@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

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
    ready: bool = False
    blockers: list[str] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)
    next_actions: list[str] = Field(default_factory=list)
    override_required: bool = False
    override_hint: str | None = None

TrainResumeInput

Bases: BaseModel

Input for train_resume tool.

Source code in src/fovux/schemas/training.py

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

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

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)
    preflight_approval_reason: str | None = None

    # 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

@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

@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

@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

@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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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
    baseline_path: Path | None = None

BenchmarkLatencyOutput

Bases: BaseModel

Output for benchmark_latency.

Source code in src/fovux/schemas/inference.py

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
    num_warmup: int = 0
    batch_size: int = 1
    input_shape: list[int] = Field(default_factory=list)
    environment: dict[str, Any] = Field(default_factory=dict)
    artifact: dict[str, Any] = Field(default_factory=dict)
    comparison: dict[str, Any] = Field(default_factory=dict)
    reproducibility_notes: list[str] = Field(default_factory=list)

Detection

Bases: BaseModel

A single model detection.

Source code in src/fovux/schemas/inference.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

class RunTagOutput(BaseModel):
    """Output for run_tag."""

    run_id: str
    tags: list[str]