"""Fail2ban jail configuration file parser and activator.

Parses the full set of fail2ban jail configuration files
(``jail.conf``, ``jail.local``, ``jail.d/*.conf``, ``jail.d/*.local``)
to discover all defined jails — both active and inactive — and provides
functions to activate or deactivate them by writing ``.local`` override
files.

Merge order (fail2ban convention):
    1. ``jail.conf``
    2. ``jail.local``
    3. ``jail.d/*.conf`` (alphabetical)
    4. ``jail.d/*.local`` (alphabetical)

Security note: the ``activate_jail`` and ``deactivate_jail`` callers must
supply a validated jail name.  This module validates the name against an
allowlist pattern before constructing any filesystem paths to prevent
directory traversal.
"""

from __future__ import annotations

import asyncio
import configparser
import contextlib
import io
import os
import re
import tempfile
from pathlib import Path
from typing import Any

import structlog

from app.models.config import (
    ActionConfig,
    ActionConfigUpdate,
    ActionCreateRequest,
    ActionListResponse,
    ActionUpdateRequest,
    ActivateJailRequest,
    AssignActionRequest,
    AssignFilterRequest,
    BantimeEscalation,
    FilterConfig,
    FilterConfigUpdate,
    FilterCreateRequest,
    FilterListResponse,
    FilterUpdateRequest,
    InactiveJail,
    InactiveJailListResponse,
    JailActivationResponse,
    JailValidationIssue,
    JailValidationResult,
    RollbackResponse,
)
from app.services import conffile_parser, jail_service
from app.utils.fail2ban_client import Fail2BanClient, Fail2BanConnectionError

log: structlog.stdlib.BoundLogger = structlog.get_logger()

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_SOCKET_TIMEOUT: float = 10.0

# Allowlist pattern for jail names used in path construction.
_SAFE_JAIL_NAME_RE: re.Pattern[str] = re.compile(
    r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)

# Sections that are not jail definitions.
_META_SECTIONS: frozenset[str] = frozenset({"INCLUDES", "DEFAULT"})

# True-ish values for the ``enabled`` key.
_TRUE_VALUES: frozenset[str] = frozenset({"true", "yes", "1"})

# False-ish values for the ``enabled`` key.
_FALSE_VALUES: frozenset[str] = frozenset({"false", "no", "0"})


# ---------------------------------------------------------------------------
# Custom exceptions
# ---------------------------------------------------------------------------


class JailNotFoundInConfigError(Exception):
    """Raised when the requested jail name is not defined in any config file."""

    def __init__(self, name: str) -> None:
        """Initialise with the jail name that was not found.

        Args:
            name: The jail name that could not be located.
        """
        self.name: str = name
        super().__init__(f"Jail not found in config files: {name!r}")


class JailAlreadyActiveError(Exception):
    """Raised when trying to activate a jail that is already active."""

    def __init__(self, name: str) -> None:
        """Initialise with the jail name.

        Args:
            name: The jail that is already active.
        """
        self.name: str = name
        super().__init__(f"Jail is already active: {name!r}")


class JailAlreadyInactiveError(Exception):
    """Raised when trying to deactivate a jail that is already inactive."""

    def __init__(self, name: str) -> None:
        """Initialise with the jail name.

        Args:
            name: The jail that is already inactive.
        """
        self.name: str = name
        super().__init__(f"Jail is already inactive: {name!r}")


class JailNameError(Exception):
    """Raised when a jail name contains invalid characters."""


class ConfigWriteError(Exception):
    """Raised when writing a ``.local`` override file fails."""


class FilterNameError(Exception):
    """Raised when a filter name contains invalid characters."""


class FilterAlreadyExistsError(Exception):
    """Raised when trying to create a filter whose ``.conf`` or ``.local`` already exists."""

    def __init__(self, name: str) -> None:
        """Initialise with the filter name that already exists.

        Args:
            name: The filter name that already exists.
        """
        self.name: str = name
        super().__init__(f"Filter already exists: {name!r}")


class FilterReadonlyError(Exception):
    """Raised when trying to delete a shipped ``.conf`` filter with no ``.local`` override."""

    def __init__(self, name: str) -> None:
        """Initialise with the filter name that cannot be deleted.

        Args:
            name: The filter name that is read-only (shipped ``.conf`` only).
        """
        self.name: str = name
        super().__init__(
            f"Filter {name!r} is a shipped default (.conf only); "
            "only user-created .local files can be deleted."
        )


class FilterInvalidRegexError(Exception):
    """Raised when a regex pattern fails to compile."""

    def __init__(self, pattern: str, error: str) -> None:
        """Initialise with the invalid pattern and the compile error.

        Args:
            pattern: The regex string that failed to compile.
            error: The ``re.error`` message.
        """
        self.pattern: str = pattern
        self.error: str = error
        super().__init__(f"Invalid regex {pattern!r}: {error}")


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------


def _safe_jail_name(name: str) -> str:
    """Validate *name* and return it unchanged or raise :class:`JailNameError`.

    Args:
        name: Proposed jail name.

    Returns:
        The name unchanged if valid.

    Raises:
        JailNameError: If *name* contains unsafe characters.
    """
    if not _SAFE_JAIL_NAME_RE.match(name):
        raise JailNameError(
            f"Jail name {name!r} contains invalid characters. "
            "Only alphanumeric characters, hyphens, underscores, and dots are "
            "allowed; must start with an alphanumeric character."
        )
    return name


def _safe_filter_name(name: str) -> str:
    """Validate *name* and return it unchanged or raise :class:`FilterNameError`.

    Args:
        name: Proposed filter name (without extension).

    Returns:
        The name unchanged if valid.

    Raises:
        FilterNameError: If *name* contains unsafe characters.
    """
    if not _SAFE_FILTER_NAME_RE.match(name):
        raise FilterNameError(
            f"Filter name {name!r} contains invalid characters. "
            "Only alphanumeric characters, hyphens, underscores, and dots are "
            "allowed; must start with an alphanumeric character."
        )
    return name


def _ordered_config_files(config_dir: Path) -> list[Path]:
    """Return all jail config files in fail2ban merge order.

    Args:
        config_dir: The fail2ban configuration root directory.

    Returns:
        List of paths in ascending priority order (later entries override
        earlier ones).
    """
    files: list[Path] = []

    jail_conf = config_dir / "jail.conf"
    if jail_conf.is_file():
        files.append(jail_conf)

    jail_local = config_dir / "jail.local"
    if jail_local.is_file():
        files.append(jail_local)

    jail_d = config_dir / "jail.d"
    if jail_d.is_dir():
        files.extend(sorted(jail_d.glob("*.conf")))
        files.extend(sorted(jail_d.glob("*.local")))

    return files


def _build_parser() -> configparser.RawConfigParser:
    """Create a :class:`configparser.RawConfigParser` for fail2ban configs.

    Returns:
        Parser with interpolation disabled and case-sensitive option names.
    """
    parser = configparser.RawConfigParser(interpolation=None, strict=False)
    # fail2ban keys are lowercase but preserve case to be safe.
    parser.optionxform = str  # type: ignore[assignment]
    return parser


def _is_truthy(value: str) -> bool:
    """Return ``True`` if *value* is a fail2ban boolean true string.

    Args:
        value: Raw string from config (e.g. ``"true"``, ``"yes"``, ``"1"``).

    Returns:
        ``True`` when the value represents enabled.
    """
    return value.strip().lower() in _TRUE_VALUES


def _parse_int_safe(value: str) -> int | None:
    """Parse *value* as int, returning ``None`` on failure.

    Args:
        value: Raw string to parse.

    Returns:
        Integer value, or ``None``.
    """
    try:
        return int(value.strip())
    except (ValueError, AttributeError):
        return None


def _parse_time_to_seconds(value: str | None, default: int) -> int:
    """Convert a fail2ban time string (e.g. ``1h``, ``10m``, ``3600``) to seconds.

    Supports the suffixes ``s`` (seconds), ``m`` (minutes), ``h`` (hours),
    ``d`` (days), ``w`` (weeks), and plain integers (already seconds).
    ``-1`` is treated as a permanent ban and returned as-is.

    Args:
        value: Raw time string from config, or ``None``.
        default: Value to return when ``value`` is absent or unparseable.

    Returns:
        Duration in seconds, or ``-1`` for permanent, or ``default`` on failure.
    """
    if not value:
        return default
    stripped = value.strip()
    if stripped == "-1":
        return -1
    multipliers: dict[str, int] = {
        "w": 604800,
        "d": 86400,
        "h": 3600,
        "m": 60,
        "s": 1,
    }
    for suffix, factor in multipliers.items():
        if stripped.endswith(suffix) and len(stripped) > 1:
            try:
                return int(stripped[:-1]) * factor
            except ValueError:
                return default
    try:
        return int(stripped)
    except ValueError:
        return default


def _parse_multiline(raw: str) -> list[str]:
    """Split a multi-line INI value into individual non-blank lines.

    Args:
        raw: Raw multi-line string from configparser.

    Returns:
        List of stripped, non-empty, non-comment strings.
    """
    result: list[str] = []
    for line in raw.splitlines():
        stripped = line.strip()
        if stripped and not stripped.startswith("#"):
            result.append(stripped)
    return result


def _resolve_filter(raw_filter: str, jail_name: str, mode: str) -> str:
    """Resolve fail2ban variable placeholders in a filter string.

    Handles the common default ``%(__name__)s[mode=%(mode)s]`` pattern that
    fail2ban uses so the filter name displayed to the user is readable.

    Args:
        raw_filter: Raw ``filter`` value from config (may contain ``%()s``).
        jail_name: The jail's section name, used to substitute ``%(__name__)s``.
        mode: The jail's ``mode`` value, used to substitute ``%(mode)s``.

    Returns:
        Human-readable filter string.
    """
    result = raw_filter.replace("%(__name__)s", jail_name)
    result = result.replace("%(mode)s", mode)
    return result


def _parse_jails_sync(
    config_dir: Path,
) -> tuple[dict[str, dict[str, str]], dict[str, str]]:
    """Synchronously parse all jail configs and return merged definitions.

    This is a CPU-bound / IO-bound sync function; callers must dispatch to
    an executor for async use.

    Args:
        config_dir: The fail2ban configuration root directory.

    Returns:
        A two-tuple ``(jails, source_files)`` where:

        - ``jails``: ``{jail_name: {key: value}}`` – merged settings for each
          jail with DEFAULT values already applied.
        - ``source_files``: ``{jail_name: str(path)}`` – path of the file that
          last defined each jail section (for display in the UI).
    """
    parser = _build_parser()
    files = _ordered_config_files(config_dir)

    # Track which file each section came from (last write wins).
    source_files: dict[str, str] = {}
    for path in files:
        try:
            single = _build_parser()
            single.read(str(path), encoding="utf-8")
            for section in single.sections():
                if section not in _META_SECTIONS:
                    source_files[section] = str(path)
        except (configparser.Error, OSError) as exc:
            log.warning("jail_config_read_error", path=str(path), error=str(exc))

    # Full merged parse: configparser applies DEFAULT values to every section.
    try:
        parser.read([str(p) for p in files], encoding="utf-8")
    except configparser.Error as exc:
        log.warning("jail_config_parse_error", error=str(exc))

    jails: dict[str, dict[str, str]] = {}
    for section in parser.sections():
        if section in _META_SECTIONS:
            continue
        try:
            # items() merges DEFAULT values automatically.
            jails[section] = dict(parser.items(section))
        except configparser.Error as exc:
            log.warning(
                "jail_section_parse_error", section=section, error=str(exc)
            )

    log.debug("jails_parsed", count=len(jails), config_dir=str(config_dir))
    return jails, source_files


def _build_inactive_jail(
    name: str,
    settings: dict[str, str],
    source_file: str,
) -> InactiveJail:
    """Construct an :class:`~app.models.config.InactiveJail` from raw settings.

    Args:
        name: Jail section name.
        settings: Merged key→value dict (DEFAULT values already applied).
        source_file: Path of the file that last defined this section.

    Returns:
        Populated :class:`~app.models.config.InactiveJail`.
    """
    raw_filter = settings.get("filter", "")
    mode = settings.get("mode", "normal")
    filter_name = _resolve_filter(raw_filter, name, mode) if raw_filter else name

    raw_action = settings.get("action", "")
    actions = _parse_multiline(raw_action) if raw_action else []

    raw_logpath = settings.get("logpath", "")
    logpath = _parse_multiline(raw_logpath) if raw_logpath else []

    enabled_raw = settings.get("enabled", "false")
    enabled = _is_truthy(enabled_raw)

    maxretry_raw = settings.get("maxretry", "")
    maxretry = _parse_int_safe(maxretry_raw)

    # Extended fields for full GUI display
    ban_time_seconds = _parse_time_to_seconds(settings.get("bantime"), 600)
    find_time_seconds = _parse_time_to_seconds(settings.get("findtime"), 600)
    log_encoding = settings.get("logencoding") or "auto"
    backend = settings.get("backend") or "auto"
    date_pattern = settings.get("datepattern") or None
    use_dns = settings.get("usedns") or "warn"
    prefregex = settings.get("prefregex") or ""
    fail_regex = _parse_multiline(settings.get("failregex", ""))
    ignore_regex = _parse_multiline(settings.get("ignoreregex", ""))

    # Ban-time escalation
    esc_increment = _is_truthy(settings.get("bantime.increment", "false"))
    esc_factor_raw = settings.get("bantime.factor")
    esc_factor = float(esc_factor_raw) if esc_factor_raw else None
    esc_formula = settings.get("bantime.formula") or None
    esc_multipliers = settings.get("bantime.multipliers") or None
    esc_max_raw = settings.get("bantime.maxtime")
    esc_max_time = _parse_time_to_seconds(esc_max_raw, 0) if esc_max_raw else None
    esc_rnd_raw = settings.get("bantime.rndtime")
    esc_rnd_time = _parse_time_to_seconds(esc_rnd_raw, 0) if esc_rnd_raw else None
    esc_overall = _is_truthy(settings.get("bantime.overalljails", "false"))
    bantime_escalation = (
        BantimeEscalation(
            increment=esc_increment,
            factor=esc_factor,
            formula=esc_formula,
            multipliers=esc_multipliers,
            max_time=esc_max_time,
            rnd_time=esc_rnd_time,
            overall_jails=esc_overall,
        )
        if esc_increment
        else None
    )

    return InactiveJail(
        name=name,
        filter=filter_name,
        actions=actions,
        port=settings.get("port") or None,
        logpath=logpath,
        bantime=settings.get("bantime") or None,
        findtime=settings.get("findtime") or None,
        maxretry=maxretry,
        ban_time_seconds=ban_time_seconds,
        find_time_seconds=find_time_seconds,
        log_encoding=log_encoding,
        backend=backend,
        date_pattern=date_pattern,
        use_dns=use_dns,
        prefregex=prefregex,
        fail_regex=fail_regex,
        ignore_regex=ignore_regex,
        bantime_escalation=bantime_escalation,
        source_file=source_file,
        enabled=enabled,
    )


async def _get_active_jail_names(socket_path: str) -> set[str]:
    """Fetch the set of currently running jail names from fail2ban.

    Returns an empty set gracefully if fail2ban is unreachable.

    Args:
        socket_path: Path to the fail2ban Unix domain socket.

    Returns:
        Set of active jail names, or empty set on connection failure.
    """
    try:
        client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT)

        def _to_dict_inner(pairs: Any) -> dict[str, Any]:
            if not isinstance(pairs, (list, tuple)):
                return {}
            result: dict[str, Any] = {}
            for item in pairs:
                try:
                    k, v = item
                    result[str(k)] = v
                except (TypeError, ValueError):
                    pass
            return result

        def _ok(response: Any) -> Any:
            code, data = response
            if code != 0:
                raise ValueError(f"fail2ban error {code}: {data!r}")
            return data

        status_raw = _ok(await client.send(["status"]))
        status_dict = _to_dict_inner(status_raw)
        jail_list_raw: str = str(status_dict.get("Jail list", "") or "").strip()
        if not jail_list_raw:
            return set()
        return {j.strip() for j in jail_list_raw.split(",") if j.strip()}
    except Fail2BanConnectionError:
        log.warning("fail2ban_unreachable_during_inactive_list")
        return set()
    except Exception as exc:  # noqa: BLE001
        log.warning(
            "fail2ban_status_error_during_inactive_list", error=str(exc)
        )
        return set()


# ---------------------------------------------------------------------------
# Validation helpers (Task 3)
# ---------------------------------------------------------------------------

# Seconds to wait between fail2ban liveness probes after a reload.
_POST_RELOAD_PROBE_INTERVAL: float = 2.0

# Maximum number of post-reload probe attempts (initial attempt + retries).
_POST_RELOAD_MAX_ATTEMPTS: int = 4


def _extract_action_base_name(action_str: str) -> str | None:
    """Return the base action name from an action assignment string.

    Returns ``None`` for complex fail2ban expressions that cannot be resolved
    to a single filename (e.g. ``%(action_)s`` interpolations or multi-token
    composite actions).

    Args:
        action_str: A single line from the jail's ``action`` setting.

    Returns:
        Simple base name suitable for a filesystem lookup, or ``None``.
    """
    if "%" in action_str or "$" in action_str:
        return None
    base = action_str.split("[")[0].strip()
    if _SAFE_ACTION_NAME_RE.match(base):
        return base
    return None


def _validate_jail_config_sync(
    config_dir: Path,
    name: str,
) -> JailValidationResult:
    """Run synchronous pre-activation checks on a jail configuration.

    Validates:
    1. Filter file existence in ``filter.d/``.
    2. Action file existence in ``action.d/`` (for resolvable action names).
    3. Regex compilation for every ``failregex`` and ``ignoreregex`` pattern.
    4. Log path existence on disk (generates warnings, not errors).

    Args:
        config_dir: The fail2ban configuration root directory.
        name: Validated jail name.

    Returns:
        :class:`~app.models.config.JailValidationResult` with any issues found.
    """
    issues: list[JailValidationIssue] = []

    all_jails, _ = _parse_jails_sync(config_dir)
    settings = all_jails.get(name)

    if settings is None:
        return JailValidationResult(
            jail_name=name,
            valid=False,
            issues=[
                JailValidationIssue(
                    field="name",
                    message=f"Jail {name!r} not found in config files.",
                )
            ],
        )

    filter_d = config_dir / "filter.d"
    action_d = config_dir / "action.d"

    # 1. Filter existence check.
    raw_filter = settings.get("filter", "")
    if raw_filter:
        mode = settings.get("mode", "normal")
        resolved = _resolve_filter(raw_filter, name, mode)
        base_filter = _extract_filter_base_name(resolved)
        if base_filter:
            conf_ok = (filter_d / f"{base_filter}.conf").is_file()
            local_ok = (filter_d / f"{base_filter}.local").is_file()
            if not conf_ok and not local_ok:
                issues.append(
                    JailValidationIssue(
                        field="filter",
                        message=(
                            f"Filter file not found: filter.d/{base_filter}.conf"
                            " (or .local)"
                        ),
                    )
                )

    # 2. Action existence check.
    raw_action = settings.get("action", "")
    if raw_action:
        for action_line in _parse_multiline(raw_action):
            action_name = _extract_action_base_name(action_line)
            if action_name:
                conf_ok = (action_d / f"{action_name}.conf").is_file()
                local_ok = (action_d / f"{action_name}.local").is_file()
                if not conf_ok and not local_ok:
                    issues.append(
                        JailValidationIssue(
                            field="action",
                            message=(
                                f"Action file not found: action.d/{action_name}.conf"
                                " (or .local)"
                            ),
                        )
                    )

    # 3. failregex compilation.
    for pattern in _parse_multiline(settings.get("failregex", "")):
        try:
            re.compile(pattern)
        except re.error as exc:
            issues.append(
                JailValidationIssue(
                    field="failregex",
                    message=f"Invalid regex pattern: {exc}",
                )
            )

    # 4. ignoreregex compilation.
    for pattern in _parse_multiline(settings.get("ignoreregex", "")):
        try:
            re.compile(pattern)
        except re.error as exc:
            issues.append(
                JailValidationIssue(
                    field="ignoreregex",
                    message=f"Invalid regex pattern: {exc}",
                )
            )

    # 5. Log path existence (warning only — paths may be created at runtime).
    raw_logpath = settings.get("logpath", "")
    if raw_logpath:
        for log_path in _parse_multiline(raw_logpath):
            # Skip glob patterns and fail2ban variable references.
            if "*" in log_path or "?" in log_path or "%(" in log_path:
                continue
            if not Path(log_path).exists():
                issues.append(
                    JailValidationIssue(
                        field="logpath",
                        message=f"Log file not found on disk: {log_path}",
                    )
                )

    valid = len(issues) == 0
    log.debug(
        "jail_validation_complete",
        jail=name,
        valid=valid,
        issue_count=len(issues),
    )
    return JailValidationResult(jail_name=name, valid=valid, issues=issues)


async def _probe_fail2ban_running(socket_path: str) -> bool:
    """Return ``True`` if the fail2ban socket responds to a ping.

    Args:
        socket_path: Path to the fail2ban Unix domain socket.

    Returns:
        ``True`` when fail2ban is reachable, ``False`` otherwise.
    """
    try:
        client = Fail2BanClient(socket_path=socket_path, timeout=5.0)
        resp = await client.send(["ping"])
        return isinstance(resp, (list, tuple)) and resp[0] == 0
    except Exception:  # noqa: BLE001
        return False


async def _wait_for_fail2ban(
    socket_path: str,
    max_wait_seconds: float = 10.0,
    poll_interval: float = 2.0,
) -> bool:
    """Poll the fail2ban socket until it responds or the timeout expires.

    Args:
        socket_path: Path to the fail2ban Unix domain socket.
        max_wait_seconds: Total time budget in seconds.
        poll_interval: Delay between probe attempts in seconds.

    Returns:
        ``True`` if fail2ban came online within the budget.
    """
    elapsed = 0.0
    while elapsed < max_wait_seconds:
        if await _probe_fail2ban_running(socket_path):
            return True
        await asyncio.sleep(poll_interval)
        elapsed += poll_interval
    return False


async def _start_daemon(start_cmd_parts: list[str]) -> bool:
    """Start the fail2ban daemon using *start_cmd_parts*.

    Uses :func:`asyncio.create_subprocess_exec` (no shell interpretation)
    to avoid command injection.

    Args:
        start_cmd_parts: Command and arguments, e.g.
            ``["fail2ban-client", "start"]``.

    Returns:
        ``True`` when the process exited with code 0.
    """
    if not start_cmd_parts:
        log.warning("fail2ban_start_cmd_empty")
        return False
    try:
        proc = await asyncio.create_subprocess_exec(
            *start_cmd_parts,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        await asyncio.wait_for(proc.wait(), timeout=30.0)
        success = proc.returncode == 0
        if not success:
            log.warning(
                "fail2ban_start_cmd_nonzero",
                cmd=start_cmd_parts,
                returncode=proc.returncode,
            )
        return success
    except (TimeoutError, OSError) as exc:
        log.warning("fail2ban_start_cmd_error", cmd=start_cmd_parts, error=str(exc))
        return False


def _write_local_override_sync(
    config_dir: Path,
    jail_name: str,
    enabled: bool,
    overrides: dict[str, Any],
) -> None:
    """Write a ``jail.d/{name}.local`` file atomically.

    Always writes to ``jail.d/{jail_name}.local``.  If the file already
    exists it is replaced entirely.  The write is atomic: content is
    written to a temp file first, then renamed into place.

    Args:
        config_dir: The fail2ban configuration root directory.
        jail_name: Validated jail name (used as filename stem).
        enabled: Value to write for ``enabled =``.
        overrides: Optional setting overrides (bantime, findtime, maxretry,
                   port, logpath).

    Raises:
        ConfigWriteError: If writing fails.
    """
    jail_d = config_dir / "jail.d"
    try:
        jail_d.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise ConfigWriteError(
            f"Cannot create jail.d directory: {exc}"
        ) from exc

    local_path = jail_d / f"{jail_name}.local"

    lines: list[str] = [
        "# Managed by BanGUI — do not edit manually",
        "",
        f"[{jail_name}]",
        "",
        f"enabled = {'true' if enabled else 'false'}",
        # Provide explicit banaction defaults so fail2ban can resolve the
        # %(banaction)s interpolation used in the built-in action_ chain.
        "banaction = iptables-multiport",
        "banaction_allports = iptables-allports",
    ]

    if overrides.get("bantime") is not None:
        lines.append(f"bantime  = {overrides['bantime']}")
    if overrides.get("findtime") is not None:
        lines.append(f"findtime = {overrides['findtime']}")
    if overrides.get("maxretry") is not None:
        lines.append(f"maxretry = {overrides['maxretry']}")
    if overrides.get("port") is not None:
        lines.append(f"port     = {overrides['port']}")
    if overrides.get("logpath"):
        paths: list[str] = overrides["logpath"]
        if paths:
            lines.append(f"logpath  = {paths[0]}")
            for p in paths[1:]:
                lines.append(f"           {p}")

    content = "\n".join(lines) + "\n"

    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=jail_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        # Clean up temp file if rename failed.
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821 — only reachable when tmp_name is set
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info(
        "jail_local_written",
        jail=jail_name,
        path=str(local_path),
        enabled=enabled,
    )


def _validate_regex_patterns(patterns: list[str]) -> None:
    """Validate each pattern in *patterns* using Python's ``re`` module.

    Args:
        patterns: List of regex strings to validate.

    Raises:
        FilterInvalidRegexError: If any pattern fails to compile.
    """
    for pattern in patterns:
        try:
            re.compile(pattern)
        except re.error as exc:
            raise FilterInvalidRegexError(pattern, str(exc)) from exc


def _write_filter_local_sync(filter_d: Path, name: str, content: str) -> None:
    """Write *content* to ``filter.d/{name}.local`` atomically.

    The write is atomic: content is written to a temp file first, then
    renamed into place.  The ``filter.d/`` directory is created if absent.

    Args:
        filter_d: Path to the ``filter.d`` directory.
        name: Validated filter base name (used as filename stem).
        content: Full serialized filter content to write.

    Raises:
        ConfigWriteError: If writing fails.
    """
    try:
        filter_d.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise ConfigWriteError(
            f"Cannot create filter.d directory: {exc}"
        ) from exc

    local_path = filter_d / f"{name}.local"
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=filter_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info("filter_local_written", filter=name, path=str(local_path))


def _set_jail_local_key_sync(
    config_dir: Path,
    jail_name: str,
    key: str,
    value: str,
) -> None:
    """Update ``jail.d/{jail_name}.local`` to set a single key in the jail section.

    If the ``.local`` file already exists it is read, the key is updated (or
    added), and the file is written back atomically without disturbing other
    settings.  If the file does not exist a new one is created containing
    only the BanGUI header comment, the jail section, and the requested key.

    Args:
        config_dir: The fail2ban configuration root directory.
        jail_name: Validated jail name (used as section name and filename stem).
        key: Config key to set inside the jail section.
        value: Config value to assign.

    Raises:
        ConfigWriteError: If writing fails.
    """
    jail_d = config_dir / "jail.d"
    try:
        jail_d.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise ConfigWriteError(
            f"Cannot create jail.d directory: {exc}"
        ) from exc

    local_path = jail_d / f"{jail_name}.local"

    parser = _build_parser()
    if local_path.is_file():
        try:
            parser.read(str(local_path), encoding="utf-8")
        except (configparser.Error, OSError) as exc:
            log.warning(
                "jail_local_read_for_update_error",
                jail=jail_name,
                error=str(exc),
            )

    if not parser.has_section(jail_name):
        parser.add_section(jail_name)
    parser.set(jail_name, key, value)

    # Serialize: write a BanGUI header then the parser output.
    buf = io.StringIO()
    buf.write("# Managed by BanGUI — do not edit manually\n\n")
    parser.write(buf)
    content = buf.getvalue()

    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=jail_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info(
        "jail_local_key_set",
        jail=jail_name,
        key=key,
        path=str(local_path),
    )


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------


async def list_inactive_jails(
    config_dir: str,
    socket_path: str,
) -> InactiveJailListResponse:
    """Return all jails defined in config files that are not currently active.

    Parses ``jail.conf``, ``jail.local``, and ``jail.d/`` following the
    fail2ban merge order.  A jail is considered inactive when:

    - Its merged ``enabled`` value is ``false`` (or absent, which defaults to
      ``false`` in fail2ban), **or**
    - Its ``enabled`` value is ``true`` in config but fail2ban does not report
      it as running.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.

    Returns:
        :class:`~app.models.config.InactiveJailListResponse` with all
        inactive jails.
    """
    loop = asyncio.get_event_loop()
    parsed_result: tuple[dict[str, dict[str, str]], dict[str, str]] = (
        await loop.run_in_executor(None, _parse_jails_sync, Path(config_dir))
    )
    all_jails, source_files = parsed_result
    active_names: set[str] = await _get_active_jail_names(socket_path)

    inactive: list[InactiveJail] = []
    for jail_name, settings in sorted(all_jails.items()):
        if jail_name in active_names:
            # fail2ban reports this jail as running — skip it.
            continue

        source = source_files.get(jail_name, config_dir)
        inactive.append(_build_inactive_jail(jail_name, settings, source))

    log.info(
        "inactive_jails_listed",
        total_defined=len(all_jails),
        active=len(active_names),
        inactive=len(inactive),
    )
    return InactiveJailListResponse(jails=inactive, total=len(inactive))


async def activate_jail(
    config_dir: str,
    socket_path: str,
    name: str,
    req: ActivateJailRequest,
) -> JailActivationResponse:
    """Enable an inactive jail and reload fail2ban.

    Performs pre-activation validation, writes ``enabled = true`` (plus any
    override values from *req*) to ``jail.d/{name}.local``, and triggers a
    full fail2ban reload.  After the reload a multi-attempt health probe
    determines whether fail2ban (and the specific jail) are still running.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Name of the jail to activate.  Must exist in the parsed config.
        req: Optional override values to write alongside ``enabled = true``.

    Returns:
        :class:`~app.models.config.JailActivationResponse` including
        ``fail2ban_running`` and ``validation_warnings`` fields.

    Raises:
        JailNameError: If *name* contains invalid characters.
        JailNotFoundInConfigError: If *name* is not defined in any config file.
        JailAlreadyActiveError: If fail2ban already reports *name* as running.
        ConfigWriteError: If writing the ``.local`` file fails.
        ~app.utils.fail2ban_client.Fail2BanConnectionError: If the fail2ban
            socket is unreachable during reload.
    """
    _safe_jail_name(name)

    loop = asyncio.get_event_loop()
    all_jails, _source_files = await loop.run_in_executor(
        None, _parse_jails_sync, Path(config_dir)
    )

    if name not in all_jails:
        raise JailNotFoundInConfigError(name)

    active_names = await _get_active_jail_names(socket_path)
    if name in active_names:
        raise JailAlreadyActiveError(name)

    # ---------------------------------------------------------------------- #
    # Pre-activation validation — collect warnings but do not block           #
    # ---------------------------------------------------------------------- #
    validation_result: JailValidationResult = await loop.run_in_executor(
        None, _validate_jail_config_sync, Path(config_dir), name
    )
    warnings: list[str] = [f"{i.field}: {i.message}" for i in validation_result.issues]
    if warnings:
        log.warning(
            "jail_activation_validation_warnings",
            jail=name,
            warnings=warnings,
        )

    overrides: dict[str, Any] = {
        "bantime": req.bantime,
        "findtime": req.findtime,
        "maxretry": req.maxretry,
        "port": req.port,
        "logpath": req.logpath,
    }

    await loop.run_in_executor(
        None,
        _write_local_override_sync,
        Path(config_dir),
        name,
        True,
        overrides,
    )

    try:
        await jail_service.reload_all(socket_path, include_jails=[name])
    except Exception as exc:  # noqa: BLE001
        log.warning("reload_after_activate_failed", jail=name, error=str(exc))

    # ---------------------------------------------------------------------- #
    # Post-reload health probe with retries                                   #
    # ---------------------------------------------------------------------- #
    fail2ban_running = False
    for attempt in range(_POST_RELOAD_MAX_ATTEMPTS):
        if attempt > 0:
            await asyncio.sleep(_POST_RELOAD_PROBE_INTERVAL)
        if await _probe_fail2ban_running(socket_path):
            fail2ban_running = True
            break

    if not fail2ban_running:
        log.warning(
            "fail2ban_down_after_activate",
            jail=name,
            message="fail2ban socket unreachable after reload — daemon may have crashed.",
        )
        return JailActivationResponse(
            name=name,
            active=False,
            fail2ban_running=False,
            validation_warnings=warnings,
            message=(
                f"Jail {name!r} was written to config but fail2ban stopped "
                "responding after reload.  The jail configuration may be invalid."
            ),
        )

    # Verify the jail actually started (config error may prevent it silently).
    post_reload_names = await _get_active_jail_names(socket_path)
    actually_running = name in post_reload_names
    if not actually_running:
        log.warning(
            "jail_activation_unverified",
            jail=name,
            message="Jail did not appear in running jails after reload.",
        )
        return JailActivationResponse(
            name=name,
            active=False,
            fail2ban_running=True,
            validation_warnings=warnings,
            message=(
                f"Jail {name!r} was written to config but did not start after "
                "reload — check the jail configuration (filters, log paths, regex)."
            ),
        )

    log.info("jail_activated", jail=name)
    return JailActivationResponse(
        name=name,
        active=True,
        fail2ban_running=True,
        validation_warnings=warnings,
        message=f"Jail {name!r} activated successfully.",
    )


async def deactivate_jail(
    config_dir: str,
    socket_path: str,
    name: str,
) -> JailActivationResponse:
    """Disable an active jail and reload fail2ban.

    Writes ``enabled = false`` to ``jail.d/{name}.local`` and triggers a
    full fail2ban reload so the jail stops immediately.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Name of the jail to deactivate.  Must exist in the parsed config.

    Returns:
        :class:`~app.models.config.JailActivationResponse`.

    Raises:
        JailNameError: If *name* contains invalid characters.
        JailNotFoundInConfigError: If *name* is not defined in any config file.
        JailAlreadyInactiveError: If fail2ban already reports *name* as not
            running.
        ConfigWriteError: If writing the ``.local`` file fails.
        ~app.utils.fail2ban_client.Fail2BanConnectionError: If the fail2ban
            socket is unreachable during reload.
    """
    _safe_jail_name(name)

    loop = asyncio.get_event_loop()
    all_jails, _source_files = await loop.run_in_executor(
        None, _parse_jails_sync, Path(config_dir)
    )

    if name not in all_jails:
        raise JailNotFoundInConfigError(name)

    active_names = await _get_active_jail_names(socket_path)
    if name not in active_names:
        raise JailAlreadyInactiveError(name)

    await loop.run_in_executor(
        None,
        _write_local_override_sync,
        Path(config_dir),
        name,
        False,
        {},
    )

    try:
        await jail_service.reload_all(socket_path, exclude_jails=[name])
    except Exception as exc:  # noqa: BLE001
        log.warning("reload_after_deactivate_failed", jail=name, error=str(exc))

    log.info("jail_deactivated", jail=name)
    return JailActivationResponse(
        name=name,
        active=False,
        message=f"Jail {name!r} deactivated successfully.",
    )


async def validate_jail_config(
    config_dir: str,
    name: str,
) -> JailValidationResult:
    """Run pre-activation validation checks on a jail configuration.

    Validates that referenced filter and action files exist in ``filter.d/``
    and ``action.d/``, that all regex patterns compile, and that declared log
    paths exist on disk.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        name: Name of the jail to validate.

    Returns:
        :class:`~app.models.config.JailValidationResult` with any issues found.

    Raises:
        JailNameError: If *name* contains invalid characters.
    """
    _safe_jail_name(name)
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        _validate_jail_config_sync,
        Path(config_dir),
        name,
    )


async def rollback_jail(
    config_dir: str,
    socket_path: str,
    name: str,
    start_cmd_parts: list[str],
) -> RollbackResponse:
    """Disable a bad jail config and restart the fail2ban daemon.

    Writes ``enabled = false`` to ``jail.d/{name}.local`` (works even when
    fail2ban is down — only a file write), then attempts to start the daemon
    with *start_cmd_parts*.  Waits up to 10 seconds for the socket to respond.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Name of the jail to disable.
        start_cmd_parts: Argument list for the daemon start command, e.g.
            ``["fail2ban-client", "start"]``.

    Returns:
        :class:`~app.models.config.RollbackResponse`.

    Raises:
        JailNameError: If *name* contains invalid characters.
        ConfigWriteError: If writing the ``.local`` file fails.
    """
    _safe_jail_name(name)

    loop = asyncio.get_event_loop()

    # Write enabled=false — this must succeed even when fail2ban is down.
    await loop.run_in_executor(
        None,
        _write_local_override_sync,
        Path(config_dir),
        name,
        False,
        {},
    )
    log.info("jail_rolled_back_disabled", jail=name)

    # Attempt to start the daemon.
    started = await _start_daemon(start_cmd_parts)
    log.info("jail_rollback_start_attempted", jail=name, start_ok=started)

    # Wait for the socket to come back.
    fail2ban_running = await _wait_for_fail2ban(
        socket_path, max_wait_seconds=10.0, poll_interval=2.0
    )

    active_jails = 0
    if fail2ban_running:
        names = await _get_active_jail_names(socket_path)
        active_jails = len(names)

    if fail2ban_running:
        log.info("jail_rollback_success", jail=name, active_jails=active_jails)
        return RollbackResponse(
            jail_name=name,
            disabled=True,
            fail2ban_running=True,
            active_jails=active_jails,
            message=(
                f"Jail {name!r} disabled and fail2ban restarted successfully "
                f"with {active_jails} active jail(s)."
            ),
        )

    log.warning("jail_rollback_fail2ban_still_down", jail=name)
    return RollbackResponse(
        jail_name=name,
        disabled=True,
        fail2ban_running=False,
        active_jails=0,
        message=(
            f"Jail {name!r} was disabled but fail2ban did not come back online. "
            "Check the fail2ban log for additional errors."
        ),
    )


# ---------------------------------------------------------------------------
# Filter discovery helpers (Task 2.1)
# ---------------------------------------------------------------------------

# Allowlist pattern for filter names used in path construction.
_SAFE_FILTER_NAME_RE: re.Pattern[str] = re.compile(
    r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)


class FilterNotFoundError(Exception):
    """Raised when the requested filter name is not found in ``filter.d/``."""

    def __init__(self, name: str) -> None:
        """Initialise with the filter name that was not found.

        Args:
            name: The filter name that could not be located.
        """
        self.name: str = name
        super().__init__(f"Filter not found: {name!r}")


def _extract_filter_base_name(filter_raw: str) -> str:
    """Extract the base filter name from a raw fail2ban filter string.

    fail2ban jail configs may specify a filter with an optional mode suffix,
    e.g. ``sshd``, ``sshd[mode=aggressive]``, or
    ``%(__name__)s[mode=%(mode)s]``.  This function strips the ``[…]`` mode
    block and any leading/trailing whitespace to return just the file-system
    base name used to look up ``filter.d/{name}.conf``.

    Args:
        filter_raw: Raw ``filter`` value from a jail config (already
                    with ``%(__name__)s`` substituted by the caller).

    Returns:
        Base filter name, e.g. ``"sshd"``.
    """
    bracket = filter_raw.find("[")
    if bracket != -1:
        return filter_raw[:bracket].strip()
    return filter_raw.strip()


def _build_filter_to_jails_map(
    all_jails: dict[str, dict[str, str]],
    active_names: set[str],
) -> dict[str, list[str]]:
    """Return a mapping of filter base name → list of active jail names.

    Iterates over every jail whose name is in *active_names*, resolves its
    ``filter`` config key, and records the jail against the base filter name.

    Args:
        all_jails: Merged jail config dict — ``{jail_name: {key: value}}``.
        active_names: Set of jail names currently running in fail2ban.

    Returns:
        ``{filter_base_name: [jail_name, …]}``.
    """
    mapping: dict[str, list[str]] = {}
    for jail_name, settings in all_jails.items():
        if jail_name not in active_names:
            continue
        raw_filter = settings.get("filter", "")
        mode = settings.get("mode", "normal")
        resolved = _resolve_filter(raw_filter, jail_name, mode) if raw_filter else jail_name
        base = _extract_filter_base_name(resolved)
        if base:
            mapping.setdefault(base, []).append(jail_name)
    return mapping


def _parse_filters_sync(
    filter_d: Path,
) -> list[tuple[str, str, str, bool, str]]:
    """Synchronously scan ``filter.d/`` and return per-filter tuples.

    Each tuple contains:

    - ``name``          — filter base name (``"sshd"``).
    - ``filename``      — actual filename (``"sshd.conf"`` or ``"sshd.local"``).
    - ``content``       — merged file content (``conf`` overridden by ``local``).
    - ``has_local``     — whether a ``.local`` override exists alongside a ``.conf``.
    - ``source_path``   — absolute path to the primary (``conf``) source file, or
                          to the ``.local`` file for user-created (local-only) filters.

    Also discovers ``.local``-only files (user-created filters with no
    corresponding ``.conf``).  These are returned with ``has_local = False``
    and ``source_path`` pointing to the ``.local`` file itself.

    Args:
        filter_d: Path to the ``filter.d`` directory.

    Returns:
        List of ``(name, filename, content, has_local, source_path)`` tuples,
        sorted by name.
    """
    if not filter_d.is_dir():
        log.warning("filter_d_not_found", path=str(filter_d))
        return []

    conf_names: set[str] = set()
    results: list[tuple[str, str, str, bool, str]] = []

    # ---- .conf-based filters (with optional .local override) ----------------
    for conf_path in sorted(filter_d.glob("*.conf")):
        if not conf_path.is_file():
            continue
        name = conf_path.stem
        filename = conf_path.name
        conf_names.add(name)
        local_path = conf_path.with_suffix(".local")
        has_local = local_path.is_file()

        try:
            content = conf_path.read_text(encoding="utf-8")
        except OSError as exc:
            log.warning(
                "filter_read_error", name=name, path=str(conf_path), error=str(exc)
            )
            continue

        if has_local:
            try:
                local_content = local_path.read_text(encoding="utf-8")
                # Append local content after conf so configparser reads local
                # values last (higher priority).
                content = content + "\n" + local_content
            except OSError as exc:
                log.warning(
                    "filter_local_read_error",
                    name=name,
                    path=str(local_path),
                    error=str(exc),
                )

        results.append((name, filename, content, has_local, str(conf_path)))

    # ---- .local-only filters (user-created, no corresponding .conf) ----------
    for local_path in sorted(filter_d.glob("*.local")):
        if not local_path.is_file():
            continue
        name = local_path.stem
        if name in conf_names:
            # Already covered above as a .conf filter with a .local override.
            continue
        try:
            content = local_path.read_text(encoding="utf-8")
        except OSError as exc:
            log.warning(
                "filter_local_read_error",
                name=name,
                path=str(local_path),
                error=str(exc),
            )
            continue
        results.append((name, local_path.name, content, False, str(local_path)))

    results.sort(key=lambda t: t[0])
    log.debug("filters_scanned", count=len(results), filter_d=str(filter_d))
    return results


# ---------------------------------------------------------------------------
# Public API — filter discovery (Task 2.1)
# ---------------------------------------------------------------------------


async def list_filters(
    config_dir: str,
    socket_path: str,
) -> FilterListResponse:
    """Return all available filters from ``filter.d/`` with active/inactive status.

    Scans ``{config_dir}/filter.d/`` for ``.conf`` files, merges any
    corresponding ``.local`` overrides, parses each file into a
    :class:`~app.models.config.FilterConfig`, and cross-references with the
    currently running jails to determine which filters are active.

    A filter is considered *active* when its base name matches the ``filter``
    field of at least one currently running jail.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.

    Returns:
        :class:`~app.models.config.FilterListResponse` with all filters
        sorted alphabetically, active ones carrying non-empty
        ``used_by_jails`` lists.
    """
    filter_d = Path(config_dir) / "filter.d"
    loop = asyncio.get_event_loop()

    # Run the synchronous scan in a thread-pool executor.
    raw_filters: list[tuple[str, str, str, bool, str]] = await loop.run_in_executor(
        None, _parse_filters_sync, filter_d
    )

    # Fetch active jail names and their configs concurrently.
    all_jails_result, active_names = await asyncio.gather(
        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
        _get_active_jail_names(socket_path),
    )
    all_jails, _source_files = all_jails_result

    filter_to_jails = _build_filter_to_jails_map(all_jails, active_names)

    filters: list[FilterConfig] = []
    for name, filename, content, has_local, source_path in raw_filters:
        cfg = conffile_parser.parse_filter_file(
            content, name=name, filename=filename
        )
        used_by = sorted(filter_to_jails.get(name, []))
        filters.append(
            FilterConfig(
                name=cfg.name,
                filename=cfg.filename,
                before=cfg.before,
                after=cfg.after,
                variables=cfg.variables,
                prefregex=cfg.prefregex,
                failregex=cfg.failregex,
                ignoreregex=cfg.ignoreregex,
                maxlines=cfg.maxlines,
                datepattern=cfg.datepattern,
                journalmatch=cfg.journalmatch,
                active=len(used_by) > 0,
                used_by_jails=used_by,
                source_file=source_path,
                has_local_override=has_local,
            )
        )

    log.info("filters_listed", total=len(filters), active=sum(1 for f in filters if f.active))
    return FilterListResponse(filters=filters, total=len(filters))


async def get_filter(
    config_dir: str,
    socket_path: str,
    name: str,
) -> FilterConfig:
    """Return a single filter from ``filter.d/`` with active/inactive status.

    Reads ``{config_dir}/filter.d/{name}.conf``, merges any ``.local``
    override, and enriches the parsed :class:`~app.models.config.FilterConfig`
    with ``active``, ``used_by_jails``, ``source_file``, and
    ``has_local_override``.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Filter base name (e.g. ``"sshd"`` or ``"sshd.conf"``).

    Returns:
        :class:`~app.models.config.FilterConfig` with status fields populated.

    Raises:
        FilterNotFoundError: If no ``{name}.conf`` or ``{name}.local`` file
            exists in ``filter.d/``.
    """
    # Normalise — strip extension if provided (.conf=5 chars, .local=6 chars).
    if name.endswith(".conf"):
        base_name = name[:-5]
    elif name.endswith(".local"):
        base_name = name[:-6]
    else:
        base_name = name

    filter_d = Path(config_dir) / "filter.d"
    conf_path = filter_d / f"{base_name}.conf"
    local_path = filter_d / f"{base_name}.local"
    loop = asyncio.get_event_loop()

    def _read() -> tuple[str, bool, str]:
        """Read filter content and return (content, has_local_override, source_path)."""
        has_local = local_path.is_file()
        if conf_path.is_file():
            content = conf_path.read_text(encoding="utf-8")
            if has_local:
                try:
                    content += "\n" + local_path.read_text(encoding="utf-8")
                except OSError as exc:
                    log.warning(
                        "filter_local_read_error",
                        name=base_name,
                        path=str(local_path),
                        error=str(exc),
                    )
            return content, has_local, str(conf_path)
        elif has_local:
            # Local-only filter: created by the user, no shipped .conf base.
            content = local_path.read_text(encoding="utf-8")
            return content, False, str(local_path)
        else:
            raise FilterNotFoundError(base_name)

    content, has_local, source_path = await loop.run_in_executor(None, _read)

    cfg = conffile_parser.parse_filter_file(
        content, name=base_name, filename=f"{base_name}.conf"
    )

    all_jails_result, active_names = await asyncio.gather(
        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
        _get_active_jail_names(socket_path),
    )
    all_jails, _source_files = all_jails_result
    filter_to_jails = _build_filter_to_jails_map(all_jails, active_names)

    used_by = sorted(filter_to_jails.get(base_name, []))
    log.info("filter_fetched", name=base_name, active=len(used_by) > 0)
    return FilterConfig(
        name=cfg.name,
        filename=cfg.filename,
        before=cfg.before,
        after=cfg.after,
        variables=cfg.variables,
        prefregex=cfg.prefregex,
        failregex=cfg.failregex,
        ignoreregex=cfg.ignoreregex,
        maxlines=cfg.maxlines,
        datepattern=cfg.datepattern,
        journalmatch=cfg.journalmatch,
        active=len(used_by) > 0,
        used_by_jails=used_by,
        source_file=source_path,
        has_local_override=has_local,
    )


# ---------------------------------------------------------------------------
# Public API — filter write operations (Task 2.2)
# ---------------------------------------------------------------------------


async def update_filter(
    config_dir: str,
    socket_path: str,
    name: str,
    req: FilterUpdateRequest,
    do_reload: bool = False,
) -> FilterConfig:
    """Update a filter's ``.local`` override with new regex/pattern values.

    Reads the current merged configuration for *name* (``conf`` + any existing
    ``local``), applies the non-``None`` fields in *req* on top of it, and
    writes the resulting definition to ``filter.d/{name}.local``.  The
    original ``.conf`` file is never modified.

    All regex patterns in *req* are validated with Python's ``re`` module
    before any write occurs.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Filter base name (e.g. ``"sshd"`` or ``"sshd.conf"``).
        req: Partial update — only non-``None`` fields are applied.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Returns:
        :class:`~app.models.config.FilterConfig` reflecting the updated state.

    Raises:
        FilterNameError: If *name* contains invalid characters.
        FilterNotFoundError: If no ``{name}.conf`` or ``{name}.local`` exists.
        FilterInvalidRegexError: If any supplied regex pattern is invalid.
        ConfigWriteError: If writing the ``.local`` file fails.
    """
    base_name = name[:-5] if name.endswith(".conf") or name.endswith(".local") else name
    _safe_filter_name(base_name)

    # Validate regex patterns before touching the filesystem.
    patterns: list[str] = []
    if req.failregex is not None:
        patterns.extend(req.failregex)
    if req.ignoreregex is not None:
        patterns.extend(req.ignoreregex)
    _validate_regex_patterns(patterns)

    # Fetch the current merged config (raises FilterNotFoundError if absent).
    current = await get_filter(config_dir, socket_path, base_name)

    # Build a FilterConfigUpdate from the request fields.
    update = FilterConfigUpdate(
        failregex=req.failregex,
        ignoreregex=req.ignoreregex,
        datepattern=req.datepattern,
        journalmatch=req.journalmatch,
    )

    merged = conffile_parser.merge_filter_update(current, update)
    content = conffile_parser.serialize_filter_config(merged)

    filter_d = Path(config_dir) / "filter.d"
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _write_filter_local_sync, filter_d, base_name, content)

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_filter_update_failed",
                filter=base_name,
                error=str(exc),
            )

    log.info("filter_updated", filter=base_name, reload=do_reload)
    return await get_filter(config_dir, socket_path, base_name)


async def create_filter(
    config_dir: str,
    socket_path: str,
    req: FilterCreateRequest,
    do_reload: bool = False,
) -> FilterConfig:
    """Create a brand-new user-defined filter in ``filter.d/{name}.local``.

    No ``.conf`` is written; fail2ban loads ``.local`` files directly.  If a
    ``.conf`` or ``.local`` file already exists for the requested name, a
    :class:`FilterAlreadyExistsError` is raised.

    All regex patterns are validated with Python's ``re`` module before
    writing.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        req: Filter name and definition fields.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Returns:
        :class:`~app.models.config.FilterConfig` for the newly created filter.

    Raises:
        FilterNameError: If ``req.name`` contains invalid characters.
        FilterAlreadyExistsError: If a ``.conf`` or ``.local`` already exists.
        FilterInvalidRegexError: If any regex pattern is invalid.
        ConfigWriteError: If writing fails.
    """
    _safe_filter_name(req.name)

    filter_d = Path(config_dir) / "filter.d"
    conf_path = filter_d / f"{req.name}.conf"
    local_path = filter_d / f"{req.name}.local"

    def _check_not_exists() -> None:
        if conf_path.is_file() or local_path.is_file():
            raise FilterAlreadyExistsError(req.name)

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _check_not_exists)

    # Validate regex patterns.
    patterns: list[str] = list(req.failregex) + list(req.ignoreregex)
    _validate_regex_patterns(patterns)

    # Build a FilterConfig and serialise it.
    cfg = FilterConfig(
        name=req.name,
        filename=f"{req.name}.local",
        failregex=req.failregex,
        ignoreregex=req.ignoreregex,
        prefregex=req.prefregex,
        datepattern=req.datepattern,
        journalmatch=req.journalmatch,
    )
    content = conffile_parser.serialize_filter_config(cfg)

    await loop.run_in_executor(None, _write_filter_local_sync, filter_d, req.name, content)

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_filter_create_failed",
                filter=req.name,
                error=str(exc),
            )

    log.info("filter_created", filter=req.name, reload=do_reload)
    # Re-fetch to get the canonical FilterConfig (source_file, active, etc.).
    return await get_filter(config_dir, socket_path, req.name)


async def delete_filter(
    config_dir: str,
    name: str,
) -> None:
    """Delete a user-created filter's ``.local`` file.

    Deletion rules:
    - If only a ``.conf`` file exists (shipped default, no user override) →
      :class:`FilterReadonlyError`.
    - If a ``.local`` file exists (whether or not a ``.conf`` also exists) →
      the ``.local`` file is deleted.  The shipped ``.conf`` is never touched.
    - If neither file exists → :class:`FilterNotFoundError`.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        name: Filter base name (e.g. ``"sshd"``).

    Raises:
        FilterNameError: If *name* contains invalid characters.
        FilterNotFoundError: If no filter file is found for *name*.
        FilterReadonlyError: If only a shipped ``.conf`` exists (no ``.local``).
        ConfigWriteError: If deletion of the ``.local`` file fails.
    """
    base_name = name[:-5] if name.endswith(".conf") or name.endswith(".local") else name
    _safe_filter_name(base_name)

    filter_d = Path(config_dir) / "filter.d"
    conf_path = filter_d / f"{base_name}.conf"
    local_path = filter_d / f"{base_name}.local"

    loop = asyncio.get_event_loop()

    def _delete() -> None:
        has_conf = conf_path.is_file()
        has_local = local_path.is_file()

        if not has_conf and not has_local:
            raise FilterNotFoundError(base_name)

        if has_conf and not has_local:
            # Shipped default — nothing user-writable to remove.
            raise FilterReadonlyError(base_name)

        try:
            local_path.unlink()
        except OSError as exc:
            raise ConfigWriteError(
                f"Failed to delete {local_path}: {exc}"
            ) from exc

        log.info("filter_local_deleted", filter=base_name, path=str(local_path))

    await loop.run_in_executor(None, _delete)


async def assign_filter_to_jail(
    config_dir: str,
    socket_path: str,
    jail_name: str,
    req: AssignFilterRequest,
    do_reload: bool = False,
) -> None:
    """Assign a filter to a jail by updating the jail's ``.local`` file.

    Writes ``filter = {req.filter_name}`` into the ``[{jail_name}]`` section
    of ``jail.d/{jail_name}.local``.  If the ``.local`` file already contains
    other settings for this jail they are preserved.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        jail_name: Name of the jail to update.
        req: Request containing the filter name to assign.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Raises:
        JailNameError: If *jail_name* contains invalid characters.
        FilterNameError: If ``req.filter_name`` contains invalid characters.
        JailNotFoundInConfigError: If *jail_name* is not defined in any config
            file.
        FilterNotFoundError: If ``req.filter_name`` does not exist in
            ``filter.d/``.
        ConfigWriteError: If writing fails.
    """
    _safe_jail_name(jail_name)
    _safe_filter_name(req.filter_name)

    loop = asyncio.get_event_loop()

    # Verify the jail exists in config.
    all_jails, _src = await loop.run_in_executor(
        None, _parse_jails_sync, Path(config_dir)
    )
    if jail_name not in all_jails:
        raise JailNotFoundInConfigError(jail_name)

    # Verify the filter exists (conf or local).
    filter_d = Path(config_dir) / "filter.d"

    def _check_filter() -> None:
        conf_exists = (filter_d / f"{req.filter_name}.conf").is_file()
        local_exists = (filter_d / f"{req.filter_name}.local").is_file()
        if not conf_exists and not local_exists:
            raise FilterNotFoundError(req.filter_name)

    await loop.run_in_executor(None, _check_filter)

    await loop.run_in_executor(
        None,
        _set_jail_local_key_sync,
        Path(config_dir),
        jail_name,
        "filter",
        req.filter_name,
    )

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_assign_filter_failed",
                jail=jail_name,
                filter=req.filter_name,
                error=str(exc),
            )

    log.info(
        "filter_assigned_to_jail",
        jail=jail_name,
        filter=req.filter_name,
        reload=do_reload,
    )


# ---------------------------------------------------------------------------
# Action discovery helpers (Task 3.1)
# ---------------------------------------------------------------------------

# Allowlist pattern for action names used in path construction.
_SAFE_ACTION_NAME_RE: re.Pattern[str] = re.compile(
    r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)


class ActionNotFoundError(Exception):
    """Raised when the requested action name is not found in ``action.d/``."""

    def __init__(self, name: str) -> None:
        """Initialise with the action name that was not found.

        Args:
            name: The action name that could not be located.
        """
        self.name: str = name
        super().__init__(f"Action not found: {name!r}")


class ActionAlreadyExistsError(Exception):
    """Raised when trying to create an action whose ``.conf`` or ``.local`` already exists."""

    def __init__(self, name: str) -> None:
        """Initialise with the action name that already exists.

        Args:
            name: The action name that already exists.
        """
        self.name: str = name
        super().__init__(f"Action already exists: {name!r}")


class ActionReadonlyError(Exception):
    """Raised when trying to delete a shipped ``.conf`` action with no ``.local`` override."""

    def __init__(self, name: str) -> None:
        """Initialise with the action name that cannot be deleted.

        Args:
            name: The action name that is read-only (shipped ``.conf`` only).
        """
        self.name: str = name
        super().__init__(
            f"Action {name!r} is a shipped default (.conf only); "
            "only user-created .local files can be deleted."
        )


class ActionNameError(Exception):
    """Raised when an action name contains invalid characters."""


def _safe_action_name(name: str) -> str:
    """Validate *name* and return it unchanged or raise :class:`ActionNameError`.

    Args:
        name: Proposed action name (without extension).

    Returns:
        The name unchanged if valid.

    Raises:
        ActionNameError: If *name* contains unsafe characters.
    """
    if not _SAFE_ACTION_NAME_RE.match(name):
        raise ActionNameError(
            f"Action name {name!r} contains invalid characters. "
            "Only alphanumeric characters, hyphens, underscores, and dots are "
            "allowed; must start with an alphanumeric character."
        )
    return name


def _build_action_to_jails_map(
    all_jails: dict[str, dict[str, str]],
    active_names: set[str],
) -> dict[str, list[str]]:
    """Return a mapping of action base name → list of active jail names.

    Iterates over every jail whose name is in *active_names*, resolves each
    entry in its ``action`` config key to an action base name (stripping
    ``[…]`` parameter blocks), and records the jail against each base name.

    Args:
        all_jails: Merged jail config dict — ``{jail_name: {key: value}}``.
        active_names: Set of jail names currently running in fail2ban.

    Returns:
        ``{action_base_name: [jail_name, …]}``.
    """
    mapping: dict[str, list[str]] = {}
    for jail_name, settings in all_jails.items():
        if jail_name not in active_names:
            continue
        raw_action = settings.get("action", "")
        if not raw_action:
            continue
        for line in raw_action.splitlines():
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            # Strip optional [key=value] parameter block to get the base name.
            bracket = stripped.find("[")
            base = stripped[:bracket].strip() if bracket != -1 else stripped
            if base:
                mapping.setdefault(base, []).append(jail_name)
    return mapping


def _parse_actions_sync(
    action_d: Path,
) -> list[tuple[str, str, str, bool, str]]:
    """Synchronously scan ``action.d/`` and return per-action tuples.

    Each tuple contains:

    - ``name``        — action base name (``"iptables"``).
    - ``filename``    — actual filename (``"iptables.conf"``).
    - ``content``     — merged file content (``conf`` overridden by ``local``).
    - ``has_local``   — whether a ``.local`` override exists alongside a ``.conf``.
    - ``source_path`` — absolute path to the primary (``conf``) source file, or
                        to the ``.local`` file for user-created (local-only) actions.

    Also discovers ``.local``-only files (user-created actions with no
    corresponding ``.conf``).

    Args:
        action_d: Path to the ``action.d`` directory.

    Returns:
        List of ``(name, filename, content, has_local, source_path)`` tuples,
        sorted by name.
    """
    if not action_d.is_dir():
        log.warning("action_d_not_found", path=str(action_d))
        return []

    conf_names: set[str] = set()
    results: list[tuple[str, str, str, bool, str]] = []

    # ---- .conf-based actions (with optional .local override) ----------------
    for conf_path in sorted(action_d.glob("*.conf")):
        if not conf_path.is_file():
            continue
        name = conf_path.stem
        filename = conf_path.name
        conf_names.add(name)
        local_path = conf_path.with_suffix(".local")
        has_local = local_path.is_file()

        try:
            content = conf_path.read_text(encoding="utf-8")
        except OSError as exc:
            log.warning(
                "action_read_error", name=name, path=str(conf_path), error=str(exc)
            )
            continue

        if has_local:
            try:
                local_content = local_path.read_text(encoding="utf-8")
                content = content + "\n" + local_content
            except OSError as exc:
                log.warning(
                    "action_local_read_error",
                    name=name,
                    path=str(local_path),
                    error=str(exc),
                )

        results.append((name, filename, content, has_local, str(conf_path)))

    # ---- .local-only actions (user-created, no corresponding .conf) ----------
    for local_path in sorted(action_d.glob("*.local")):
        if not local_path.is_file():
            continue
        name = local_path.stem
        if name in conf_names:
            continue
        try:
            content = local_path.read_text(encoding="utf-8")
        except OSError as exc:
            log.warning(
                "action_local_read_error",
                name=name,
                path=str(local_path),
                error=str(exc),
            )
            continue
        results.append((name, local_path.name, content, False, str(local_path)))

    results.sort(key=lambda t: t[0])
    log.debug("actions_scanned", count=len(results), action_d=str(action_d))
    return results


def _append_jail_action_sync(
    config_dir: Path,
    jail_name: str,
    action_entry: str,
) -> None:
    """Append an action entry to the ``action`` key in ``jail.d/{jail_name}.local``.

    If the ``.local`` file already contains an ``action`` key under the jail
    section, the new entry is appended as an additional line (multi-line
    configparser format) unless it is already present.  If no ``action`` key
    exists, one is created.

    Args:
        config_dir: The fail2ban configuration root directory.
        jail_name: Validated jail name.
        action_entry: Full action string including any ``[…]`` parameters.

    Raises:
        ConfigWriteError: If writing fails.
    """
    jail_d = config_dir / "jail.d"
    try:
        jail_d.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise ConfigWriteError(
            f"Cannot create jail.d directory: {exc}"
        ) from exc

    local_path = jail_d / f"{jail_name}.local"

    parser = _build_parser()
    if local_path.is_file():
        try:
            parser.read(str(local_path), encoding="utf-8")
        except (configparser.Error, OSError) as exc:
            log.warning(
                "jail_local_read_for_update_error",
                jail=jail_name,
                error=str(exc),
            )

    if not parser.has_section(jail_name):
        parser.add_section(jail_name)

    existing_raw = parser.get(jail_name, "action") if parser.has_option(jail_name, "action") else ""
    existing_lines = [
        line.strip()
        for line in existing_raw.splitlines()
        if line.strip() and not line.strip().startswith("#")
    ]

    # Extract base names from existing entries for duplicate checking.
    def _base(entry: str) -> str:
        bracket = entry.find("[")
        return entry[:bracket].strip() if bracket != -1 else entry.strip()

    new_base = _base(action_entry)
    if not any(_base(e) == new_base for e in existing_lines):
        existing_lines.append(action_entry)

    if existing_lines:
        # configparser multi-line: continuation lines start with whitespace.
        new_value = existing_lines[0] + "".join(
            f"\n           {line}" for line in existing_lines[1:]
        )
        parser.set(jail_name, "action", new_value)
    else:
        parser.set(jail_name, "action", action_entry)

    buf = io.StringIO()
    buf.write("# Managed by BanGUI — do not edit manually\n\n")
    parser.write(buf)
    content = buf.getvalue()

    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=jail_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info(
        "jail_action_appended",
        jail=jail_name,
        action=action_entry,
        path=str(local_path),
    )


def _remove_jail_action_sync(
    config_dir: Path,
    jail_name: str,
    action_name: str,
) -> None:
    """Remove an action entry from the ``action`` key in ``jail.d/{jail_name}.local``.

    Reads the ``.local`` file, removes any ``action`` entries whose base name
    matches *action_name*, and writes the result back atomically.  If no
    ``.local`` file exists, this is a no-op.

    Args:
        config_dir: The fail2ban configuration root directory.
        jail_name: Validated jail name.
        action_name: Base name of the action to remove (without ``[…]``).

    Raises:
        ConfigWriteError: If writing fails.
    """
    jail_d = config_dir / "jail.d"
    local_path = jail_d / f"{jail_name}.local"

    if not local_path.is_file():
        return

    parser = _build_parser()
    try:
        parser.read(str(local_path), encoding="utf-8")
    except (configparser.Error, OSError) as exc:
        log.warning(
            "jail_local_read_for_update_error",
            jail=jail_name,
            error=str(exc),
        )
        return

    if not parser.has_section(jail_name) or not parser.has_option(jail_name, "action"):
        return

    existing_raw = parser.get(jail_name, "action")
    existing_lines = [
        line.strip()
        for line in existing_raw.splitlines()
        if line.strip() and not line.strip().startswith("#")
    ]

    def _base(entry: str) -> str:
        bracket = entry.find("[")
        return entry[:bracket].strip() if bracket != -1 else entry.strip()

    filtered = [e for e in existing_lines if _base(e) != action_name]

    if len(filtered) == len(existing_lines):
        # Action was not found — silently return (idempotent).
        return

    if filtered:
        new_value = filtered[0] + "".join(
            f"\n           {line}" for line in filtered[1:]
        )
        parser.set(jail_name, "action", new_value)
    else:
        parser.remove_option(jail_name, "action")

    buf = io.StringIO()
    buf.write("# Managed by BanGUI — do not edit manually\n\n")
    parser.write(buf)
    content = buf.getvalue()

    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=jail_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info(
        "jail_action_removed",
        jail=jail_name,
        action=action_name,
        path=str(local_path),
    )


def _write_action_local_sync(action_d: Path, name: str, content: str) -> None:
    """Write *content* to ``action.d/{name}.local`` atomically.

    The write is atomic: content is written to a temp file first, then
    renamed into place.  The ``action.d/`` directory is created if absent.

    Args:
        action_d: Path to the ``action.d`` directory.
        name: Validated action base name (used as filename stem).
        content: Full serialized action content to write.

    Raises:
        ConfigWriteError: If writing fails.
    """
    try:
        action_d.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        raise ConfigWriteError(
            f"Cannot create action.d directory: {exc}"
        ) from exc

    local_path = action_d / f"{name}.local"
    try:
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=action_d,
            delete=False,
            suffix=".tmp",
        ) as tmp:
            tmp.write(content)
            tmp_name = tmp.name
        os.replace(tmp_name, local_path)
    except OSError as exc:
        with contextlib.suppress(OSError):
            os.unlink(tmp_name)  # noqa: F821
        raise ConfigWriteError(
            f"Failed to write {local_path}: {exc}"
        ) from exc

    log.info("action_local_written", action=name, path=str(local_path))


# ---------------------------------------------------------------------------
# Public API — action discovery (Task 3.1)
# ---------------------------------------------------------------------------


async def list_actions(
    config_dir: str,
    socket_path: str,
) -> ActionListResponse:
    """Return all available actions from ``action.d/`` with active/inactive status.

    Scans ``{config_dir}/action.d/`` for ``.conf`` files, merges any
    corresponding ``.local`` overrides, parses each file into an
    :class:`~app.models.config.ActionConfig`, and cross-references with the
    currently running jails to determine which actions are active.

    An action is considered *active* when its base name appears in the
    ``action`` field of at least one currently running jail.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.

    Returns:
        :class:`~app.models.config.ActionListResponse` with all actions
        sorted alphabetically, active ones carrying non-empty
        ``used_by_jails`` lists.
    """
    action_d = Path(config_dir) / "action.d"
    loop = asyncio.get_event_loop()

    raw_actions: list[tuple[str, str, str, bool, str]] = await loop.run_in_executor(
        None, _parse_actions_sync, action_d
    )

    all_jails_result, active_names = await asyncio.gather(
        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
        _get_active_jail_names(socket_path),
    )
    all_jails, _source_files = all_jails_result

    action_to_jails = _build_action_to_jails_map(all_jails, active_names)

    actions: list[ActionConfig] = []
    for name, filename, content, has_local, source_path in raw_actions:
        cfg = conffile_parser.parse_action_file(
            content, name=name, filename=filename
        )
        used_by = sorted(action_to_jails.get(name, []))
        actions.append(
            ActionConfig(
                name=cfg.name,
                filename=cfg.filename,
                before=cfg.before,
                after=cfg.after,
                actionstart=cfg.actionstart,
                actionstop=cfg.actionstop,
                actioncheck=cfg.actioncheck,
                actionban=cfg.actionban,
                actionunban=cfg.actionunban,
                actionflush=cfg.actionflush,
                definition_vars=cfg.definition_vars,
                init_vars=cfg.init_vars,
                active=len(used_by) > 0,
                used_by_jails=used_by,
                source_file=source_path,
                has_local_override=has_local,
            )
        )

    log.info("actions_listed", total=len(actions), active=sum(1 for a in actions if a.active))
    return ActionListResponse(actions=actions, total=len(actions))


async def get_action(
    config_dir: str,
    socket_path: str,
    name: str,
) -> ActionConfig:
    """Return a single action from ``action.d/`` with active/inactive status.

    Reads ``{config_dir}/action.d/{name}.conf``, merges any ``.local``
    override, and enriches the parsed :class:`~app.models.config.ActionConfig`
    with ``active``, ``used_by_jails``, ``source_file``, and
    ``has_local_override``.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Action base name (e.g. ``"iptables"`` or ``"iptables.conf"``).

    Returns:
        :class:`~app.models.config.ActionConfig` with status fields populated.

    Raises:
        ActionNotFoundError: If no ``{name}.conf`` or ``{name}.local`` file
            exists in ``action.d/``.
    """
    if name.endswith(".conf"):
        base_name = name[:-5]
    elif name.endswith(".local"):
        base_name = name[:-6]
    else:
        base_name = name

    action_d = Path(config_dir) / "action.d"
    conf_path = action_d / f"{base_name}.conf"
    local_path = action_d / f"{base_name}.local"
    loop = asyncio.get_event_loop()

    def _read() -> tuple[str, bool, str]:
        """Read action content and return (content, has_local_override, source_path)."""
        has_local = local_path.is_file()
        if conf_path.is_file():
            content = conf_path.read_text(encoding="utf-8")
            if has_local:
                try:
                    content += "\n" + local_path.read_text(encoding="utf-8")
                except OSError as exc:
                    log.warning(
                        "action_local_read_error",
                        name=base_name,
                        path=str(local_path),
                        error=str(exc),
                    )
            return content, has_local, str(conf_path)
        elif has_local:
            content = local_path.read_text(encoding="utf-8")
            return content, False, str(local_path)
        else:
            raise ActionNotFoundError(base_name)

    content, has_local, source_path = await loop.run_in_executor(None, _read)

    cfg = conffile_parser.parse_action_file(
        content, name=base_name, filename=f"{base_name}.conf"
    )

    all_jails_result, active_names = await asyncio.gather(
        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
        _get_active_jail_names(socket_path),
    )
    all_jails, _source_files = all_jails_result
    action_to_jails = _build_action_to_jails_map(all_jails, active_names)

    used_by = sorted(action_to_jails.get(base_name, []))
    log.info("action_fetched", name=base_name, active=len(used_by) > 0)
    return ActionConfig(
        name=cfg.name,
        filename=cfg.filename,
        before=cfg.before,
        after=cfg.after,
        actionstart=cfg.actionstart,
        actionstop=cfg.actionstop,
        actioncheck=cfg.actioncheck,
        actionban=cfg.actionban,
        actionunban=cfg.actionunban,
        actionflush=cfg.actionflush,
        definition_vars=cfg.definition_vars,
        init_vars=cfg.init_vars,
        active=len(used_by) > 0,
        used_by_jails=used_by,
        source_file=source_path,
        has_local_override=has_local,
    )


# ---------------------------------------------------------------------------
# Public API — action write operations (Task 3.2)
# ---------------------------------------------------------------------------


async def update_action(
    config_dir: str,
    socket_path: str,
    name: str,
    req: ActionUpdateRequest,
    do_reload: bool = False,
) -> ActionConfig:
    """Update an action's ``.local`` override with new lifecycle command values.

    Reads the current merged configuration for *name* (``conf`` + any existing
    ``local``), applies the non-``None`` fields in *req* on top of it, and
    writes the resulting definition to ``action.d/{name}.local``.  The
    original ``.conf`` file is never modified.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        name: Action base name (e.g. ``"iptables"`` or ``"iptables.conf"``).
        req: Partial update — only non-``None`` fields are applied.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Returns:
        :class:`~app.models.config.ActionConfig` reflecting the updated state.

    Raises:
        ActionNameError: If *name* contains invalid characters.
        ActionNotFoundError: If no ``{name}.conf`` or ``{name}.local`` exists.
        ConfigWriteError: If writing the ``.local`` file fails.
    """
    base_name = name[:-5] if name.endswith((".conf", ".local")) else name
    _safe_action_name(base_name)

    current = await get_action(config_dir, socket_path, base_name)

    update = ActionConfigUpdate(
        actionstart=req.actionstart,
        actionstop=req.actionstop,
        actioncheck=req.actioncheck,
        actionban=req.actionban,
        actionunban=req.actionunban,
        actionflush=req.actionflush,
        definition_vars=req.definition_vars,
        init_vars=req.init_vars,
    )

    merged = conffile_parser.merge_action_update(current, update)
    content = conffile_parser.serialize_action_config(merged)

    action_d = Path(config_dir) / "action.d"
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _write_action_local_sync, action_d, base_name, content)

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_action_update_failed",
                action=base_name,
                error=str(exc),
            )

    log.info("action_updated", action=base_name, reload=do_reload)
    return await get_action(config_dir, socket_path, base_name)


async def create_action(
    config_dir: str,
    socket_path: str,
    req: ActionCreateRequest,
    do_reload: bool = False,
) -> ActionConfig:
    """Create a brand-new user-defined action in ``action.d/{name}.local``.

    No ``.conf`` is written; fail2ban loads ``.local`` files directly.  If a
    ``.conf`` or ``.local`` file already exists for the requested name, an
    :class:`ActionAlreadyExistsError` is raised.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        req: Action name and definition fields.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Returns:
        :class:`~app.models.config.ActionConfig` for the newly created action.

    Raises:
        ActionNameError: If ``req.name`` contains invalid characters.
        ActionAlreadyExistsError: If a ``.conf`` or ``.local`` already exists.
        ConfigWriteError: If writing fails.
    """
    _safe_action_name(req.name)

    action_d = Path(config_dir) / "action.d"
    conf_path = action_d / f"{req.name}.conf"
    local_path = action_d / f"{req.name}.local"

    def _check_not_exists() -> None:
        if conf_path.is_file() or local_path.is_file():
            raise ActionAlreadyExistsError(req.name)

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, _check_not_exists)

    cfg = ActionConfig(
        name=req.name,
        filename=f"{req.name}.local",
        actionstart=req.actionstart,
        actionstop=req.actionstop,
        actioncheck=req.actioncheck,
        actionban=req.actionban,
        actionunban=req.actionunban,
        actionflush=req.actionflush,
        definition_vars=req.definition_vars,
        init_vars=req.init_vars,
    )
    content = conffile_parser.serialize_action_config(cfg)

    await loop.run_in_executor(None, _write_action_local_sync, action_d, req.name, content)

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_action_create_failed",
                action=req.name,
                error=str(exc),
            )

    log.info("action_created", action=req.name, reload=do_reload)
    return await get_action(config_dir, socket_path, req.name)


async def delete_action(
    config_dir: str,
    name: str,
) -> None:
    """Delete a user-created action's ``.local`` file.

    Deletion rules:
    - If only a ``.conf`` file exists (shipped default, no user override) →
      :class:`ActionReadonlyError`.
    - If a ``.local`` file exists (whether or not a ``.conf`` also exists) →
      only the ``.local`` file is deleted.
    - If neither file exists → :class:`ActionNotFoundError`.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        name: Action base name (e.g. ``"iptables"``).

    Raises:
        ActionNameError: If *name* contains invalid characters.
        ActionNotFoundError: If no action file is found for *name*.
        ActionReadonlyError: If only a shipped ``.conf`` exists (no ``.local``).
        ConfigWriteError: If deletion of the ``.local`` file fails.
    """
    base_name = name[:-5] if name.endswith((".conf", ".local")) else name
    _safe_action_name(base_name)

    action_d = Path(config_dir) / "action.d"
    conf_path = action_d / f"{base_name}.conf"
    local_path = action_d / f"{base_name}.local"

    loop = asyncio.get_event_loop()

    def _delete() -> None:
        has_conf = conf_path.is_file()
        has_local = local_path.is_file()

        if not has_conf and not has_local:
            raise ActionNotFoundError(base_name)

        if has_conf and not has_local:
            raise ActionReadonlyError(base_name)

        try:
            local_path.unlink()
        except OSError as exc:
            raise ConfigWriteError(
                f"Failed to delete {local_path}: {exc}"
            ) from exc

        log.info("action_local_deleted", action=base_name, path=str(local_path))

    await loop.run_in_executor(None, _delete)


async def assign_action_to_jail(
    config_dir: str,
    socket_path: str,
    jail_name: str,
    req: AssignActionRequest,
    do_reload: bool = False,
) -> None:
    """Add an action to a jail by updating the jail's ``.local`` file.

    Appends ``{req.action_name}[{params}]`` (or just ``{req.action_name}`` when
    no params are given) to the ``action`` key in the ``[{jail_name}]`` section
    of ``jail.d/{jail_name}.local``.  If the action is already listed it is not
    duplicated.  If the ``.local`` file does not exist it is created.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        jail_name: Name of the jail to update.
        req: Request containing the action name and optional parameters.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Raises:
        JailNameError: If *jail_name* contains invalid characters.
        ActionNameError: If ``req.action_name`` contains invalid characters.
        JailNotFoundInConfigError: If *jail_name* is not defined in any config
            file.
        ActionNotFoundError: If ``req.action_name`` does not exist in
            ``action.d/``.
        ConfigWriteError: If writing fails.
    """
    _safe_jail_name(jail_name)
    _safe_action_name(req.action_name)

    loop = asyncio.get_event_loop()

    all_jails, _src = await loop.run_in_executor(
        None, _parse_jails_sync, Path(config_dir)
    )
    if jail_name not in all_jails:
        raise JailNotFoundInConfigError(jail_name)

    action_d = Path(config_dir) / "action.d"

    def _check_action() -> None:
        if (
            not (action_d / f"{req.action_name}.conf").is_file()
            and not (action_d / f"{req.action_name}.local").is_file()
        ):
            raise ActionNotFoundError(req.action_name)

    await loop.run_in_executor(None, _check_action)

    # Build the action string with optional parameters.
    if req.params:
        param_str = ", ".join(f"{k}={v}" for k, v in sorted(req.params.items()))
        action_entry = f"{req.action_name}[{param_str}]"
    else:
        action_entry = req.action_name

    await loop.run_in_executor(
        None,
        _append_jail_action_sync,
        Path(config_dir),
        jail_name,
        action_entry,
    )

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_assign_action_failed",
                jail=jail_name,
                action=req.action_name,
                error=str(exc),
            )

    log.info(
        "action_assigned_to_jail",
        jail=jail_name,
        action=req.action_name,
        reload=do_reload,
    )


async def remove_action_from_jail(
    config_dir: str,
    socket_path: str,
    jail_name: str,
    action_name: str,
    do_reload: bool = False,
) -> None:
    """Remove an action from a jail's ``.local`` config.

    Reads ``jail.d/{jail_name}.local``, removes the line(s) that reference
    ``{action_name}`` from the ``action`` key (including any ``[…]`` parameter
    blocks), and writes the file back atomically.

    Args:
        config_dir: Absolute path to the fail2ban configuration directory.
        socket_path: Path to the fail2ban Unix domain socket.
        jail_name: Name of the jail to update.
        action_name: Base name of the action to remove.
        do_reload: When ``True``, trigger a full fail2ban reload after writing.

    Raises:
        JailNameError: If *jail_name* contains invalid characters.
        ActionNameError: If *action_name* contains invalid characters.
        JailNotFoundInConfigError: If *jail_name* is not defined in any config.
        ConfigWriteError: If writing fails.
    """
    _safe_jail_name(jail_name)
    _safe_action_name(action_name)

    loop = asyncio.get_event_loop()

    all_jails, _src = await loop.run_in_executor(
        None, _parse_jails_sync, Path(config_dir)
    )
    if jail_name not in all_jails:
        raise JailNotFoundInConfigError(jail_name)

    await loop.run_in_executor(
        None,
        _remove_jail_action_sync,
        Path(config_dir),
        jail_name,
        action_name,
    )

    if do_reload:
        try:
            await jail_service.reload_all(socket_path)
        except Exception as exc:  # noqa: BLE001
            log.warning(
                "reload_after_remove_action_failed",
                jail=jail_name,
                action=action_name,
                error=str(exc),
            )

    log.info(
        "action_removed_from_jail",
        jail=jail_name,
        action=action_name,
        reload=do_reload,
    )