"""Whole-repository integrity verification for ``muse verify``.

Walks every reachable commit from every branch ref (or a single branch when
``branch`` is specified) and performs five tiers of integrity checking:

1. **Ref integrity** — every branch ref file points to an existing commit.
2. **Commit → snapshot integrity** — every commit's ``snapshot_id`` resolves.
3. **Snapshot → object integrity** — every object path in every manifest exists.
4. **Object hash integrity** — re-hashes each object file and compares against
   its declared ID (the SHA-256 content address).  Optional; controlled by the
   ``check_objects`` flag.  Skipped when ``check_objects=False`` for speed.
5. **Signature integrity** — for every commit that carries a non-empty
   ``signature`` and ``agent_id``, the HMAC-SHA256 signature is verified
   against the agent key stored under ``.muse/keys/<agent_id>.key``.

The walk is BFS across all reachable commits from all branches; each commit is
checked exactly once even when reachable from multiple branches.

Security model::

    Branch ref files are validated before being read:
    - Symlinks inside ``.muse/refs/heads/`` are silently skipped — a symlink
      could point to a file outside the repository boundary.
    - Ref file content is capped at 65 bytes (SHA-256 hex + newline); anything
      larger is treated as an invalid ref rather than read into memory.

    Missing agent keys are reported as ``kind="key_missing"`` warnings in the
    failure list — distinct from ``kind="signature"`` (genuine tamper detection).
    Agents can filter by ``kind`` to distinguish key-rotation events from
    corruption.

Walk budget::

    ``_MAX_COMMITS = 500_000``.  When the budget is reached (using ``>=`` so
    exactly 500 000 commits are visited) a warning is logged and the walk stops.
    The result reflects only the commits visited up to that point.
"""

from __future__ import annotations

import hashlib
import json
import logging
import pathlib
import urllib.error
import urllib.request
from collections import deque
from typing import Any, Literal, TypedDict

from muse.core.attestation import (
    AttestationVerifyError,
    IdentityChainResult,
    IdentityRecord,
    MuseIdentityProvider,
    get_attestation_provider,
    get_identity_provider,
)
from muse.core.object_store import object_path
from muse.core.provenance import provenance_payload, verify_commit_ed25519
from muse.core.store import read_commit, read_snapshot

logger = logging.getLogger(__name__)

_CHUNK_SIZE = 65_536
_MAX_COMMITS = 500_000
# Maximum bytes we'll read from a ref file.  A valid SHA-256 hex string is 64
# chars + optional newline = 65 bytes.  Anything larger is treated as corrupt.
_MAX_REF_BYTES = 65


class VerifyFailure(TypedDict):
    """A single integrity failure detected during ``muse verify``.

    ``kind`` is one of:

    - ``"ref"``         — branch ref file is malformed or missing
    - ``"commit"``      — commit file is missing or unreadable
    - ``"snapshot"``    — snapshot file is missing or unreadable
    - ``"object"``      — object file is missing or its hash does not match
    - ``"signature"``   — Ed25519 signature verification failed (tamper detected)
    - ``"key_missing"`` — signer_public_key absent; signature could not be verified
    """

    kind: Literal["ref", "commit", "snapshot", "object", "signature", "key_missing"]
    id: str
    error: str


class VerifyResult(TypedDict):
    """Aggregate result of a full repository integrity walk."""

    refs_checked: int
    commits_checked: int
    snapshots_checked: int
    objects_checked: int
    signatures_checked: int
    all_ok: bool
    failures: list[VerifyFailure]


def _branch_refs(
    root: pathlib.Path,
    *,
    branch: str | None = None,
) -> list[tuple[str, str]]:
    """Return ``[(branch_name, commit_id)]`` for branch ref files.

    When *branch* is given only that branch's ref is returned.  Otherwise all
    refs under ``.muse/refs/heads/`` are returned.

    Security:
        Symlinks are silently skipped — a symlink could point outside the
        repository boundary.  Ref file content is capped at ``_MAX_REF_BYTES``
        (65 bytes); oversized files are treated as invalid refs.
    """
    heads_dir = root / ".muse" / "refs" / "heads"
    if not heads_dir.exists():
        return []

    if branch is not None:
        ref_file = heads_dir / branch
        if not ref_file.exists() or ref_file.is_symlink():
            return []
        raw_bytes = ref_file.read_bytes()[:_MAX_REF_BYTES]
        raw = raw_bytes.decode("utf-8", errors="replace").strip()
        return [(branch, raw)] if raw else []

    refs: list[tuple[str, str]] = []
    for ref_file in sorted(heads_dir.rglob("*")):
        if ref_file.is_symlink():
            logger.warning("⚠️ verify: skipping symlink ref: %s", ref_file)
            continue
        if not ref_file.is_file():
            continue
        raw_bytes = ref_file.read_bytes()[:_MAX_REF_BYTES]
        raw = raw_bytes.decode("utf-8", errors="replace").strip()
        if raw:
            br = str(ref_file.relative_to(heads_dir).as_posix())
            refs.append((br, raw))
    return refs


def _rehash_object(path: pathlib.Path) -> str:
    """Re-compute SHA-256 of the object file at *path* using streaming reads."""
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(_CHUNK_SIZE), b""):
            h.update(chunk)
    return h.hexdigest()


def run_verify(
    root: pathlib.Path,
    *,
    check_objects: bool = True,
    branch: str | None = None,
    fail_fast: bool = False,
) -> VerifyResult:
    """Perform a full repository integrity walk.

    Args:
        root:          Repository root (the directory containing ``.muse/``).
        check_objects: When ``True`` (the default), every object file is
                       re-hashed and its digest compared against its declared
                       content address.  Pass ``False`` for a faster
                       existence-only check.
        branch:        When given, only the named branch is walked.  All
                       other branches are ignored.  Useful for targeted
                       verification of a single branch without paying the
                       full BFS cost.
        fail_fast:     When ``True``, the walk stops after the first failure
                       is recorded.  Useful in CI pipelines where the
                       presence of any failure is sufficient to fail the job.

    Returns:
        A :class:`VerifyResult` summarising what was checked and any failures.
        Failures with ``kind="key_missing"`` indicate that an agent key could
        not be found (possible key rotation); failures with
        ``kind="signature"`` indicate that a signature did not verify
        (possible tampering).
    """
    failures: list[VerifyFailure] = []
    refs_checked = 0
    commits_checked = 0
    snapshots_checked = 0
    objects_checked = 0
    signatures_checked = 0

    # Track which objects and snapshots we've already verified to avoid
    # re-checking the same content multiple times (shared objects / snapshots).
    verified_snapshots: set[str] = set()
    verified_objects: set[str] = set()

    # Phase 1: enumerate branch refs and collect tip commit IDs.
    branch_refs = _branch_refs(root, branch=branch)
    tip_commit_ids: list[str] = []

    for br, commit_id in branch_refs:
        refs_checked += 1
        # Validate the ref format: exactly 64 lowercase hex characters.
        if len(commit_id) != 64 or not all(c in "0123456789abcdef" for c in commit_id):
            failures.append(VerifyFailure(
                kind="ref",
                id=br,
                error=f"invalid commit ID in ref: {commit_id!r}",
            ))
            if fail_fast:
                return _make_result(
                    refs_checked, commits_checked, snapshots_checked,
                    objects_checked, signatures_checked, failures,
                )
            continue
        tip_commit_ids.append(commit_id)

    # Phase 2: BFS walk of the commit DAG.
    visited: set[str] = set()
    queue: deque[str] = deque(tip_commit_ids)

    while queue:
        cid = queue.popleft()
        if cid in visited:
            continue
        visited.add(cid)

        if len(visited) >= _MAX_COMMITS:
            logger.warning(
                "⚠️ verify: reached %d-commit limit — stopping early", _MAX_COMMITS
            )
            break

        commit = read_commit(root, cid)
        if commit is None:
            failures.append(VerifyFailure(
                kind="commit",
                id=cid,
                error="commit file missing or unreadable",
            ))
            if fail_fast:
                break
            continue

        commits_checked += 1

        # Phase 5: verify Ed25519 signature when the commit carries one.
        # format_version == 7: Ed25519 signature with embedded public key.
        # format_version <= 6: legacy HMAC (no longer verifiable) — skip silently.
        # A commit is considered signed when signature and signer_public_key are both set.
        if commit.signature and commit.format_version >= 7:
            if not commit.signer_public_key:
                logger.warning(
                    "⚠️ verify: signed commit %s missing signer_public_key — cannot verify",
                    cid[:12],
                )
                failures.append(VerifyFailure(
                    kind="key_missing",
                    id=cid,
                    error=(
                        f"signed commit {cid[:12]} (format_version=7) is missing "
                        "signer_public_key — cannot verify Ed25519 signature"
                    ),
                ))
                if fail_fast:
                    break
            else:
                import base64
                signatures_checked += 1
                signed_input = provenance_payload(
                    cid,
                    author=commit.author,
                    agent_id=commit.agent_id,
                    model_id=commit.model_id,
                    toolchain_id=commit.toolchain_id,
                    prompt_hash=commit.prompt_hash,
                    committed_at=commit.committed_at.isoformat(),
                )
                padded = commit.signer_public_key + "=" * (-len(commit.signer_public_key) % 4)
                try:
                    pub_bytes = base64.urlsafe_b64decode(padded)
                except Exception:
                    pub_bytes = b""
                if not pub_bytes or not verify_commit_ed25519(signed_input, commit.signature, pub_bytes):
                    failures.append(VerifyFailure(
                        kind="signature",
                        id=cid,
                        error=(
                            f"Ed25519 signature INVALID for commit {cid[:12]} "
                            f"(agent {commit.agent_id!r}, key {commit.signer_key_id!r}) "
                            "— commit record may have been tampered with"
                        ),
                    ))
                    if fail_fast:
                        break

        # Phase 3: check snapshot.
        snap_id = commit.snapshot_id
        if snap_id not in verified_snapshots:
            verified_snapshots.add(snap_id)
            snap = read_snapshot(root, snap_id)
            if snap is None:
                failures.append(VerifyFailure(
                    kind="snapshot",
                    id=snap_id,
                    error=f"snapshot missing (referenced by commit {cid[:12]})",
                ))
                if fail_fast:
                    break
            else:
                snapshots_checked += 1

                # Phase 4: check objects referenced by this snapshot.
                for rel_path, obj_id in snap.manifest.items():
                    if obj_id in verified_objects:
                        continue
                    verified_objects.add(obj_id)

                    obj_file = object_path(root, obj_id)
                    if not obj_file.exists():
                        failures.append(VerifyFailure(
                            kind="object",
                            id=obj_id,
                            error=f"object file missing (path={rel_path})",
                        ))
                        if fail_fast:
                            break
                        continue

                    objects_checked += 1

                    if check_objects:
                        actual = _rehash_object(obj_file)
                        if actual != obj_id:
                            failures.append(VerifyFailure(
                                kind="object",
                                id=obj_id,
                                error=(
                                    f"hash mismatch: expected {obj_id[:12]} "
                                    f"got {actual[:12]} — data corruption detected"
                                ),
                            ))
                            if fail_fast:
                                break
                if fail_fast and failures:
                    break

        # Enqueue parents for the BFS walk.
        if commit.parent_commit_id:
            queue.append(commit.parent_commit_id)
        if commit.parent2_commit_id:
            queue.append(commit.parent2_commit_id)

        if fail_fast and failures:
            break

    return _make_result(
        refs_checked, commits_checked, snapshots_checked,
        objects_checked, signatures_checked, failures,
    )


def _make_result(
    refs_checked: int,
    commits_checked: int,
    snapshots_checked: int,
    objects_checked: int,
    signatures_checked: int,
    failures: list[VerifyFailure],
) -> VerifyResult:
    """Construct a :class:`VerifyResult` from the walk counters."""
    return VerifyResult(
        refs_checked=refs_checked,
        commits_checked=commits_checked,
        snapshots_checked=snapshots_checked,
        objects_checked=objects_checked,
        signatures_checked=signatures_checked,
        all_ok=len(failures) == 0,
        failures=failures,
    )


# ===========================================================================
# Phase 7.5: per-commit attestation verification
# ===========================================================================


#: Tier values for the attestation verification result.
TIER_UNSIGNED: int = 0
TIER_SIGNED_ONLY: int = 1
TIER_HMAC_ATTESTED: int = 2
TIER_ICP_ANCHORED: int = 3


class AttestationVerifyResult(TypedDict, total=False):
    """Per-commit attestation verification result.

    Returned by :func:`verify_attestation`.  Maps directly to the four-tier
    CLI output of ``muse verify-commit --attest``:

    * ``tier == 0`` (UNSIGNED)         — no Ed25519 signature.
    * ``tier == 1`` (SIGNED_ONLY)      — Ed25519 OK, no attestation record.
    * ``tier == 2`` (HMAC_ATTESTED)    — Ed25519 + HMAC valid, no anchor proof.
    * ``tier == 3`` (ICP_ANCHORED)     — Ed25519 + HMAC + ICP canister anchor.

    Attributes:
        valid:           ``True`` iff ``tier >= 1`` and the verifier completed
                         the chain WITHOUT raising.  ``False`` only for
                         provably-bad records (e.g. tampered HMAC, mismatched
                         anchor content).
        reason:          Human-readable diagnostic.  Empty when
                         ``tier >= 1`` and ``valid=True``.  When the verifier
                         could not complete a partial check, this carries
                         a stable code prefix like ``"ICP_UNREACHABLE — ..."``
                         so the CLI can surface "verification incomplete"
                         distinctly from "verified" or "invalid".
        tier:            One of :data:`TIER_UNSIGNED`,
                         :data:`TIER_SIGNED_ONLY`, :data:`TIER_HMAC_ATTESTED`,
                         :data:`TIER_ICP_ANCHORED`.
        hmac_valid:      ``True`` iff the HMAC signature on the attestation
                         record verified.  ``False`` for unsigned commits or
                         provably-bad HMACs.
        icp_anchored:    ``True`` iff the canister returned a 200 for
                         ``GET /attest/<id>`` AND the anchored copy matches
                         the local record (same id, content_hash, path).
        anchor_tx_id:    Anchor receipt's tx_id when anchored, else ``""``.
        identity_depth:  Identity-chain depth validated.  ``0`` for
                         single-signer commits; non-zero for org-quorum
                         commits (Phase 7.7).
        provider_id:     Domain string of the provider used (e.g.
                         ``"knowtation"``), or ``""`` when no provider was
                         consulted.
        partial:         ``True`` when the verifier could not complete the
                         full chain (e.g. ICP unreachable).  When ``True``,
                         the CLI MUST report the partial state — never
                         silently degrade to "signed only".
    """

    valid: bool
    reason: str
    tier: int
    hmac_valid: bool
    icp_anchored: bool
    anchor_tx_id: str
    identity_depth: int
    provider_id: str
    partial: bool


def _check_ed25519_signature(commit: Any) -> tuple[bool, str]:
    """Verify the commit's Ed25519 signature.

    Args:
        commit: A ``CommitRecord`` from :func:`muse.core.store.read_commit`.

    Returns:
        ``(valid, reason)`` — ``valid`` is ``True`` iff the signature exists
        and verified.  When ``False``, ``reason`` describes why.
    """
    if not commit.signature or commit.format_version < 7:
        return False, "no Ed25519 signature on commit"
    if not commit.signer_public_key:
        return False, "signed commit missing signer_public_key"
    import base64
    payload = provenance_payload(
        commit.commit_id,
        author=commit.author,
        agent_id=commit.agent_id,
        model_id=commit.model_id,
        toolchain_id=commit.toolchain_id,
        prompt_hash=commit.prompt_hash,
        committed_at=commit.committed_at.isoformat(),
    )
    padded = commit.signer_public_key + "=" * (-len(commit.signer_public_key) % 4)
    try:
        pub_bytes = base64.urlsafe_b64decode(padded)
    except Exception:
        return False, "signer_public_key is not valid base64url"
    if not verify_commit_ed25519(payload, commit.signature, pub_bytes):
        return False, "Ed25519 signature INVALID — commit may have been tampered with"
    return True, ""


def _check_icp_anchor(
    record: dict[str, Any],
    canister_id: str,
    *,
    timeout: float = 5.0,
) -> tuple[bool, str | None, str]:
    """Check whether *record* is anchored on the ICP canister.

    Performs a ``GET /attest/<id>`` and verifies that the response body
    matches the local record on the fields that should be invariant
    (``id``, ``content_hash``, ``path``, ``action``).

    Args:
        record:      Local attestation record dict.
        canister_id: Validated canister ID.
        timeout:     HTTP socket timeout (seconds).

    Returns:
        ``(anchored, tx_id, reason)``.

        * ``anchored=True``   — canister returned a matching record;
          ``tx_id`` populated; ``reason=""``.
        * ``anchored=False``  — canister returned a *different* record;
          ``tx_id=None``; ``reason`` explains the mismatch.
        * ``(None, None, "ICP_UNREACHABLE — ...")`` — verifier could not
          complete the check; caller must surface this as partial, not as
          "anchored=False".  Returned via the typed sentinel — the function
          raises :class:`AttestationVerifyError` instead.

    Raises:
        AttestationVerifyError: When the canister is unreachable or the
            response is malformed — the caller surfaces this as
            ``partial=True``.  Never silently returns ``anchored=False``.
    """
    from muse.plugins.knowtation.icp_push_hook import (
        canister_http_get_url,
        fetch_attestation_via_http,
    )
    rid = str(record.get("id", ""))
    if not rid:
        raise AttestationVerifyError("record has no id", code="MISSING_ID")
    try:
        anchored = fetch_attestation_via_http(canister_id, rid, timeout=timeout)
    except OSError as exc:
        raise AttestationVerifyError(
            f"ICP canister unreachable: {exc}", code="ICP_UNREACHABLE"
        ) from exc
    if anchored is None:
        return False, None, "canister has no record for this id (404)"
    # Cross-check invariant fields.
    for field in ("id", "content_hash", "path", "action"):
        if str(record.get(field, "")) != str(anchored.get(field, "")):
            return (
                False, None,
                f"anchored {field!r} differs from local record",
            )
    tx_id = str(anchored.get("seq", ""))
    return True, f"icp-seq-{tx_id}" if tx_id else "icp-anchored", ""


def verify_attestation(
    commit_record: Any,
    repo_path: pathlib.Path,
    config: dict[str, Any] | None = None,
) -> AttestationVerifyResult:
    """Verify the four-tier attestation chain for a single commit.

    Tiers:

    * 0 — unsigned (no Ed25519 signature)
    * 1 — signed only (Ed25519 OK but no attestation record)
    * 2 — signed + HMAC-attested (AIR record present and HMAC valid)
    * 3 — signed + HMAC-attested + ICP-anchored (canister returned matching
      record on GET /attest/<id>)

    Partial-verification handling (security-critical):
    When the verifier cannot complete a step — typically because the ICP
    canister is unreachable — it sets ``partial=True`` and embeds a
    structured reason like ``"ICP_UNREACHABLE — ..."`` *without* silently
    degrading the tier.  The CLI MUST surface this as "incomplete", never
    as "verified" or "signed only".

    Args:
        commit_record: A ``CommitRecord`` (from :func:`muse.core.store.read_commit`).
        repo_path:     Repository root.
        config:        Optional config dict.  Reads:
            - ``"icp_canister_id"``: override the canister ID.
            - ``"check_icp"``: bool, whether to attempt the canister query
              (default ``True`` if an attestation record is present).

    Returns:
        An :class:`AttestationVerifyResult`.

    Raises:
        Never — partial verifications are surfaced via the ``partial``
        field, NOT via exceptions.  The Phase 7.5 contract is:
        "Never silently degrade".  Exceptions are caught and converted
        to ``partial=True`` results with the appropriate reason code.
    """
    cfg = config or {}
    result: AttestationVerifyResult = {
        "valid": False,
        "reason": "",
        "tier": TIER_UNSIGNED,
        "hmac_valid": False,
        "icp_anchored": False,
        "anchor_tx_id": "",
        "identity_depth": 0,
        "provider_id": "",
        "partial": False,
    }

    # ── Tier 0 → 1: Ed25519 signature ────────────────────────────────────
    sig_ok, sig_reason = _check_ed25519_signature(commit_record)
    if not sig_ok:
        result["reason"] = sig_reason or "unsigned"
        return result
    result["tier"] = TIER_SIGNED_ONLY
    result["valid"] = True

    # ── Tier 1 → 2: HMAC attestation ─────────────────────────────────────
    metadata = commit_record.metadata or {}
    blob = metadata.get("attestation")
    if not blob:
        return result
    try:
        record = json.loads(blob)
    except json.JSONDecodeError as exc:
        result["partial"] = True
        result["reason"] = f"MALFORMED_RECORD — attestation JSON unparseable: {exc}"
        return result
    if not isinstance(record, dict):
        result["partial"] = True
        result["reason"] = "MALFORMED_RECORD — attestation is not a JSON object"
        return result

    provider_id = str(record.get("provider_id", ""))
    result["provider_id"] = provider_id
    provider = get_attestation_provider(provider_id)
    if provider is None:
        result["partial"] = True
        result["reason"] = (
            f"NO_PROVIDER — domain {provider_id!r} has no registered "
            "attestation provider; cannot verify HMAC"
        )
        return result

    synthetic_receipt: dict[str, Any] = {
        "tx_id": "verify-only",
        "anchor_url": "",
        "anchored_at": str(record.get("timestamp", "")),
        "provider_id": provider_id,
    }
    try:
        verify_result = provider.verify(record, synthetic_receipt)  # type: ignore[arg-type]
    except AttestationVerifyError as exc:
        result["partial"] = True
        result["reason"] = f"{exc.code} — {exc}"
        return result
    if not verify_result.get("valid"):
        result["partial"] = False
        result["reason"] = (
            f"HMAC_INVALID — {verify_result.get('reason') or 'provider returned valid=False'}"
        )
        result["valid"] = False
        return result

    result["hmac_valid"] = True
    result["tier"] = TIER_HMAC_ATTESTED
    result["identity_depth"] = int(verify_result.get("depth", 0) or 0)

    # ── Tier 2 → 3: ICP anchor verification ──────────────────────────────
    if not cfg.get("check_icp", True):
        return result

    canister_id = str(
        cfg.get("icp_canister_id")
        or _resolve_default_canister_id()
    )
    try:
        anchored, tx_id, anchor_reason = _check_icp_anchor(
            record, canister_id, timeout=float(cfg.get("icp_timeout", 5.0))
        )
    except AttestationVerifyError as exc:
        result["partial"] = True
        result["reason"] = f"{exc.code} — cannot confirm anchor; HMAC valid"
        return result

    if anchored:
        result["icp_anchored"] = True
        result["anchor_tx_id"] = tx_id or ""
        result["tier"] = TIER_ICP_ANCHORED
        return result

    result["reason"] = f"ICP_NOT_ANCHORED — {anchor_reason}"
    return result


def _resolve_default_canister_id() -> str:
    """Return the default ICP canister ID, lazily importing the constant."""
    from muse.plugins.knowtation.icp_push_hook import DEFAULT_CANISTER_ID
    return DEFAULT_CANISTER_ID


# ---------------------------------------------------------------------------
# Phase 7.7 — identity chain walker and quorum-signature verifier
# ---------------------------------------------------------------------------


#: Hard cap on identity-chain walk depth.  Prevents pathological recursion
#: caused by adversarial deep org-of-orgs nesting.  10 is enough for any
#: realistic governance hierarchy and bounds verification cost.
DEFAULT_IDENTITY_DEPTH_LIMIT: int = 10


class QuorumVerifyResult(TypedDict, total=False):
    """Result of :func:`verify_quorum_signatures` for a single commit.

    Attributes:
        valid:          ``True`` iff the number of independently-verified
                        member signatures is at least the live
                        ``quorum_threshold`` AND the org identity chain
                        validated AND no signer was revoked.
        reason:         Human-readable diagnostic.  Empty when
                        ``valid=True``.  When ``valid=False``, names the
                        exact failure (``BELOW_THRESHOLD``,
                        ``SIGNATURE_INVALID``, ``UNKNOWN_SIGNER``,
                        ``CYCLE_DETECTED``, ``REVOKED_MEMBER``,
                        ``IDENTITY_PROVIDER_MISSING``).
        threshold:      Live quorum threshold from the org identity record
                        (NOT the threshold copied into commit metadata —
                        that copy is for audit only).
        valid_signers:  Number of member signatures that verified against
                        their member identity's public key.
        total_signers:  Number of signers listed in ``commit.metadata['quorum']``.
        depth:          Maximum identity-chain depth walked while resolving
                        members.  Surfaced for audit.
        partial:        ``True`` when the verifier could not complete the
                        check (e.g. identity provider unreachable).  When
                        ``True``, the CLI MUST NOT report "verified" — the
                        result is "incomplete".
        org_handle:     The org handle from commit metadata (echoed for
                        machine-readable output).
    """

    valid: bool
    reason: str
    threshold: int
    valid_signers: int
    total_signers: int
    depth: int
    partial: bool
    org_handle: str


def verify_identity_chain(
    handle: str,
    provider: MuseIdentityProvider,
    *,
    depth_limit: int = DEFAULT_IDENTITY_DEPTH_LIMIT,
) -> IdentityChainResult:
    """Walk the identity chain rooted at *handle* using *provider*.

    Performs a depth-bounded BFS:

    * Resolves *handle* via ``provider.get_identity``.
    * For org records, recursively resolves every member.
    * Maintains a seen-set to detect cycles defensively (the identity
      domain rejects cycles at write time, but a tampered local cache
      could still introduce one).

    Args:
        handle:      Root handle to walk from.
        provider:    A :class:`MuseIdentityProvider` (typically resolved via
            :func:`muse.core.attestation.get_identity_provider`).
        depth_limit: Maximum walk depth.  When the BFS reaches this depth
            the walk terminates with ``valid=False`` and reason
            ``"DEPTH_EXCEEDED"``.

    Returns:
        :class:`IdentityChainResult` summarising the walk.

    Raises:
        Never — any provider error is wrapped into ``valid=False`` plus
        a stable reason code.  Callers receive a complete result.
    """
    result: IdentityChainResult = {
        "valid": False, "depth": 0, "reason": "", "cycle_detected": False,
    }
    seen: set[str] = set()
    stack: list[tuple[str, int]] = [(handle, 1)]
    max_depth = 0

    while stack:
        cur, depth = stack.pop()
        if depth > depth_limit:
            result["depth"] = max_depth
            result["reason"] = f"DEPTH_EXCEEDED — chain deeper than {depth_limit}"
            return result
        if cur in seen:
            result["depth"] = max_depth
            result["cycle_detected"] = True
            result["reason"] = f"CYCLE_DETECTED — handle {cur!r} appears twice on the chain"
            return result
        seen.add(cur)
        max_depth = max(max_depth, depth)
        try:
            record = provider.get_identity(cur)
        except AttestationVerifyError as exc:
            result["depth"] = max_depth
            result["reason"] = f"{exc.code} — {exc}"
            return result
        if record is None:
            result["depth"] = max_depth
            result["reason"] = f"UNKNOWN_HANDLE — no identity record for {cur!r}"
            return result
        if record.get("revoked_at"):
            result["depth"] = max_depth
            result["reason"] = f"REVOKED — {cur!r} revoked at {record['revoked_at']}"
            return result
        for m in record.get("members", []) or []:
            stack.append((m, depth + 1))

    result["valid"] = True
    result["depth"] = max_depth
    return result


def _verify_member_signature(
    signer: dict[str, str],
    member_record: IdentityRecord,
    payload: str,
) -> tuple[bool, str]:
    """Verify a single member's Ed25519 signature on *payload*.

    Cross-checks the signer's stated ``public_key`` against the member's
    canonical ``public_key`` from the identity record — a mismatch is a
    hard failure (a malicious commit could otherwise embed a key they
    control alongside a real member's handle).

    Args:
        signer:        ``{"handle", "public_key", "signature"}`` dict from
            commit metadata.
        member_record: Canonical :class:`IdentityRecord` for the same
            handle, fetched live from the identity provider.
        payload:       The :func:`muse.core.provenance.provenance_payload`
            digest that the member signed.

    Returns:
        ``(valid, reason)``.  ``valid=True`` only when the public key
        matches the canonical record AND the Ed25519 signature verifies.
    """
    import base64

    declared_pk = signer.get("public_key", "")
    canonical_pk = member_record.get("public_key", "")
    if not canonical_pk:
        return False, f"member {signer.get('handle')!r} has no canonical public_key"
    if declared_pk != canonical_pk:
        return False, (
            f"signer {signer.get('handle')!r} public_key does not match "
            "canonical identity record public_key"
        )

    sig_b64 = signer.get("signature", "")
    if not sig_b64:
        return False, "missing signature"

    padded = canonical_pk + "=" * (-len(canonical_pk) % 4)
    try:
        pub_bytes = base64.urlsafe_b64decode(padded)
    except Exception:
        return False, "canonical public_key is not valid base64url"
    if not verify_commit_ed25519(payload, sig_b64, pub_bytes):
        return False, "Ed25519 signature INVALID"
    return True, ""


def verify_quorum_signatures(
    commit_record: Any,
    repo_path: pathlib.Path,
    config: dict[str, Any] | None = None,
) -> QuorumVerifyResult:
    """Verify that *commit_record* satisfies its org's quorum threshold.

    The verifier:

    1. Reads ``commit.metadata["quorum"]`` and parses the canonical shape
       produced by
       :func:`muse.plugins.knowtation.quorum.build_quorum_metadata`.
    2. Resolves the org identity record live via the registered identity
       provider — does NOT trust the threshold embedded in commit metadata
       (which is for audit only; an attacker could otherwise reduce it
       client-side).
    3. Walks the org identity chain via :func:`verify_identity_chain`,
       confirming all members exist, none are revoked, and no cycle
       appears in the membership graph.
    4. For each declared signer, fetches the member's identity record,
       compares the embedded public_key to the canonical key (rejecting
       any mismatch as forgery), and verifies the Ed25519 signature over
       the same :func:`muse.core.provenance.provenance_payload` the
       primary signer signed.
    5. Counts unique valid member signatures.  Result is
       ``valid=True`` iff the count is at least the live
       ``quorum_threshold`` AND no signer is unknown / outside the
       member set.

    Threats handled (per the Phase 7.7 spec):

    * **Partial-quorum forgery** — only valid member signatures count;
      ``floor(threshold - 1)`` valid sigs is rejected.
    * **Signer collusion below threshold** — same as above; counting is
      deterministic.
    * **Signature stripping** — when commit metadata contains fewer valid
      sigs than the live threshold the result is ``valid=False`` with
      reason ``BELOW_THRESHOLD``.  No silent pass.
    * **Replay across branches** — the signed payload binds the
      ``commit_id`` so it cannot be reused for a different commit.
    * **Cycle-introducing membership** — :func:`verify_identity_chain`
      sets ``cycle_detected=True`` and the verifier rejects.

    Args:
        commit_record: A ``CommitRecord`` from
            :func:`muse.core.store.read_commit`.
        repo_path:     Repository root.  Reserved for future per-repo
            identity caches (currently unused — the identity provider
            owns its own resolution).
        config:        Optional dict.  Reads:

            * ``"identity_domain"`` — domain string to resolve the
              identity provider with (default ``"knowtation"``).
            * ``"depth_limit"`` — override
              :data:`DEFAULT_IDENTITY_DEPTH_LIMIT`.

    Returns:
        :class:`QuorumVerifyResult`.

    Raises:
        Never — provider errors and partial-verifications are converted
        to ``partial=True``/``valid=False`` with stable reason codes.
    """
    cfg = config or {}
    result: QuorumVerifyResult = {
        "valid": False, "reason": "", "threshold": 0,
        "valid_signers": 0, "total_signers": 0, "depth": 0,
        "partial": False, "org_handle": "",
    }

    metadata = commit_record.metadata or {}
    blob = metadata.get("quorum")
    if not blob:
        result["reason"] = "NO_QUORUM_METADATA — commit has no metadata['quorum']"
        return result
    try:
        quorum = json.loads(blob) if isinstance(blob, str) else blob
    except json.JSONDecodeError as exc:
        result["partial"] = True
        result["reason"] = f"MALFORMED_QUORUM — {exc}"
        return result
    if not isinstance(quorum, dict):
        result["partial"] = True
        result["reason"] = "MALFORMED_QUORUM — not a JSON object"
        return result

    org_handle = str(quorum.get("org_handle", ""))
    result["org_handle"] = org_handle
    if not org_handle.startswith("@"):
        result["reason"] = f"INVALID_ORG_HANDLE — {org_handle!r} does not start with '@'"
        return result

    signers = quorum.get("signers", [])
    if not isinstance(signers, list) or not signers:
        result["reason"] = "MALFORMED_QUORUM — signers must be a non-empty list"
        return result
    result["total_signers"] = len(signers)

    domain = str(cfg.get("identity_domain", "knowtation"))
    provider = get_identity_provider(domain)
    if provider is None:
        result["partial"] = True
        result["reason"] = (
            f"IDENTITY_PROVIDER_MISSING — no identity provider registered "
            f"for domain {domain!r}; cannot verify quorum"
        )
        return result

    depth_limit = int(cfg.get("depth_limit", DEFAULT_IDENTITY_DEPTH_LIMIT))

    # Step 1 — fetch and validate the org record (live, not from metadata).
    try:
        org_record = provider.get_identity(org_handle)
    except AttestationVerifyError as exc:
        result["partial"] = True
        result["reason"] = f"{exc.code} — fetching org {org_handle!r}: {exc}"
        return result
    if org_record is None:
        result["reason"] = f"UNKNOWN_ORG — no identity record for {org_handle!r}"
        return result
    if org_record.get("kind") != "org":
        result["reason"] = (
            f"NOT_AN_ORG — handle {org_handle!r} kind is "
            f"{org_record.get('kind')!r}, expected 'org'"
        )
        return result
    if org_record.get("revoked_at"):
        result["reason"] = (
            f"ORG_REVOKED — {org_handle!r} revoked at "
            f"{org_record['revoked_at']}"
        )
        return result

    # Step 2 — walk the identity chain (cycle detection + member resolution).
    chain = verify_identity_chain(org_handle, provider, depth_limit=depth_limit)
    result["depth"] = chain.get("depth", 0)
    if not chain.get("valid"):
        if chain.get("cycle_detected"):
            result["reason"] = chain.get("reason", "CYCLE_DETECTED")
        else:
            result["reason"] = chain.get("reason", "CHAIN_INVALID")
        return result

    # Step 3 — recompute the signed payload from canonical commit fields.
    payload = provenance_payload(
        commit_record.commit_id,
        author=getattr(commit_record, "author", "") or "",
        agent_id=getattr(commit_record, "agent_id", "") or "",
        model_id=getattr(commit_record, "model_id", "") or "",
        toolchain_id=getattr(commit_record, "toolchain_id", "") or "",
        prompt_hash=getattr(commit_record, "prompt_hash", "") or "",
        committed_at=commit_record.committed_at.isoformat(),
    )

    # Step 4 — verify each declared signer's signature.
    threshold = int(org_record.get("quorum_threshold", 0))
    result["threshold"] = threshold
    member_set = set(org_record.get("members", []) or [])
    counted: set[str] = set()

    for signer in signers:
        if not isinstance(signer, dict):
            result["reason"] = "MALFORMED_QUORUM — signer entry is not a dict"
            return result
        handle = str(signer.get("handle", ""))
        if handle not in member_set:
            result["reason"] = (
                f"UNKNOWN_SIGNER — {handle!r} is not a member of org {org_handle!r}"
            )
            return result
        if handle in counted:
            result["reason"] = (
                f"DUPLICATE_SIGNER — {handle!r} appears twice; duplicates do not"
                " count toward quorum"
            )
            return result

        try:
            member_record = provider.get_identity(handle)
        except AttestationVerifyError as exc:
            result["partial"] = True
            result["reason"] = f"{exc.code} — fetching member {handle!r}: {exc}"
            return result
        if member_record is None:
            result["reason"] = f"UNKNOWN_MEMBER — identity record missing for {handle!r}"
            return result
        if member_record.get("revoked_at"):
            result["reason"] = (
                f"REVOKED_MEMBER — {handle!r} revoked at "
                f"{member_record['revoked_at']}"
            )
            return result

        ok, why = _verify_member_signature(signer, member_record, payload)
        if not ok:
            result["reason"] = f"SIGNATURE_INVALID — {handle!r}: {why}"
            return result
        counted.add(handle)

    result["valid_signers"] = len(counted)
    if len(counted) < threshold:
        result["reason"] = (
            f"BELOW_THRESHOLD — {len(counted)} valid signatures, "
            f"threshold is {threshold}"
        )
        return result

    result["valid"] = True
    return result