john_wrapper.py

"""Small, process-based wrapper for the repository's MeshCore John format.

This module deliberately deals in MeshCore packet samples rather than generic
password-hash input.  John remains an external process so the web server stays
responsive while a decode job is running.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import signal
import subprocess
import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Callable, Iterable


CHANNEL_CHARACTER_SETS = ("alpha", "numeric", "hyphen")
CHANNEL_CHARACTER_SET_VALUES = {
    "alpha": "abcdefghijklmnopqrstuvwxyz",
    "numeric": "0123456789",
    "hyphen": "-",
}


def normalize_channel_character_sets(values: Iterable[str] | None = None) -> tuple[str, ...]:
    """Return character-set names in the stable order used by the GPU search."""
    if values is None:
        return CHANNEL_CHARACTER_SETS
    if isinstance(values, str):
        raise JohnWrapperError("character_sets must be an array")
    try:
        selected = set(values)
    except TypeError as exc:
        raise JohnWrapperError("character_sets must be an array") from exc
    if not selected or any(not isinstance(value, str) for value in selected):
        raise JohnWrapperError("select at least one channel character set")
    unknown = selected.difference(CHANNEL_CHARACTER_SETS)
    if unknown:
        raise JohnWrapperError("character_sets may only contain alpha, numeric, and hyphen")
    return tuple(character_set for character_set in CHANNEL_CHARACTER_SETS if character_set in selected)


def channel_alphabet(values: Iterable[str] | None = None) -> str:
    return "".join(CHANNEL_CHARACTER_SET_VALUES[value] for value in normalize_channel_character_sets(values))


class JohnWrapperError(RuntimeError):
    """Raised when a MeshCore John job cannot be started or inspected."""


class JohnJobBusyError(JohnWrapperError):
    """Raised when another John-backed job already owns the process slot."""

    def __init__(self, active_kind: str) -> None:
        self.active_kind = active_kind
        super().__init__(f"a {active_kind.replace('_', ' ')} is already running")


class JohnProcessCoordinator:
    """Keep all John-backed jobs in this application mutually exclusive."""

    def __init__(self) -> None:
        self._condition = threading.Condition()
        self._active_kind: str | None = None
        self._active_token: str | None = None
        self._cancel_active: Callable[[], None] | None = None

    def acquire(self, kind: str, cancel_active: Callable[[], None]) -> str:
        with self._condition:
            if self._active_kind is not None:
                raise JohnJobBusyError(self._active_kind)
            token = uuid.uuid4().hex
            self._active_kind = kind
            self._active_token = token
            self._cancel_active = cancel_active
            return token

    def release(self, token: str) -> None:
        with self._condition:
            if token != self._active_token:
                return
            self._active_kind = None
            self._active_token = None
            self._cancel_active = None
            self._condition.notify_all()

    def active_kind(self) -> str | None:
        with self._condition:
            return self._active_kind

    def cancel_active(self, expected_kind: str, timeout: float = 5.0) -> None:
        """Ask the active job to stop and wait until its John process is gone."""
        with self._condition:
            if self._active_kind is None:
                return
            if self._active_kind != expected_kind:
                raise JohnJobBusyError(self._active_kind)
            cancel = self._cancel_active
            deadline = time.monotonic() + timeout
        if cancel is not None:
            cancel()
        with self._condition:
            while self._active_kind == expected_kind:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise JohnWrapperError(
                        f"could not stop the running {expected_kind.replace('_', ' ')}"
                    )
                self._condition.wait(remaining)


class JohnDecodeJob:
    """Run one constrained MeshCore channel-name search and expose its status."""

    # John emits a progress line once per second; polling a separate status
    # process more often than that only adds process-launch overhead.
    STATUS_INTERVAL = 1.0
    MAX_OUTPUT_LINES = 40
    ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
    CURRENT_KEY_RE = re.compile(
        r"(?:\d+(?:\.\d+)?[KMG]?(?:[CcPp])?/s)\s+(#[^\s]+)",
        re.IGNORECASE,
    )
    CURRENT_KEY_RANGE_RE = re.compile(r"(#[a-z0-9-]+)\.\.(#[a-z0-9-]+)", re.IGNORECASE)
    HASH_RATE_RE = re.compile(r"(\d+(?:\.\d+)?)([KMG]?)[Cc]/s", re.IGNORECASE)
    # John puts the cracking percentage in its own token, but GPU sensor
    # fields may append values such as ``util:98%`` to the same line.  Only
    # accept a percentage that is not part of a word or ``name:value`` field.
    PROGRESS_PERCENT_RE = re.compile(
        r"(?<![\w:])(\d+(?:\.\d+)?)%(?=\s|$)"
    )
    BODY_LENGTH_RE = re.compile(r"Trying channel body length (\d+)")

    def __init__(
        self,
        script_path: Path,
        packet_samples: dict[int, list[bytes]],
        max_body_length: int = 29,
        status_interval: float = STATUS_INTERVAL,
        popen: Callable[..., subprocess.Popen[str]] = subprocess.Popen,
        command_runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
        packet_file: Path | None = None,
        session_prefix: str | None = None,
        start_body_length: int = 1,
        character_sets: Iterable[str] | None = None,
        on_update: Callable[[dict[str, object]], None] | None = None,
    ) -> None:
        if not packet_samples:
            raise JohnWrapperError("at least one unknown group must be selected")
        if not 1 <= max_body_length <= 29:
            raise JohnWrapperError("maximum body length must be between 1 and 29")
        if not 1 <= start_body_length <= max_body_length:
            raise JohnWrapperError("start body length must be within the configured range")
        if not script_path.is_file():
            raise JohnWrapperError(f"John wrapper script not found: {script_path}")

        self.script_path = script_path
        self.packet_samples = packet_samples
        self.max_body_length = max_body_length
        self.selected_character_sets = normalize_channel_character_sets(character_sets)
        self.channel_alphabet = channel_alphabet(self.selected_character_sets)
        self.status_interval = status_interval
        self._popen = popen
        self._run_command = command_runner
        self._lock = threading.RLock()
        self._process: subprocess.Popen[str] | None = None
        self._thread: threading.Thread | None = None
        self._poll_thread: threading.Thread | None = None
        self._stop_requested = threading.Event()
        self._pause_requested = threading.Event()
        self._started_at = 0.0
        self._phase_started_at: float | None = None
        self._finished_at: float | None = None
        self._session_prefix = session_prefix or f"meshcore-decode-{uuid.uuid4().hex}"
        self._start_body_length = start_body_length
        self._current_body_length = start_body_length
        self._on_update = on_update
        self._current_session: str | None = None
        self._temporary_packet_file: Path | None = None
        self._state = "queued"
        self._phase = "Preparing packet samples"
        self._current_key: str | None = None
        self._current_hash_rate: float | None = None
        self._current_progress_percent: float | None = None
        self._output: list[str] = []
        self._recovered_candidates: list[str] = []
        self._error: str | None = None
        self._packet_file = packet_file

    @property
    def selected_hmac_ids(self) -> list[int]:
        return sorted(self.packet_samples)

    def start(self) -> None:
        with self._lock:
            if self._thread is not None:
                raise JohnWrapperError("John decode job has already started")
            self._started_at = time.monotonic()
            self._phase_started_at = self._started_at
            self._thread = threading.Thread(target=self._run, daemon=True)
            self._thread.start()

    def stop(self) -> None:
        self.pause()

    def pause(self) -> None:
        self._pause_requested.set()
        self._stop_requested.set()
        with self._lock:
            process = self._process
        if process is not None and process.poll() is None:
            try:
                os.killpg(process.pid, signal.SIGTERM)
            except (AttributeError, OSError):
                process.terminate()
        thread = self._thread
        if thread is not None:
            thread.join(timeout=3)

    def _notify(self) -> None:
        if self._on_update is not None:
            try:
                self._on_update(self.status())
            except Exception:
                pass

    def status(self) -> dict[str, object]:
        with self._lock:
            now = self._finished_at or time.monotonic()
            elapsed = max(0.0, now - self._started_at) if self._started_at else 0.0
            phase_started_at = getattr(self, "_phase_started_at", None)
            phase_elapsed = (
                max(0.0, now - phase_started_at)
                if phase_started_at is not None
                else 0.0
            )
            phase_progress = self._phase_progress_percent()
            return {
                "running": self._state in {"queued", "running"},
                "state": self._state,
                "selected_hmac_ids": [f"0x{value:02X}" for value in self.selected_hmac_ids],
                "selected_character_sets": list(
                    getattr(self, "selected_character_sets", CHANNEL_CHARACTER_SETS)
                ),
                "selected_count": sum(len(samples) for samples in self.packet_samples.values()),
                "phase": self._phase,
                "current_body_length": self._current_body_length,
                "current_key": self._current_key,
                "current_hash_rate": (
                    round(self._current_hash_rate, 1)
                    if getattr(self, "_current_hash_rate", None) is not None
                    else None
                ),
                "current_progress_percent": (
                    round(self._current_progress_percent, 2)
                    if getattr(self, "_current_progress_percent", None) is not None
                    else None
                ),
                "phase_progress_percent": (
                    round(phase_progress, 2)
                    if phase_progress is not None
                    else None
                ),
                "elapsed_seconds": round(elapsed, 1),
                "phase_elapsed_seconds": round(phase_elapsed, 1),
                "recovered_channels": list(self._recovered_candidates),
                "error": self._error,
                "output": list(self._output[-self.MAX_OUTPUT_LINES:]),
            }

    def _phase_progress_percent(self) -> float | None:
        """Calculate progress within the active body-length search.

        John’s reported percentage belongs to the current exact-length mask.
        The MeshCore plugin exposes the current candidate (or candidate range),
        which lets us calculate the percentage for that body length directly.
        """
        if not self._phase.startswith("Testing channel names"):
            return None
        current_key = getattr(self, "_current_key", None)
        if not current_key:
            return 0.0
        candidate = str(current_key).rsplit("..", 1)[-1]
        if not re.fullmatch(r"#[a-z0-9-]+", candidate):
            return 0.0
        alphabet = getattr(self, "channel_alphabet", channel_alphabet())
        indexes = {character: index for index, character in enumerate(alphabet)}
        body = candidate[1:]
        if not body or any(character not in indexes for character in body):
            return 0.0
        rank = 0
        for character in body:
            rank = rank * len(alphabet) + indexes[character]
        total = len(alphabet) ** len(body)
        return min(100.0, (rank + 1) * 100.0 / total)

    @classmethod
    def parse_show_output(cls, output: str) -> list[str]:
        """Extract channel candidates from John ``--show`` output."""
        candidates: list[str] = []
        for line in output.splitlines():
            _hash, separator, candidate = line.strip().partition(":")
            if not separator:
                continue
            candidate = candidate.strip()
            if not candidate.startswith("#") or "\n" in candidate or len(candidate) > 64:
                continue
            if candidate not in candidates:
                candidates.append(candidate)
        return candidates

    def _run(self) -> None:
        packet_file: Path | None = None
        try:
            packet_file = self._write_packet_file()
            self._temporary_packet_file = packet_file
            if self._pause_requested.is_set():
                with self._lock:
                    self._state = "paused"
                    self._phase = "Paused"
                self._notify()
                return
            command = [
                str(self.script_path),
                "crack",
                str(packet_file),
                str(self._start_body_length),
                str(self.max_body_length),
            ]
            environment = os.environ.copy()
            environment["SESSION_PREFIX"] = self._session_prefix
            environment["MESCORE_CHANNEL_ALPHABET"] = self.channel_alphabet
            with self._lock:
                self._state = "running"
                self._phase = "Starting John the Ripper"
                self._phase_started_at = time.monotonic()
                self._process = self._popen(
                    command,
                    cwd=str(self.script_path.parent.parent),
                    env=environment,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    text=True,
                    bufsize=1,
                    start_new_session=True,
                )
                process = self._process

            reader = threading.Thread(target=self._read_output, args=(process,), daemon=True)
            reader.start()
            self._poll_thread = threading.Thread(target=self._poll_status, daemon=True)
            self._poll_thread.start()
            return_code = process.wait()
            reader.join(timeout=2)
            self._stop_requested.set()
            if self._poll_thread is not None:
                self._poll_thread.join(timeout=2)

            if self._pause_requested.is_set():
                with self._lock:
                    self._state = "paused"
                    self._phase = "Paused"
                self._notify()
                return
            if return_code != 0:
                with self._lock:
                    detail = next(
                        (line for line in reversed(self._output) if line.lower().startswith("error:")),
                        None,
                    )
                raise JohnWrapperError(
                    detail.removeprefix("error:").strip()
                    if detail
                    else f"John decode exited with status {return_code}"
                )

            with self._lock:
                self._phase = "Reading recovered channel names"
            show_output = self._show_results(packet_file, environment)
            candidates = self.parse_show_output(show_output)
            with self._lock:
                self._recovered_candidates = candidates
                self._current_key = None
                self._state = "complete"
                self._phase = "Complete"
            self._notify()
        except Exception as exc:  # worker errors are reported through status()
            with self._lock:
                self._state = "paused" if self._pause_requested.is_set() else "failed"
                self._phase = "Paused" if self._pause_requested.is_set() else "Failed"
                self._error = str(exc)
            self._notify()
        finally:
            with self._lock:
                self._finished_at = time.monotonic()
                self._process = None
            if packet_file is not None and self._packet_file is None:
                packet_file.unlink(missing_ok=True)

    def _write_packet_file(self) -> Path:
        if self._packet_file is not None and self._packet_file.exists():
            return self._packet_file
        if self._packet_file is None:
            handle = tempfile.NamedTemporaryFile(
                mode="w",
                encoding="ascii",
                prefix="meshcore-john-",
                suffix=".txt",
                delete=False,
            )
            path = Path(handle.name)
        else:
            self._packet_file.parent.mkdir(parents=True, exist_ok=True)
            handle = self._packet_file.open("w", encoding="ascii")
            path = self._packet_file
        try:
            with handle:
                for samples in self.packet_samples.values():
                    for packet in samples:
                        handle.write(packet.hex().upper() + "\n")
        except Exception:
            path.unlink(missing_ok=True)
            raise
        return path

    def _read_output(self, process: subprocess.Popen[str]) -> None:
        if process.stdout is None:
            return
        for line in process.stdout:
            self._record_output(line.strip())

    def _record_output(self, line: str) -> None:
        if not line:
            return
        line = self.ANSI_ESCAPE_RE.sub("", line).strip()
        if not line:
            return
        with self._lock:
            self._output.append(line)
            self._output = self._output[-self.MAX_OUTPUT_LINES:]
            body_match = self.BODY_LENGTH_RE.search(line)
            if body_match:
                body_length = body_match.group(1)
                self._current_body_length = int(body_length)
                self._phase = f"Testing channel names with {int(body_length) + 1} characters"
                self._current_session = f"{self._session_prefix}-{body_length}"
                self._current_key = None
                self._current_progress_percent = 0.0
                self._phase_started_at = time.monotonic()
            key_match = self.CURRENT_KEY_RE.search(line)
            if key_match:
                current_key = key_match.group(1)
                self._current_key = current_key
                # Prefer the exact candidate length in the phase display.
                # Range renderings (for example ``#aa..#-a``) are not
                # concrete candidates.
                if ".." not in current_key:
                    body_length = max(1, len(current_key) - 1)
                else:
                    range_key_match = self.CURRENT_KEY_RANGE_RE.fullmatch(current_key)
                    body_length = None
                    if range_key_match:
                        first_key, last_key = range_key_match.groups()
                        if len(first_key) == len(last_key):
                            body_length = max(1, len(first_key) - 1)
                if body_length is not None:
                    if body_length != self._current_body_length:
                        self._current_progress_percent = 0.0
                        self._phase_started_at = time.monotonic()
                    self._current_body_length = body_length
                    self._phase = f"Testing channel names with {body_length + 1} characters"
            rate_matches = self.HASH_RATE_RE.findall(line)
            if rate_matches:
                value, unit = rate_matches[-1]
                multiplier = {"": 1, "K": 1_000, "M": 1_000_000, "G": 1_000_000_000}[unit.upper()]
                self._current_hash_rate = float(value) * multiplier
            progress_matches = self.PROGRESS_PERCENT_RE.findall(line)
            if progress_matches:
                reported_progress = min(
                    100.0, max(0.0, float(progress_matches[-1]))
                )
                current_progress = getattr(self, "_current_progress_percent", None)
                # John status snapshots can lag behind the live process output.
                # Do not let a stale snapshot move the displayed progress
                # backwards within the active body-length phase.
                if current_progress is None or reported_progress >= current_progress:
                    self._current_progress_percent = reported_progress
        self._notify()

    def _poll_status(self) -> None:
        while not self._stop_requested.wait(self.status_interval):
            with self._lock:
                session = self._current_session
                process = self._process
            if process is None or process.poll() is not None or session is None:
                continue
            try:
                result = self._run_command(
                    [self._john_binary(), f"--status={session}"],
                    cwd=str(self._john_run_directory()),
                    capture_output=True,
                    text=True,
                    timeout=max(0.5, self.status_interval),
                    check=False,
                )
            except (OSError, subprocess.TimeoutExpired):
                continue
            status_output = "\n".join(part for part in (result.stdout, result.stderr) if part)
            for line in status_output.splitlines():
                self._record_output(line.strip())
            self.refresh_recovered_candidates()

    def refresh_recovered_candidates(self) -> None:
        with self._lock:
            packet_file = self._packet_file or self._temporary_packet_file
        if packet_file is None or not packet_file.exists():
            return
        try:
            show_output = self._show_results(packet_file, os.environ.copy())
        except (OSError, subprocess.TimeoutExpired, JohnWrapperError):
            return
        candidates = self.parse_show_output(show_output)
        if not candidates:
            return
        with self._lock:
            merged = list(self._recovered_candidates)
            for candidate in candidates:
                if candidate not in merged:
                    merged.append(candidate)
            changed = merged != self._recovered_candidates
            self._recovered_candidates = merged
        if changed:
            self._notify()

    def _show_results(self, packet_file: Path, environment: dict[str, str]) -> str:
        result = self._run_command(
            [str(self.script_path), "show", str(packet_file)],
            cwd=str(self.script_path.parent.parent),
            env=environment,
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
        if result.returncode != 0:
            raise JohnWrapperError(
                "John could not read recovered names: "
                + (result.stderr or result.stdout or f"status {result.returncode}").strip()
            )
        return result.stdout

    def _john_run_directory(self) -> Path:
        john_dir = Path(os.environ.get("JOHN_DIR", self.script_path.parent / ".john-jumbo"))
        return john_dir / "run"

    def _john_binary(self) -> str:
        return str(self._john_run_directory() / "john")


class JohnVanityJob:
    """Generate one Ed25519 identity matching public-key prefix and suffix."""

    MAX_OUTPUT_LINES = 40

    def __init__(
        self,
        script_path: Path,
        public_key_prefix: str,
        counter_digits: int = 16,
        public_key_suffix: str = "",
        popen: Callable[..., subprocess.Popen[str]] = subprocess.Popen,
        on_update: Callable[[dict[str, object]], None] | None = None,
    ) -> None:
        prefix = public_key_prefix.strip().lower()
        suffix = public_key_suffix.strip().lower()
        if not re.fullmatch(r"[0-9a-f]{1,16}", prefix):
            raise JohnWrapperError("vanity prefix must contain 1 to 16 hexadecimal characters")
        if not re.fullmatch(r"[0-9a-f]{0,16}", suffix):
            raise JohnWrapperError("vanity suffix must contain 0 to 16 hexadecimal characters")
        if len(prefix) + len(suffix) > 64:
            raise JohnWrapperError("vanity prefix and suffix cannot exceed 64 hexadecimal characters")
        if not 1 <= counter_digits <= 16:
            raise JohnWrapperError("vanity counter length must be between 1 and 16")
        if not script_path.is_file():
            raise JohnWrapperError(f"John wrapper script not found: {script_path}")

        self.script_path = script_path
        self.public_key_prefix = prefix
        self.public_key_suffix = suffix
        self.counter_digits = counter_digits
        self._popen = popen
        self._on_update = on_update
        self._lock = threading.RLock()
        self._process: subprocess.Popen[str] | None = None
        self._thread: threading.Thread | None = None
        self._started_at = 0.0
        self._finished_at: float | None = None
        self._stop_requested = threading.Event()
        self._state = "queued"
        self._phase = "Waiting to generate vanity key"
        self._output: list[str] = []
        self._seed: str | None = None
        self._private_key: str | None = None
        self._generated_public_key: str | None = None
        self._error: str | None = None

    def start(self) -> None:
        with self._lock:
            if self._thread is not None:
                raise JohnWrapperError("vanity key job has already started")
            self._started_at = time.monotonic()
            self._thread = threading.Thread(target=self._run, daemon=True)
            self._thread.start()

    def stop(self) -> None:
        self._stop_requested.set()
        with self._lock:
            process = self._process
        if process is not None and process.poll() is None:
            try:
                os.killpg(process.pid, signal.SIGTERM)
            except (AttributeError, OSError):
                process.terminate()

    def status(self) -> dict[str, object]:
        with self._lock:
            now = self._finished_at or time.monotonic()
            elapsed = max(0.0, now - self._started_at) if self._started_at else 0.0
            return {
                "running": self._state in {"queued", "running"},
                "state": self._state,
                "phase": self._phase,
                "public_key_prefix": self.public_key_prefix,
                "public_key_suffix": self.public_key_suffix,
                "counter_digits": self.counter_digits,
                "elapsed_seconds": round(elapsed, 1),
                "seed": self._seed,
                "private_key": self._private_key,
                "public_key": self._generated_public_key,
                "error": self._error,
                "output": list(self._output[-self.MAX_OUTPUT_LINES:]),
            }

    def _notify(self) -> None:
        if self._on_update is not None:
            try:
                self._on_update(self.status())
            except Exception:
                pass

    def _run(self) -> None:
        environment = os.environ.copy()
        command = [
            str(self.script_path),
            "vanity",
            f"{self.public_key_prefix}|{self.public_key_suffix}",
            str(self.counter_digits),
        ]
        try:
            with self._lock:
                self._state = "running"
                self._phase = "Generating vanity key with John the Ripper"
                self._process = self._popen(
                    command,
                    cwd=str(self.script_path.parent.parent),
                    env=environment,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    text=True,
                    bufsize=1,
                    start_new_session=True,
                )
                process = self._process
                if self._stop_requested.is_set() and process.poll() is None:
                    try:
                        os.killpg(process.pid, signal.SIGTERM)
                    except (AttributeError, OSError):
                        process.terminate()
            if process.stdout is not None:
                for line in process.stdout:
                    self._record_output(line.strip())
            return_code = process.wait()
            if self._stop_requested.is_set():
                with self._lock:
                    self._state = "cancelled"
                    self._phase = "Generation cancelled"
                self._notify()
                return
            if return_code != 0:
                detail = next(
                    (line for line in reversed(self._output) if line.lower().startswith("error:")),
                    None,
                )
                raise JohnWrapperError(
                    detail.removeprefix("error:").strip()
                    if detail
                    else f"John vanity generation exited with status {return_code}"
                )
            with self._lock:
                if not self._seed or not self._private_key or not self._generated_public_key:
                    raise JohnWrapperError("John completed without returning a complete key")
                self._state = "complete"
                self._phase = "Key generated"
            self._notify()
        except Exception as exc:
            with self._lock:
                self._state = "failed"
                self._phase = "Generation failed"
                self._error = str(exc)
            self._notify()
        finally:
            with self._lock:
                self._finished_at = time.monotonic()
                self._process = None

    def _record_output(self, line: str) -> None:
        if not line:
            return
        with self._lock:
            self._output.append(line)
            self._output = self._output[-self.MAX_OUTPUT_LINES:]
            key, separator, value = line.partition("=")
            if not separator:
                return
            value = value.strip().lower()
            if key == "seed":
                self._seed = value
            elif key == "private":
                self._private_key = value
            elif key == "public":
                self._generated_public_key = value.upper()
        self._notify()


class JohnDecodeManager:
    """Serialize decode jobs and apply validated candidates to the dashboard."""

    # The candidate numbering depends on the selected alphabet. Persisted John
    # sessions cannot safely resume across a changed search configuration.
    SEARCH_VERSION = 2

    def __init__(
        self,
        decoder,
        state,
        script_path: Path,
        max_body_length: int = 29,
        job_factory=JohnDecodeJob,
        job_file: Path | None = None,
        coordinator: JohnProcessCoordinator | None = None,
    ) -> None:
        self.decoder = decoder
        self.state = state
        self.script_path = script_path
        self.max_body_length = max_body_length
        self.job_factory = job_factory
        self.job_file = job_file or (state.capture_path.parent / "john-decode-job.json")
        self.coordinator = coordinator
        self._lock = threading.RLock()
        self._job: JohnDecodeJob | None = None
        self._lease: str | None = None
        self._apply_thread: threading.Thread | None = None
        self._applied_candidates: set[str] = set()
        self._metadata: dict[str, object] | None = self._load_metadata()

    def _load_metadata(self) -> dict[str, object] | None:
        if not self.job_file.exists():
            return None
        try:
            metadata = json.loads(self.job_file.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            return None
        if not isinstance(metadata, dict) or not isinstance(metadata.get("selected_hmac_ids"), list):
            return None
        if metadata.get("state") == "running":
            metadata["state"] = "paused"
            metadata["phase"] = "Paused after application restart"
            metadata["running"] = False
            self._write_metadata(metadata)
        return metadata

    def recover_completed_job(self) -> None:
        """Repair a completed job whose recovered names were not persisted."""
        with self._lock:
            metadata = self._metadata
            if self._job is not None or metadata is None or metadata.get("state") != "complete":
                return
        recovered = metadata.get("recovered_channels")
        if isinstance(recovered, list) and recovered:
            self._apply_completed_metadata(metadata)
            return

        packet_file = Path(str(metadata.get("packet_file", "")))
        if not packet_file.is_file():
            return
        try:
            result = subprocess.run(
                [str(self.script_path), "show", str(packet_file)],
                cwd=str(self.script_path.parent.parent),
                env=os.environ.copy(),
                capture_output=True,
                text=True,
                timeout=30,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired):
            return
        if result.returncode != 0:
            return
        candidates = JohnDecodeJob.parse_show_output(result.stdout)
        if not candidates:
            return
        with self._lock:
            if self._metadata is None or self._metadata.get("state") != "complete":
                return
            self._metadata["recovered_channels"] = candidates
            self._metadata["current_key"] = None
            self._write_metadata(self._metadata)
            metadata = self._metadata
        self._apply_completed_metadata(metadata)

    def _apply_completed_metadata(self, metadata: dict[str, object]) -> None:
        try:
            packet_samples = self._packet_samples_from_metadata(metadata)
        except JohnWrapperError:
            return
        recovered = metadata.get("recovered_channels", [])
        if not isinstance(recovered, list):
            return
        for candidate in recovered:
            self._apply_candidate(str(candidate), packet_samples)

    def _write_metadata(self, metadata: dict[str, object]) -> None:
        self.job_file.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.job_file.with_suffix(".tmp")
        temporary.write_text(json.dumps(metadata, separators=(",", ":")) + "\n", encoding="utf-8")
        temporary.replace(self.job_file)

    @staticmethod
    def _metadata_status(metadata: dict[str, object]) -> dict[str, object]:
        selected_character_sets = normalize_channel_character_sets(
            metadata.get("selected_character_sets")
        )
        return {
            "running": metadata.get("state") == "running",
            "state": metadata.get("state", "idle"),
            "selected_hmac_ids": [f"0x{int(value):02X}" for value in metadata.get("selected_hmac_ids", [])],
            "selected_character_sets": list(selected_character_sets),
            "selected_count": int(metadata.get("selected_count", 0)),
            "phase": metadata.get("phase", "Idle"),
            "current_key": metadata.get("current_key"),
            "current_hash_rate": metadata.get("current_hash_rate"),
            "current_progress_percent": metadata.get("current_progress_percent"),
            "phase_progress_percent": metadata.get("phase_progress_percent"),
            "elapsed_seconds": float(metadata.get("elapsed_seconds", 0.0)),
            "phase_elapsed_seconds": float(metadata.get("phase_elapsed_seconds", 0.0)),
            "recovered_channels": list(metadata.get("recovered_channels", [])),
            "error": metadata.get("error"),
            "output": list(metadata.get("output", [])),
        }

    def _packet_samples_from_metadata(self, metadata: dict[str, object]) -> dict[int, list[bytes]]:
        packet_path = Path(str(metadata.get("packet_file", "")))
        if not packet_path.is_file():
            raise JohnWrapperError("saved John packet samples are unavailable; start a new decode")
        samples: dict[int, list[bytes]] = {}
        for line in packet_path.read_text(encoding="ascii").splitlines():
            try:
                packet = bytes.fromhex(line.strip())
            except ValueError:
                continue
            if len(packet) < 3:
                continue
            group_id = self._packet_hmac_id(packet)
            if group_id is None:
                continue
            samples.setdefault(group_id, []).append(packet)
        selected = [int(value) for value in metadata["selected_hmac_ids"]]
        restored = {value: samples[value] for value in selected if value in samples}
        missing = [value for value in selected if value not in restored]
        if missing:
            labels = ", ".join(f"0x{value:02X}" for value in missing)
            raise JohnWrapperError(f"saved John samples are incomplete: {labels}")
        return restored

    @staticmethod
    def _write_packet_samples(packet_file: Path, packet_samples: dict[int, list[bytes]]) -> None:
        packet_file.parent.mkdir(parents=True, exist_ok=True)
        packet_file.write_text(
            "".join(
                f"{packet.hex().upper()}\n"
                for hmac_id in sorted(packet_samples)
                for packet in packet_samples[hmac_id]
            ),
            encoding="ascii",
        )

    @staticmethod
    def _packet_hmac_id(packet: bytes) -> int | None:
        if len(packet) < 2:
            return None
        route_type = packet[0] & 0x03
        offset = 1 + (4 if route_type in (0x00, 0x03) else 0)
        if len(packet) <= offset:
            return None
        path_length = packet[offset]
        offset += 1 + (path_length & 0x3F) * ((path_length >> 6) + 1)
        return packet[offset] if len(packet) > offset else None

    def _job_updated(self, status: dict[str, object]) -> None:
        candidates_to_apply: list[str] = []
        packet_samples: dict[int, list[bytes]] | None = None
        with self._lock:
            if self._metadata is None:
                return
            self._metadata.update({
                "running": status["running"],
                "state": status["state"],
                "phase": status["phase"],
                "current_key": status["current_key"],
                "current_hash_rate": status.get("current_hash_rate", self._metadata.get("current_hash_rate")),
                "current_progress_percent": status.get(
                    "current_progress_percent", self._metadata.get("current_progress_percent")
                ),
                "phase_progress_percent": status.get(
                    "phase_progress_percent", self._metadata.get("phase_progress_percent")
                ),
                "elapsed_seconds": status["elapsed_seconds"],
                "phase_elapsed_seconds": status.get(
                    "phase_elapsed_seconds", self._metadata.get("phase_elapsed_seconds", 0.0)
                ),
                "recovered_channels": status["recovered_channels"],
                "error": status["error"],
                "output": status["output"],
                "current_body_length": status.get("current_body_length", self._metadata.get("current_body_length", 1)),
            })
            self._write_metadata(self._metadata)
            job = self._job
            if job is not None and isinstance(status.get("recovered_channels"), list):
                packet_samples = job.packet_samples
                for candidate in status["recovered_channels"]:
                    candidate = str(candidate)
                    if candidate not in self._applied_candidates:
                        self._applied_candidates.add(candidate)
                        candidates_to_apply.append(candidate)
        if packet_samples is not None:
            for candidate in candidates_to_apply:
                self._apply_candidate(candidate, packet_samples)
        if not status["running"]:
            self._release_lease()

    def _release_lease(self) -> None:
        with self._lock:
            lease = self._lease
            self._lease = None
        if lease is not None and self.coordinator is not None:
            self.coordinator.release(lease)

    def start(
        self,
        hmac_ids: Iterable[int],
        character_sets: Iterable[str] | None = None,
    ) -> dict[str, object]:
        selected = sorted(set(hmac_ids))
        if not selected or any(not 0 <= value <= 255 for value in selected):
            raise JohnWrapperError("select at least one valid unknown-group ID")
        selected_character_sets = normalize_channel_character_sets(character_sets)
        with self._lock:
            if self._job is not None and self._job.status()["running"]:
                raise JohnWrapperError("a John decode is already running")
            resuming = (
                self._metadata is not None
                and self._metadata.get("version") == self.SEARCH_VERSION
                and self._metadata.get("state") == "paused"
                and [int(value) for value in self._metadata.get("selected_hmac_ids", [])] == selected
                and normalize_channel_character_sets(self._metadata.get("selected_character_sets"))
                == selected_character_sets
            )
            if resuming:
                packet_samples = self._packet_samples_from_metadata(self._metadata)
                session_prefix = str(self._metadata["session_prefix"])
                packet_file = Path(str(self._metadata["packet_file"]))
                current_samples = self.state.unknown_channel_samples(selected)
                if (
                    all(value in current_samples for value in selected)
                    and sum(len(samples) for samples in current_samples.values())
                    >= sum(len(samples) for samples in packet_samples.values())
                ):
                    packet_samples = current_samples
                    self._write_packet_samples(packet_file, packet_samples)
                start_body_length = int(self._metadata.get("current_body_length", 1))
            else:
                packet_samples = self.state.unknown_channel_samples(selected)
                missing = [value for value in selected if value not in packet_samples]
                if missing:
                    labels = ", ".join(f"0x{value:02X}" for value in missing)
                    raise JohnWrapperError(f"unknown-group samples are no longer available: {labels}")
                session_prefix = f"meshcore-decode-{uuid.uuid4().hex}"
                packet_file = self.job_file.parent / "john" / f"{session_prefix}.packets.txt"
                start_body_length = 1
            lease = None
            if self.coordinator is not None:
                lease = self.coordinator.acquire(
                    "channel_name_decode",
                    lambda: self.pause(),
                )
            if not resuming:
                self._discard_metadata()
            self._metadata = {
                "version": self.SEARCH_VERSION,
                "running": True,
                "state": "running",
                "phase": "Preparing packet samples",
                "selected_hmac_ids": selected,
                "selected_character_sets": list(selected_character_sets),
                "selected_count": sum(len(samples) for samples in packet_samples.values()),
                "current_key": None,
                "current_hash_rate": None,
                "current_progress_percent": None,
                "phase_progress_percent": 0.0,
                "elapsed_seconds": 0.0,
                "phase_elapsed_seconds": 0.0,
                "recovered_channels": [],
                "error": None,
                "output": [],
                "session_prefix": session_prefix,
                "packet_file": str(packet_file),
                "max_body_length": self.max_body_length,
                "current_body_length": start_body_length,
            }
            self._applied_candidates = set()
            self._lease = lease
            try:
                self._write_metadata(self._metadata)
                self._job = self.job_factory(
                    self.script_path,
                    packet_samples,
                    max_body_length=self.max_body_length,
                    character_sets=selected_character_sets,
                    packet_file=packet_file,
                    session_prefix=session_prefix,
                    start_body_length=start_body_length,
                    on_update=self._job_updated,
                )
                self._job.start()
            except Exception:
                self._release_lease()
                raise
            refresh_recovered = getattr(self._job, "refresh_recovered_candidates", None)
            if refresh_recovered is not None:
                refresh_recovered()
            self._apply_thread = threading.Thread(target=self._apply_when_done, daemon=True)
            self._apply_thread.start()
            return self._job.status()

    def status(self) -> dict[str, object]:
        with self._lock:
            if self._job is None and self._metadata is not None:
                return self._metadata_status(self._metadata)
            if self._job is None:
                return {
                    "running": False,
                    "state": "idle",
                    "selected_hmac_ids": [],
                    "selected_character_sets": list(CHANNEL_CHARACTER_SETS),
                    "selected_count": 0,
                    "phase": "Idle",
                    "current_key": None,
                    "current_hash_rate": None,
                    "current_progress_percent": None,
                    "phase_progress_percent": None,
                    "elapsed_seconds": 0.0,
                    "phase_elapsed_seconds": 0.0,
                    "recovered_channels": [],
                    "error": None,
                    "output": [],
                }
            return self._job.status()

    def stop(self) -> None:
        with self._lock:
            job = self._job
        if job is not None:
            job.pause()
        elif self._metadata is not None and self._metadata.get("state") == "running":
            self._metadata["state"] = "paused"
            self._metadata["running"] = False
            self._metadata["phase"] = "Paused after application shutdown"
            self._write_metadata(self._metadata)

    def pause(self) -> dict[str, object]:
        with self._lock:
            job = self._job
        if job is not None and job.status()["running"]:
            job.pause()
        return self.status()

    def _discard_metadata(self) -> None:
        if self._metadata is not None:
            packet_file = Path(str(self._metadata.get("packet_file", "")))
            packet_file.unlink(missing_ok=True)
        self._metadata = None

    def _apply_when_done(self) -> None:
        with self._lock:
            job = self._job
        if job is None:
            return
        while job.status()["running"]:
            time.sleep(0.1)
        job_status = job.status()
        self._release_lease()
        if job_status["state"] != "complete":
            return
        candidates = job_status["recovered_channels"]
        if not isinstance(candidates, list):
            return
        for candidate in candidates:
            self._apply_candidate(str(candidate), job.packet_samples)
        with self._lock:
            if self._metadata is not None:
                self._metadata["state"] = "complete"
                self._metadata["running"] = False
                self._metadata["phase"] = "Complete"
                self._write_metadata(self._metadata)

    def _apply_candidate(self, candidate: str, packet_samples: dict[int, list[bytes]]) -> None:
        try:
            channel = self.decoder.normalize_channel(candidate)
        except ValueError:
            return
        secret = hashlib.sha256(channel.encode("utf-8")).digest()[:16]
        channel_hash = hashlib.sha256(secret).digest()[0]
        for hmac_id, samples in packet_samples.items():
            if channel_hash != hmac_id:
                continue
            already_known = channel in self.decoder.known_channels()
            if not already_known:
                try:
                    self.decoder.add_known_channel(channel)
                except (OSError, ValueError):
                    continue
            valid = any(
                (decoded := self.decoder.decode(packet)) is not None
                and decoded.get("channel") == channel
                for packet in samples
            )
            if valid:
                self.state.backfill_decoded_messages(self.decoder, {channel})
                self.state.remove_unknown_channel(hmac_id)
                continue
            if not already_known:
                try:
                    self.decoder.remove_known_channel(channel)
                except (KeyError, OSError, ValueError):
                    pass