comicbox.online_session¶

"""
OnlineSession — Codex-facing API for batch online metadata tagging.

Wraps the existing Comicbox.run_online_lookup() per-file flow with a
Codex-friendly surface:

- Per-source credential validation up front (fail-fast).
- Direct :class:`MatchMode` API (no string alias layer; ``ASK`` is
  rejected since the session has no built-in prompt resolver for it).
- One programmatic PromptHandler per session that the matcher calls
  whenever it would otherwise hit the CLI questionary prompt.
- Event stream via on_event= for UI feedback.
- Cancellation token: callers stop the batch between files.

The internal matcher / selector / rate-limit machinery is untouched;
this module is a façade.
"""

from __future__ import annotations

import threading
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias, runtime_checkable

from comicbox.box import Comicbox
from comicbox.config import get_config
from comicbox.config.settings import (
    MatchMode,
    OnlineAuthSettings,
    OnlineLookupSettings,
    OnlineSourceCredentials,
    Prompts,
)
from comicbox.events import FileError, PromptDeferred, PromptResolvedFromCache

# OnlineConfigurationError keeps its historical comicbox.online_session
# import path; the definition lives in comicbox.exceptions so it shares
# the ComicboxError base.
from comicbox.exceptions import OnlineConfigurationError, OnlineLookupAbortedError

# Re-exported so batch callers get the run estimator off the same Codex-facing
# façade as the session it estimates; the implementation lives in
# comicbox.online_estimate.
from comicbox.online_estimate import (
    SOURCE_RATE_PER_MINUTE,
    RunEstimate,
    estimate_run,
)

if TYPE_CHECKING:
    from collections.abc import Callable, Iterable, Iterator, Sequence
    from pathlib import Path

    from comicbox.config.settings import ComicboxSettings
    from comicbox.events import EventHandler
    from comicbox.formats.base.online.profile import Candidate, ComicProfile
    from comicbox.formats.base.online.selector import (
        SelectorAction,
        SelectorContext,
        SelectorResult,
    )


__all__ = (
    "SOURCE_RATE_PER_MINUTE",
    "BatchedPromptHandler",
    "DeferredPrompt",
    "MatchMode",
    "OnlineConfigurationError",
    "OnlineCredentials",
    "OnlinePrompt",
    "OnlineResult",
    "OnlineSession",
    "PromptHandler",
    "PromptResponse",
    "RunEstimate",
    "SourceName",
    "estimate_run",
)


# --- public types -----------------------------------------------------------


SourceName: TypeAlias = Literal["metron", "comicvine"]

_KNOWN_SOURCES: frozenset[str] = frozenset({"metron", "comicvine"})


@dataclass(frozen=True, slots=True)
class OnlineCredentials:
[docs]
    """Credentials for the online sources Codex may enable."""

    metron_user: str | None = None
    metron_password: str | None = None
    metron_url: str | None = None
    comicvine_key: str | None = None
    comicvine_url: str | None = None


@dataclass(frozen=True, slots=True)
class OnlinePrompt:
[docs]
    """
    Codex-facing handoff for an ambiguous match.

    The PromptHandler receives one of these and returns a PromptResponse.
    """

    path: Path | None
    source: str
    profile_summary: dict[str, Any]
    candidates: tuple[Candidate, ...]
    mode: MatchMode
    unattended: bool


@dataclass(frozen=True, slots=True)
class PromptResponse:
[docs]
    """
    Codex-side decision for one prompt.

    ``action`` mirrors :data:`SelectorAction`; ``payload`` is the index for
    ``choose``, the ``"<source>:<id>"`` string for ``manual``, the new
    :class:`MatchMode` value (e.g. ``"auto"``) for ``set_policy``, or
    ``None`` otherwise.
    """

    action: Literal["choose", "skip", "manual", "abort", "set_unattended", "set_policy"]
    payload: int | str | None = None


@runtime_checkable
class PromptHandler(Protocol):
[docs]
    """
    Codex's prompt-resolution callback.

    All handlers must implement :meth:`request` — the single-prompt form
    used by every code path that ships today. Handlers MAY additionally
    implement :meth:`request_many` to opt into the future batched-prompt
    delivery path (plan §3.4 / step 8): when the engine grows the
    per-session prompt queue, handlers exposing ``request_many`` get the
    full pending window in one call; handlers without it continue to be
    invoked one prompt at a time. v1 delivers prompts via ``request``
    only — see ``tasks/prepare-for-codex-writing/03-batched-prompt-handler.md``
    for the deferred implementation plan.
    """

    def request(self, prompt: OnlinePrompt) -> PromptResponse:
[docs]
        """Resolve one prompt; return the chosen action/payload."""
        ...

    # request_many is intentionally omitted from this Protocol body so
    # existing single-method handlers structurally match. Handlers that
    # implement it should match BatchedPromptHandler below; callers detect
    # support via isinstance() / hasattr("request_many").


@runtime_checkable
class BatchedPromptHandler(Protocol):
[docs]
    """
    A PromptHandler that also supports batched prompt delivery.

    Codex's defer-mode UI naturally yields a window of prompts the user
    will resolve together. When the engine gains the per-session prompt
    queue (step 8), it will prefer ``request_many`` over ``request`` for
    handlers structurally matching this Protocol.
    """

    def request(self, prompt: OnlinePrompt) -> PromptResponse:
[docs]
        """Resolve one prompt; return the chosen action/payload."""
        ...
[docs]
    def request_many(self, prompts: Sequence[OnlinePrompt]) -> Sequence[PromptResponse]:
        """
        Resolve a window of prompts in one call.

        Return one response per prompt, in the same order. The engine
        will not interleave ``request`` and ``request_many`` calls
        within a single drain window.
        """
        ...


@dataclass(frozen=True, slots=True)
class OnlineResult:
[docs]
    """
    Per-file outcome of an online tagging call.

    ``tags`` is the file's full merged metadata, present even when the
    lookup applied nothing new (skip, no-match, deferred prompt) — it is
    NOT evidence of a match. ``matched`` is: it's True only when an
    online source actually contributed metadata, so writers should gate
    on it to avoid re-writing a file with its own existing tags.
    """

    path: Path
    tags: dict[str, Any] | None = None
    error: BaseException | None = None
    cancelled: bool = False
    matched: bool = False


@dataclass(frozen=True, slots=True)
class _CachedResolution:
    """A previously-handled prompt's outcome, keyed by series-level fingerprint."""

    action: SelectorAction
    payload: int | str | None
    # For "choose", the candidate's volume_id at the time the user picked.
    # We re-map to whichever candidate in the new prompt shares it.
    chosen_volume_id: int | None = None


@dataclass(frozen=True, slots=True)
class DeferredPrompt:
[docs]
    """
    A prompt the session skipped under defer_prompts mode.

    Captures everything Codex needs to render the prompt later in a
    review-tagging UI: the file it came from, source, candidates, mode
    context, and the fingerprint used to key the dedup cache. Codex
    feeds the user's resolution back via :meth:`OnlineSession.preload_resolution`.
    """

    path: Path | None
    source: str
    fingerprint: str
    profile_summary: dict[str, Any]
    candidates: tuple[Candidate, ...]
    mode: MatchMode
    unattended: bool


# --- session ---------------------------------------------------------------


class OnlineSession:
[docs]
    """
    Codex-friendly façade over Comicbox's online lookup.

    Construction validates per-source credentials and pre-computes the
    ComicboxSettings layer that each per-file Comicbox instance will see.
    Mutable state — ``mode``, ``unattended``, the cancel token — lives on
    the instance and may be updated from any thread.
    """

    def __init__(  # noqa: PLR0913
        self,
        *,
        sources: Iterable[str] = ("metron", "comicvine"),
        credentials: OnlineCredentials | None = None,
        mode: MatchMode = MatchMode.AUTO,
        unattended: bool = False,
        prompt_handler: PromptHandler | None = None,
        on_event: EventHandler | None = None,
        rematch: bool = False,
        first_wins: bool = True,
        defer_prompts: bool = False,
        series_batching: bool = True,
    ) -> None:
        """Validate inputs, build per-session state. See class docstring."""
        # Order is run priority: the first source runs first and, under
        # first_wins, its match ends the lookup for that comic.
        self._sources = tuple(dict.fromkeys(sources))
        self._validate_sources(self._sources)
        self._credentials = credentials or OnlineCredentials()
        self._validate_credentials(self._sources, self._credentials)
        self._mode: MatchMode = self._validate_mode(mode)
        self._unattended = unattended
        self._prompt_handler = prompt_handler
        self._on_event = on_event
        self._rematch = rematch
        self._first_wins = first_wins
        self._defer_prompts = defer_prompts
        self._series_batching = series_batching
        # Read config files / env exactly once per session. _build_config
        # used to call get_config() per file — wasted disk I/O on big
        # batches plus a behavioral surprise where a config-file edit
        # mid-batch changed settings for the remaining files.
        self._base_settings: ComicboxSettings = get_config()

        # Cancel token. Set when cancel() is called; checked between files
        # in tag_many() and consulted by the wired retry sleep
        # (_retry_sleep_wait), which aborts an in-flight rate-limit wait.
        self._cancel = threading.Event()
        self._state_lock = threading.Lock()

        # Prompt-dedup cache. Keyed by fingerprint of (source, normalized
        # series, sorted distinct candidate volume_ids); the stored entry
        # is the PromptResponse the handler returned the first time we
        # saw this fingerprint. For "choose" actions we also record the
        # candidate's volume_id so we can re-map the index when the new
        # prompt's candidate list is ordered differently. Per-session,
        # in-memory only — disk persistence is out of scope for v1.
        self._prompt_cache: dict[str, _CachedResolution] = {}
        self._prompt_cache_lock = threading.Lock()

        # Defer-prompts queue. When ``defer_prompts`` is set, the bridged
        # selector skips the user's PromptHandler and queues the prompt
        # here instead. Codex's flow: drain via ``deferred_prompts()`` at
        # end-of-batch, present in a review UI, then seed resolutions
        # back via ``preload_resolution()`` and re-run the affected files.
        self._deferred: list[DeferredPrompt] = []
        self._deferred_lock = threading.Lock()

        # Series cache (plan §3.10). Keyed by (source_name, series
        # fingerprint); value is the resolved volume_id. Populated by
        # the matcher on each cold-path acceptance; consulted before the
        # cold-path search on subsequent comics of the same series so
        # we make one `search` call per series instead of one per issue.
        # In-memory only per the resolved open question.
        self._series_cache: dict[tuple[str, str], int] = {}
        self._series_cache_lock = threading.Lock()

    # -- mutable session state ----------------------------------------------

    @property
    def mode(self) -> MatchMode:
[docs]
        """Current session mode (read-only; mutate via set_mode())."""
        with self._state_lock:
            return self._mode

    @property
    def unattended(self) -> bool:
[docs]
        """Current session unattended flag."""
        with self._state_lock:
            return self._unattended

    def set_mode(self, mode: MatchMode) -> None:
[docs]
        """Change the session mode for subsequent file lookups."""
        validated = self._validate_mode(mode)
        with self._state_lock:
            self._mode = validated

    def set_unattended(self, *, unattended: bool) -> None:
[docs]
        """Toggle the unattended flag for subsequent file lookups."""
        with self._state_lock:
            self._unattended = unattended

    def cancel(self) -> None:
[docs]
        """Stop accepting new files. In-flight lookup runs to completion."""
        self._cancel.set()

    @property
    def cancelled(self) -> bool:
[docs]
        """Whether cancel() has been called."""
        return self._cancel.is_set()

    # -- deferred prompts ---------------------------------------------------

    @property
    def defer_prompts(self) -> bool:
[docs]
        """Whether the session is currently in defer-prompts mode."""
        return self._defer_prompts

    def set_defer_prompts(self, *, defer: bool) -> None:
[docs]
        """Toggle defer-prompts mode for subsequent file lookups."""
        self._defer_prompts = defer

    def deferred_prompts(self) -> tuple[DeferredPrompt, ...]:
[docs]
        """Snapshot the queued deferred prompts. Codex's review-UI input."""
        with self._deferred_lock:
            return tuple(self._deferred)

    def clear_deferred_prompts(self) -> None:
[docs]
        """Drop the deferred-prompt queue without resolving them."""
        with self._deferred_lock:
            self._deferred.clear()

    def preload_resolution(
[docs]
        self,
        fingerprint: str,
        *,
        action: Literal["choose", "skip", "manual"],
        payload: int | str | None = None,
        chosen_volume_id: int | None = None,
    ) -> None:
        """
        Seed the dedup cache so a re-run auto-resolves the fingerprint.

        Codex's defer-mode flow: drain ``deferred_prompts()`` after the
        batch, present them in a review UI, call ``preload_resolution()``
        for each one the user resolves, then re-tag the affected files
        (with defer_prompts toggled off if desired). The cache hit fires
        the same way it does for in-batch dedup. A non-empty cache is
        enough to install the bridged selector — but with no handler and
        defer_prompts off, a cache MISS (the re-search produced a
        different candidate set than the fingerprint was minted from)
        raises rather than falling back to the interactive CLI prompt.
        """
        entry = _CachedResolution(
            action=action, payload=payload, chosen_volume_id=chosen_volume_id
        )
        with self._prompt_cache_lock:
            self._prompt_cache[fingerprint] = entry

    # -- series-first batching ----------------------------------------------

    def preload_series_resolution(
[docs]
        self, *, source: str, series_fingerprint: str, volume_id: int
    ) -> None:
        """
        Seed the series cache from outside the session.

        Codex's use case: persist resolved series across runs in its own
        DB and replay them at session-start to skip the cold-path search
        entirely on subsequent imports. Same first-writer-wins semantic
        as in-batch population — if the key is already cached, this is
        a no-op.
        """
        key = (source, series_fingerprint)
        with self._series_cache_lock:
            self._series_cache.setdefault(key, volume_id)

    def series_cache_snapshot(self) -> dict[tuple[str, str], int]:
[docs]
        """Snapshot the current series cache. Useful for persistence."""
        with self._series_cache_lock:
            return dict(self._series_cache)

    def clear_series_cache(self) -> None:
[docs]
        """Drop every series resolution from the cache."""
        with self._series_cache_lock:
            self._series_cache.clear()

    # -- rate limits --------------------------------------------------------

    def rate_limit_status(self) -> dict[str, dict[str, Any]]:
[docs]
        """
        Snapshot of each enabled source's current rate-limit budget.

        v1 stub: returns an empty dict for every enabled source. The
        wiring to mokkari / simyan rate-limit buckets lands in the
        follow-up commit (see plan §3.5 / §3.6).
        """
        return {name: {} for name in self._sources}

    # -- per-file tagging ---------------------------------------------------

    def tag(self, path: Path) -> OnlineResult:
[docs]
        """Tag one file synchronously."""
        if self._cancel.is_set():
            return OnlineResult(path=path, cancelled=True)
        try:
            tags, matched = self._run_one(path)
        except OnlineLookupAbortedError:
            # The handler answered "abort" (or a cancelled retry sleep
            # aborted an in-flight lookup). Abort means "abort the entire
            # run", not "skip this file": trip the cancel token so
            # tag_many drains the remaining paths as cancelled.
            self.cancel()
            return OnlineResult(path=path, cancelled=True)
        except Exception as exc:
            if self._on_event is not None:
                self._on_event(FileError(path=path, error=str(exc)))
            return OnlineResult(path=path, error=exc)
        return OnlineResult(path=path, tags=tags, matched=matched)

    def tag_many(self, paths: Iterable[Path]) -> Iterator[OnlineResult]:
[docs]
        """
        Tag many files sequentially. Stops accepting new ones on cancel.

        When ``series_batching`` is on (the default), paths are reordered
        upfront by a lightweight filename-derived fingerprint so issues of
        the same series cluster together. The first issue of each cluster
        triggers the cold-path search; the remaining issues hit the
        per-session series cache and skip the search entirely. Order
        within a cluster is the original input order; the cluster order
        itself is deterministic (sorted by fingerprint) so re-runs
        produce identical cache-key sequences.
        """
        path_list = list(paths)
        if self._series_batching:
            path_list = sorted(path_list, key=_filename_series_fingerprint)
        for path in path_list:
            if self._cancel.is_set():
                yield OnlineResult(path=path, cancelled=True)
                continue
            yield self.tag(path)

    # -- internals ----------------------------------------------------------

    def _run_one(self, path: Path) -> tuple[dict[str, Any], bool]:
        config = self._build_config()
        with Comicbox(path, config=config) as cb:
            # Bridge the selector when we have a handler, when defer
            # mode is on (defer produces no handler call but still needs
            # to intercept the prompt to queue it), or when resolutions
            # were preloaded — the bridged selector is the only consumer
            # of the prompt cache, so a preload-only replay session
            # (codex's resolve flow) needs it too.
            if (
                self._prompt_handler is not None
                or self._defer_prompts
                or self._prompt_cache
            ):
                cb.set_online_selector(self._bridged_selector())
            if self._on_event is not None:
                cb.set_event_handler(self._on_event)
            if self._series_batching:
                cb.set_series_cache(self._series_cache)
            cb.set_retry_sleep(self._retry_sleep_wait)
            matched = cb.run_online_lookup()
            payload = cb.to_dict()
        return payload.get("comicbox", {}), matched

    def _retry_sleep_wait(self, delay: float) -> None:
        """
        Wait out a retry delay, aborting the lookup when cancel() fires.

        Rate-limit recovery can sleep for minutes per attempt; plain
        time.sleep would make cancel() wait out the whole budget while
        the archive sits open. Waiting on the cancel event instead lets
        a caller interrupt mid-retry; raising aborts the in-flight
        lookup, which tag() converts into a cancelled result.
        """
        if self._cancel.wait(delay):
            reason = "online: session cancelled during retry wait"
            raise OnlineLookupAbortedError(reason)

    def _bridged_selector(self) -> Callable[..., SelectorResult]:
        """
        Adapt the user's PromptHandler into a SelectorCallback.

        Translates Comicbox's positional SelectorCallback signature into
        a Codex-facing OnlinePrompt / PromptResponse pair. Consults the
        per-session prompt cache first; on cache miss falls through to
        defer-mode (queue + skip) or the user's handler, depending on
        ``defer_prompts``. Resolutions from the handler path are stored
        for next time.
        """
        handler = self._prompt_handler

        def _selector(
            profile: ComicProfile,
            candidates: Sequence[Candidate],
            ctx: SelectorContext,
        ) -> SelectorResult:
            cand_tuple = tuple(candidates)
            fingerprint = _prompt_fingerprint(ctx.source, profile, cand_tuple)
            cached = self._lookup_cached_prompt(fingerprint, cand_tuple)
            if cached is not None:
                if self._on_event is not None:
                    self._on_event(
                        PromptResolvedFromCache(
                            path=ctx.file_path,
                            source=ctx.source,
                            prompt_id=f"{ctx.source}:{id(candidates)}",
                            action=cached[0],
                            fingerprint=fingerprint,
                        )
                    )
                return cached
            if self._defer_prompts:
                self._defer_prompt(fingerprint, profile, cand_tuple, ctx)
                return ("skip", None)
            if handler is None:
                msg = (
                    "OnlineSession has no PromptHandler and defer_prompts "
                    "is disabled; cannot resolve an ambiguous match"
                )
                raise RuntimeError(msg)
            prompt = OnlinePrompt(
                path=ctx.file_path,
                source=ctx.source,
                profile_summary=_summarise_profile(profile),
                candidates=cand_tuple,
                mode=self.mode,
                unattended=self.unattended,
            )
            response = handler.request(prompt)
            self._sync_session_state(response)
            self._store_prompt_resolution(fingerprint, response, cand_tuple)
            return (response.action, response.payload)

        return _selector

    def _sync_session_state(self, response: PromptResponse) -> None:
        """
        Mirror session-level prompt actions into the session's own state.

        The box applies set_policy / set_unattended to its per-file config,
        but _run_one rebuilds that config for every file from the session's
        _mode / _unattended — without this sync the handler's decision
        would silently revert on the next file, despite PromptResponse
        documenting these as session-level actions.
        """
        if response.action == "set_unattended":
            self.set_unattended(unattended=True)
            return
        if response.action != "set_policy" or not isinstance(response.payload, str):
            return
        try:
            mode = MatchMode(response.payload)
        except ValueError:
            # Malformed payload: the box logs and declines it; nothing to sync.
            return
        if mode is MatchMode.ASK:
            # The session's set_mode rejects ASK (no built-in CLI prompt);
            # the box still honors it for the in-flight file.
            return
        self.set_mode(mode)

    def _defer_prompt(
        self,
        fingerprint: str,
        profile: ComicProfile,
        candidates: tuple[Candidate, ...],
        ctx: SelectorContext,
    ) -> None:
        """Queue this prompt for later resolution and emit PromptDeferred."""
        deferred = DeferredPrompt(
            path=ctx.file_path,
            source=ctx.source,
            fingerprint=fingerprint,
            profile_summary=_summarise_profile(profile),
            candidates=candidates,
            mode=self.mode,
            unattended=self.unattended,
        )
        with self._deferred_lock:
            self._deferred.append(deferred)
        if self._on_event is not None:
            self._on_event(
                PromptDeferred(
                    path=ctx.file_path,
                    source=ctx.source,
                    prompt_id=f"{ctx.source}:{id(candidates)}",
                    fingerprint=fingerprint,
                    n_candidates=len(candidates),
                )
            )

    def _lookup_cached_prompt(
        self, fingerprint: str, candidates: tuple[Candidate, ...]
    ) -> SelectorResult | None:
        """
        Match a fingerprint against the cache and re-map the choice.

        For ``choose`` we re-map the cached volume_id to whichever candidate
        in this prompt's list shares it. If no candidate matches, treat
        this as a cache miss and let the handler fire fresh.
        """
        with self._prompt_cache_lock:
            cached = self._prompt_cache.get(fingerprint)
        if cached is None:
            return None
        if cached.action != "choose":
            return (cached.action, cached.payload)
        if cached.chosen_volume_id is None:
            return None
        for i, cand in enumerate(candidates):
            if cand.volume_id is not None and cand.volume_id == cached.chosen_volume_id:
                return ("choose", i)
        return None

    def _store_prompt_resolution(
        self,
        fingerprint: str,
        response: PromptResponse,
        candidates: tuple[Candidate, ...],
    ) -> None:
        """Cache a fresh resolution under its fingerprint."""
        chosen_volume_id: int | None = None
        if response.action == "choose" and isinstance(response.payload, int):
            try:
                chosen_volume_id = candidates[response.payload].volume_id
            except IndexError:
                chosen_volume_id = None
        # Session-level actions (set_unattended / set_policy / abort) are
        # not cached: they apply once, not as a deferred decision.
        if response.action in {"set_unattended", "set_policy", "abort"}:
            return
        entry = _CachedResolution(
            action=response.action,
            payload=response.payload,
            chosen_volume_id=chosen_volume_id,
        )
        with self._prompt_cache_lock:
            self._prompt_cache[fingerprint] = entry

    def _build_config(self) -> ComicboxSettings:
        """
        Build a ComicboxSettings from the session's tagging preferences.

        Layered as ``replace`` over the default settings rather than fed
        through ``get_config(Namespace(...))`` because some of the fields
        we need to set (``enabled``, ``sources``, per-source auth dict)
        live on runtime-only fields the CLI namespace parser ignores.

        ``mode`` / ``unattended`` are session-mutable, so the cheap
        ``replace`` layering stays per-file; the disk read happened once
        in ``__init__``.
        """
        base = self._base_settings
        new_lookup = OnlineLookupSettings(
            enabled=True,
            sources=self._sources,
            match=self.mode,
            prompts=Prompts.NEVER if self.unattended else Prompts.ASK,
            rematch=self._rematch,
            first_wins=self._first_wins,
        )
        new_auth = OnlineAuthSettings(sources=self._build_auth_sources())
        new_online = replace(base.online, lookup=new_lookup, auth=new_auth)
        return replace(base, online=new_online)

    def _build_auth_sources(self) -> dict[str, OnlineSourceCredentials]:
        creds = self._credentials
        result: dict[str, OnlineSourceCredentials] = {}
        if "metron" in self._sources:
            result["metron"] = OnlineSourceCredentials(
                user=creds.metron_user,
                password=creds.metron_password,
                url=creds.metron_url,
            )
        if "comicvine" in self._sources:
            result["comicvine"] = OnlineSourceCredentials(
                key=creds.comicvine_key,
                url=creds.comicvine_url,
            )
        return result

    # -- validation ---------------------------------------------------------

    @staticmethod
    def _validate_sources(sources: tuple[str, ...]) -> None:
        if not sources:
            msg = "OnlineSession requires at least one source"
            raise OnlineConfigurationError(msg)
        unknown = set(sources) - _KNOWN_SOURCES
        if unknown:
            msg = (
                f"Unknown online sources: {sorted(unknown)}. "
                f"Expected a subset of {sorted(_KNOWN_SOURCES)}."
            )
            raise OnlineConfigurationError(msg)

    @staticmethod
    def _validate_credentials(
        sources: tuple[str, ...], creds: OnlineCredentials
    ) -> None:
        missing: list[str] = []
        if "metron" in sources and not (creds.metron_user and creds.metron_password):
            missing.append("metron requires metron_user and metron_password")
        if "comicvine" in sources and not creds.comicvine_key:
            missing.append("comicvine requires comicvine_key")
        if missing:
            msg = "Missing credentials for enabled online source(s):\n  " + "\n  ".join(
                missing
            )
            raise OnlineConfigurationError(msg)

    @staticmethod
    def _validate_mode(mode: MatchMode) -> MatchMode:
        if not isinstance(mode, MatchMode):  # pyright: ignore[reportUnnecessaryIsInstance]
            msg = f"OnlineSession.mode must be a MatchMode enum value; got {mode!r}."  # pyright: ignore[reportUnreachable]
            raise OnlineConfigurationError(msg)
        if mode is MatchMode.ASK:
            msg = (
                "OnlineSession.mode does not accept MatchMode.ASK; "
                "the session has no built-in prompt resolver. Use a "
                "PromptHandler or defer_prompts=True instead."
            )
            raise OnlineConfigurationError(msg)
        return mode


def _summarise_profile(profile: ComicProfile) -> dict[str, Any]:
    """Pluck the user-displayable fields off a ComicProfile."""
    fields_of_interest = (
        "series",
        "volume",
        "issue",
        "year",
        "publisher",
        "filename",
    )
    return {f: getattr(profile, f, None) for f in fields_of_interest}


def _prompt_fingerprint(
    source: str, profile: ComicProfile, candidates: tuple[Candidate, ...]
) -> str:
    """
    Build a deterministic key identifying the disambiguation question.

    Composed of the source name, series-level profile fields, and the sorted
    set of candidate volume_ids. Different issues of the same series collapse
    to the same fingerprint because their candidate volume_ids match —
    enabling the cache to auto-apply the user's prior series choice to
    every subsequent issue of that run.
    """
    series = (profile.series or "").strip().lower()
    publisher = (profile.publisher or "").strip().lower()
    volume_ids = tuple(
        sorted({c.volume_id for c in candidates if c.volume_id is not None})
    )
    # Fall back to candidate issue_ids when no volume_id is present (some
    # sources don't expose one) — this gives a strictly stricter
    # fingerprint, equivalent to "same exact candidate list."
    if not volume_ids:
        volume_ids = tuple(sorted(c.issue_id for c in candidates))
    parts = (source, series, str(profile.year or ""), publisher, repr(volume_ids))
    return "|".join(parts)


def _filename_series_fingerprint(path: Path) -> str:
    """
    Lightweight series fingerprint derived from the filename alone.

    Used by ``tag_many`` to group same-series comics together *before*
    opening any archives. Falls back to the bare filename when comicfn2dict
    can't extract a series — those files won't benefit from series
    batching but won't break it either (they sort to a deterministic
    "by filename" bucket).
    """
    from comicfn2dict import comicfn2dict

    try:
        parsed = comicfn2dict(path.name)
    except Exception:  # pragma: no cover — comicfn2dict is permissive
        return path.name.lower()
    series = str(parsed.get("series") or "").strip().lower()
    year = str(parsed.get("year") or "")
    return f"{series}|{year}" if series else f"~{path.name.lower()}"