"""Tests for Phase 4.4 — generic pre-merge hook framework + Knowtation hook.

Covers all 7 Rule-#0 tiers:

1. **Unit**           — Frozen dataclass, port validation, no-op path,
                        registry lookup, registration helper.
2. **Integration**    — Registry round-trip, HTTP success, HTTP failure
                        captured into ``error`` field, multi-domain
                        isolation.
3. **End-to-end**     — ``muse merge --consolidate`` on a knowtation
                        repo invokes the hook (mock); without the flag
                        the hook is silent.
4. **Stress**         — 100 sequential no-op invocations under 1 s;
                        10-thread concurrent invocations do not deadlock.
5. **Data-integrity** — Hook outputs are deterministic; isolated
                        registry instances do not leak state.
6. **Performance**    — 1 000 ``get_hook_registry()`` calls in < 100 ms.
7. **Security**       — Port-injection rejection (``"3000; rm -rf /"``);
                        out-of-range rejection (``0`` and ``65536``);
                        background-thread exceptions never propagate.
"""

from __future__ import annotations

import dataclasses
import datetime
import http.server
import json
import logging
import os
import pathlib
import socket
import threading
import time
import uuid
from typing import Any, Generator
from unittest import mock

import pytest

from muse.core.merge_hooks import (
    MergeHookPlugin,
    MergeHookRegistry,
    MergeHookResult,
    get_hook_registry,
)
from muse.plugins.knowtation.hooks import (
    API_PATH,
    DEFAULT_PORT,
    ENV_PORT,
    HOOK_NAME,
    HTTP_TIMEOUT_SECONDS,
    PORT_MAX,
    PORT_MIN,
    KnowtationMergeHook,
    _resolve_port,
    register_knowtation_hook,
)
from tests.cli_test_helper import CliRunner

# ────────────────────────────────────────────────────────────────────────────
# Constants and shared fixtures
# ────────────────────────────────────────────────────────────────────────────

_DUMMY_ROOT = pathlib.Path("/tmp/knowtation-merge-hooks-test-root")
_OURS_ID = "a" * 64
_THEIRS_ID = "b" * 64
_BRANCH = "feat/test"


@pytest.fixture
def fresh_registry() -> MergeHookRegistry:
    """Return a brand-new :class:`MergeHookRegistry` for isolated tests.

    Tests that need to avoid touching the process-wide singleton use
    this fixture to guarantee no state bleeds between cases.
    """
    return MergeHookRegistry()


@pytest.fixture
def clean_env_port() -> Generator[None, None, None]:
    """Remove ``KNOWTATION_HUB_PORT`` for the test, restore afterwards."""
    original = os.environ.pop(ENV_PORT, None)
    try:
        yield
    finally:
        if original is None:
            os.environ.pop(ENV_PORT, None)
        else:
            os.environ[ENV_PORT] = original


# ────────────────────────────────────────────────────────────────────────────
# Tier 1 — Unit
# ────────────────────────────────────────────────────────────────────────────


class TestMergeHookResultUnit:
    """Frozen-dataclass invariants and field shape."""

    def test_result_is_frozen_dataclass(self) -> None:
        """Setattr must raise FrozenInstanceError per @dataclass(frozen=True)."""
        result = MergeHookResult(
            hook_name="x", fired=True, out_of_band=True,
            message="ok", error=None,
        )
        with pytest.raises(dataclasses.FrozenInstanceError):
            result.fired = False  # type: ignore[misc]

    def test_result_field_types_and_values(self) -> None:
        """Construction requires every field; types match declared annotations."""
        result = MergeHookResult(
            hook_name="knowtation-consolidate",
            fired=False,
            out_of_band=False,
            message="msg",
            error=None,
        )
        assert result.hook_name == "knowtation-consolidate"
        assert result.fired is False
        assert result.out_of_band is False
        assert result.message == "msg"
        assert result.error is None

    def test_result_supports_error_string(self) -> None:
        """`error` accepts a string for diagnostic information."""
        result = MergeHookResult(
            hook_name="x", fired=False, out_of_band=False,
            message="failed", error="boom",
        )
        assert result.error == "boom"

    def test_result_equality_is_value_based(self) -> None:
        """Two MergeHookResult with identical fields must compare equal."""
        a = MergeHookResult("x", True, True, "msg", None)
        b = MergeHookResult("x", True, True, "msg", None)
        assert a == b


class TestKnowtationMergeHookUnit:
    """Behaviour of the Knowtation-specific hook in isolation."""

    def test_consolidate_false_returns_noop_result(
        self, clean_env_port: None
    ) -> None:
        """When --consolidate is not requested, the hook fires no I/O."""
        hook = KnowtationMergeHook()
        result = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=False,
        )
        assert result.hook_name == HOOK_NAME
        assert result.fired is False
        assert result.out_of_band is False
        assert result.error is None
        assert "not requested" in result.message

    def test_port_validation_rejects_zero(
        self, clean_env_port: None
    ) -> None:
        """Port 0 is reserved (IANA) and must be rejected."""
        os.environ[ENV_PORT] = "0"
        hook = KnowtationMergeHook()
        result = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        assert result.fired is False
        assert result.error is not None
        assert "out of range" in result.error.lower()

    def test_port_validation_rejects_above_max(
        self, clean_env_port: None
    ) -> None:
        """Port 99999 is above the max TCP port (65535)."""
        os.environ[ENV_PORT] = "99999"
        hook = KnowtationMergeHook()
        result = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        assert result.fired is False
        assert result.error is not None
        assert "out of range" in result.error.lower()

    def test_port_validation_accepts_default(
        self, clean_env_port: None
    ) -> None:
        """No env var → default port (3000)."""
        port, err = _resolve_port()
        assert err is None
        assert port == DEFAULT_PORT

    def test_port_validation_accepts_explicit_3000(
        self, clean_env_port: None
    ) -> None:
        """Explicit 3000 round-trips cleanly."""
        os.environ[ENV_PORT] = "3000"
        port, err = _resolve_port()
        assert err is None
        assert port == 3000

    def test_port_min_and_max_constants_are_correct(self) -> None:
        """Sanity guard for the port-range literals."""
        assert PORT_MIN == 1
        assert PORT_MAX == 65535


class TestRegistrySingletonUnit:
    """Singleton + helper-function behaviour."""

    def test_get_hook_registry_returns_singleton(self) -> None:
        """Subsequent calls must return the same instance."""
        a = get_hook_registry()
        b = get_hook_registry()
        assert a is b

    def test_get_returns_none_for_unregistered_domain(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        """Unknown domains resolve to None, never raising."""
        assert fresh_registry.get("nonexistent-domain-xyz") is None

    def test_register_knowtation_hook_registers_under_knowtation(self) -> None:
        """Helper must register the hook under the canonical key."""
        register_knowtation_hook()
        plugin = get_hook_registry().get("knowtation")
        assert plugin is not None
        assert isinstance(plugin, KnowtationMergeHook)

    def test_register_rejects_empty_domain(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        """Empty domain string must raise ValueError."""
        with pytest.raises(ValueError):
            fresh_registry.register("", KnowtationMergeHook())

    def test_protocol_is_runtime_checkable(self) -> None:
        """KnowtationMergeHook must satisfy MergeHookPlugin via isinstance."""
        assert isinstance(KnowtationMergeHook(), MergeHookPlugin)


# ────────────────────────────────────────────────────────────────────────────
# Tier 2 — Integration
# ────────────────────────────────────────────────────────────────────────────


class _MockHubServer:
    """Tiny in-process HTTP server that captures POSTed bodies for tests.

    Listens on localhost on an OS-assigned free port; tests pass that
    port via ``KNOWTATION_HUB_PORT`` and then assert that the hook
    actually delivered the request.
    """

    def __init__(self) -> None:
        self.received: list[dict[str, Any]] = []
        self._lock = threading.Lock()
        recv_lock = self._lock
        recv_sink = self.received

        class _Handler(http.server.BaseHTTPRequestHandler):
            def do_POST(self) -> None:  # noqa: N802 — http.server convention
                length = int(self.headers.get("Content-Length", "0") or "0")
                body = self.rfile.read(length) if length else b""
                with recv_lock:
                    recv_sink.append({
                        "path": self.path,
                        "method": self.command,
                        "body": body,
                        "content_type": self.headers.get("Content-Type", ""),
                    })
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(b'{"ok":true}')

            def log_message(self, *args: Any, **kwargs: Any) -> None:
                """Silence access logs during tests."""
                return

        self._handler_cls = _Handler
        self._server = http.server.HTTPServer(("127.0.0.1", 0), self._handler_cls)
        self.port = self._server.server_address[1]
        self._thread = threading.Thread(
            target=self._server.serve_forever,
            name="mock-hub-server",
            daemon=True,
        )

    def start(self) -> None:
        self._thread.start()

    def stop(self) -> None:
        self._server.shutdown()
        self._server.server_close()
        self._thread.join(timeout=5.0)

    def wait_for_request(self, timeout: float = 5.0) -> bool:
        """Block until at least one request is captured, or *timeout*."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            with self._lock:
                if self.received:
                    return True
            time.sleep(0.01)
        return False


@pytest.fixture
def mock_hub_server() -> Generator[_MockHubServer, None, None]:
    """Start a one-shot in-process mock hub; tear down after the test."""
    server = _MockHubServer()
    server.start()
    try:
        yield server
    finally:
        server.stop()


def _free_tcp_port() -> int:
    """Return an OS-assigned free TCP port for test use."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


class TestRegistryIntegration:
    """End-to-end through the registry, with mocked HTTP where appropriate."""

    def test_run_pre_merge_consolidate_false_returns_noop(
        self, fresh_registry: MergeHookRegistry, clean_env_port: None
    ) -> None:
        fresh_registry.register("knowtation", KnowtationMergeHook())
        results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "knowtation", _OURS_ID, _THEIRS_ID, _BRANCH,
            consolidate=False,
        )
        assert len(results) == 1
        assert results[0].fired is False
        assert results[0].out_of_band is False
        assert results[0].error is None

    def test_run_pre_merge_no_plugin_returns_empty_list(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        assert fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "no-such-domain", _OURS_ID, _THEIRS_ID, _BRANCH,
            consolidate=True,
        ) == []

    def test_run_pre_merge_consolidate_true_dispatches_http(
        self,
        fresh_registry: MergeHookRegistry,
        mock_hub_server: _MockHubServer,
        clean_env_port: None,
    ) -> None:
        os.environ[ENV_PORT] = str(mock_hub_server.port)
        fresh_registry.register("knowtation", KnowtationMergeHook())
        results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "knowtation", _OURS_ID, _THEIRS_ID, _BRANCH,
            consolidate=True,
        )
        assert len(results) == 1
        assert results[0].fired is True
        assert results[0].out_of_band is True
        assert results[0].error is None
        assert mock_hub_server.wait_for_request(timeout=5.0)
        captured = mock_hub_server.received[0]
        assert captured["path"] == API_PATH
        assert captured["method"] == "POST"
        assert captured["body"] == b"{}"
        assert "application/json" in captured["content_type"]

    def test_run_pre_merge_http_failure_captured_in_error_via_log(
        self,
        fresh_registry: MergeHookRegistry,
        clean_env_port: None,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """Connection refused must NOT raise; the warning is logged."""
        port = _free_tcp_port()
        os.environ[ENV_PORT] = str(port)
        fresh_registry.register("knowtation", KnowtationMergeHook())
        with caplog.at_level(logging.WARNING, logger="muse.plugins.knowtation.hooks"):
            results = fresh_registry.run_pre_merge(
                _DUMMY_ROOT, "knowtation", _OURS_ID, _THEIRS_ID, _BRANCH,
                consolidate=True,
            )
        assert len(results) == 1
        assert results[0].fired is True
        assert results[0].out_of_band is True
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if any("knowtation-consolidate" in rec.message for rec in caplog.records):
                break
            time.sleep(0.05)
        assert any(
            "knowtation-consolidate" in rec.message and "fail" in rec.message.lower()
            for rec in caplog.records
        ), f"expected a WARNING log for connection failure; got: {[r.message for r in caplog.records]}"

    def test_run_pre_merge_plugin_raises_is_caught_into_error(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        """A buggy plugin that raises gets captured into a synthetic result."""

        class _BoomPlugin:
            def pre_merge_hook(
                self, root, ours_commit_id, theirs_commit_id, branch,
                *, consolidate: bool = False,
            ) -> MergeHookResult:
                raise RuntimeError("synthetic boom")

        fresh_registry.register("buggy", _BoomPlugin())  # type: ignore[arg-type]
        results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "buggy", _OURS_ID, _THEIRS_ID, _BRANCH,
            consolidate=True,
        )
        assert len(results) == 1
        assert results[0].fired is False
        assert results[0].error is not None
        assert "RuntimeError" in results[0].error
        assert "synthetic boom" in results[0].error

    def test_multiple_domains_register_independently(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        """Multi-domain registration: each plugin is reachable by name."""

        class _A:
            def pre_merge_hook(self, root, ours_commit_id, theirs_commit_id,
                               branch, *, consolidate: bool = False
                               ) -> MergeHookResult:
                return MergeHookResult("a", True, False, "a-msg", None)

        class _B:
            def pre_merge_hook(self, root, ours_commit_id, theirs_commit_id,
                               branch, *, consolidate: bool = False
                               ) -> MergeHookResult:
                return MergeHookResult("b", True, False, "b-msg", None)

        fresh_registry.register("alpha", _A())  # type: ignore[arg-type]
        fresh_registry.register("beta", _B())  # type: ignore[arg-type]

        a_results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "alpha", _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        b_results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "beta", _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        assert a_results[0].hook_name == "a"
        assert b_results[0].hook_name == "b"
        assert fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "gamma", _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        ) == []


# ────────────────────────────────────────────────────────────────────────────
# Tier 3 — End-to-end: muse merge --consolidate / without
# ────────────────────────────────────────────────────────────────────────────


def _init_knowtation_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
    """Create a minimal knowtation-domain repo on disk."""
    muse_dir = tmp_path / ".muse"
    muse_dir.mkdir()
    repo_id = str(uuid.uuid4())
    (muse_dir / "repo.json").write_text(
        json.dumps({
            "repo_id": repo_id,
            "domain": "knowtation",
            "default_branch": "main",
            "created_at": "2025-01-01T00:00:00+00:00",
        }),
        encoding="utf-8",
    )
    (muse_dir / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
    (muse_dir / "refs" / "heads").mkdir(parents=True)
    (muse_dir / "snapshots").mkdir()
    (muse_dir / "commits").mkdir()
    (muse_dir / "objects").mkdir()
    return tmp_path, repo_id


def _make_simple_commit(
    root: pathlib.Path, repo_id: str, branch: str, message: str,
    manifest: dict[str, str] | None = None,
) -> str:
    from muse.core.snapshot import compute_commit_id, compute_snapshot_id
    from muse.core.store import (
        CommitRecord, SnapshotRecord, write_commit, write_snapshot,
    )
    ref_file = root / ".muse" / "refs" / "heads" / branch
    parent_id = ref_file.read_text().strip() if ref_file.exists() else None
    m = manifest or {}
    snap_id = compute_snapshot_id(m)
    committed_at = datetime.datetime.now(datetime.timezone.utc)
    commit_id = compute_commit_id(
        parent_ids=[parent_id] if parent_id else [],
        snapshot_id=snap_id, message=message,
        committed_at_iso=committed_at.isoformat(),
    )
    write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m))
    write_commit(root, CommitRecord(
        commit_id=commit_id, repo_id=repo_id, branch=branch,
        snapshot_id=snap_id, message=message, committed_at=committed_at,
        parent_commit_id=parent_id,
    ))
    ref_file.parent.mkdir(parents=True, exist_ok=True)
    ref_file.write_text(commit_id, encoding="utf-8")
    return commit_id


class TestCliEndToEnd:
    """Drive ``muse merge --consolidate`` via the CLI runner."""

    def test_merge_with_consolidate_invokes_hook(
        self,
        tmp_path: pathlib.Path,
        mock_hub_server: _MockHubServer,
        clean_env_port: None,
    ) -> None:
        os.environ[ENV_PORT] = str(mock_hub_server.port)
        # Ensure the knowtation hook is registered on the singleton.
        register_knowtation_hook()
        root, repo_id = _init_knowtation_repo(tmp_path)
        base_id = _make_simple_commit(root, repo_id, "main", "base")
        (root / ".muse" / "refs" / "heads" / "feature").write_text(base_id)
        _make_simple_commit(root, repo_id, "feature", "feat")
        runner = CliRunner()
        env = {"MUSE_REPO_ROOT": str(root), ENV_PORT: str(mock_hub_server.port)}
        result = runner.invoke(
            None, ["merge", "--consolidate", "--format", "json", "feature"],
            env=env, catch_exceptions=False,
        )
        assert result.exit_code == 0, result.output
        data = json.loads(result.output)
        assert data["status"] in ("fast_forward", "up_to_date", "merged")
        hook_record = data.get("consolidation_hook")
        assert hook_record is not None, (
            f"expected consolidation_hook in JSON output, got keys: {list(data)}"
        )
        assert hook_record["hook_name"] == HOOK_NAME
        assert hook_record["fired"] is True
        assert hook_record["out_of_band"] is True
        assert hook_record["error"] is None
        assert mock_hub_server.wait_for_request(timeout=5.0)

    def test_merge_without_consolidate_does_not_fire_hook(
        self,
        tmp_path: pathlib.Path,
        mock_hub_server: _MockHubServer,
        clean_env_port: None,
    ) -> None:
        os.environ[ENV_PORT] = str(mock_hub_server.port)
        register_knowtation_hook()
        root, repo_id = _init_knowtation_repo(tmp_path)
        base_id = _make_simple_commit(root, repo_id, "main", "base")
        (root / ".muse" / "refs" / "heads" / "feature").write_text(base_id)
        _make_simple_commit(root, repo_id, "feature", "feat")
        runner = CliRunner()
        env = {"MUSE_REPO_ROOT": str(root), ENV_PORT: str(mock_hub_server.port)}
        result = runner.invoke(
            None, ["merge", "--format", "json", "feature"],
            env=env, catch_exceptions=False,
        )
        assert result.exit_code == 0
        data = json.loads(result.output)
        assert "consolidation_hook" not in data
        time.sleep(0.5)
        assert mock_hub_server.received == []


# ────────────────────────────────────────────────────────────────────────────
# Tier 4 — Stress
# ────────────────────────────────────────────────────────────────────────────


class TestStress:
    """Throughput and concurrency."""

    def test_100_sequential_noop_invocations_under_one_second(
        self, fresh_registry: MergeHookRegistry, clean_env_port: None
    ) -> None:
        fresh_registry.register("knowtation", KnowtationMergeHook())
        start = time.perf_counter()
        for _ in range(100):
            fresh_registry.run_pre_merge(
                _DUMMY_ROOT, "knowtation", _OURS_ID, _THEIRS_ID, _BRANCH,
                consolidate=False,
            )
        elapsed = time.perf_counter() - start
        assert elapsed < 1.0, f"100 noop hooks took {elapsed:.3f}s (budget 1.0s)"

    def test_concurrent_invocations_do_not_deadlock(
        self, fresh_registry: MergeHookRegistry, clean_env_port: None
    ) -> None:
        fresh_registry.register("knowtation", KnowtationMergeHook())
        completed: list[bool] = []
        errors: list[BaseException] = []
        completion_lock = threading.Lock()

        def _worker() -> None:
            try:
                for _ in range(20):
                    fresh_registry.run_pre_merge(
                        _DUMMY_ROOT, "knowtation", _OURS_ID, _THEIRS_ID, _BRANCH,
                        consolidate=False,
                    )
                with completion_lock:
                    completed.append(True)
            except BaseException as exc:  # noqa: BLE001
                with completion_lock:
                    errors.append(exc)

        threads = [threading.Thread(target=_worker) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10.0)
            assert not t.is_alive(), "thread failed to join — possible deadlock"
        assert errors == []
        assert len(completed) == 10


# ────────────────────────────────────────────────────────────────────────────
# Tier 5 — Data integrity
# ────────────────────────────────────────────────────────────────────────────


class TestDataIntegrity:
    """Determinism and isolation."""

    def test_noop_result_is_deterministic_across_calls(
        self, clean_env_port: None
    ) -> None:
        hook = KnowtationMergeHook()
        a = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=False,
        )
        b = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=False,
        )
        assert a == b
        assert dataclasses.asdict(a) == dataclasses.asdict(b)

    def test_fresh_registries_do_not_share_state(self) -> None:
        a = MergeHookRegistry()
        b = MergeHookRegistry()
        a.register("knowtation", KnowtationMergeHook())
        assert b.get("knowtation") is None
        assert a.get("knowtation") is not None

    def test_unregister_removes_state_from_isolated_instance(self) -> None:
        reg = MergeHookRegistry()
        reg.register("knowtation", KnowtationMergeHook())
        assert reg.get("knowtation") is not None
        assert reg.unregister("knowtation") is True
        assert reg.get("knowtation") is None
        assert reg.unregister("knowtation") is False


# ────────────────────────────────────────────────────────────────────────────
# Tier 6 — Performance
# ────────────────────────────────────────────────────────────────────────────


class TestPerformance:
    """Singleton lookup latency budget."""

    def test_1000_get_hook_registry_calls_under_100ms(self) -> None:
        start = time.perf_counter()
        for _ in range(1000):
            get_hook_registry()
        elapsed = time.perf_counter() - start
        assert elapsed < 0.100, (
            f"1000 get_hook_registry() calls took {elapsed * 1000:.2f}ms "
            f"(budget 100ms)"
        )


# ────────────────────────────────────────────────────────────────────────────
# Tier 7 — Security
# ────────────────────────────────────────────────────────────────────────────


class TestSecurity:
    """Input validation, injection, and error containment."""

    def test_port_injection_string_with_shell_metacharacters_rejected(
        self, clean_env_port: None
    ) -> None:
        """``"3000; rm -rf /"`` MUST fail integer parse and emit no request."""
        os.environ[ENV_PORT] = "3000; rm -rf /"
        port, err = _resolve_port()
        assert port is None
        assert err is not None
        hook = KnowtationMergeHook()
        with mock.patch(
            "muse.plugins.knowtation.hooks._post_consolidate_in_background"
        ) as spawn_mock:
            result = hook.pre_merge_hook(
                _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
            )
        spawn_mock.assert_not_called()
        assert result.fired is False
        assert result.error is not None
        assert "valid integer" in result.error.lower()

    def test_port_zero_rejected_via_env(self, clean_env_port: None) -> None:
        os.environ[ENV_PORT] = "0"
        port, err = _resolve_port()
        assert port is None
        assert err is not None
        assert "out of range" in err.lower()

    def test_port_65536_rejected_via_env(self, clean_env_port: None) -> None:
        os.environ[ENV_PORT] = "65536"
        port, err = _resolve_port()
        assert port is None
        assert err is not None
        assert "out of range" in err.lower()

    def test_url_is_built_only_from_validated_int_no_user_string(
        self, clean_env_port: None, mock_hub_server: _MockHubServer
    ) -> None:
        """Even with an injected env value, the URL never contains user text."""
        os.environ[ENV_PORT] = str(mock_hub_server.port)
        captured_urls: list[str] = []

        def _capture(url: str) -> threading.Thread:
            captured_urls.append(url)
            t = threading.Thread(target=lambda: None)
            t.start()
            return t

        with mock.patch(
            "muse.plugins.knowtation.hooks._post_consolidate_in_background",
            side_effect=_capture,
        ):
            hook = KnowtationMergeHook()
            hook.pre_merge_hook(
                _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
            )
        assert len(captured_urls) == 1
        url = captured_urls[0]
        expected = f"http://localhost:{mock_hub_server.port}{API_PATH}"
        assert url == expected
        assert ";" not in url
        assert "rm" not in url
        assert "&" not in url
        assert "$" not in url

    def test_background_thread_exception_does_not_propagate(
        self, clean_env_port: None
    ) -> None:
        """A broken HTTP call must not raise into the merge process."""
        port = _free_tcp_port()
        os.environ[ENV_PORT] = str(port)
        hook = KnowtationMergeHook()
        result = hook.pre_merge_hook(
            _DUMMY_ROOT, _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        assert isinstance(result, MergeHookResult)
        assert result.fired is True
        assert result.out_of_band is True
        time.sleep(0.5)

    def test_run_pre_merge_with_buggy_plugin_returns_error_result(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        class _Boom:
            def pre_merge_hook(
                self, root, ours_commit_id, theirs_commit_id, branch,
                *, consolidate: bool = False,
            ) -> MergeHookResult:
                raise ValueError("oops")

        fresh_registry.register("evil", _Boom())  # type: ignore[arg-type]
        results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "evil", _OURS_ID, _THEIRS_ID, _BRANCH, consolidate=True,
        )
        assert len(results) == 1
        assert results[0].error is not None
        assert "ValueError" in results[0].error

    def test_plugin_returning_wrong_type_is_caught(
        self, fresh_registry: MergeHookRegistry
    ) -> None:
        """A misbehaving plugin returning the wrong type does not crash callers."""

        class _BadType:
            def pre_merge_hook(
                self, root, ours_commit_id, theirs_commit_id, branch,
                *, consolidate: bool = False,
            ) -> Any:
                return "not a MergeHookResult"

        fresh_registry.register("badtype", _BadType())  # type: ignore[arg-type]
        results = fresh_registry.run_pre_merge(
            _DUMMY_ROOT, "badtype", _OURS_ID, _THEIRS_ID, _BRANCH,
            consolidate=True,
        )
        assert len(results) == 1
        assert results[0].error is not None
        assert "MergeHookResult" in results[0].error