filesystem_backend

FileSystem backend for PRIMARY mode.

This backend reads and writes model reference JSON files directly on the local filesystem. It is the source of truth for PRIMARY mode instances and never interacts with GitHub.

CategoryMetadataPopulationResult

Bases: BaseModel

Result from ensure_category_metadata_populated method.

Source code in src/horde_model_reference/backends/filesystem_backend.py

class CategoryMetadataPopulationResult(BaseModel):
    """Result from ensure_category_metadata_populated method."""

    category_metadata_initialized: bool = Field(description="Whether v2 CategoryMetadata was initialized")
    legacy_metadata_initialized: bool = Field(description="Whether legacy CategoryMetadata was initialized")
    models_updated: int = Field(description="Number of models that had metadata populated")
    timestamp_used: int = Field(description="Unix timestamp used for metadata population")

category_metadata_initialized class-attribute instance-attribute

category_metadata_initialized: bool = Field(
    description="Whether v2 CategoryMetadata was initialized"
)

legacy_metadata_initialized class-attribute instance-attribute

legacy_metadata_initialized: bool = Field(
    description="Whether legacy CategoryMetadata was initialized"
)

models_updated class-attribute instance-attribute

models_updated: int = Field(
    description="Number of models that had metadata populated"
)

timestamp_used class-attribute instance-attribute

timestamp_used: int = Field(
    description="Unix timestamp used for metadata population"
)

AllMetadataPopulationResult

Bases: BaseModel

Result from ensure_all_metadata_populated method.

Source code in src/horde_model_reference/backends/filesystem_backend.py

class AllMetadataPopulationResult(BaseModel):
    """Result from ensure_all_metadata_populated method."""

    categories_processed: list[str] = Field(
        description="List of category names that were processed",
        default_factory=list,
    )
    total_categories: int = Field(
        description="Total number of categories processed",
        default=0,
    )
    total_models_updated: int = Field(
        description="Total number of models updated across all categories",
        default=0,
    )
    total_metadata_initialized: int = Field(
        description="Total number of metadata files initialized",
        default=0,
    )

categories_processed class-attribute instance-attribute

categories_processed: list[str] = Field(
    description="List of category names that were processed",
    default_factory=list,
)

total_categories class-attribute instance-attribute

total_categories: int = Field(
    description="Total number of categories processed",
    default=0,
)

total_models_updated class-attribute instance-attribute

total_models_updated: int = Field(
    description="Total number of models updated across all categories",
    default=0,
)

total_metadata_initialized class-attribute instance-attribute

total_metadata_initialized: int = Field(
    description="Total number of metadata files initialized",
    default=0,
)

FileSystemBackend

Bases: ReplicaBackendBase

Backend that reads/writes model references directly on the local filesystem.

Source code in src/horde_model_reference/backends/filesystem_backend.py

class FileSystemBackend(ReplicaBackendBase):
    """Backend that reads/writes model references directly on the local filesystem."""

    def __init__(
        self,
        *,
        base_path: str | Path = horde_model_reference_paths.base_path,
        cache_ttl_seconds: int = 60,
        replicate_mode: ReplicateMode = ReplicateMode.PRIMARY,
        skip_startup_metadata_population: bool = False,
        audit_writer: AuditTrailWriter | None = None,
    ) -> None:
        """Initialize the FileSystem backend.

        Args:
            base_path: Base path for model reference files.
            cache_ttl_seconds: TTL for internal cache in seconds.
            replicate_mode: Must be PRIMARY.
            skip_startup_metadata_population: If True, skip automatic metadata population on startup.
                This is used when GitHub seeding will handle metadata population instead.
            audit_writer: Optional AuditTrailWriter for emitting audit events on CRUD operations.

        Raises:
            ValueError: If replicate_mode is not PRIMARY.

        """
        if replicate_mode != ReplicateMode.PRIMARY:
            raise ValueError(
                "FileSystemBackend can only be used in PRIMARY mode. "
                "For REPLICA mode, use GitHubBackend or HTTPBackend."
            )
        super().__init__(mode=replicate_mode, cache_ttl_seconds=cache_ttl_seconds)

        self.base_path = Path(base_path)
        self._metadata_manager = MetadataManager(self.base_path)
        self._audit_writer = audit_writer

        logger.debug(f"FileSystemBackend initialized with base_path={self.base_path}")

        # Create empty files for categories that have no legacy format available
        # This ensures consistent behavior between CI (fresh environment) and local (may have existing files)
        from horde_model_reference.meta_consts import get_no_legacy_format_categories

        for category in get_no_legacy_format_categories():
            if not isinstance(category, MODEL_REFERENCE_CATEGORY):
                continue
            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            if file_path and not file_path.exists():
                file_path.parent.mkdir(parents=True, exist_ok=True)
                file_path.write_text("{}")
                logger.info(f"Created empty file for {category} (no legacy format available)")

        # Populate metadata on startup if not skipped
        if not skip_startup_metadata_population:
            logger.info("Running startup metadata population check")
            self.ensure_all_metadata_populated()
        else:
            logger.debug("Skipping startup metadata population (will be handled by GitHub seeding)")

    def _resolve_legacy_text_generation_path(self) -> tuple[Path, bool]:
        """Return the legacy text_generation path and whether it is CSV-based."""
        csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
            MODEL_REFERENCE_CATEGORY.text_generation,
            base_path=self.base_path,
        )
        json_path = csv_path.with_name("text_generation.json")

        # If canonical_format is legacy, prefer CSV if it exists, otherwise JSON if it exists
        if horde_model_reference_settings.canonical_format == CanonicalFormat.LEGACY:
            if csv_path.exists():
                return csv_path, True
            if json_path.exists():
                return json_path, False

            # Neither file exists, default to CSV path
            return csv_path, True

        # Otherwise, prefer CSV if it exists, else JSON if it exists
        if csv_path.exists():
            return csv_path, True
        if json_path.exists():
            return json_path, False

        # Default to CSV path if nothing exists
        return csv_path, True

    @override
    def _get_file_path_for_validation(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
        """Return the file path for mtime validation.

        Args:
            category: The category to get the file path for.

        Returns:
            Path | None: Path to file for mtime validation.

        """
        return horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

    @override
    def _get_legacy_file_path_for_validation(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
        """Return the legacy file path for mtime validation.

        Args:
            category: The category to get the legacy file path for.

        Returns:
            Path | None: Path to legacy file for mtime validation.

        """
        return horde_model_reference_paths.get_legacy_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

    def _mark_category_modified(self, category: MODEL_REFERENCE_CATEGORY, file_path: Path) -> None:
        """Mark a category as modified after a write operation.

        This invalidates the cache and triggers callbacks to notify manager.

        Args:
            category: Category that was modified.
            file_path: Path to the file that was modified.

        """
        # Use mark_stale() to trigger callbacks, not _invalidate_cache()
        self.mark_stale(category)
        logger.debug(f"Marked category {category} as modified")

    def _mark_legacy_category_modified(self, category: MODEL_REFERENCE_CATEGORY, legacy_file_path: Path) -> None:
        """Mark a legacy category as modified after a write operation.

        This invalidates the legacy cache and triggers callbacks.

        Args:
            category: Category that was modified.
            legacy_file_path: Path to the legacy file that was modified.

        """
        # Use mark_stale() to trigger callbacks
        self.mark_stale(category)
        logger.debug(f"Marked legacy category {category} as modified")

    def _read_legacy_csv_to_dict(self, file_path: Path) -> dict[str, Any]:
        """Read legacy CSV file (models.csv format) and convert to dict format.

        Uses the shared ``csv_rows_to_legacy_dict`` to replicate convert.py exactly,
        including defaults.json merging, instruct_format, correct field ordering,
        and backend prefix generation (3 entries per model).

        Args:
            file_path: Path to the legacy CSV file.

        Returns:
            dict[str, Any]: Model data with 3 entries per CSV row (matching db.json format).

        """
        parsed_rows, parse_issues = parse_legacy_text_csv_file(file_path)
        for issue in parse_issues:
            logger.warning(f"Legacy CSV issue for {issue.row_identifier}: {issue.message}")

        data = csv_rows_to_legacy_dict(parsed_rows, with_backend_prefixes=True)
        logger.debug(f"Read {len(data)} models from legacy CSV (with backend prefixes) from {file_path}")
        return data

    def _append_audit_event(
        self,
        *,
        domain: CanonicalFormat,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        operation: AuditOperation,
        payload: AuditPayload,
        logical_user_id: str | None,
        request_id: str | None,
    ) -> None:
        if self._audit_writer is None or logical_user_id is None:
            return

        try:
            self._audit_writer.append_event(
                domain=domain,
                category=category.value,
                model_name=model_name,
                operation=operation,
                logical_user_id=logical_user_id,
                payload=payload,
                request_id=request_id,
            )
        except OSError as exc:  # pragma: no cover - audit writes must not break CRUD
            logger.warning(f"Failed to append {domain} audit event for {category}/{model_name}: {exc}")

    def _read_csv_to_dict(self, file_path: Path) -> dict[str, Any]:
        """Read CSV file and convert to dict format (grouped by base name, no backend prefixes).

        This reads the grouped CSV format and returns a dict with one entry per base model.
        No backend prefix duplicates are generated here - that only happens during GitHub sync.

        Args:
            file_path: Path to the CSV file.

        Returns:
            dict[str, Any]: Model data with one entry per base model.

        Raises:
            Exception: If CSV parsing fails.

        """
        data: dict[str, Any] = {}
        parsed_rows, parse_issues = parse_legacy_text_csv_file(file_path)
        for issue in parse_issues:
            logger.warning(f"CSV issue for {issue.row_identifier}: {issue.message}")

        for csv_row in parsed_rows:
            name = csv_row.name
            model_name = name.split("/")[1] if "/" in name else name
            tags = set(csv_row.tags)
            if csv_row.style:
                tags.add(csv_row.style)
            tags.add(f"{round(csv_row.parameters_bn, 0):.0f}B")

            settings_dict = dict(csv_row.settings) if csv_row.settings is not None else {}

            display_name = csv_row.display_name
            if not display_name:
                display_name = re.sub(r" +", " ", re.sub(r"[-_]", " ", model_name)).strip()

            record: dict[str, Any] = {
                "name": name,
                "model_name": model_name,
                "parameters": csv_row.parameters,
                "description": csv_row.description,
                "version": csv_row.version,
                "style": csv_row.style,
                "nsfw": csv_row.nsfw,
                "baseline": csv_row.baseline,
                "url": csv_row.url,
                "tags": sorted(tags),
                "settings": settings_dict,
                "display_name": display_name,
            }

            record = {k: v for k, v in record.items() if v or v is False}
            data[name] = record

        logger.debug(f"Read {len(data)} models from CSV (grouped, no backend prefixes) from {file_path}")
        return data

    def _write_dict_to_csv(self, data: dict[str, Any], file_path: Path) -> None:
        """Write dict format to CSV file (removes backend prefix duplicates).

        This writes the grouped CSV format with one entry per base model.
        Any backend-prefixed entries in the input are filtered out.

        Args:
            data: Model data dict (may contain backend-prefixed duplicates).
            file_path: Path to write the CSV file.

        Raises:
            Exception: If CSV writing fails.

        """
        from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

        # Filter out backend-prefixed entries
        base_models: dict[str, Any] = {}
        for model_name, record in data.items():
            if has_legacy_text_backend_prefix(model_name):
                logger.debug(f"Skipping backend-prefixed entry during CSV write: {model_name}")
                continue
            base_models[model_name] = record

        # Convert to CSV rows
        csv_rows: list[dict[str, str]] = []

        for model_name, record in base_models.items():
            # Extract parameters in billions
            parameters_int = record.get("parameters", 0)
            params_bn = float(parameters_int) / 1_000_000_000

            # Extract tags and remove auto-generated ones
            tags = record.get("tags", [])
            tags_set = set(tags) if isinstance(tags, list) else set()

            # Remove style tag
            style = record.get("style")
            if style and style in tags_set:
                tags_set.discard(style)

            # Remove parameter size tag
            size_tag = f"{round(params_bn, 0):.0f}B"
            tags_set.discard(size_tag)

            # Serialize settings to compact JSON
            settings = record.get("settings", {})
            settings_str = json.dumps(settings, separators=(",", ":")) if settings else ""

            # Check if display_name is auto-generated (if so, omit it)
            display_name = record.get("display_name", "")
            model_name_field = record.get("model_name", model_name)
            auto_display = re.sub(r" +", " ", re.sub(r"[-_]", " ", model_name_field)).strip()
            if display_name == auto_display:
                display_name = ""

            csv_row = {
                "name": model_name,
                "parameters_bn": f"{params_bn:.1f}",
                "description": record.get("description", ""),
                "version": record.get("version", ""),
                "style": style or "",
                "nsfw": str(record.get("nsfw", False)).lower(),
                "baseline": record.get("baseline", ""),
                "url": record.get("url", ""),
                "tags": ",".join(sorted(tags_set)),
                "settings": settings_str,
                "display_name": display_name,
            }

            csv_rows.append(csv_row)

        # Write CSV file
        fieldnames = [
            "name",
            "parameters_bn",
            "description",
            "version",
            "style",
            "nsfw",
            "baseline",
            "url",
            "tags",
            "settings",
            "display_name",
        ]

        with open(file_path, "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(csv_rows)

        logger.debug(f"Wrote {len(csv_rows)} models to CSV (grouped, no backend prefixes) to {file_path}")

    @override
    def fetch_category(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        *,
        force_refresh: bool = False,
    ) -> dict[str, Any] | None:
        """Fetch model reference data for a specific category from filesystem.

        All v2 format files (including text_generation.json) are in JSON format.
        CSV format is only used for legacy files (legacy/models.csv).

        Args:
            category: The category to fetch.
            force_refresh: If True, bypass cache and read from disk.

        Returns:
            dict[str, Any] | None: The model reference data, or None if file doesn't exist.

        """
        with self._lock:
            if not (force_refresh or self.should_fetch_data(category)):
                return self._get_from_cache(category)

            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )

            if not file_path or not file_path.exists():
                logger.debug(f"File not found for {category}: {file_path}")
                self._store_in_cache(category, None)
                return None

            try:
                # All v2 files are JSON format (including text_generation.json)
                with open(file_path, encoding="utf-8") as f:
                    data: dict[str, Any] = json.load(f)

                self._store_in_cache(category, data)
                logger.debug(f"Loaded {category} from {file_path}")
                return data

            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read {file_path}: {e}")
                self._invalidate_cache(category)
                return None

    @override
    def fetch_all_categories(
        self,
        *,
        force_refresh: bool = False,
    ) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]:
        """Fetch model reference data for all categories.

        Args:
            force_refresh: If True, bypass cache for all categories.

        Returns:
            dict mapping categories to their model reference data.

        """
        result: dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None] = {}

        for category in MODEL_REFERENCE_CATEGORY:
            result[category] = self.fetch_category(category, force_refresh=force_refresh)

        return result

    async def fetch_category_async(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        *,
        httpx_client: httpx.AsyncClient | None = None,
        force_refresh: bool = False,
    ) -> dict[str, Any] | None:
        """Asynchronously fetch model reference data for a category.

        Note: File I/O is still synchronous as async file I/O doesn't provide
        significant benefits for small JSON files.

        Args:
            category: The category to fetch.
            httpx_client: Optional httpx async client for downloads.
            force_refresh: If True, bypass cache.

        Returns:
            dict[str, Any] | None: The model reference data.

        """
        async with self._async_lock:
            if not (force_refresh or self.should_fetch_data(category)):
                return self._get_from_cache(category)

            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            if not file_path or not file_path.exists():
                logger.debug(f"File not found for {category}: {file_path}")
                self._store_in_cache(category, None)
                return None
            try:
                async with aiofiles.open(file_path, encoding="utf-8") as f:
                    content = await f.read()
                    data: dict[str, Any] = json.loads(content)

                self._store_in_cache(category, data)
                logger.debug(f"Loaded {category} from {file_path} asynchronously")
                return data
            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read {file_path} asynchronously: {e}")
                self._invalidate_cache(category)
                return None

    @override
    async def fetch_all_categories_async(
        self,
        *,
        httpx_client: httpx.AsyncClient | None = None,
        force_refresh: bool = False,
    ) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]:
        """Asynchronously fetch all categories (delegates to sync method)."""
        result: dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None] = {}

        for category in MODEL_REFERENCE_CATEGORY:
            result[category] = await self.fetch_category_async(
                category,
                httpx_client=httpx_client,
                force_refresh=force_refresh,
            )

        return result

    @override
    def get_category_file_path(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
        """Get the file path for a category's data.

        Args:
            category: The category to get path for.

        Returns:
            Path | None: Path to the JSON file, or None if not configured.

        """
        return horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

    @override
    def get_all_category_file_paths(self) -> dict[MODEL_REFERENCE_CATEGORY, Path | None]:
        """Get file paths for all categories.

        Returns:
            dict: Mapping of categories to their file paths.

        """
        return horde_model_reference_paths.get_all_model_reference_file_paths(base_path=self.base_path)

    @override
    def get_legacy_json(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        redownload: bool = False,
    ) -> dict[str, Any] | None:
        """Get legacy format data from legacy/ folder.

        For text_generation category, reads from CSV format (models.csv).
        For other categories, reads from JSON format.

        Args:
            category: Category to retrieve.
            redownload: If True, bypass cache and read from disk.

        Returns:
            dict[str, Any] | None: The legacy format data, or None if file doesn't exist.

        """
        with self._lock:
            # Check cache first unless redownload
            if not redownload:
                legacy_dict, _ = self._get_legacy_from_cache(category)
                if legacy_dict is not None:
                    return legacy_dict

            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                legacy_file_path, is_csv = self._resolve_legacy_text_generation_path()
            else:
                legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                    category,
                    base_path=self.base_path,
                )
                is_csv = False

            if not legacy_file_path.exists():
                logger.debug(f"Legacy file not found for {category}: {legacy_file_path}")
                self._store_legacy_in_cache(category, None, None)
                return None

            try:
                # Handle CSV format for text_generation
                if category == MODEL_REFERENCE_CATEGORY.text_generation:
                    if is_csv:
                        data = self._read_legacy_csv_to_dict(legacy_file_path)
                    else:
                        with open(legacy_file_path, encoding="utf-8") as f:
                            data = cast(dict[str, Any], json.load(f))

                    # Generate JSON string for cache
                    content = json.dumps(data, indent=2, ensure_ascii=False)
                    self._store_legacy_in_cache(category, data, content)
                    logger.debug(f"Loaded legacy CSV for {category} from {legacy_file_path}")
                    return data

                # Handle JSON format for other categories
                with open(legacy_file_path, encoding="utf-8") as f:
                    content = f.read()
                    data = cast(dict[str, Any], json.loads(content))

                self._store_legacy_in_cache(category, data, content)
                logger.debug(f"Loaded legacy JSON for {category} from {legacy_file_path}")
                return data

            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read legacy file {legacy_file_path}: {e}")
                self._invalidate_legacy_cache(category)
                return None

    @override
    def get_legacy_json_string(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        redownload: bool = False,
    ) -> str | None:
        """Get legacy format data as JSON string from legacy/ folder.

        For text_generation category, reads CSV and converts to JSON string.
        For other categories, reads JSON format directly.

        Args:
            category: Category to retrieve.
            redownload: If True, bypass cache and read from disk.

        Returns:
            str | None: The legacy format as JSON string, or None if file doesn't exist.

        """
        with self._lock:
            # Check cache first unless redownload
            if not redownload:
                _, legacy_string = self._get_legacy_from_cache(category)
                if legacy_string is not None:
                    return legacy_string

            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                legacy_file_path, is_csv = self._resolve_legacy_text_generation_path()
            else:
                legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                    category,
                    base_path=self.base_path,
                )
                is_csv = False

            if not legacy_file_path.exists():
                logger.debug(f"Legacy file not found for {category}: {legacy_file_path}")
                self._store_legacy_in_cache(category, None, None)
                return None

            try:
                # Handle CSV format for text_generation
                if category == MODEL_REFERENCE_CATEGORY.text_generation:
                    if is_csv:
                        data = self._read_legacy_csv_to_dict(legacy_file_path)
                    else:
                        with open(legacy_file_path, encoding="utf-8") as f:
                            data = cast(dict[str, Any], json.load(f))
                    # Generate JSON string
                    content = json.dumps(data, indent=2, ensure_ascii=False)
                    self._store_legacy_in_cache(category, data, content)
                    logger.debug(f"Loaded legacy CSV string for {category} from {legacy_file_path}")
                    return content

                # Handle JSON format for other categories
                with open(legacy_file_path, encoding="utf-8") as f:
                    content = f.read()
                    data = cast(dict[str, Any], json.loads(content))

                self._store_legacy_in_cache(category, data, content)
                logger.debug(f"Loaded legacy JSON string for {category} from {legacy_file_path}")
                return content

            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read legacy file {legacy_file_path}: {e}")
                self._invalidate_legacy_cache(category)
                return None

    @override
    def supports_writes(self) -> bool:
        """Check if backend supports writes (always True for PRIMARY filesystem).

        Returns:
            bool: Always True.

        """
        return True

    @override
    def supports_metadata(self) -> bool:
        """Check if backend supports metadata tracking (always True for PRIMARY filesystem).

        Returns:
            bool: Always True.

        """
        return True

    @override
    def update_model(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        record_dict: dict[str, Any],
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Update or create a model reference.

        Modifies the JSON file on disk atomically for all categories (v2 format is always JSON).
        Preserves created_at and created_by metadata on updates.

        Args:
            category: The category to update.
            model_name: The name of the model to update or create.
            record_dict: The model record data as a dictionary.
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        Raises:
            FileNotFoundError: If the category file path is not configured.

        """
        from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

        with self._lock:
            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )

            if not file_path:
                raise FileNotFoundError(f"No file path configured for category {category}")

            # Read existing data (v2 format is always JSON, including text_generation)
            existing_data: dict[str, Any]
            if file_path.exists():
                try:
                    with open(file_path, encoding="utf-8") as f:
                        existing_data = json.load(f)
                except json.JSONDecodeError as e:
                    # V2 files should always be valid JSON - surface corruption errors
                    logger.error(
                        f"Invalid JSON in v2 file {file_path}. This indicates data corruption. "
                        f"V2 format is always JSON, including text_generation.json. Error: {e}"
                    )
                    raise
                except OSError as e:
                    logger.error(f"Failed to read {file_path}: {e}")
                    raise
            else:
                existing_data = {}
                file_path.parent.mkdir(parents=True, exist_ok=True)

            # For text_generation, filter out backend prefix entries before updating
            if category == MODEL_REFERENCE_CATEGORY.text_generation and has_legacy_text_backend_prefix(model_name):
                logger.warning(
                    f"Attempted to update backend-prefixed model {model_name} in text_generation. "
                    "Backend prefixes are not stored internally - update the base model instead."
                )
                # Don't raise an error, just skip the update
                return

            # For text_generation, validate/transform the record
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                from horde_model_reference.text_model_write_processor import TextModelWriteProcessor

                processor = TextModelWriteProcessor()
                record_dict = processor.validate_and_transform(model_name, record_dict)

            # Determine if this is a create or update operation
            is_update = model_name in existing_data
            operation_type = OperationType.UPDATE if is_update else OperationType.CREATE
            previous_record = copy.deepcopy(existing_data[model_name]) if is_update else None

            # Handle per-model metadata
            if is_update:
                # Preserve creation metadata if model already exists
                existing_model = existing_data[model_name]
                self._metadata_manager.model_metadata.preserve_creation_fields(existing_model, record_dict)
                # Set updated_at timestamp
                self._metadata_manager.model_metadata.set_update_timestamp(record_dict)
            else:
                # For new models, ensure timestamps are populated (without overwriting existing values)
                self._metadata_manager.model_metadata.ensure_metadata_populated(record_dict)

            record_snapshot = copy.deepcopy(record_dict)
            existing_data[model_name] = record_snapshot

            temp_path = file_path.with_suffix(f".tmp.{time.time()}")
            try:
                # Write to temp file (v2 format is always JSON)
                with open(temp_path, "w", encoding="utf-8") as f:
                    json.dump(existing_data, f, indent=2, ensure_ascii=False)
                    f.flush()
                    try:
                        import os

                        os.fsync(f.fileno())
                    except OSError:
                        pass

                # Atomic replace
                if file_path.exists():
                    backup_path = file_path.with_suffix(".bak")
                    file_path.replace(backup_path)
                    temp_path.replace(file_path)
                    with contextlib.suppress(OSError):
                        backup_path.unlink()
                else:
                    temp_path.replace(file_path)

                logger.info(f"Updated model {model_name} in category {category} at {file_path}")

                # Record metadata for observability (centralized hook point)
                self._metadata_manager.record_v2_operation(
                    category=category,
                    operation=operation_type,
                    model_name=model_name,
                    success=True,
                    backend_type=self.__class__.__name__,
                )

                if logical_user_id is not None and self._audit_writer is not None:
                    if is_update and previous_record is not None:
                        payload = AuditPayload.from_update(previous_record, record_snapshot)
                        audit_operation = AuditOperation.UPDATE
                    elif is_update:
                        payload = AuditPayload.from_create(record_snapshot)
                        audit_operation = AuditOperation.UPDATE
                    else:
                        payload = AuditPayload.from_create(record_snapshot)
                        audit_operation = AuditOperation.CREATE

                    self._append_audit_event(
                        domain=CanonicalFormat.v2,
                        category=category,
                        model_name=model_name,
                        operation=audit_operation,
                        payload=payload,
                        logical_user_id=logical_user_id,
                        request_id=request_id,
                    )

                self._mark_category_modified(category, file_path)

            except (OSError, ValueError, TypeError) as e:
                try:
                    if temp_path.exists():
                        temp_path.unlink()
                except OSError:
                    pass
                logger.error(f"Failed to update model {model_name} in {category}: {e}")
                raise

    @override
    def delete_model(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Delete a model reference.

        Removes the model from the JSON file on disk atomically for all categories
        (v2 format is always JSON).

        Args:
            category: The category containing the model.
            model_name: The name of the model to delete.
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        Raises:
            FileNotFoundError: If the category file doesn't exist.
            KeyError: If the model doesn't exist in the category.

        """
        with self._lock:
            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )

            existing_data: dict[str, Any]
            if not file_path or not file_path.exists():
                raise FileNotFoundError(f"Category file not found: {file_path}")

            # Read existing data (v2 format is always JSON, including text_generation)
            try:
                with open(file_path, encoding="utf-8") as f:
                    existing_data = json.load(f)
            except json.JSONDecodeError as e:
                # V2 files should always be valid JSON - surface corruption errors
                logger.error(
                    f"Invalid JSON in v2 file {file_path}. This indicates data corruption. "
                    f"V2 format is always JSON, including text_generation.json. Error: {e}"
                )
                raise
            except OSError as e:
                logger.error(f"Failed to read {file_path}: {e}")
                raise

            if model_name not in existing_data:
                raise KeyError(f"Model {model_name} not found in category {category}")

            deleted_snapshot = copy.deepcopy(existing_data[model_name])
            del existing_data[model_name]

            temp_path = file_path.with_suffix(f".tmp.{time.time()}")
            try:
                # Write to temp file (v2 format is always JSON)
                with open(temp_path, "w", encoding="utf-8") as f:
                    json.dump(existing_data, f, indent=2, ensure_ascii=False)
                    f.flush()
                    try:
                        import os

                        os.fsync(f.fileno())
                    except OSError:
                        pass

                backup_path = file_path.with_suffix(".bak")
                file_path.replace(backup_path)
                temp_path.replace(file_path)

                with contextlib.suppress(OSError):
                    backup_path.unlink()

                logger.info(f"Deleted model {model_name} from category {category} at {file_path}")

                # Record metadata for observability (centralized hook point)
                self._metadata_manager.record_v2_operation(
                    category=category,
                    operation=OperationType.DELETE,
                    model_name=model_name,
                    success=True,
                    backend_type=self.__class__.__name__,
                )

                if logical_user_id is not None and self._audit_writer is not None:
                    payload = AuditPayload.from_delete(deleted_snapshot)
                    self._append_audit_event(
                        domain=CanonicalFormat.v2,
                        category=category,
                        model_name=model_name,
                        operation=AuditOperation.DELETE,
                        payload=payload,
                        logical_user_id=logical_user_id,
                        request_id=request_id,
                    )

                self._mark_category_modified(category, file_path)

            except (OSError, ValueError, TypeError) as e:
                try:
                    if temp_path.exists():
                        temp_path.unlink()
                except OSError:
                    pass
                logger.error(f"Failed to delete model {model_name} from {category}: {e}")
                raise

    @override
    def supports_legacy_writes(self) -> bool:
        """Check if backend supports legacy format writes.

        Returns True only when canonical_format='LEGACY' in settings.

        Returns:
            bool: True if legacy writes are supported.

        """
        from horde_model_reference import horde_model_reference_settings

        return horde_model_reference_settings.canonical_format == CanonicalFormat.LEGACY

    @override
    def update_model_legacy(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        record_dict: dict[str, Any],
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Update or create a model reference in legacy format.

        This method modifies the legacy format JSON file on disk atomically.

        Args:
            category: The category to update.
            model_name: The name of the model to update or create.
            record_dict: The model record data in legacy format as a dictionary.
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        Raises:
            FileNotFoundError: If the legacy category file path is not configured.
            RuntimeError: If canonical_format is not set to 'LEGACY'.

        """
        from horde_model_reference import horde_model_reference_settings

        if not self.supports_legacy_writes():
            raise RuntimeError(
                "Legacy writes are only supported when canonical_format='LEGACY'. "
                f"Current setting: canonical_format='{horde_model_reference_settings.canonical_format}'"
            )

        with self._lock:
            # text_generation uses CSV as source of truth — route to dedicated handler
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                self._update_text_generation_csv(
                    model_name,
                    record_dict,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )
                return

            legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            target_write_path = legacy_file_path

            if not legacy_file_path:
                raise FileNotFoundError(f"No legacy file path configured for category {category}")

            existing_data: dict[str, Any]
            if legacy_file_path.exists():
                try:
                    with open(legacy_file_path, encoding="utf-8") as f:
                        existing_data = json.load(f)
                except (OSError, json.JSONDecodeError) as e:
                    logger.error(f"Failed to read {legacy_file_path}: {e}")
                    raise
            else:
                existing_data = {}
                target_write_path.parent.mkdir(parents=True, exist_ok=True)

            # Determine if this is a create or update operation
            is_update = model_name in existing_data
            operation_type = OperationType.UPDATE if is_update else OperationType.CREATE
            previous_record = copy.deepcopy(existing_data.get(model_name)) if is_update else None
            record_snapshot = copy.deepcopy(record_dict)

            existing_data[model_name] = record_snapshot

            temp_path = target_write_path.with_suffix(f".tmp.{time.time()}")
            try:
                with open(temp_path, "w", encoding="utf-8") as f:
                    json.dump(existing_data, f, indent=2, ensure_ascii=False)
                    f.flush()
                    try:
                        import os

                        os.fsync(f.fileno())
                    except OSError:
                        pass

                if target_write_path.exists():
                    backup_path = target_write_path.with_suffix(".bak")
                    target_write_path.replace(backup_path)
                    temp_path.replace(target_write_path)
                    with contextlib.suppress(OSError):
                        backup_path.unlink()
                else:
                    temp_path.replace(target_write_path)

                logger.info(f"Updated legacy model {model_name} in category {category} at {target_write_path}")

                self._metadata_manager.record_legacy_operation(
                    category=category,
                    operation=operation_type,
                    model_name=model_name,
                    success=True,
                    backend_type=self.__class__.__name__,
                )

                if logical_user_id is not None and self._audit_writer is not None:
                    if is_update and previous_record is not None:
                        payload = AuditPayload.from_update(previous_record, record_snapshot)
                        audit_operation = AuditOperation.UPDATE
                    else:
                        payload = AuditPayload.from_create(record_snapshot)
                        audit_operation = AuditOperation.CREATE
                    self._append_audit_event(
                        domain=CanonicalFormat.LEGACY,
                        category=category,
                        model_name=model_name,
                        operation=audit_operation,
                        payload=payload,
                        logical_user_id=logical_user_id,
                        request_id=request_id,
                    )

                self._mark_legacy_category_modified(category, target_write_path)

            except (OSError, ValueError, TypeError) as e:
                try:
                    if temp_path.exists():
                        temp_path.unlink()
                except OSError:
                    pass
                logger.error(f"Failed to update legacy model {model_name} in {category}: {e}")
                raise

    def _update_text_generation_csv(
        self,
        model_name: str,
        record_dict: dict[str, Any],
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Update a text_generation model by writing CSV (not JSON) to models.csv.

        Reads the existing CSV, validates/transforms the record, updates the row list,
        writes CSV back, and regenerates the cached dict representation.

        Args:
            model_name: The base model name (no backend prefix).
            record_dict: The model record data.
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        """
        from horde_model_reference.text_model_write_processor import TextModelWriteProcessor

        category = MODEL_REFERENCE_CATEGORY.text_generation
        csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        # Read existing CSV rows
        existing_rows: list[TextCSVRow] = []
        if csv_path.exists():
            existing_rows, parse_issues = parse_legacy_text_csv_file(csv_path)
            for issue in parse_issues:
                logger.warning(f"Legacy CSV parse issue for {issue.row_identifier}: {issue.message}")

        # Validate and transform the incoming record
        processor = TextModelWriteProcessor()
        record_dict = processor.validate_and_transform(model_name, record_dict)

        # Convert the validated record to a CSV row
        new_row = legacy_record_to_csv_row(model_name, record_dict)

        # Find and replace existing row, or append
        row_index: int | None = None
        previous_record: dict[str, Any] | None = None
        for i, row in enumerate(existing_rows):
            if row.name == model_name:
                row_index = i
                break

        if row_index is not None:
            # Capture previous record for audit before replacing
            old_dict = csv_rows_to_legacy_dict([existing_rows[row_index]], with_backend_prefixes=False)
            previous_record = old_dict.get(model_name)
            existing_rows[row_index] = new_row
            operation_type = OperationType.UPDATE
        else:
            existing_rows.append(new_row)
            operation_type = OperationType.CREATE

        # Write CSV back
        csv_path.parent.mkdir(parents=True, exist_ok=True)
        write_legacy_text_csv(existing_rows, csv_path)

        # Regenerate the full dict for cache
        full_data = csv_rows_to_legacy_dict(existing_rows, with_backend_prefixes=True)
        content = json.dumps(full_data, indent=2, ensure_ascii=False)
        self._store_legacy_in_cache(category, full_data, content)

        record_snapshot = copy.deepcopy(record_dict)
        logger.info(f"Updated legacy text_generation model {model_name} in CSV at {csv_path}")

        self._metadata_manager.record_legacy_operation(
            category=category,
            operation=operation_type,
            model_name=model_name,
            success=True,
            backend_type=self.__class__.__name__,
        )

        if logical_user_id is not None and self._audit_writer is not None:
            if operation_type == OperationType.UPDATE and previous_record is not None:
                payload = AuditPayload.from_update(previous_record, record_snapshot)
                audit_operation = AuditOperation.UPDATE
            else:
                payload = AuditPayload.from_create(record_snapshot)
                audit_operation = AuditOperation.CREATE
            self._append_audit_event(
                domain=CanonicalFormat.LEGACY,
                category=category,
                model_name=model_name,
                operation=audit_operation,
                payload=payload,
                logical_user_id=logical_user_id,
                request_id=request_id,
            )

        self._mark_legacy_category_modified(category, csv_path)

    def _delete_text_generation_csv(
        self,
        model_name: str,
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Delete a text_generation model from CSV, preserving CSV format.

        Args:
            model_name: The base model name (no backend prefix).
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        Raises:
            FileNotFoundError: If the CSV file doesn't exist.
            KeyError: If the model doesn't exist.

        """
        category = MODEL_REFERENCE_CATEGORY.text_generation
        csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        if not csv_path.exists():
            raise FileNotFoundError(f"Legacy CSV file not found: {csv_path}")

        existing_rows, parse_issues = parse_legacy_text_csv_file(csv_path)
        for issue in parse_issues:
            logger.warning(f"Legacy CSV parse issue for {issue.row_identifier}: {issue.message}")

        # Find and remove the row
        row_index: int | None = None
        for i, row in enumerate(existing_rows):
            if row.name == model_name:
                row_index = i
                break

        if row_index is None:
            raise KeyError(f"Model {model_name} not found in legacy text_generation CSV")

        deleted_row = existing_rows.pop(row_index)

        # Write CSV back
        write_legacy_text_csv(existing_rows, csv_path)

        # Regenerate the full dict for cache
        full_data = csv_rows_to_legacy_dict(existing_rows, with_backend_prefixes=True)
        content = json.dumps(full_data, indent=2, ensure_ascii=False)
        self._store_legacy_in_cache(category, full_data, content)

        # Capture the deleted record for audit
        deleted_dict = csv_rows_to_legacy_dict([deleted_row], with_backend_prefixes=False)
        deleted_snapshot = deleted_dict.get(model_name, {})

        logger.info(f"Deleted legacy text_generation model {model_name} from CSV at {csv_path}")

        self._metadata_manager.record_legacy_operation(
            category=category,
            operation=OperationType.DELETE,
            model_name=model_name,
            success=True,
            backend_type=self.__class__.__name__,
        )

        if logical_user_id is not None and self._audit_writer is not None:
            payload = AuditPayload.from_delete(deleted_snapshot)
            self._append_audit_event(
                domain=CanonicalFormat.LEGACY,
                category=category,
                model_name=model_name,
                operation=AuditOperation.DELETE,
                payload=payload,
                logical_user_id=logical_user_id,
                request_id=request_id,
            )

        self._mark_legacy_category_modified(category, csv_path)

    @override
    def delete_model_legacy(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Delete a model reference from legacy format files.

        This method removes the model from the legacy format JSON file on disk atomically.

        Args:
            category: The category containing the model.
            model_name: The name of the model to delete.
            logical_user_id: Optional logical user ID for audit logging.
            request_id: Optional request ID for audit logging.

        Raises:
            FileNotFoundError: If the legacy category file doesn't exist.
            KeyError: If the model doesn't exist in the category.
            RuntimeError: If canonical_format is not set to 'LEGACY'.

        """
        from horde_model_reference import horde_model_reference_settings

        if not self.supports_legacy_writes():
            raise RuntimeError(
                "Legacy writes are only supported when canonical_format='LEGACY'. "
                f"Current setting: canonical_format='{horde_model_reference_settings.canonical_format}'"
            )

        with self._lock:
            # text_generation uses CSV as source of truth — route to dedicated handler
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                self._delete_text_generation_csv(
                    model_name,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )
                return

            legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            target_write_path = legacy_file_path

            existing_data: dict[str, Any]
            if not legacy_file_path or not legacy_file_path.exists():
                raise FileNotFoundError(f"Legacy category file not found: {legacy_file_path}")

            try:
                with open(legacy_file_path, encoding="utf-8") as f:
                    existing_data = json.load(f)
            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read {legacy_file_path}: {e}")
                raise

            if model_name not in existing_data:
                raise KeyError(f"Model {model_name} not found in legacy category {category}")

            deleted_snapshot = copy.deepcopy(existing_data[model_name])
            del existing_data[model_name]

            temp_path = target_write_path.with_suffix(f".tmp.{time.time()}")
            try:
                with open(temp_path, "w", encoding="utf-8") as f:
                    json.dump(existing_data, f, indent=2, ensure_ascii=False)
                    f.flush()
                    try:
                        import os

                        os.fsync(f.fileno())
                    except OSError:
                        pass

                if target_write_path.exists():
                    backup_path = target_write_path.with_suffix(".bak")
                    target_write_path.replace(backup_path)
                    temp_path.replace(target_write_path)
                    with contextlib.suppress(OSError):
                        backup_path.unlink()
                else:
                    temp_path.replace(target_write_path)

                logger.info(f"Deleted legacy model {model_name} from category {category} at {target_write_path}")

                self._metadata_manager.record_legacy_operation(
                    category=category,
                    operation=OperationType.DELETE,
                    model_name=model_name,
                    success=True,
                    backend_type=self.__class__.__name__,
                )

                if logical_user_id is not None and self._audit_writer is not None:
                    payload = AuditPayload.from_delete(deleted_snapshot)
                    self._append_audit_event(
                        domain=CanonicalFormat.LEGACY,
                        category=category,
                        model_name=model_name,
                        operation=AuditOperation.DELETE,
                        payload=payload,
                        logical_user_id=logical_user_id,
                        request_id=request_id,
                    )

                self._mark_legacy_category_modified(category, target_write_path)

            except (OSError, ValueError, TypeError) as e:
                try:
                    if temp_path.exists():
                        temp_path.unlink()
                except OSError:
                    pass
                logger.error(f"Failed to delete legacy model {model_name} from {category}: {e}")
                raise

    def _populate_model_metadata(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        timestamp: int | None = None,
    ) -> int:
        """Populate missing per-model metadata fields in a category's JSON file.

        This method scans all models in a category file and ensures each has:
        - metadata.created_at (if missing)
        - metadata.updated_at (if missing)

        Args:
            category: The category to populate metadata for.
            timestamp: The timestamp to use for created_at/updated_at. If None, uses current time.

        Returns:
            int: Number of models that had metadata populated.

        """
        with self._lock:
            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )

            if not file_path or not file_path.exists():
                logger.debug(f"Category file not found for {category}, skipping metadata population")
                return 0

            if timestamp is None:
                timestamp = int(time.time())

            try:
                with open(file_path, encoding="utf-8") as f:
                    data: dict[str, Any] = json.load(f)
            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read {file_path} for metadata population: {e}")
                return 0

            models_updated = 0
            for _model_name, model_data in data.items():
                if not isinstance(model_data, dict):
                    continue

                # Use ModelMetadataManager to ensure metadata is populated
                if self._metadata_manager.model_metadata.ensure_metadata_populated(model_data, timestamp):
                    models_updated += 1

            if models_updated == 0:
                logger.trace(f"No models needed metadata population in {category}")
                return 0

            temp_path = file_path.with_suffix(f".tmp.{time.time()}")
            try:
                with open(temp_path, "w", encoding="utf-8") as f:
                    json.dump(data, f, indent=2, ensure_ascii=False)
                    f.flush()
                    try:
                        import os

                        os.fsync(f.fileno())
                    except OSError:
                        pass

                backup_path = file_path.with_suffix(".bak")
                file_path.replace(backup_path)
                temp_path.replace(file_path)

                with contextlib.suppress(OSError):
                    backup_path.unlink()

                logger.info(f"Populated metadata for {models_updated} models in {category}")
                self._mark_category_modified(category, file_path)

                return models_updated

            except (OSError, ValueError, TypeError) as e:
                try:
                    if temp_path.exists():
                        temp_path.unlink()
                except OSError:
                    pass
                logger.error(f"Failed to write metadata-populated file for {category}: {e}")
                return 0

    def ensure_category_metadata_populated(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        timestamp: int | None = None,
    ) -> CategoryMetadataPopulationResult:
        """Ensure both CategoryMetadata and per-model metadata are populated for a category.

        This method:
        1. Checks if CategoryMetadata exists for both v2 and legacy formats
        2. Initializes CategoryMetadata if missing
        3. Populates per-model metadata fields in JSON files
        4. Uses the same timestamp for both backend and model-level metadata

        Args:
            category: The category to ensure metadata for.
            timestamp: Optional timestamp to use. If None, uses current time.

        Returns:
            dict with keys:
                - "category_metadata_initialized": bool
                - "legacy_metadata_initialized": bool
                - "models_updated": int
                - "timestamp_used": int

        """
        with self._lock:
            if timestamp is None:
                timestamp = int(time.time())

            result = CategoryMetadataPopulationResult(
                category_metadata_initialized=False,
                legacy_metadata_initialized=False,
                models_updated=0,
                timestamp_used=timestamp,
            )

            # Get or initialize v2 CategoryMetadata
            v2_metadata = self._metadata_manager.get_or_initialize_v2_metadata(
                category=category,
                backend_type=self.__class__.__name__,
            )

            # Check if it was just created (no prior data file existed)
            if v2_metadata.initialization_time == v2_metadata.last_updated:
                result.category_metadata_initialized = True
                logger.trace(f"Initialized v2 CategoryMetadata for {category}")

            # Use initialization_time from metadata
            timestamp = v2_metadata.initialization_time
            result.timestamp_used = timestamp

            # Get or initialize legacy CategoryMetadata
            legacy_metadata = self._metadata_manager.get_or_initialize_legacy_metadata(
                category=category,
                backend_type=self.__class__.__name__,
            )

            # Check if it was just created
            if legacy_metadata.initialization_time == legacy_metadata.last_updated:
                result.legacy_metadata_initialized = True
                logger.trace(f"Initialized legacy CategoryMetadata for {category}")

            # Populate per-model metadata using the determined timestamp
            models_updated = self._populate_model_metadata(category, timestamp)
            result.models_updated = models_updated

            if result.category_metadata_initialized or result.legacy_metadata_initialized or models_updated > 0:
                logger.info(
                    f"Metadata population for {category}: "
                    f"v2_meta={result.category_metadata_initialized}, "
                    f"legacy_meta={result.legacy_metadata_initialized}, "
                    f"models={models_updated}"
                )

            return result

    def ensure_all_metadata_populated(self) -> AllMetadataPopulationResult:
        """Ensure metadata is populated for all categories that have files.

        Scans all category files and ensures:
        1. CategoryMetadata exists (both v2 and legacy formats)
        2. All model records have metadata fields populated

        This is called:
        - On FileSystemBackend initialization (PRIMARY mode)
        - After GitHub seeding completes

        Returns:
            AllMetadataPopulationResult with summary of metadata population.

        """
        with self._lock:
            result = AllMetadataPopulationResult()

            logger.info("Starting metadata population scan for all categories")

            for category in MODEL_REFERENCE_CATEGORY:
                file_path = horde_model_reference_paths.get_model_reference_file_path(
                    category,
                    base_path=self.base_path,
                )

                # Skip categories that don't have files
                if not file_path or not file_path.exists():
                    logger.debug(f"Skipping {category} - no file found")
                    continue

                # Ensure metadata for this category
                category_result = self.ensure_category_metadata_populated(category)

                if (
                    category_result.category_metadata_initialized
                    or category_result.legacy_metadata_initialized
                    or category_result.models_updated > 0
                ):
                    result.categories_processed.append(category.value)
                    result.total_categories += 1
                    result.total_models_updated += category_result.models_updated

                    if category_result.category_metadata_initialized:
                        result.total_metadata_initialized += 1
                    if category_result.legacy_metadata_initialized:
                        result.total_metadata_initialized += 1

            if result.total_categories > 0:
                logger.info(
                    f"Metadata population complete: "
                    f"{result.total_categories} categories processed, "
                    f"{result.total_models_updated} models updated, "
                    f"{result.total_metadata_initialized} metadata files initialized"
                )
            else:
                logger.debug("No metadata population needed - all files already have metadata")

            return result

    @override
    def get_legacy_metadata(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
        """Get legacy format metadata for a specific category.

        Args:
            category: The category to get metadata for.

        Returns:
            CategoryMetadata | None: The legacy metadata, or None if not available.

        """
        return self._metadata_manager.get_legacy_metadata(category)

    @override
    async def get_legacy_metadata_async(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
        """Asynchronously get legacy format metadata for a specific category.

        Args:
            category: The category to get metadata for.

        Returns:
            CategoryMetadata | None: The legacy metadata, or None if not available.

        """
        return self._metadata_manager.get_legacy_metadata(category)

    @override
    def get_metadata(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
        """Get v2 format metadata for a specific category.

        Args:
            category: The category to get metadata for.

        Returns:
            CategoryMetadata | None: The v2 metadata, or None if not available.

        """
        return self._metadata_manager.get_v2_metadata(category)

    @override
    async def get_metadata_async(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
        """Asynchronously get v2 format metadata for a specific category.

        Args:
            category: The category to get metadata for.

        Returns:
            CategoryMetadata | None: The v2 metadata, or None if not available.

        """
        return self._metadata_manager.get_v2_metadata(category)

    @override
    def get_all_legacy_metadata(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
        """Get legacy format metadata for all categories.

        Returns:
            dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

        """
        return self._metadata_manager.get_all_legacy_metadata()

    @override
    async def get_all_legacy_metadata_async(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
        """Asynchronously get legacy format metadata for all categories.

        Returns:
            dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

        """
        return self._metadata_manager.get_all_legacy_metadata()

    @override
    def get_all_metadata(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
        """Get v2 format metadata for all categories.

        Returns:
            dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

        """
        return self._metadata_manager.get_all_v2_metadata()

    @override
    async def get_all_metadata_async(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
        """Asynchronously get v2 format metadata for all categories.

        Returns:
            dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

        """
        return self._metadata_manager.get_all_v2_metadata()

base_path instance-attribute

base_path = Path(base_path)

_metadata_manager instance-attribute

_metadata_manager = MetadataManager(base_path)

_audit_writer instance-attribute

_audit_writer = audit_writer

_replicate_mode class-attribute instance-attribute

_replicate_mode = mode

_invalidation_callbacks instance-attribute

_invalidation_callbacks: list[
    Callable[[MODEL_REFERENCE_CATEGORY], None]
] = []

replicate_mode property

replicate_mode: ReplicateMode

Get the replicate mode of this backend.

_cache_ttl_seconds instance-attribute

_cache_ttl_seconds = cache_ttl_seconds

_cache instance-attribute

_cache: dict[
    MODEL_REFERENCE_CATEGORY, dict[str, Any] | None
] = {}

_category_timestamps instance-attribute

_category_timestamps: dict[
    MODEL_REFERENCE_CATEGORY, float
] = {}

_last_known_mtimes instance-attribute

_last_known_mtimes: dict[
    MODEL_REFERENCE_CATEGORY, float
] = {}

_stale_categories instance-attribute

_stale_categories: set[MODEL_REFERENCE_CATEGORY] = set()

_legacy_json_cache instance-attribute

_legacy_json_cache: dict[
    MODEL_REFERENCE_CATEGORY, dict[str, Any] | None
] = {}

_legacy_json_string_cache instance-attribute

_legacy_json_string_cache: dict[
    MODEL_REFERENCE_CATEGORY, str | None
] = {}

_legacy_cache_timestamps instance-attribute

_legacy_cache_timestamps: dict[
    MODEL_REFERENCE_CATEGORY, float
] = {}

_legacy_last_known_mtimes instance-attribute

_legacy_last_known_mtimes: dict[
    MODEL_REFERENCE_CATEGORY, float
] = {}

_stale_legacy_categories instance-attribute

_stale_legacy_categories: set[MODEL_REFERENCE_CATEGORY] = (
    set()
)

_lock instance-attribute

_lock = RLock()

_async_lock instance-attribute

_async_lock: Lock = Lock()

cache_ttl_seconds property

cache_ttl_seconds: int | None

The cache TTL currently enforced for category payloads.

lock property

lock: RLock

Thread-safe lock shared by subclasses for critical sections.

async_lock property

async_lock: Lock | None

Asyncio lock usable by subclasses when coordinating coroutines.

__init__

__init__(
    *,
    base_path: str
    | Path = horde_model_reference_paths.base_path,
    cache_ttl_seconds: int = 60,
    replicate_mode: ReplicateMode = ReplicateMode.PRIMARY,
    skip_startup_metadata_population: bool = False,
    audit_writer: AuditTrailWriter | None = None,
) -> None

Initialize the FileSystem backend.

Parameters:

• base_path (str | Path, default: base_path ) –

  Base path for model reference files.

• cache_ttl_seconds (int, default: 60 ) –

  TTL for internal cache in seconds.

• replicate_mode (ReplicateMode, default: PRIMARY ) –

  Must be PRIMARY.

• skip_startup_metadata_population (bool, default: False ) –

  If True, skip automatic metadata population on startup. This is used when GitHub seeding will handle metadata population instead.

• audit_writer (AuditTrailWriter | None, default: None ) –

  Optional AuditTrailWriter for emitting audit events on CRUD operations.

Raises:

• ValueError –

  If replicate_mode is not PRIMARY.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def __init__(
    self,
    *,
    base_path: str | Path = horde_model_reference_paths.base_path,
    cache_ttl_seconds: int = 60,
    replicate_mode: ReplicateMode = ReplicateMode.PRIMARY,
    skip_startup_metadata_population: bool = False,
    audit_writer: AuditTrailWriter | None = None,
) -> None:
    """Initialize the FileSystem backend.

    Args:
        base_path: Base path for model reference files.
        cache_ttl_seconds: TTL for internal cache in seconds.
        replicate_mode: Must be PRIMARY.
        skip_startup_metadata_population: If True, skip automatic metadata population on startup.
            This is used when GitHub seeding will handle metadata population instead.
        audit_writer: Optional AuditTrailWriter for emitting audit events on CRUD operations.

    Raises:
        ValueError: If replicate_mode is not PRIMARY.

    """
    if replicate_mode != ReplicateMode.PRIMARY:
        raise ValueError(
            "FileSystemBackend can only be used in PRIMARY mode. "
            "For REPLICA mode, use GitHubBackend or HTTPBackend."
        )
    super().__init__(mode=replicate_mode, cache_ttl_seconds=cache_ttl_seconds)

    self.base_path = Path(base_path)
    self._metadata_manager = MetadataManager(self.base_path)
    self._audit_writer = audit_writer

    logger.debug(f"FileSystemBackend initialized with base_path={self.base_path}")

    # Create empty files for categories that have no legacy format available
    # This ensures consistent behavior between CI (fresh environment) and local (may have existing files)
    from horde_model_reference.meta_consts import get_no_legacy_format_categories

    for category in get_no_legacy_format_categories():
        if not isinstance(category, MODEL_REFERENCE_CATEGORY):
            continue
        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )
        if file_path and not file_path.exists():
            file_path.parent.mkdir(parents=True, exist_ok=True)
            file_path.write_text("{}")
            logger.info(f"Created empty file for {category} (no legacy format available)")

    # Populate metadata on startup if not skipped
    if not skip_startup_metadata_population:
        logger.info("Running startup metadata population check")
        self.ensure_all_metadata_populated()
    else:
        logger.debug("Skipping startup metadata population (will be handled by GitHub seeding)")

_resolve_legacy_text_generation_path

_resolve_legacy_text_generation_path() -> tuple[Path, bool]

Return the legacy text_generation path and whether it is CSV-based.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _resolve_legacy_text_generation_path(self) -> tuple[Path, bool]:
    """Return the legacy text_generation path and whether it is CSV-based."""
    csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
        MODEL_REFERENCE_CATEGORY.text_generation,
        base_path=self.base_path,
    )
    json_path = csv_path.with_name("text_generation.json")

    # If canonical_format is legacy, prefer CSV if it exists, otherwise JSON if it exists
    if horde_model_reference_settings.canonical_format == CanonicalFormat.LEGACY:
        if csv_path.exists():
            return csv_path, True
        if json_path.exists():
            return json_path, False

        # Neither file exists, default to CSV path
        return csv_path, True

    # Otherwise, prefer CSV if it exists, else JSON if it exists
    if csv_path.exists():
        return csv_path, True
    if json_path.exists():
        return json_path, False

    # Default to CSV path if nothing exists
    return csv_path, True

_get_file_path_for_validation

_get_file_path_for_validation(
    category: MODEL_REFERENCE_CATEGORY,
) -> Path | None

Return the file path for mtime validation.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get the file path for.

Returns:

• Path | None –

  Path | None: Path to file for mtime validation.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def _get_file_path_for_validation(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
    """Return the file path for mtime validation.

    Args:
        category: The category to get the file path for.

    Returns:
        Path | None: Path to file for mtime validation.

    """
    return horde_model_reference_paths.get_model_reference_file_path(
        category,
        base_path=self.base_path,
    )

_get_legacy_file_path_for_validation

_get_legacy_file_path_for_validation(
    category: MODEL_REFERENCE_CATEGORY,
) -> Path | None

Return the legacy file path for mtime validation.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get the legacy file path for.

Returns:

• Path | None –

  Path | None: Path to legacy file for mtime validation.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def _get_legacy_file_path_for_validation(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
    """Return the legacy file path for mtime validation.

    Args:
        category: The category to get the legacy file path for.

    Returns:
        Path | None: Path to legacy file for mtime validation.

    """
    return horde_model_reference_paths.get_legacy_model_reference_file_path(
        category,
        base_path=self.base_path,
    )

_mark_category_modified

_mark_category_modified(
    category: MODEL_REFERENCE_CATEGORY, file_path: Path
) -> None

Mark a category as modified after a write operation.

This invalidates the cache and triggers callbacks to notify manager.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  Category that was modified.

• file_path (Path) –

  Path to the file that was modified.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _mark_category_modified(self, category: MODEL_REFERENCE_CATEGORY, file_path: Path) -> None:
    """Mark a category as modified after a write operation.

    This invalidates the cache and triggers callbacks to notify manager.

    Args:
        category: Category that was modified.
        file_path: Path to the file that was modified.

    """
    # Use mark_stale() to trigger callbacks, not _invalidate_cache()
    self.mark_stale(category)
    logger.debug(f"Marked category {category} as modified")

_mark_legacy_category_modified

_mark_legacy_category_modified(
    category: MODEL_REFERENCE_CATEGORY,
    legacy_file_path: Path,
) -> None

Mark a legacy category as modified after a write operation.

This invalidates the legacy cache and triggers callbacks.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  Category that was modified.

• legacy_file_path (Path) –

  Path to the legacy file that was modified.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _mark_legacy_category_modified(self, category: MODEL_REFERENCE_CATEGORY, legacy_file_path: Path) -> None:
    """Mark a legacy category as modified after a write operation.

    This invalidates the legacy cache and triggers callbacks.

    Args:
        category: Category that was modified.
        legacy_file_path: Path to the legacy file that was modified.

    """
    # Use mark_stale() to trigger callbacks
    self.mark_stale(category)
    logger.debug(f"Marked legacy category {category} as modified")

_read_legacy_csv_to_dict

_read_legacy_csv_to_dict(file_path: Path) -> dict[str, Any]

Read legacy CSV file (models.csv format) and convert to dict format.

Uses the shared csv_rows_to_legacy_dict to replicate convert.py exactly, including defaults.json merging, instruct_format, correct field ordering, and backend prefix generation (3 entries per model).

Parameters:

• file_path (Path) –

  Path to the legacy CSV file.

Returns:

• dict[str, Any] –

  dict[str, Any]: Model data with 3 entries per CSV row (matching db.json format).

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _read_legacy_csv_to_dict(self, file_path: Path) -> dict[str, Any]:
    """Read legacy CSV file (models.csv format) and convert to dict format.

    Uses the shared ``csv_rows_to_legacy_dict`` to replicate convert.py exactly,
    including defaults.json merging, instruct_format, correct field ordering,
    and backend prefix generation (3 entries per model).

    Args:
        file_path: Path to the legacy CSV file.

    Returns:
        dict[str, Any]: Model data with 3 entries per CSV row (matching db.json format).

    """
    parsed_rows, parse_issues = parse_legacy_text_csv_file(file_path)
    for issue in parse_issues:
        logger.warning(f"Legacy CSV issue for {issue.row_identifier}: {issue.message}")

    data = csv_rows_to_legacy_dict(parsed_rows, with_backend_prefixes=True)
    logger.debug(f"Read {len(data)} models from legacy CSV (with backend prefixes) from {file_path}")
    return data

_append_audit_event

_append_audit_event(
    *,
    domain: CanonicalFormat,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    operation: AuditOperation,
    payload: AuditPayload,
    logical_user_id: str | None,
    request_id: str | None,
) -> None

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _append_audit_event(
    self,
    *,
    domain: CanonicalFormat,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    operation: AuditOperation,
    payload: AuditPayload,
    logical_user_id: str | None,
    request_id: str | None,
) -> None:
    if self._audit_writer is None or logical_user_id is None:
        return

    try:
        self._audit_writer.append_event(
            domain=domain,
            category=category.value,
            model_name=model_name,
            operation=operation,
            logical_user_id=logical_user_id,
            payload=payload,
            request_id=request_id,
        )
    except OSError as exc:  # pragma: no cover - audit writes must not break CRUD
        logger.warning(f"Failed to append {domain} audit event for {category}/{model_name}: {exc}")

_read_csv_to_dict

_read_csv_to_dict(file_path: Path) -> dict[str, Any]

Read CSV file and convert to dict format (grouped by base name, no backend prefixes).

This reads the grouped CSV format and returns a dict with one entry per base model. No backend prefix duplicates are generated here - that only happens during GitHub sync.

Parameters:

• file_path (Path) –

  Path to the CSV file.

Returns:

• dict[str, Any] –

  dict[str, Any]: Model data with one entry per base model.

Raises:

• Exception –

  If CSV parsing fails.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _read_csv_to_dict(self, file_path: Path) -> dict[str, Any]:
    """Read CSV file and convert to dict format (grouped by base name, no backend prefixes).

    This reads the grouped CSV format and returns a dict with one entry per base model.
    No backend prefix duplicates are generated here - that only happens during GitHub sync.

    Args:
        file_path: Path to the CSV file.

    Returns:
        dict[str, Any]: Model data with one entry per base model.

    Raises:
        Exception: If CSV parsing fails.

    """
    data: dict[str, Any] = {}
    parsed_rows, parse_issues = parse_legacy_text_csv_file(file_path)
    for issue in parse_issues:
        logger.warning(f"CSV issue for {issue.row_identifier}: {issue.message}")

    for csv_row in parsed_rows:
        name = csv_row.name
        model_name = name.split("/")[1] if "/" in name else name
        tags = set(csv_row.tags)
        if csv_row.style:
            tags.add(csv_row.style)
        tags.add(f"{round(csv_row.parameters_bn, 0):.0f}B")

        settings_dict = dict(csv_row.settings) if csv_row.settings is not None else {}

        display_name = csv_row.display_name
        if not display_name:
            display_name = re.sub(r" +", " ", re.sub(r"[-_]", " ", model_name)).strip()

        record: dict[str, Any] = {
            "name": name,
            "model_name": model_name,
            "parameters": csv_row.parameters,
            "description": csv_row.description,
            "version": csv_row.version,
            "style": csv_row.style,
            "nsfw": csv_row.nsfw,
            "baseline": csv_row.baseline,
            "url": csv_row.url,
            "tags": sorted(tags),
            "settings": settings_dict,
            "display_name": display_name,
        }

        record = {k: v for k, v in record.items() if v or v is False}
        data[name] = record

    logger.debug(f"Read {len(data)} models from CSV (grouped, no backend prefixes) from {file_path}")
    return data

_write_dict_to_csv

_write_dict_to_csv(
    data: dict[str, Any], file_path: Path
) -> None

Write dict format to CSV file (removes backend prefix duplicates).

This writes the grouped CSV format with one entry per base model. Any backend-prefixed entries in the input are filtered out.

Parameters:

• data (dict[str, Any]) –

  Model data dict (may contain backend-prefixed duplicates).

• file_path (Path) –

  Path to write the CSV file.

Raises:

• Exception –

  If CSV writing fails.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _write_dict_to_csv(self, data: dict[str, Any], file_path: Path) -> None:
    """Write dict format to CSV file (removes backend prefix duplicates).

    This writes the grouped CSV format with one entry per base model.
    Any backend-prefixed entries in the input are filtered out.

    Args:
        data: Model data dict (may contain backend-prefixed duplicates).
        file_path: Path to write the CSV file.

    Raises:
        Exception: If CSV writing fails.

    """
    from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

    # Filter out backend-prefixed entries
    base_models: dict[str, Any] = {}
    for model_name, record in data.items():
        if has_legacy_text_backend_prefix(model_name):
            logger.debug(f"Skipping backend-prefixed entry during CSV write: {model_name}")
            continue
        base_models[model_name] = record

    # Convert to CSV rows
    csv_rows: list[dict[str, str]] = []

    for model_name, record in base_models.items():
        # Extract parameters in billions
        parameters_int = record.get("parameters", 0)
        params_bn = float(parameters_int) / 1_000_000_000

        # Extract tags and remove auto-generated ones
        tags = record.get("tags", [])
        tags_set = set(tags) if isinstance(tags, list) else set()

        # Remove style tag
        style = record.get("style")
        if style and style in tags_set:
            tags_set.discard(style)

        # Remove parameter size tag
        size_tag = f"{round(params_bn, 0):.0f}B"
        tags_set.discard(size_tag)

        # Serialize settings to compact JSON
        settings = record.get("settings", {})
        settings_str = json.dumps(settings, separators=(",", ":")) if settings else ""

        # Check if display_name is auto-generated (if so, omit it)
        display_name = record.get("display_name", "")
        model_name_field = record.get("model_name", model_name)
        auto_display = re.sub(r" +", " ", re.sub(r"[-_]", " ", model_name_field)).strip()
        if display_name == auto_display:
            display_name = ""

        csv_row = {
            "name": model_name,
            "parameters_bn": f"{params_bn:.1f}",
            "description": record.get("description", ""),
            "version": record.get("version", ""),
            "style": style or "",
            "nsfw": str(record.get("nsfw", False)).lower(),
            "baseline": record.get("baseline", ""),
            "url": record.get("url", ""),
            "tags": ",".join(sorted(tags_set)),
            "settings": settings_str,
            "display_name": display_name,
        }

        csv_rows.append(csv_row)

    # Write CSV file
    fieldnames = [
        "name",
        "parameters_bn",
        "description",
        "version",
        "style",
        "nsfw",
        "baseline",
        "url",
        "tags",
        "settings",
        "display_name",
    ]

    with open(file_path, "w", newline="", encoding="utf-8") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(csv_rows)

    logger.debug(f"Wrote {len(csv_rows)} models to CSV (grouped, no backend prefixes) to {file_path}")

fetch_category

fetch_category(
    category: MODEL_REFERENCE_CATEGORY,
    *,
    force_refresh: bool = False,
) -> dict[str, Any] | None

Fetch model reference data for a specific category from filesystem.

All v2 format files (including text_generation.json) are in JSON format. CSV format is only used for legacy files (legacy/models.csv).

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to fetch.

• force_refresh (bool, default: False ) –

  If True, bypass cache and read from disk.

Returns:

• dict[str, Any] | None –

  dict[str, Any] | None: The model reference data, or None if file doesn't exist.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def fetch_category(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    *,
    force_refresh: bool = False,
) -> dict[str, Any] | None:
    """Fetch model reference data for a specific category from filesystem.

    All v2 format files (including text_generation.json) are in JSON format.
    CSV format is only used for legacy files (legacy/models.csv).

    Args:
        category: The category to fetch.
        force_refresh: If True, bypass cache and read from disk.

    Returns:
        dict[str, Any] | None: The model reference data, or None if file doesn't exist.

    """
    with self._lock:
        if not (force_refresh or self.should_fetch_data(category)):
            return self._get_from_cache(category)

        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        if not file_path or not file_path.exists():
            logger.debug(f"File not found for {category}: {file_path}")
            self._store_in_cache(category, None)
            return None

        try:
            # All v2 files are JSON format (including text_generation.json)
            with open(file_path, encoding="utf-8") as f:
                data: dict[str, Any] = json.load(f)

            self._store_in_cache(category, data)
            logger.debug(f"Loaded {category} from {file_path}")
            return data

        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read {file_path}: {e}")
            self._invalidate_cache(category)
            return None

fetch_all_categories

fetch_all_categories(
    *, force_refresh: bool = False
) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]

Fetch model reference data for all categories.

Parameters:

• force_refresh (bool, default: False ) –

  If True, bypass cache for all categories.

Returns:

• dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None] –

  dict mapping categories to their model reference data.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def fetch_all_categories(
    self,
    *,
    force_refresh: bool = False,
) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]:
    """Fetch model reference data for all categories.

    Args:
        force_refresh: If True, bypass cache for all categories.

    Returns:
        dict mapping categories to their model reference data.

    """
    result: dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None] = {}

    for category in MODEL_REFERENCE_CATEGORY:
        result[category] = self.fetch_category(category, force_refresh=force_refresh)

    return result

fetch_category_async async

fetch_category_async(
    category: MODEL_REFERENCE_CATEGORY,
    *,
    httpx_client: AsyncClient | None = None,
    force_refresh: bool = False,
) -> dict[str, Any] | None

Asynchronously fetch model reference data for a category.

Note: File I/O is still synchronous as async file I/O doesn't provide significant benefits for small JSON files.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to fetch.

• httpx_client (AsyncClient | None, default: None ) –

  Optional httpx async client for downloads.

• force_refresh (bool, default: False ) –

  If True, bypass cache.

Returns:

• dict[str, Any] | None –

  dict[str, Any] | None: The model reference data.

Source code in src/horde_model_reference/backends/filesystem_backend.py

async def fetch_category_async(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    *,
    httpx_client: httpx.AsyncClient | None = None,
    force_refresh: bool = False,
) -> dict[str, Any] | None:
    """Asynchronously fetch model reference data for a category.

    Note: File I/O is still synchronous as async file I/O doesn't provide
    significant benefits for small JSON files.

    Args:
        category: The category to fetch.
        httpx_client: Optional httpx async client for downloads.
        force_refresh: If True, bypass cache.

    Returns:
        dict[str, Any] | None: The model reference data.

    """
    async with self._async_lock:
        if not (force_refresh or self.should_fetch_data(category)):
            return self._get_from_cache(category)

        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )
        if not file_path or not file_path.exists():
            logger.debug(f"File not found for {category}: {file_path}")
            self._store_in_cache(category, None)
            return None
        try:
            async with aiofiles.open(file_path, encoding="utf-8") as f:
                content = await f.read()
                data: dict[str, Any] = json.loads(content)

            self._store_in_cache(category, data)
            logger.debug(f"Loaded {category} from {file_path} asynchronously")
            return data
        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read {file_path} asynchronously: {e}")
            self._invalidate_cache(category)
            return None

fetch_all_categories_async async

fetch_all_categories_async(
    *,
    httpx_client: AsyncClient | None = None,
    force_refresh: bool = False,
) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]

Asynchronously fetch all categories (delegates to sync method).

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
async def fetch_all_categories_async(
    self,
    *,
    httpx_client: httpx.AsyncClient | None = None,
    force_refresh: bool = False,
) -> dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None]:
    """Asynchronously fetch all categories (delegates to sync method)."""
    result: dict[MODEL_REFERENCE_CATEGORY, dict[str, Any] | None] = {}

    for category in MODEL_REFERENCE_CATEGORY:
        result[category] = await self.fetch_category_async(
            category,
            httpx_client=httpx_client,
            force_refresh=force_refresh,
        )

    return result

get_category_file_path

get_category_file_path(
    category: MODEL_REFERENCE_CATEGORY,
) -> Path | None

Get the file path for a category's data.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get path for.

Returns:

• Path | None –

  Path | None: Path to the JSON file, or None if not configured.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_category_file_path(self, category: MODEL_REFERENCE_CATEGORY) -> Path | None:
    """Get the file path for a category's data.

    Args:
        category: The category to get path for.

    Returns:
        Path | None: Path to the JSON file, or None if not configured.

    """
    return horde_model_reference_paths.get_model_reference_file_path(
        category,
        base_path=self.base_path,
    )

get_all_category_file_paths

get_all_category_file_paths() -> dict[
    MODEL_REFERENCE_CATEGORY, Path | None
]

Get file paths for all categories.

Returns:

• dict ( dict[MODEL_REFERENCE_CATEGORY, Path | None] ) –

  Mapping of categories to their file paths.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_all_category_file_paths(self) -> dict[MODEL_REFERENCE_CATEGORY, Path | None]:
    """Get file paths for all categories.

    Returns:
        dict: Mapping of categories to their file paths.

    """
    return horde_model_reference_paths.get_all_model_reference_file_paths(base_path=self.base_path)

get_legacy_json

get_legacy_json(
    category: MODEL_REFERENCE_CATEGORY,
    redownload: bool = False,
) -> dict[str, Any] | None

Get legacy format data from legacy/ folder.

For text_generation category, reads from CSV format (models.csv). For other categories, reads from JSON format.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  Category to retrieve.

• redownload (bool, default: False ) –

  If True, bypass cache and read from disk.

Returns:

• dict[str, Any] | None –

  dict[str, Any] | None: The legacy format data, or None if file doesn't exist.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_legacy_json(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    redownload: bool = False,
) -> dict[str, Any] | None:
    """Get legacy format data from legacy/ folder.

    For text_generation category, reads from CSV format (models.csv).
    For other categories, reads from JSON format.

    Args:
        category: Category to retrieve.
        redownload: If True, bypass cache and read from disk.

    Returns:
        dict[str, Any] | None: The legacy format data, or None if file doesn't exist.

    """
    with self._lock:
        # Check cache first unless redownload
        if not redownload:
            legacy_dict, _ = self._get_legacy_from_cache(category)
            if legacy_dict is not None:
                return legacy_dict

        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            legacy_file_path, is_csv = self._resolve_legacy_text_generation_path()
        else:
            legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            is_csv = False

        if not legacy_file_path.exists():
            logger.debug(f"Legacy file not found for {category}: {legacy_file_path}")
            self._store_legacy_in_cache(category, None, None)
            return None

        try:
            # Handle CSV format for text_generation
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                if is_csv:
                    data = self._read_legacy_csv_to_dict(legacy_file_path)
                else:
                    with open(legacy_file_path, encoding="utf-8") as f:
                        data = cast(dict[str, Any], json.load(f))

                # Generate JSON string for cache
                content = json.dumps(data, indent=2, ensure_ascii=False)
                self._store_legacy_in_cache(category, data, content)
                logger.debug(f"Loaded legacy CSV for {category} from {legacy_file_path}")
                return data

            # Handle JSON format for other categories
            with open(legacy_file_path, encoding="utf-8") as f:
                content = f.read()
                data = cast(dict[str, Any], json.loads(content))

            self._store_legacy_in_cache(category, data, content)
            logger.debug(f"Loaded legacy JSON for {category} from {legacy_file_path}")
            return data

        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read legacy file {legacy_file_path}: {e}")
            self._invalidate_legacy_cache(category)
            return None

get_legacy_json_string

get_legacy_json_string(
    category: MODEL_REFERENCE_CATEGORY,
    redownload: bool = False,
) -> str | None

Get legacy format data as JSON string from legacy/ folder.

For text_generation category, reads CSV and converts to JSON string. For other categories, reads JSON format directly.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  Category to retrieve.

• redownload (bool, default: False ) –

  If True, bypass cache and read from disk.

Returns:

• str | None –

  str | None: The legacy format as JSON string, or None if file doesn't exist.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_legacy_json_string(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    redownload: bool = False,
) -> str | None:
    """Get legacy format data as JSON string from legacy/ folder.

    For text_generation category, reads CSV and converts to JSON string.
    For other categories, reads JSON format directly.

    Args:
        category: Category to retrieve.
        redownload: If True, bypass cache and read from disk.

    Returns:
        str | None: The legacy format as JSON string, or None if file doesn't exist.

    """
    with self._lock:
        # Check cache first unless redownload
        if not redownload:
            _, legacy_string = self._get_legacy_from_cache(category)
            if legacy_string is not None:
                return legacy_string

        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            legacy_file_path, is_csv = self._resolve_legacy_text_generation_path()
        else:
            legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
                category,
                base_path=self.base_path,
            )
            is_csv = False

        if not legacy_file_path.exists():
            logger.debug(f"Legacy file not found for {category}: {legacy_file_path}")
            self._store_legacy_in_cache(category, None, None)
            return None

        try:
            # Handle CSV format for text_generation
            if category == MODEL_REFERENCE_CATEGORY.text_generation:
                if is_csv:
                    data = self._read_legacy_csv_to_dict(legacy_file_path)
                else:
                    with open(legacy_file_path, encoding="utf-8") as f:
                        data = cast(dict[str, Any], json.load(f))
                # Generate JSON string
                content = json.dumps(data, indent=2, ensure_ascii=False)
                self._store_legacy_in_cache(category, data, content)
                logger.debug(f"Loaded legacy CSV string for {category} from {legacy_file_path}")
                return content

            # Handle JSON format for other categories
            with open(legacy_file_path, encoding="utf-8") as f:
                content = f.read()
                data = cast(dict[str, Any], json.loads(content))

            self._store_legacy_in_cache(category, data, content)
            logger.debug(f"Loaded legacy JSON string for {category} from {legacy_file_path}")
            return content

        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read legacy file {legacy_file_path}: {e}")
            self._invalidate_legacy_cache(category)
            return None

supports_writes

supports_writes() -> bool

Check if backend supports writes (always True for PRIMARY filesystem).

Returns:

• bool ( bool ) –

  Always True.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def supports_writes(self) -> bool:
    """Check if backend supports writes (always True for PRIMARY filesystem).

    Returns:
        bool: Always True.

    """
    return True

supports_metadata

supports_metadata() -> bool

Check if backend supports metadata tracking (always True for PRIMARY filesystem).

Returns:

• bool ( bool ) –

  Always True.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def supports_metadata(self) -> bool:
    """Check if backend supports metadata tracking (always True for PRIMARY filesystem).

    Returns:
        bool: Always True.

    """
    return True

update_model

update_model(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Update or create a model reference.

Modifies the JSON file on disk atomically for all categories (v2 format is always JSON). Preserves created_at and created_by metadata on updates.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to update.

• model_name (str) –

  The name of the model to update or create.

• record_dict (dict[str, Any]) –

  The model record data as a dictionary.

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Raises:

• FileNotFoundError –

  If the category file path is not configured.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def update_model(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Update or create a model reference.

    Modifies the JSON file on disk atomically for all categories (v2 format is always JSON).
    Preserves created_at and created_by metadata on updates.

    Args:
        category: The category to update.
        model_name: The name of the model to update or create.
        record_dict: The model record data as a dictionary.
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    Raises:
        FileNotFoundError: If the category file path is not configured.

    """
    from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

    with self._lock:
        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        if not file_path:
            raise FileNotFoundError(f"No file path configured for category {category}")

        # Read existing data (v2 format is always JSON, including text_generation)
        existing_data: dict[str, Any]
        if file_path.exists():
            try:
                with open(file_path, encoding="utf-8") as f:
                    existing_data = json.load(f)
            except json.JSONDecodeError as e:
                # V2 files should always be valid JSON - surface corruption errors
                logger.error(
                    f"Invalid JSON in v2 file {file_path}. This indicates data corruption. "
                    f"V2 format is always JSON, including text_generation.json. Error: {e}"
                )
                raise
            except OSError as e:
                logger.error(f"Failed to read {file_path}: {e}")
                raise
        else:
            existing_data = {}
            file_path.parent.mkdir(parents=True, exist_ok=True)

        # For text_generation, filter out backend prefix entries before updating
        if category == MODEL_REFERENCE_CATEGORY.text_generation and has_legacy_text_backend_prefix(model_name):
            logger.warning(
                f"Attempted to update backend-prefixed model {model_name} in text_generation. "
                "Backend prefixes are not stored internally - update the base model instead."
            )
            # Don't raise an error, just skip the update
            return

        # For text_generation, validate/transform the record
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            from horde_model_reference.text_model_write_processor import TextModelWriteProcessor

            processor = TextModelWriteProcessor()
            record_dict = processor.validate_and_transform(model_name, record_dict)

        # Determine if this is a create or update operation
        is_update = model_name in existing_data
        operation_type = OperationType.UPDATE if is_update else OperationType.CREATE
        previous_record = copy.deepcopy(existing_data[model_name]) if is_update else None

        # Handle per-model metadata
        if is_update:
            # Preserve creation metadata if model already exists
            existing_model = existing_data[model_name]
            self._metadata_manager.model_metadata.preserve_creation_fields(existing_model, record_dict)
            # Set updated_at timestamp
            self._metadata_manager.model_metadata.set_update_timestamp(record_dict)
        else:
            # For new models, ensure timestamps are populated (without overwriting existing values)
            self._metadata_manager.model_metadata.ensure_metadata_populated(record_dict)

        record_snapshot = copy.deepcopy(record_dict)
        existing_data[model_name] = record_snapshot

        temp_path = file_path.with_suffix(f".tmp.{time.time()}")
        try:
            # Write to temp file (v2 format is always JSON)
            with open(temp_path, "w", encoding="utf-8") as f:
                json.dump(existing_data, f, indent=2, ensure_ascii=False)
                f.flush()
                try:
                    import os

                    os.fsync(f.fileno())
                except OSError:
                    pass

            # Atomic replace
            if file_path.exists():
                backup_path = file_path.with_suffix(".bak")
                file_path.replace(backup_path)
                temp_path.replace(file_path)
                with contextlib.suppress(OSError):
                    backup_path.unlink()
            else:
                temp_path.replace(file_path)

            logger.info(f"Updated model {model_name} in category {category} at {file_path}")

            # Record metadata for observability (centralized hook point)
            self._metadata_manager.record_v2_operation(
                category=category,
                operation=operation_type,
                model_name=model_name,
                success=True,
                backend_type=self.__class__.__name__,
            )

            if logical_user_id is not None and self._audit_writer is not None:
                if is_update and previous_record is not None:
                    payload = AuditPayload.from_update(previous_record, record_snapshot)
                    audit_operation = AuditOperation.UPDATE
                elif is_update:
                    payload = AuditPayload.from_create(record_snapshot)
                    audit_operation = AuditOperation.UPDATE
                else:
                    payload = AuditPayload.from_create(record_snapshot)
                    audit_operation = AuditOperation.CREATE

                self._append_audit_event(
                    domain=CanonicalFormat.v2,
                    category=category,
                    model_name=model_name,
                    operation=audit_operation,
                    payload=payload,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )

            self._mark_category_modified(category, file_path)

        except (OSError, ValueError, TypeError) as e:
            try:
                if temp_path.exists():
                    temp_path.unlink()
            except OSError:
                pass
            logger.error(f"Failed to update model {model_name} in {category}: {e}")
            raise

delete_model

delete_model(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Delete a model reference.

Removes the model from the JSON file on disk atomically for all categories (v2 format is always JSON).

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category containing the model.

• model_name (str) –

  The name of the model to delete.

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Raises:

• FileNotFoundError –

  If the category file doesn't exist.

• KeyError –

  If the model doesn't exist in the category.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def delete_model(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Delete a model reference.

    Removes the model from the JSON file on disk atomically for all categories
    (v2 format is always JSON).

    Args:
        category: The category containing the model.
        model_name: The name of the model to delete.
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    Raises:
        FileNotFoundError: If the category file doesn't exist.
        KeyError: If the model doesn't exist in the category.

    """
    with self._lock:
        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        existing_data: dict[str, Any]
        if not file_path or not file_path.exists():
            raise FileNotFoundError(f"Category file not found: {file_path}")

        # Read existing data (v2 format is always JSON, including text_generation)
        try:
            with open(file_path, encoding="utf-8") as f:
                existing_data = json.load(f)
        except json.JSONDecodeError as e:
            # V2 files should always be valid JSON - surface corruption errors
            logger.error(
                f"Invalid JSON in v2 file {file_path}. This indicates data corruption. "
                f"V2 format is always JSON, including text_generation.json. Error: {e}"
            )
            raise
        except OSError as e:
            logger.error(f"Failed to read {file_path}: {e}")
            raise

        if model_name not in existing_data:
            raise KeyError(f"Model {model_name} not found in category {category}")

        deleted_snapshot = copy.deepcopy(existing_data[model_name])
        del existing_data[model_name]

        temp_path = file_path.with_suffix(f".tmp.{time.time()}")
        try:
            # Write to temp file (v2 format is always JSON)
            with open(temp_path, "w", encoding="utf-8") as f:
                json.dump(existing_data, f, indent=2, ensure_ascii=False)
                f.flush()
                try:
                    import os

                    os.fsync(f.fileno())
                except OSError:
                    pass

            backup_path = file_path.with_suffix(".bak")
            file_path.replace(backup_path)
            temp_path.replace(file_path)

            with contextlib.suppress(OSError):
                backup_path.unlink()

            logger.info(f"Deleted model {model_name} from category {category} at {file_path}")

            # Record metadata for observability (centralized hook point)
            self._metadata_manager.record_v2_operation(
                category=category,
                operation=OperationType.DELETE,
                model_name=model_name,
                success=True,
                backend_type=self.__class__.__name__,
            )

            if logical_user_id is not None and self._audit_writer is not None:
                payload = AuditPayload.from_delete(deleted_snapshot)
                self._append_audit_event(
                    domain=CanonicalFormat.v2,
                    category=category,
                    model_name=model_name,
                    operation=AuditOperation.DELETE,
                    payload=payload,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )

            self._mark_category_modified(category, file_path)

        except (OSError, ValueError, TypeError) as e:
            try:
                if temp_path.exists():
                    temp_path.unlink()
            except OSError:
                pass
            logger.error(f"Failed to delete model {model_name} from {category}: {e}")
            raise

supports_legacy_writes

supports_legacy_writes() -> bool

Check if backend supports legacy format writes.

Returns True only when canonical_format='LEGACY' in settings.

Returns:

• bool ( bool ) –

  True if legacy writes are supported.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def supports_legacy_writes(self) -> bool:
    """Check if backend supports legacy format writes.

    Returns True only when canonical_format='LEGACY' in settings.

    Returns:
        bool: True if legacy writes are supported.

    """
    from horde_model_reference import horde_model_reference_settings

    return horde_model_reference_settings.canonical_format == CanonicalFormat.LEGACY

update_model_legacy

update_model_legacy(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Update or create a model reference in legacy format.

This method modifies the legacy format JSON file on disk atomically.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to update.

• model_name (str) –

  The name of the model to update or create.

• record_dict (dict[str, Any]) –

  The model record data in legacy format as a dictionary.

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Raises:

• FileNotFoundError –

  If the legacy category file path is not configured.

• RuntimeError –

  If canonical_format is not set to 'LEGACY'.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def update_model_legacy(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Update or create a model reference in legacy format.

    This method modifies the legacy format JSON file on disk atomically.

    Args:
        category: The category to update.
        model_name: The name of the model to update or create.
        record_dict: The model record data in legacy format as a dictionary.
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    Raises:
        FileNotFoundError: If the legacy category file path is not configured.
        RuntimeError: If canonical_format is not set to 'LEGACY'.

    """
    from horde_model_reference import horde_model_reference_settings

    if not self.supports_legacy_writes():
        raise RuntimeError(
            "Legacy writes are only supported when canonical_format='LEGACY'. "
            f"Current setting: canonical_format='{horde_model_reference_settings.canonical_format}'"
        )

    with self._lock:
        # text_generation uses CSV as source of truth — route to dedicated handler
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            self._update_text_generation_csv(
                model_name,
                record_dict,
                logical_user_id=logical_user_id,
                request_id=request_id,
            )
            return

        legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
            category,
            base_path=self.base_path,
        )
        target_write_path = legacy_file_path

        if not legacy_file_path:
            raise FileNotFoundError(f"No legacy file path configured for category {category}")

        existing_data: dict[str, Any]
        if legacy_file_path.exists():
            try:
                with open(legacy_file_path, encoding="utf-8") as f:
                    existing_data = json.load(f)
            except (OSError, json.JSONDecodeError) as e:
                logger.error(f"Failed to read {legacy_file_path}: {e}")
                raise
        else:
            existing_data = {}
            target_write_path.parent.mkdir(parents=True, exist_ok=True)

        # Determine if this is a create or update operation
        is_update = model_name in existing_data
        operation_type = OperationType.UPDATE if is_update else OperationType.CREATE
        previous_record = copy.deepcopy(existing_data.get(model_name)) if is_update else None
        record_snapshot = copy.deepcopy(record_dict)

        existing_data[model_name] = record_snapshot

        temp_path = target_write_path.with_suffix(f".tmp.{time.time()}")
        try:
            with open(temp_path, "w", encoding="utf-8") as f:
                json.dump(existing_data, f, indent=2, ensure_ascii=False)
                f.flush()
                try:
                    import os

                    os.fsync(f.fileno())
                except OSError:
                    pass

            if target_write_path.exists():
                backup_path = target_write_path.with_suffix(".bak")
                target_write_path.replace(backup_path)
                temp_path.replace(target_write_path)
                with contextlib.suppress(OSError):
                    backup_path.unlink()
            else:
                temp_path.replace(target_write_path)

            logger.info(f"Updated legacy model {model_name} in category {category} at {target_write_path}")

            self._metadata_manager.record_legacy_operation(
                category=category,
                operation=operation_type,
                model_name=model_name,
                success=True,
                backend_type=self.__class__.__name__,
            )

            if logical_user_id is not None and self._audit_writer is not None:
                if is_update and previous_record is not None:
                    payload = AuditPayload.from_update(previous_record, record_snapshot)
                    audit_operation = AuditOperation.UPDATE
                else:
                    payload = AuditPayload.from_create(record_snapshot)
                    audit_operation = AuditOperation.CREATE
                self._append_audit_event(
                    domain=CanonicalFormat.LEGACY,
                    category=category,
                    model_name=model_name,
                    operation=audit_operation,
                    payload=payload,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )

            self._mark_legacy_category_modified(category, target_write_path)

        except (OSError, ValueError, TypeError) as e:
            try:
                if temp_path.exists():
                    temp_path.unlink()
            except OSError:
                pass
            logger.error(f"Failed to update legacy model {model_name} in {category}: {e}")
            raise

_update_text_generation_csv

_update_text_generation_csv(
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Update a text_generation model by writing CSV (not JSON) to models.csv.

Reads the existing CSV, validates/transforms the record, updates the row list, writes CSV back, and regenerates the cached dict representation.

Parameters:

• model_name (str) –

  The base model name (no backend prefix).

• record_dict (dict[str, Any]) –

  The model record data.

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _update_text_generation_csv(
    self,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Update a text_generation model by writing CSV (not JSON) to models.csv.

    Reads the existing CSV, validates/transforms the record, updates the row list,
    writes CSV back, and regenerates the cached dict representation.

    Args:
        model_name: The base model name (no backend prefix).
        record_dict: The model record data.
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    """
    from horde_model_reference.text_model_write_processor import TextModelWriteProcessor

    category = MODEL_REFERENCE_CATEGORY.text_generation
    csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
        category,
        base_path=self.base_path,
    )

    # Read existing CSV rows
    existing_rows: list[TextCSVRow] = []
    if csv_path.exists():
        existing_rows, parse_issues = parse_legacy_text_csv_file(csv_path)
        for issue in parse_issues:
            logger.warning(f"Legacy CSV parse issue for {issue.row_identifier}: {issue.message}")

    # Validate and transform the incoming record
    processor = TextModelWriteProcessor()
    record_dict = processor.validate_and_transform(model_name, record_dict)

    # Convert the validated record to a CSV row
    new_row = legacy_record_to_csv_row(model_name, record_dict)

    # Find and replace existing row, or append
    row_index: int | None = None
    previous_record: dict[str, Any] | None = None
    for i, row in enumerate(existing_rows):
        if row.name == model_name:
            row_index = i
            break

    if row_index is not None:
        # Capture previous record for audit before replacing
        old_dict = csv_rows_to_legacy_dict([existing_rows[row_index]], with_backend_prefixes=False)
        previous_record = old_dict.get(model_name)
        existing_rows[row_index] = new_row
        operation_type = OperationType.UPDATE
    else:
        existing_rows.append(new_row)
        operation_type = OperationType.CREATE

    # Write CSV back
    csv_path.parent.mkdir(parents=True, exist_ok=True)
    write_legacy_text_csv(existing_rows, csv_path)

    # Regenerate the full dict for cache
    full_data = csv_rows_to_legacy_dict(existing_rows, with_backend_prefixes=True)
    content = json.dumps(full_data, indent=2, ensure_ascii=False)
    self._store_legacy_in_cache(category, full_data, content)

    record_snapshot = copy.deepcopy(record_dict)
    logger.info(f"Updated legacy text_generation model {model_name} in CSV at {csv_path}")

    self._metadata_manager.record_legacy_operation(
        category=category,
        operation=operation_type,
        model_name=model_name,
        success=True,
        backend_type=self.__class__.__name__,
    )

    if logical_user_id is not None and self._audit_writer is not None:
        if operation_type == OperationType.UPDATE and previous_record is not None:
            payload = AuditPayload.from_update(previous_record, record_snapshot)
            audit_operation = AuditOperation.UPDATE
        else:
            payload = AuditPayload.from_create(record_snapshot)
            audit_operation = AuditOperation.CREATE
        self._append_audit_event(
            domain=CanonicalFormat.LEGACY,
            category=category,
            model_name=model_name,
            operation=audit_operation,
            payload=payload,
            logical_user_id=logical_user_id,
            request_id=request_id,
        )

    self._mark_legacy_category_modified(category, csv_path)

_delete_text_generation_csv

_delete_text_generation_csv(
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Delete a text_generation model from CSV, preserving CSV format.

Parameters:

• model_name (str) –

  The base model name (no backend prefix).

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Raises:

• FileNotFoundError –

  If the CSV file doesn't exist.

• KeyError –

  If the model doesn't exist.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _delete_text_generation_csv(
    self,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Delete a text_generation model from CSV, preserving CSV format.

    Args:
        model_name: The base model name (no backend prefix).
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    Raises:
        FileNotFoundError: If the CSV file doesn't exist.
        KeyError: If the model doesn't exist.

    """
    category = MODEL_REFERENCE_CATEGORY.text_generation
    csv_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
        category,
        base_path=self.base_path,
    )

    if not csv_path.exists():
        raise FileNotFoundError(f"Legacy CSV file not found: {csv_path}")

    existing_rows, parse_issues = parse_legacy_text_csv_file(csv_path)
    for issue in parse_issues:
        logger.warning(f"Legacy CSV parse issue for {issue.row_identifier}: {issue.message}")

    # Find and remove the row
    row_index: int | None = None
    for i, row in enumerate(existing_rows):
        if row.name == model_name:
            row_index = i
            break

    if row_index is None:
        raise KeyError(f"Model {model_name} not found in legacy text_generation CSV")

    deleted_row = existing_rows.pop(row_index)

    # Write CSV back
    write_legacy_text_csv(existing_rows, csv_path)

    # Regenerate the full dict for cache
    full_data = csv_rows_to_legacy_dict(existing_rows, with_backend_prefixes=True)
    content = json.dumps(full_data, indent=2, ensure_ascii=False)
    self._store_legacy_in_cache(category, full_data, content)

    # Capture the deleted record for audit
    deleted_dict = csv_rows_to_legacy_dict([deleted_row], with_backend_prefixes=False)
    deleted_snapshot = deleted_dict.get(model_name, {})

    logger.info(f"Deleted legacy text_generation model {model_name} from CSV at {csv_path}")

    self._metadata_manager.record_legacy_operation(
        category=category,
        operation=OperationType.DELETE,
        model_name=model_name,
        success=True,
        backend_type=self.__class__.__name__,
    )

    if logical_user_id is not None and self._audit_writer is not None:
        payload = AuditPayload.from_delete(deleted_snapshot)
        self._append_audit_event(
            domain=CanonicalFormat.LEGACY,
            category=category,
            model_name=model_name,
            operation=AuditOperation.DELETE,
            payload=payload,
            logical_user_id=logical_user_id,
            request_id=request_id,
        )

    self._mark_legacy_category_modified(category, csv_path)

delete_model_legacy

delete_model_legacy(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Delete a model reference from legacy format files.

This method removes the model from the legacy format JSON file on disk atomically.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category containing the model.

• model_name (str) –

  The name of the model to delete.

• logical_user_id (str | None, default: None ) –

  Optional logical user ID for audit logging.

• request_id (str | None, default: None ) –

  Optional request ID for audit logging.

Raises:

• FileNotFoundError –

  If the legacy category file doesn't exist.

• KeyError –

  If the model doesn't exist in the category.

• RuntimeError –

  If canonical_format is not set to 'LEGACY'.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def delete_model_legacy(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Delete a model reference from legacy format files.

    This method removes the model from the legacy format JSON file on disk atomically.

    Args:
        category: The category containing the model.
        model_name: The name of the model to delete.
        logical_user_id: Optional logical user ID for audit logging.
        request_id: Optional request ID for audit logging.

    Raises:
        FileNotFoundError: If the legacy category file doesn't exist.
        KeyError: If the model doesn't exist in the category.
        RuntimeError: If canonical_format is not set to 'LEGACY'.

    """
    from horde_model_reference import horde_model_reference_settings

    if not self.supports_legacy_writes():
        raise RuntimeError(
            "Legacy writes are only supported when canonical_format='LEGACY'. "
            f"Current setting: canonical_format='{horde_model_reference_settings.canonical_format}'"
        )

    with self._lock:
        # text_generation uses CSV as source of truth — route to dedicated handler
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            self._delete_text_generation_csv(
                model_name,
                logical_user_id=logical_user_id,
                request_id=request_id,
            )
            return

        legacy_file_path = horde_model_reference_paths.get_legacy_model_reference_file_path(
            category,
            base_path=self.base_path,
        )
        target_write_path = legacy_file_path

        existing_data: dict[str, Any]
        if not legacy_file_path or not legacy_file_path.exists():
            raise FileNotFoundError(f"Legacy category file not found: {legacy_file_path}")

        try:
            with open(legacy_file_path, encoding="utf-8") as f:
                existing_data = json.load(f)
        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read {legacy_file_path}: {e}")
            raise

        if model_name not in existing_data:
            raise KeyError(f"Model {model_name} not found in legacy category {category}")

        deleted_snapshot = copy.deepcopy(existing_data[model_name])
        del existing_data[model_name]

        temp_path = target_write_path.with_suffix(f".tmp.{time.time()}")
        try:
            with open(temp_path, "w", encoding="utf-8") as f:
                json.dump(existing_data, f, indent=2, ensure_ascii=False)
                f.flush()
                try:
                    import os

                    os.fsync(f.fileno())
                except OSError:
                    pass

            if target_write_path.exists():
                backup_path = target_write_path.with_suffix(".bak")
                target_write_path.replace(backup_path)
                temp_path.replace(target_write_path)
                with contextlib.suppress(OSError):
                    backup_path.unlink()
            else:
                temp_path.replace(target_write_path)

            logger.info(f"Deleted legacy model {model_name} from category {category} at {target_write_path}")

            self._metadata_manager.record_legacy_operation(
                category=category,
                operation=OperationType.DELETE,
                model_name=model_name,
                success=True,
                backend_type=self.__class__.__name__,
            )

            if logical_user_id is not None and self._audit_writer is not None:
                payload = AuditPayload.from_delete(deleted_snapshot)
                self._append_audit_event(
                    domain=CanonicalFormat.LEGACY,
                    category=category,
                    model_name=model_name,
                    operation=AuditOperation.DELETE,
                    payload=payload,
                    logical_user_id=logical_user_id,
                    request_id=request_id,
                )

            self._mark_legacy_category_modified(category, target_write_path)

        except (OSError, ValueError, TypeError) as e:
            try:
                if temp_path.exists():
                    temp_path.unlink()
            except OSError:
                pass
            logger.error(f"Failed to delete legacy model {model_name} from {category}: {e}")
            raise

_populate_model_metadata

_populate_model_metadata(
    category: MODEL_REFERENCE_CATEGORY,
    timestamp: int | None = None,
) -> int

Populate missing per-model metadata fields in a category's JSON file.

This method scans all models in a category file and ensures each has: - metadata.created_at (if missing) - metadata.updated_at (if missing)

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to populate metadata for.

• timestamp (int | None, default: None ) –

  The timestamp to use for created_at/updated_at. If None, uses current time.

Returns:

• int ( int ) –

  Number of models that had metadata populated.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def _populate_model_metadata(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    timestamp: int | None = None,
) -> int:
    """Populate missing per-model metadata fields in a category's JSON file.

    This method scans all models in a category file and ensures each has:
    - metadata.created_at (if missing)
    - metadata.updated_at (if missing)

    Args:
        category: The category to populate metadata for.
        timestamp: The timestamp to use for created_at/updated_at. If None, uses current time.

    Returns:
        int: Number of models that had metadata populated.

    """
    with self._lock:
        file_path = horde_model_reference_paths.get_model_reference_file_path(
            category,
            base_path=self.base_path,
        )

        if not file_path or not file_path.exists():
            logger.debug(f"Category file not found for {category}, skipping metadata population")
            return 0

        if timestamp is None:
            timestamp = int(time.time())

        try:
            with open(file_path, encoding="utf-8") as f:
                data: dict[str, Any] = json.load(f)
        except (OSError, json.JSONDecodeError) as e:
            logger.error(f"Failed to read {file_path} for metadata population: {e}")
            return 0

        models_updated = 0
        for _model_name, model_data in data.items():
            if not isinstance(model_data, dict):
                continue

            # Use ModelMetadataManager to ensure metadata is populated
            if self._metadata_manager.model_metadata.ensure_metadata_populated(model_data, timestamp):
                models_updated += 1

        if models_updated == 0:
            logger.trace(f"No models needed metadata population in {category}")
            return 0

        temp_path = file_path.with_suffix(f".tmp.{time.time()}")
        try:
            with open(temp_path, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=2, ensure_ascii=False)
                f.flush()
                try:
                    import os

                    os.fsync(f.fileno())
                except OSError:
                    pass

            backup_path = file_path.with_suffix(".bak")
            file_path.replace(backup_path)
            temp_path.replace(file_path)

            with contextlib.suppress(OSError):
                backup_path.unlink()

            logger.info(f"Populated metadata for {models_updated} models in {category}")
            self._mark_category_modified(category, file_path)

            return models_updated

        except (OSError, ValueError, TypeError) as e:
            try:
                if temp_path.exists():
                    temp_path.unlink()
            except OSError:
                pass
            logger.error(f"Failed to write metadata-populated file for {category}: {e}")
            return 0

ensure_category_metadata_populated

ensure_category_metadata_populated(
    category: MODEL_REFERENCE_CATEGORY,
    timestamp: int | None = None,
) -> CategoryMetadataPopulationResult

Ensure both CategoryMetadata and per-model metadata are populated for a category.

This method: 1. Checks if CategoryMetadata exists for both v2 and legacy formats 2. Initializes CategoryMetadata if missing 3. Populates per-model metadata fields in JSON files 4. Uses the same timestamp for both backend and model-level metadata

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to ensure metadata for.

• timestamp (int | None, default: None ) –

  Optional timestamp to use. If None, uses current time.

Returns:

• CategoryMetadataPopulationResult –

  dict with keys: - "category_metadata_initialized": bool - "legacy_metadata_initialized": bool - "models_updated": int - "timestamp_used": int

Source code in src/horde_model_reference/backends/filesystem_backend.py

def ensure_category_metadata_populated(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    timestamp: int | None = None,
) -> CategoryMetadataPopulationResult:
    """Ensure both CategoryMetadata and per-model metadata are populated for a category.

    This method:
    1. Checks if CategoryMetadata exists for both v2 and legacy formats
    2. Initializes CategoryMetadata if missing
    3. Populates per-model metadata fields in JSON files
    4. Uses the same timestamp for both backend and model-level metadata

    Args:
        category: The category to ensure metadata for.
        timestamp: Optional timestamp to use. If None, uses current time.

    Returns:
        dict with keys:
            - "category_metadata_initialized": bool
            - "legacy_metadata_initialized": bool
            - "models_updated": int
            - "timestamp_used": int

    """
    with self._lock:
        if timestamp is None:
            timestamp = int(time.time())

        result = CategoryMetadataPopulationResult(
            category_metadata_initialized=False,
            legacy_metadata_initialized=False,
            models_updated=0,
            timestamp_used=timestamp,
        )

        # Get or initialize v2 CategoryMetadata
        v2_metadata = self._metadata_manager.get_or_initialize_v2_metadata(
            category=category,
            backend_type=self.__class__.__name__,
        )

        # Check if it was just created (no prior data file existed)
        if v2_metadata.initialization_time == v2_metadata.last_updated:
            result.category_metadata_initialized = True
            logger.trace(f"Initialized v2 CategoryMetadata for {category}")

        # Use initialization_time from metadata
        timestamp = v2_metadata.initialization_time
        result.timestamp_used = timestamp

        # Get or initialize legacy CategoryMetadata
        legacy_metadata = self._metadata_manager.get_or_initialize_legacy_metadata(
            category=category,
            backend_type=self.__class__.__name__,
        )

        # Check if it was just created
        if legacy_metadata.initialization_time == legacy_metadata.last_updated:
            result.legacy_metadata_initialized = True
            logger.trace(f"Initialized legacy CategoryMetadata for {category}")

        # Populate per-model metadata using the determined timestamp
        models_updated = self._populate_model_metadata(category, timestamp)
        result.models_updated = models_updated

        if result.category_metadata_initialized or result.legacy_metadata_initialized or models_updated > 0:
            logger.info(
                f"Metadata population for {category}: "
                f"v2_meta={result.category_metadata_initialized}, "
                f"legacy_meta={result.legacy_metadata_initialized}, "
                f"models={models_updated}"
            )

        return result

ensure_all_metadata_populated

ensure_all_metadata_populated() -> (
    AllMetadataPopulationResult
)

Ensure metadata is populated for all categories that have files.

Scans all category files and ensures: 1. CategoryMetadata exists (both v2 and legacy formats) 2. All model records have metadata fields populated

This is called: - On FileSystemBackend initialization (PRIMARY mode) - After GitHub seeding completes

Returns:

• AllMetadataPopulationResult –

  AllMetadataPopulationResult with summary of metadata population.

Source code in src/horde_model_reference/backends/filesystem_backend.py

def ensure_all_metadata_populated(self) -> AllMetadataPopulationResult:
    """Ensure metadata is populated for all categories that have files.

    Scans all category files and ensures:
    1. CategoryMetadata exists (both v2 and legacy formats)
    2. All model records have metadata fields populated

    This is called:
    - On FileSystemBackend initialization (PRIMARY mode)
    - After GitHub seeding completes

    Returns:
        AllMetadataPopulationResult with summary of metadata population.

    """
    with self._lock:
        result = AllMetadataPopulationResult()

        logger.info("Starting metadata population scan for all categories")

        for category in MODEL_REFERENCE_CATEGORY:
            file_path = horde_model_reference_paths.get_model_reference_file_path(
                category,
                base_path=self.base_path,
            )

            # Skip categories that don't have files
            if not file_path or not file_path.exists():
                logger.debug(f"Skipping {category} - no file found")
                continue

            # Ensure metadata for this category
            category_result = self.ensure_category_metadata_populated(category)

            if (
                category_result.category_metadata_initialized
                or category_result.legacy_metadata_initialized
                or category_result.models_updated > 0
            ):
                result.categories_processed.append(category.value)
                result.total_categories += 1
                result.total_models_updated += category_result.models_updated

                if category_result.category_metadata_initialized:
                    result.total_metadata_initialized += 1
                if category_result.legacy_metadata_initialized:
                    result.total_metadata_initialized += 1

        if result.total_categories > 0:
            logger.info(
                f"Metadata population complete: "
                f"{result.total_categories} categories processed, "
                f"{result.total_models_updated} models updated, "
                f"{result.total_metadata_initialized} metadata files initialized"
            )
        else:
            logger.debug("No metadata population needed - all files already have metadata")

        return result

get_legacy_metadata

get_legacy_metadata(
    category: MODEL_REFERENCE_CATEGORY,
) -> CategoryMetadata

Get legacy format metadata for a specific category.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get metadata for.

Returns:

• CategoryMetadata –

  CategoryMetadata | None: The legacy metadata, or None if not available.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_legacy_metadata(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
    """Get legacy format metadata for a specific category.

    Args:
        category: The category to get metadata for.

    Returns:
        CategoryMetadata | None: The legacy metadata, or None if not available.

    """
    return self._metadata_manager.get_legacy_metadata(category)

get_legacy_metadata_async async

get_legacy_metadata_async(
    category: MODEL_REFERENCE_CATEGORY,
) -> CategoryMetadata

Asynchronously get legacy format metadata for a specific category.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get metadata for.

Returns:

• CategoryMetadata –

  CategoryMetadata | None: The legacy metadata, or None if not available.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
async def get_legacy_metadata_async(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
    """Asynchronously get legacy format metadata for a specific category.

    Args:
        category: The category to get metadata for.

    Returns:
        CategoryMetadata | None: The legacy metadata, or None if not available.

    """
    return self._metadata_manager.get_legacy_metadata(category)

get_metadata

get_metadata(
    category: MODEL_REFERENCE_CATEGORY,
) -> CategoryMetadata

Get v2 format metadata for a specific category.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get metadata for.

Returns:

• CategoryMetadata –

  CategoryMetadata | None: The v2 metadata, or None if not available.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_metadata(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
    """Get v2 format metadata for a specific category.

    Args:
        category: The category to get metadata for.

    Returns:
        CategoryMetadata | None: The v2 metadata, or None if not available.

    """
    return self._metadata_manager.get_v2_metadata(category)

get_metadata_async async

get_metadata_async(
    category: MODEL_REFERENCE_CATEGORY,
) -> CategoryMetadata

Asynchronously get v2 format metadata for a specific category.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to get metadata for.

Returns:

• CategoryMetadata –

  CategoryMetadata | None: The v2 metadata, or None if not available.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
async def get_metadata_async(self, category: MODEL_REFERENCE_CATEGORY) -> CategoryMetadata:
    """Asynchronously get v2 format metadata for a specific category.

    Args:
        category: The category to get metadata for.

    Returns:
        CategoryMetadata | None: The v2 metadata, or None if not available.

    """
    return self._metadata_manager.get_v2_metadata(category)

get_all_legacy_metadata

get_all_legacy_metadata() -> dict[
    MODEL_REFERENCE_CATEGORY, CategoryMetadata
]

Get legacy format metadata for all categories.

Returns:

• dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata] –

  dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_all_legacy_metadata(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
    """Get legacy format metadata for all categories.

    Returns:
        dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

    """
    return self._metadata_manager.get_all_legacy_metadata()

get_all_legacy_metadata_async async

get_all_legacy_metadata_async() -> dict[
    MODEL_REFERENCE_CATEGORY, CategoryMetadata
]

Asynchronously get legacy format metadata for all categories.

Returns:

• dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata] –

  dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
async def get_all_legacy_metadata_async(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
    """Asynchronously get legacy format metadata for all categories.

    Returns:
        dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their legacy metadata.

    """
    return self._metadata_manager.get_all_legacy_metadata()

get_all_metadata

get_all_metadata() -> dict[
    MODEL_REFERENCE_CATEGORY, CategoryMetadata
]

Get v2 format metadata for all categories.

Returns:

• dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata] –

  dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
def get_all_metadata(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
    """Get v2 format metadata for all categories.

    Returns:
        dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

    """
    return self._metadata_manager.get_all_v2_metadata()

get_all_metadata_async async

get_all_metadata_async() -> dict[
    MODEL_REFERENCE_CATEGORY, CategoryMetadata
]

Asynchronously get v2 format metadata for all categories.

Returns:

• dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata] –

  dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

Source code in src/horde_model_reference/backends/filesystem_backend.py

@override
async def get_all_metadata_async(self) -> dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]:
    """Asynchronously get v2 format metadata for all categories.

    Returns:
        dict[MODEL_REFERENCE_CATEGORY, CategoryMetadata]: Mapping of categories to their v2 metadata.

    """
    return self._metadata_manager.get_all_v2_metadata()

needs_refresh

needs_refresh(category: MODEL_REFERENCE_CATEGORY) -> bool

Source code in src/horde_model_reference/backends/replica_backend_base.py

@override
def needs_refresh(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    with self._lock:
        if category in self._stale_categories:
            logger.debug(f"Category {category} marked stale, needs refresh")
            return True

        last_updated = self._category_timestamps.get(category)

    if last_updated is None:
        # No timestamp means no data has been fetched/cached yet.
        # This is not a "refresh" scenario - it's an initial fetch scenario.
        # Callers should handle initial fetch separately from refresh logic.
        logger.debug(f"Category {category} has no timestamp, no refresh needed (not yet fetched)")
        return False

    if self._cache_ttl_seconds is not None:
        cache_stale = (time.time() - last_updated) > self._cache_ttl_seconds
        if cache_stale:
            logger.debug(f"Category {category} cache is stale, needs refresh")
            self.mark_stale(category)
            return True

    file_path = self._get_file_path_for_validation(category)
    if file_path and file_path.exists():
        try:
            current_mtime = file_path.stat().st_mtime
            last_known = self._last_known_mtimes.get(category, 0.0)
            if current_mtime != last_known:
                logger.debug(f"File {file_path.name} mtime changed, needs refresh")
                self.mark_stale(category)
                return True
        except Exception:
            return True

    return False

register_invalidation_callback

register_invalidation_callback(
    callback: Callable[[MODEL_REFERENCE_CATEGORY], None],
) -> None

Register a callback to be called when a category is invalidated.

This allows external components (like ModelReferenceManager) to be notified when cached data becomes stale and needs to be refreshed.

Parameters:

• callback (Callable[[MODEL_REFERENCE_CATEGORY], None]) –

  Function to call with the invalidated category.

Source code in src/horde_model_reference/backends/base.py

def register_invalidation_callback(
    self,
    callback: Callable[[MODEL_REFERENCE_CATEGORY], None],
) -> None:
    """Register a callback to be called when a category is invalidated.

    This allows external components (like ModelReferenceManager) to be notified
    when cached data becomes stale and needs to be refreshed.

    Args:
        callback: Function to call with the invalidated category.

    """
    self._invalidation_callbacks.append(callback)
    logger.debug(f"Registered invalidation callback: {getattr(callback, '__name__', repr(callback))}")

_notify_invalidation

_notify_invalidation(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Notify all registered callbacks that a category has been invalidated.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category that was invalidated.

Source code in src/horde_model_reference/backends/base.py

def _notify_invalidation(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Notify all registered callbacks that a category has been invalidated.

    Args:
        category: The category that was invalidated.

    """
    for callback in self._invalidation_callbacks:
        try:
            callback(category)
        except Exception as e:
            cb_name = getattr(callback, "__name__", repr(callback))
            logger.error(f"Invalidation callback {cb_name} failed for {category}: {e}")

_mark_stale_impl

_mark_stale_impl(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Source code in src/horde_model_reference/backends/replica_backend_base.py

@override
def _mark_stale_impl(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    with self._lock:
        self._stale_categories.add(category)

mark_stale

mark_stale(category: MODEL_REFERENCE_CATEGORY) -> None

Mark a category's data as stale, requiring refresh on next access.

This method calls the backend-specific implementation and then notifies all registered callbacks.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to mark as stale.

Implementation Note

The base class provides this public implementation. Subclasses should override _mark_stale_impl() instead of this method.

See Also

• _mark_stale_impl(): Backend-specific staleness tracking
• register_invalidation_callback(): Register callbacks for invalidation events

Source code in src/horde_model_reference/backends/base.py

def mark_stale(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Mark a category's data as stale, requiring refresh on next access.

    This method calls the backend-specific implementation and then notifies
    all registered callbacks.

    Args:
        category: The category to mark as stale.

    Implementation Note:
        The base class provides this public implementation. Subclasses should override
        [_mark_stale_impl()][(c)._mark_stale_impl]
        instead of this method.

    See Also:
        - [_mark_stale_impl()][(c)._mark_stale_impl]: Backend-specific staleness tracking
        - [register_invalidation_callback()][(c).register_invalidation_callback]:
          Register callbacks for invalidation events

    """
    self._mark_stale_impl(category)
    self._notify_invalidation(category)

support_any_writes

support_any_writes() -> bool

Check if this backend supports any write operations (v2 or legacy).

Returns:

• bool ( bool ) –

  True if any write operations are supported, False otherwise.

Source code in src/horde_model_reference/backends/base.py

def support_any_writes(self) -> bool:
    """Check if this backend supports any write operations (v2 or legacy).

    Returns:
        bool: True if any write operations are supported, False otherwise.

    """
    return self.supports_writes() or self.supports_legacy_writes()

supports_cache_warming

supports_cache_warming() -> bool

Check if this backend supports cache warming operations.

Cache warming pre-populates the cache with data to improve initial request performance. Typically only backends with distributed caching (like Redis) support this.

Returns:

• bool ( bool ) –

  True if cache warming is supported, False otherwise.

Source code in src/horde_model_reference/backends/base.py

def supports_cache_warming(self) -> bool:
    """Check if this backend supports cache warming operations.

    Cache warming pre-populates the cache with data to improve initial request performance.
    Typically only backends with distributed caching (like Redis) support this.

    Returns:
        bool: True if cache warming is supported, False otherwise.

    """
    return False

supports_health_checks

supports_health_checks() -> bool

Check if this backend supports health check operations.

Health checks verify that the backend's external dependencies (Redis, databases, etc.) are accessible and functioning correctly.

Returns:

• bool ( bool ) –

  True if health checks are supported, False otherwise.

Source code in src/horde_model_reference/backends/base.py

def supports_health_checks(self) -> bool:
    """Check if this backend supports health check operations.

    Health checks verify that the backend's external dependencies (Redis, databases, etc.)
    are accessible and functioning correctly.

    Returns:
        bool: True if health checks are supported, False otherwise.

    """
    return False

supports_statistics

supports_statistics() -> bool

Check if this backend supports statistics retrieval.

Statistics provide insights into backend performance, cache hits/misses, etc.

Returns:

• bool ( bool ) –

  True if statistics are supported, False otherwise.

Source code in src/horde_model_reference/backends/base.py

def supports_statistics(self) -> bool:
    """Check if this backend supports statistics retrieval.

    Statistics provide insights into backend performance, cache hits/misses, etc.

    Returns:
        bool: True if statistics are supported, False otherwise.

    """
    return False

update_model_from_base_model

update_model_from_base_model(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_model: BaseModel,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Update or create a model reference from a pydantic BaseModel.

This is an optional method that write-capable backends can implement. Read-only backends should leave the default implementation which raises NotImplementedError.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to update.

• model_name (str) –

  The name of the model to update or create.

• record_model (BaseModel) –

  The model record data as a pydantic BaseModel.

• logical_user_id (str | None, default: None ) –

  Immutable Horde user id for auditing contexts (optional).

• request_id (str | None, default: None ) –

  Optional tracing/idempotency identifier for audit correlation.

Raises:

• NotImplementedError –

  If the backend does not support write operations.

Implementation Note

The base class provides this implementation automatically. It: 1. Checks supports_writes() returns True 2. Converts the pydantic model to dict using model_dump(exclude_unset=True) 3. Calls update_model() with the dictionary

Backends that support writes typically don't need to override this method.

See Also

• update_model(): Update from dictionary (implement this)
• supports_writes(): Feature detection method

Source code in src/horde_model_reference/backends/base.py

def update_model_from_base_model(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_model: BaseModel,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Update or create a model reference from a pydantic BaseModel.

    This is an optional method that write-capable backends can implement.
    Read-only backends should leave the default implementation which raises NotImplementedError.

    Args:
        category: The category to update.
        model_name: The name of the model to update or create.
        record_model: The model record data as a pydantic BaseModel.
        logical_user_id: Immutable Horde user id for auditing contexts (optional).
        request_id: Optional tracing/idempotency identifier for audit correlation.

    Raises:
        NotImplementedError: If the backend does not support write operations.

    Implementation Note:
        The base class provides this implementation automatically. It:
        1. Checks [supports_writes()][(c).supports_writes] returns `True`
        2. Converts the pydantic model to dict using `model_dump(exclude_unset=True)`
        3. Calls [update_model()][(c).update_model] with the dictionary

        Backends that support writes typically don't need to override this method.

    See Also:
        - [update_model()][(c).update_model]: Update from dictionary (implement this)
        - [supports_writes()][(c).supports_writes]: Feature detection method

    """
    if not self.supports_writes():
        raise NotImplementedError(f"{self.__class__.__name__} does not support write operations")

    record_dict = record_model.model_dump(exclude_unset=True)
    self.update_model(
        category,
        model_name,
        record_dict,
        logical_user_id=logical_user_id,
        request_id=request_id,
    )

update_model_legacy_from_base_model

update_model_legacy_from_base_model(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_model: BaseModel,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Update or create a model reference in legacy format from a pydantic BaseModel.

This is an optional method that legacy-write-capable backends can implement. Only available when canonical_format='LEGACY' in PRIMARY mode.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to update.

• model_name (str) –

  The name of the model to update or create.

• record_model (BaseModel) –

  The model record data as a pydantic BaseModel.

• logical_user_id (str | None, default: None ) –

  Immutable Horde user id for auditing contexts (optional).

• request_id (str | None, default: None ) –

  Optional tracing/idempotency identifier for audit correlation.

Raises:

• NotImplementedError –

  If the backend does not support legacy write operations.

Source code in src/horde_model_reference/backends/base.py

def update_model_legacy_from_base_model(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_model: BaseModel,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Update or create a model reference in legacy format from a pydantic BaseModel.

    This is an optional method that legacy-write-capable backends can implement.
    Only available when canonical_format='LEGACY' in PRIMARY mode.

    Args:
        category: The category to update.
        model_name: The name of the model to update or create.
        record_model: The model record data as a pydantic BaseModel.
        logical_user_id: Immutable Horde user id for auditing contexts (optional).
        request_id: Optional tracing/idempotency identifier for audit correlation.

    Raises:
        NotImplementedError: If the backend does not support legacy write operations.

    """
    if not self.supports_legacy_writes():
        raise NotImplementedError(f"{self.__class__.__name__} does not support legacy write operations")

    record_dict = record_model.model_dump(exclude_unset=True)
    self.update_model_legacy(
        category,
        model_name,
        record_dict,
        logical_user_id=logical_user_id,
        request_id=request_id,
    )

warm_cache

warm_cache() -> None

Pre-populate cache with all categories for faster initial requests.

This is an optional method that backends with cache warming support can implement. Backends without cache warming should leave the default implementation.

Raises:

• NotImplementedError –

  If the backend does not support cache warming.

Source code in src/horde_model_reference/backends/base.py

def warm_cache(self) -> None:
    """Pre-populate cache with all categories for faster initial requests.

    This is an optional method that backends with cache warming support can implement.
    Backends without cache warming should leave the default implementation.

    Raises:
        NotImplementedError: If the backend does not support cache warming.

    """
    raise NotImplementedError(f"{self.__class__.__name__} does not support cache warming")

warm_cache_async async

warm_cache_async() -> None

Asynchronously pre-populate cache with all categories for faster initial requests.

This is an optional method that backends with cache warming support can implement. Backends without cache warming should leave the default implementation.

Raises:

• NotImplementedError –

  If the backend does not support async cache warming.

Source code in src/horde_model_reference/backends/base.py

async def warm_cache_async(self) -> None:
    """Asynchronously pre-populate cache with all categories for faster initial requests.

    This is an optional method that backends with cache warming support can implement.
    Backends without cache warming should leave the default implementation.

    Raises:
        NotImplementedError: If the backend does not support async cache warming.

    """
    raise NotImplementedError(f"{self.__class__.__name__} does not support async cache warming")

health_check

health_check() -> bool

Check the health of the backend's external dependencies.

This is an optional method that backends with health check support can implement. Backends without external dependencies should leave the default implementation.

Returns:

• bool ( bool ) –

  True if healthy, False otherwise.

Raises:

• NotImplementedError –

  If the backend does not support health checks.

Source code in src/horde_model_reference/backends/base.py

def health_check(self) -> bool:
    """Check the health of the backend's external dependencies.

    This is an optional method that backends with health check support can implement.
    Backends without external dependencies should leave the default implementation.

    Returns:
        bool: True if healthy, False otherwise.

    Raises:
        NotImplementedError: If the backend does not support health checks.

    """
    raise NotImplementedError(f"{self.__class__.__name__} does not support health checks")

get_statistics

get_statistics() -> dict[str, Any]

Get backend performance and usage statistics.

This is an optional method that backends with statistics support can implement. The structure of returned statistics is backend-specific.

Returns:

• dict[str, Any] –

  dict[str, Any]: Backend-specific statistics.

Raises:

• NotImplementedError –

  If the backend does not support statistics.

Source code in src/horde_model_reference/backends/base.py

def get_statistics(self) -> dict[str, Any]:
    """Get backend performance and usage statistics.

    This is an optional method that backends with statistics support can implement.
    The structure of returned statistics is backend-specific.

    Returns:
        dict[str, Any]: Backend-specific statistics.

    Raises:
        NotImplementedError: If the backend does not support statistics.

    """
    raise NotImplementedError(f"{self.__class__.__name__} does not support statistics")

get_replicate_mode

get_replicate_mode() -> ReplicateMode

Get the replication mode of this backend.

Returns:

• ReplicateMode ( ReplicateMode ) –

  The replicate mode (PRIMARY or REPLICA).

Source code in src/horde_model_reference/backends/base.py

def get_replicate_mode(self) -> ReplicateMode:
    """Get the replication mode of this backend.

    Returns:
        ReplicateMode: The replicate mode (PRIMARY or REPLICA).

    """
    return self._replicate_mode

_mark_category_fresh

_mark_category_fresh(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Record that we hold a fresh cache entry for category.

Also updates mtime if a file path is provided by the subclass.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to mark as fresh.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _mark_category_fresh(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Record that we hold a fresh cache entry for *category*.

    Also updates mtime if a file path is provided by the subclass.

    Args:
        category: The category to mark as fresh.

    """
    self._category_timestamps[category] = time.time()
    self._stale_categories.discard(category)

    file_path = self._get_file_path_for_validation(category)
    if file_path and file_path.exists():
        try:
            self._last_known_mtimes[category] = file_path.stat().st_mtime
        except Exception:
            self._last_known_mtimes[category] = 0.0

    logger.debug(f"Marked category {category} as fresh")

_invalidate_category_timestamp

_invalidate_category_timestamp(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Drop timestamp knowledge for category without adjusting payloads.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _invalidate_category_timestamp(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Drop timestamp knowledge for *category* without adjusting payloads."""
    self._category_timestamps.pop(category, None)

has_cached_data

has_cached_data(category: MODEL_REFERENCE_CATEGORY) -> bool

Check if any data has been cached for this category.

This is a simple existence check that doesn't validate freshness. Use this for initial fetch detection: "Have we loaded this at least once?"

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to check.

Returns:

• bool ( bool ) –

  True if data exists in cache (may be stale), False if never loaded.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def has_cached_data(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    """Check if any data has been cached for this category.

    This is a simple existence check that doesn't validate freshness.
    Use this for initial fetch detection: "Have we loaded this at least once?"

    Args:
        category: The category to check.

    Returns:
        bool: True if data exists in cache (may be stale), False if never loaded.

    """
    with self._lock:
        return category in self._cache

is_cache_valid

is_cache_valid(category: MODEL_REFERENCE_CATEGORY) -> bool

Check if cached data exists and is still valid for the given category.

This method performs comprehensive validation to determine if cached data can be used without refetching. It's primarily used internally by cache retrieval methods but can also be called directly for validation checks.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to validate.

Returns:

• bool –

  True if cache exists and all validation checks pass, False otherwise.

Validation Steps

The method performs checks in the following order:

1. Explicit Staleness: Returns False if category is in _stale_categories
2. Cache Existence: Returns False if category has never been cached
3. Timestamp Existence: Returns False if no timestamp recorded
4. TTL Expiration: Checks if cache_ttl_seconds exceeded (calls mark_stale() if expired)
5. File Modification: Compares current file mtime with cached mtime (calls mark_stale() if changed)
6. Custom Validation: Calls _additional_cache_validation() for subclass-specific checks

Side Effects

When staleness is detected (TTL expiration or mtime change), this method calls mark_stale() to trigger invalidation callbacks and notify the manager.

Related Methods

• has_cached_data(): Simple existence check, ignores validity
• should_fetch_data(): Combined check for "should I fetch?" (initial OR refresh)
• needs_refresh(): Checks if cached data should be refetched (staleness only)

Return Value Semantics

• Returns False for both "no data" and "stale data" cases
• Use has_cached_data() to distinguish between these cases
• Use needs_refresh() to check staleness without considering initial fetch

Note

This method is thread-safe and uses the internal _lock for synchronization.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def is_cache_valid(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    """Check if cached data exists and is still valid for the given category.

    This method performs comprehensive validation to determine if cached data can be
    used without refetching. It's primarily used internally by cache retrieval methods
    but can also be called directly for validation checks.

    Args:
        category: The category to validate.

    Returns:
        True if cache exists and all validation checks pass, False otherwise.

    Validation Steps:
        The method performs checks in the following order:

        1. **Explicit Staleness**: Returns False if category is in `_stale_categories`
        2. **Cache Existence**: Returns False if category has never been cached
        3. **Timestamp Existence**: Returns False if no timestamp recorded
        4. **TTL Expiration**: Checks if `cache_ttl_seconds` exceeded (calls `mark_stale()` if expired)
        5. **File Modification**: Compares current file mtime with cached mtime (calls `mark_stale()` if changed)
        6. **Custom Validation**: Calls `_additional_cache_validation()` for subclass-specific checks

    Side Effects:
        When staleness is detected (TTL expiration or mtime change), this method calls
        `mark_stale()` to trigger invalidation callbacks and notify the manager.

    Related Methods:
        - `has_cached_data()`: Simple existence check, ignores validity
        - `should_fetch_data()`: Combined check for "should I fetch?" (initial OR refresh)
        - `needs_refresh()`: Checks if cached data should be refetched (staleness only)

    Return Value Semantics:
        - Returns `False` for both "no data" and "stale data" cases
        - Use `has_cached_data()` to distinguish between these cases
        - Use `needs_refresh()` to check staleness without considering initial fetch


    Note:
        This method is thread-safe and uses the internal `_lock` for synchronization.

    """
    with self._lock:
        if category in self._stale_categories:
            logger.debug(f"Category {category} marked stale, cache invalid")
            return False

        if category not in self._cache:
            return False

        last_updated = self._category_timestamps.get(category)

    if last_updated is None:
        logger.debug(f"Category {category} has no timestamp, considering cache invalid")
        return False

    if self._cache_ttl_seconds is not None:
        elapsed = time.time() - last_updated
        if elapsed > self._cache_ttl_seconds:
            logger.debug(f"Category {category} TTL expired ({elapsed}s > {self._cache_ttl_seconds}s)")
            self.mark_stale(category)
            return False

    file_path = self._get_file_path_for_validation(category)
    if file_path and file_path.exists():
        try:
            current_mtime = file_path.stat().st_mtime
            last_known = self._last_known_mtimes.get(category, 0.0)
            if current_mtime != last_known:
                logger.debug(
                    f"File {file_path.name} mtime changed "
                    f"(current={current_mtime}, cached={last_known}), cache invalid"
                )
                self.mark_stale(category)
                return False
        except Exception:
            return False

    if not self._additional_cache_validation(category):
        logger.debug(f"Category {category} failed additional validation")
        return False

    return True

should_fetch_data

should_fetch_data(
    category: MODEL_REFERENCE_CATEGORY,
) -> bool

Determine if data should be fetched (initial load OR refresh).

This is a convenience method that combines both initial fetch detection and refresh detection into a single check. Use this when you want to know "should I fetch data now?" regardless of whether it's an initial load or a refresh.

This is equivalent to: not is_cache_valid(category) or needs_refresh(category) but handles the logic more efficiently.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to check.

Returns:

• bool ( bool ) –

  True if data should be fetched (either initial or refresh), False if cached data is valid and fresh.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def should_fetch_data(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    """Determine if data should be fetched (initial load OR refresh).

    This is a convenience method that combines both initial fetch detection
    and refresh detection into a single check. Use this when you want to
    know "should I fetch data now?" regardless of whether it's an initial
    load or a refresh.

    This is equivalent to: `not is_cache_valid(category) or needs_refresh(category)`
    but handles the logic more efficiently.

    Args:
        category: The category to check.

    Returns:
        bool: True if data should be fetched (either initial or refresh),
              False if cached data is valid and fresh.

    """
    return not self.is_cache_valid(category)

_set_cache_ttl_seconds

_set_cache_ttl_seconds(ttl_seconds: int | None) -> None

Allow subclasses to tweak TTL after initialization if desired.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _set_cache_ttl_seconds(self, ttl_seconds: int | None) -> None:
    """Allow subclasses to tweak TTL after initialization if desired."""
    self._cache_ttl_seconds = ttl_seconds

_additional_cache_validation

_additional_cache_validation(
    category: MODEL_REFERENCE_CATEGORY,
) -> bool

Perform additional cache validation.

Subclasses can override this to add custom validation logic beyond TTL and mtime checks. This is called during is_cache_valid().

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to validate.

Returns:

• bool ( bool ) –

  True if cache is valid, False to invalidate.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _additional_cache_validation(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    """Perform additional cache validation.

    Subclasses can override this to add custom validation logic beyond
    TTL and mtime checks. This is called during `is_cache_valid()`.

    Args:
        category: The category to validate.

    Returns:
        bool: True if cache is valid, False to invalidate.

    """
    return True

_fetch_with_cache

_fetch_with_cache(
    category: MODEL_REFERENCE_CATEGORY,
    fetch_fn: Callable[[], dict[str, Any] | None],
    *,
    force_refresh: bool = False,
) -> dict[str, Any] | None

Implement standard fetch pattern with automatic caching.

This helper method implements the recommended fetch pattern: 1. Check cache if not forcing refresh 2. Return cached data if valid 3. Fetch data using provided function 4. Store in cache and return

Use this in your fetch_category() implementations to avoid boilerplate.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to fetch.

• fetch_fn (Callable[[], dict[str, Any] | None]) –

  Callable that fetches the data (no args, returns dict or None).

• force_refresh (bool, default: False ) –

  If True, skip cache check and force fetch.

Returns:

• dict[str, Any] | None –

  dict[str, Any] | None: The fetched/cached data.

Example

def fetch_category(self, category, *, force_refresh=False): return self._fetch_with_cache( category, lambda: self._fetch_from_source(category), force_refresh=force_refresh )

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _fetch_with_cache(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    fetch_fn: Callable[[], dict[str, Any] | None],
    *,
    force_refresh: bool = False,
) -> dict[str, Any] | None:
    """Implement standard fetch pattern with automatic caching.

    This helper method implements the recommended fetch pattern:
    1. Check cache if not forcing refresh
    2. Return cached data if valid
    3. Fetch data using provided function
    4. Store in cache and return

    Use this in your fetch_category() implementations to avoid boilerplate.

    Args:
        category: The category to fetch.
        fetch_fn: Callable that fetches the data (no args, returns dict or None).
        force_refresh: If True, skip cache check and force fetch.

    Returns:
        dict[str, Any] | None: The fetched/cached data.

    Example:
        def fetch_category(self, category, *, force_refresh=False):
            return self._fetch_with_cache(
                category,
                lambda: self._fetch_from_source(category),
                force_refresh=force_refresh
            )

    """
    # Check cache first unless force refresh
    if not force_refresh:
        cached_data = self._get_from_cache(category)
        if cached_data is not None:
            return cached_data

    # Fetch data
    data = fetch_fn()

    # Store in cache
    if data is not None:
        self._store_in_cache(category, data)
    else:
        # Store None to indicate "checked but not found"
        self._store_in_cache(category, None)

    return data

_get_from_cache

_get_from_cache(
    category: MODEL_REFERENCE_CATEGORY,
) -> dict[str, Any] | None

Get data from cache if valid.

This is the primary method subclasses should use to retrieve cached data. It handles all validation logic internally, including initial fetch detection (returns None if data has never been loaded).

This method determines if an INITIAL fetch is needed by checking cache existence. Use needs_refresh() to check if existing cached data should be RE-fetched.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to retrieve from cache.

Returns:

• dict[str, Any] | None –

  dict[str, Any] | None: Cached data if valid, None if cache miss (initial fetch needed) or cache invalid (refresh needed).

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _get_from_cache(self, category: MODEL_REFERENCE_CATEGORY) -> dict[str, Any] | None:
    """Get data from cache if valid.

    This is the primary method subclasses should use to retrieve cached data.
    It handles all validation logic internally, including initial fetch detection
    (returns None if data has never been loaded).

    This method determines if an INITIAL fetch is needed by checking cache existence.
    Use `needs_refresh()` to check if existing cached data should be RE-fetched.

    Args:
        category: The category to retrieve from cache.

    Returns:
        dict[str, Any] | None: Cached data if valid, None if cache miss (initial fetch needed)
                               or cache invalid (refresh needed).

    """
    with self._lock:
        if self.is_cache_valid(category):
            logger.debug(f"Cache hit for {category}")
            return self._cache.get(category)

        logger.debug(f"Cache miss for {category}")
        return None

_store_in_cache

_store_in_cache(
    category: MODEL_REFERENCE_CATEGORY,
    data: dict[str, Any] | None,
) -> None

Store data in cache and mark category as fresh.

This is the primary method subclasses should use to store fetched data. It handles timestamp updates and mtime tracking internally.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to store.

• data (dict[str, Any] | None) –

  The data to cache, or None if category has no data.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _store_in_cache(self, category: MODEL_REFERENCE_CATEGORY, data: dict[str, Any] | None) -> None:
    """Store data in cache and mark category as fresh.

    This is the primary method subclasses should use to store fetched data.
    It handles timestamp updates and mtime tracking internally.

    Args:
        category: The category to store.
        data: The data to cache, or None if category has no data.

    """
    with self._lock:
        self._cache[category] = data
        # Only mark as fresh if we actually have data
        # None values indicate failed loads and should not prevent retries
        if data is not None:
            self._mark_category_fresh(category)
            logger.debug(f"Stored {category} in cache")
        else:
            logger.debug(f"Stored None for {category}, not marking as fresh")

_invalidate_cache

_invalidate_cache(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Invalidate cache for a category without deleting the data.

This marks the category as stale, forcing a refetch on next access.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to invalidate.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _invalidate_cache(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Invalidate cache for a category without deleting the data.

    This marks the category as stale, forcing a refetch on next access.

    Args:
        category: The category to invalidate.

    """
    with self._lock:
        self._stale_categories.add(category)
        self._category_timestamps.pop(category, None)
        logger.debug(f"Invalidated cache for {category}")

_mark_legacy_category_fresh

_mark_legacy_category_fresh(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Record that we hold a fresh legacy cache entry for category.

Also updates legacy file mtime if a path is provided by the subclass.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to mark as fresh.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _mark_legacy_category_fresh(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Record that we hold a fresh legacy cache entry for *category*.

    Also updates legacy file mtime if a path is provided by the subclass.

    Args:
        category: The category to mark as fresh.

    """
    self._legacy_cache_timestamps[category] = time.time()
    self._stale_legacy_categories.discard(category)

    legacy_file_path = self._get_legacy_file_path_for_validation(category)
    if legacy_file_path and legacy_file_path.exists():
        try:
            self._legacy_last_known_mtimes[category] = legacy_file_path.stat().st_mtime
        except Exception:
            self._legacy_last_known_mtimes[category] = 0.0

    logger.debug(f"Marked legacy category {category} as fresh")

is_legacy_cache_valid

is_legacy_cache_valid(
    category: MODEL_REFERENCE_CATEGORY,
) -> bool

Return True if the legacy cache for category is considered fresh.

Performs validation checks for legacy format cache: 1. Staleness check (explicit invalidation) 2. Cache existence check (dict or string) 3. TTL expiration check 4. File mtime check (if legacy file path provided by subclass)

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to validate.

Returns:

• bool ( bool ) –

  True if legacy cache is valid and can be used.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def is_legacy_cache_valid(self, category: MODEL_REFERENCE_CATEGORY) -> bool:
    """Return True if the legacy cache for *category* is considered fresh.

    Performs validation checks for legacy format cache:
    1. Staleness check (explicit invalidation)
    2. Cache existence check (dict or string)
    3. TTL expiration check
    4. File mtime check (if legacy file path provided by subclass)

    Args:
        category: The category to validate.

    Returns:
        bool: True if legacy cache is valid and can be used.

    """
    with self._lock:
        if category in self._stale_legacy_categories:
            logger.debug(f"Legacy category {category} marked stale, cache invalid")
            return False

        if category not in self._legacy_json_cache and category not in self._legacy_json_string_cache:
            return False

        last_updated = self._legacy_cache_timestamps.get(category)

    if last_updated is None:
        logger.debug(f"Legacy category {category} has no timestamp, considering cache invalid")
        return False

    if self._cache_ttl_seconds is not None:
        elapsed = time.time() - last_updated
        if elapsed > self._cache_ttl_seconds:
            logger.debug(f"Legacy category {category} TTL expired ({elapsed}s > {self._cache_ttl_seconds}s)")
            return False

    legacy_file_path = self._get_legacy_file_path_for_validation(category)
    if legacy_file_path and legacy_file_path.exists():
        try:
            current_mtime = legacy_file_path.stat().st_mtime
            last_known = self._legacy_last_known_mtimes.get(category, 0.0)
            if current_mtime != last_known:
                logger.debug(
                    f"Legacy file {legacy_file_path.name} mtime changed "
                    f"(current={current_mtime}, cached={last_known}), cache invalid"
                )
                return False
        except Exception:
            return False

    return True

_get_legacy_from_cache

_get_legacy_from_cache(
    category: MODEL_REFERENCE_CATEGORY,
) -> tuple[dict[str, Any] | None, str | None]

Get legacy data from cache if valid.

Returns both dict and string representations of legacy JSON.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to retrieve from cache.

Returns:

• tuple[dict[str, Any] | None, str | None] –

  tuple[dict | None, str | None]: (legacy_dict, legacy_string) or (None, None) if cache miss.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _get_legacy_from_cache(
    self,
    category: MODEL_REFERENCE_CATEGORY,
) -> tuple[dict[str, Any] | None, str | None]:
    """Get legacy data from cache if valid.

    Returns both dict and string representations of legacy JSON.

    Args:
        category: The category to retrieve from cache.

    Returns:
        tuple[dict | None, str | None]: (legacy_dict, legacy_string) or (None, None) if cache miss.

    """
    with self._lock:
        if self.is_legacy_cache_valid(category):
            logger.debug(f"Legacy cache hit for {category}")
            return (
                self._legacy_json_cache.get(category),
                self._legacy_json_string_cache.get(category),
            )

        logger.debug(f"Legacy cache miss for {category}")
        return None, None

_store_legacy_in_cache

_store_legacy_in_cache(
    category: MODEL_REFERENCE_CATEGORY,
    legacy_dict: dict[str, Any] | None,
    legacy_string: str | None,
) -> None

Store legacy data in cache and mark category as fresh.

Stores both dict and string representations of legacy JSON.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to store.

• legacy_dict (dict[str, Any] | None) –

  The legacy JSON as a dict, or None.

• legacy_string (str | None) –

  The legacy JSON as a string, or None.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _store_legacy_in_cache(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    legacy_dict: dict[str, Any] | None,
    legacy_string: str | None,
) -> None:
    """Store legacy data in cache and mark category as fresh.

    Stores both dict and string representations of legacy JSON.

    Args:
        category: The category to store.
        legacy_dict: The legacy JSON as a dict, or None.
        legacy_string: The legacy JSON as a string, or None.

    """
    with self._lock:
        self._legacy_json_cache[category] = legacy_dict
        self._legacy_json_string_cache[category] = legacy_string
        # Only mark as fresh if we actually have data
        # None values indicate failed loads and should not prevent retries
        if legacy_dict is not None or legacy_string is not None:
            self._mark_legacy_category_fresh(category)
            logger.debug(f"Stored legacy {category} in cache")
        else:
            logger.debug(f"Stored None for legacy {category}, not marking as fresh")

_invalidate_legacy_cache

_invalidate_legacy_cache(
    category: MODEL_REFERENCE_CATEGORY,
) -> None

Invalidate legacy cache for a category without deleting the data.

This marks the category as stale, forcing a refetch on next access.

Parameters:

• category (MODEL_REFERENCE_CATEGORY) –

  The category to invalidate.

Source code in src/horde_model_reference/backends/replica_backend_base.py

def _invalidate_legacy_cache(self, category: MODEL_REFERENCE_CATEGORY) -> None:
    """Invalidate legacy cache for a category without deleting the data.

    This marks the category as stale, forcing a refetch on next access.

    Args:
        category: The category to invalidate.

    """
    with self._lock:
        self._stale_legacy_categories.add(category)
        self._legacy_cache_timestamps.pop(category, None)
        logger.debug(f"Invalidated legacy cache for {category}")