"""Proximo MCP server.

Exposes Proxmox management (REST API) and in-container exec (ssh+pct) as MCP tools.

Verified 2026-06-07 against the official `mcp` Python SDK (FastMCP): import path,
`@mcp.tool()` decorator, type-hinted params, and dict returns are current (v1.x).

Ethical spine:
- In-container exec (ct_*) is OFF by default — API-only is the safe default; enable with PROXIMO_ENABLE_EXEC.
- Every tool call is audited *with its real outcome* (errors recorded, not assumed "ok").
- Every mutating tool (pve_guest_power, ct_exec, ct_psql) is confirm-gated.
- The CTID allowlist is enforced fail-closed in the exec backend.
- Secrets are never read or logged here.
"""

from __future__ import annotations

import os
import sys
import time
from collections.abc import Callable, Iterable
from datetime import UTC, datetime
from functools import cache, lru_cache
from typing import Annotated, Any

from mcp.server.fastmcp import FastMCP
from pydantic import Field

from . import __version__
from .audit import AuditLedger, find_rotation_archive, looks_like_head, open_ledger
from .audit_anchor import AnchorError
from .backends import ApiBackend, ExecBackend, ProximoError, _check_vmid
from .config import ProximoConfig, load_env_file
from .consent import consent_id_for, enforce_consent, set_pending_consent
from .contain import enforce_containment
from .envelope import begin_operation, enforce_envelope_forbid, enforce_envelope_rate
from .lease import enforce_lease
from .pbs import (
    PbsBackend,
    PbsConfig,
)
from .pdm import (
    PdmBackend,
    PdmConfig,
)
from .planning import (
    Plan,
    command_fingerprint,
    plan_exec,
    plan_psql,
    sql_fingerprint,
    undo_snapname,
)
from .pmg import (
    PmgBackend,
    PmgConfig,
)
from .provenance import enforce_scope
from .qemu_agent import (
    plan_agent_exec,
)
from .taint import fence_output, is_adversarial, mark_tainted, taint_tracking_on
from .targets import (
    active_target,
    ledger_remote,
    load_registry,
    resolve_target_fields,
    target_aware,
)

BANNER = (
    "Proximo — the ethical Proxmox MCP\n"
    '  "Win the crowd and you will win your freedom."  ·  Strength and honor.\n'
)

mcp = FastMCP("proximo")
# FastMCP leaves the low-level Server.version=None, so the `initialize` handshake would advertise the
# MCP SDK's version. Set Proximo's own version instead, so clients see the real server version.
mcp._mcp_server.version = __version__


def tool(*d_args: Any, **d_kwargs: Any):
    """Target-aware tool decorator: like FastMCP's, but the tool also advertises
    `proximo_target` and routes the call to that registered box (via the active-target
    contextvar). Apply to every plane tool. Instance-level tools that act on THIS Proximo
    (e.g. audit_verify) intentionally keep the plain FastMCP decorator — they have no
    remote to target.
    """
    inner = mcp.tool(*d_args, **d_kwargs)

    def deco(fn):
        return inner(target_aware(fn))

    return deco


def _resolve_pve_config(target_name: str | None) -> ProximoConfig:
    """The active PVE config: None => env box (unchanged); a name => that registry remote."""
    if target_name is None:
        return ProximoConfig.from_env()
    return ProximoConfig.from_target(resolve_target_fields(target_name, "pve"))


@cache
def _pve_backends(target_name: str | None) -> tuple[ProximoConfig, ApiBackend, ExecBackend]:
    """Build + cache the PVE config and backends per target (registry is small/bounded)."""
    cfg = _resolve_pve_config(target_name)
    return cfg, ApiBackend(cfg), ExecBackend(cfg)


@lru_cache(maxsize=1)
def _instance_ledger() -> AuditLedger:
    """The PROVE ledger is ONE chain for this Proximo instance — built from the env audit_* config,
    never per-target, so every target's ops record to the same tamper-evident chain. Uses
    from_env_ledger() (audit fields only) so a pure-targets deployment stands the ledger up WITHOUT
    the single-target PVE API triple — otherwise `proximo doctor --target X` dies on a missing
    PROXIMO_API_BASE_URL the ledger never needed."""
    return open_ledger(ProximoConfig.from_env_ledger())


def _svc() -> tuple[ProximoConfig, ApiBackend, ExecBackend, AuditLedger]:
    """Config + backends for the ACTIVE pve target (contextvar; None => env), plus the one
    instance ledger. Backends are cached per target; the ledger is the single instance chain.

    STRICT by design: a pve_* tool body calls this and uses the backend, so a non-pve active
    target (e.g. someone aimed a pve_* tool at a pbs target) RAISES here (kind safety) rather
    than silently hitting the env box. Ledger-only callers use _ledger(), which tolerates that.
    """
    cfg, api, exec_backend = _pve_backends(active_target())
    return cfg, api, exec_backend, _instance_ledger()


def _ledger() -> AuditLedger:
    """The instance PROVE ledger (one chain), plane-independent.

    Reads _svc()[3] so the tests' _svc mock still injects a test ledger. Tolerates a non-pve
    active target (a pbs_*/pmg_*/pdm_* tool's ledger call, where _svc's pve resolution raises)
    by falling back to the instance ledger directly. This is the seam the ledger helpers use.

    The broad ProximoError catch is intentional: a non-PVE tool's ledger acquisition must not
    fail because the (unrelated) env PVE backend is misconfigured (e.g. verify_tls off w/o a CA).
    A genuine PVE problem still surfaces loudly when a pve_* tool runs, and at config-load warning
    time — it is not silently lost, only kept out of an unrelated plane's path."""
    try:
        return _svc()[3]
    except ProximoError:
        return _instance_ledger()


def _svc_cache_clear() -> None:
    """Clear every per-target backend cache (all four planes) and the instance-ledger cache.
    Preserves the `_svc.cache_clear()` API used by the tests; one call = a full reset."""
    _pve_backends.cache_clear()
    _pbs_backends.cache_clear()
    _pmg_backends.cache_clear()
    _pdm_backends.cache_clear()
    _instance_ledger.cache_clear()


_svc.cache_clear = _svc_cache_clear  # type: ignore[attr-defined]  # preserve existing test API


def _resolve_pbs_config(target_name: str | None) -> PbsConfig:
    if target_name is None:
        return PbsConfig.from_env()
    return PbsConfig.from_target(resolve_target_fields(target_name, "pbs"))


@cache
def _pbs_backends(target_name: str | None) -> tuple[PbsConfig, PbsBackend]:
    cfg = _resolve_pbs_config(target_name)
    return cfg, PbsBackend(cfg)


def _pbs() -> tuple[PbsConfig, PbsBackend]:
    """Lazily build the PBS backend — only when a pbs_* tool is called.

    Separate service from the PVE host: needs PROXIMO_PBS_* env (fails loud if unset).
    PBS ops still record to the SAME tamper-evident ledger via _audited/_plan (_svc's
    AuditLedger) so PROVE remains one coherent chain across PVE and PBS actions.
    """
    return _pbs_backends(active_target())


_pbs.cache_clear = _pbs_backends.cache_clear  # type: ignore[attr-defined]


def _resolve_pmg_config(target_name: str | None) -> PmgConfig:
    if target_name is None:
        return PmgConfig.from_env()
    return PmgConfig.from_target(resolve_target_fields(target_name, "pmg"))


@cache
def _pmg_backends(target_name: str | None) -> tuple[PmgConfig, PmgBackend]:
    cfg = _resolve_pmg_config(target_name)
    return cfg, PmgBackend(cfg)


def _pmg() -> tuple[PmgConfig, PmgBackend]:
    """Lazily build the PMG backend — only when a pmg_* tool is called.

    Separate service from the PVE host: needs PROXIMO_PMG_* env (fails loud if unset).
    PMG ops still record to the SAME tamper-evident ledger via _audited/_plan (_svc's
    AuditLedger) so PROVE remains one coherent chain across PVE, PBS, and PMG actions.
    """
    return _pmg_backends(active_target())


_pmg.cache_clear = _pmg_backends.cache_clear  # type: ignore[attr-defined]


def _resolve_pdm_config(target_name: str | None) -> PdmConfig:
    if target_name is None:
        return PdmConfig.from_env()
    return PdmConfig.from_target(resolve_target_fields(target_name, "pdm"))


@cache
def _pdm_backends(target_name: str | None) -> tuple[PdmConfig, PdmBackend]:
    cfg = _resolve_pdm_config(target_name)
    return cfg, PdmBackend(cfg)


def _pdm() -> tuple[PdmConfig, PdmBackend]:
    """Lazily build the PDM backend — only when a pdm_* tool is called.

    Separate service from the PVE host: needs PROXIMO_PDM_* env (fails loud if unset).
    PDM ops still record to the SAME tamper-evident ledger via _audited (_svc's
    AuditLedger) so PROVE remains one coherent chain across PVE, PBS, PMG, and PDM actions.
    """
    return _pdm_backends(active_target())


_pdm.cache_clear = _pdm_backends.cache_clear  # type: ignore[attr-defined]


def _untrusted_detail(action: str, detail: dict | None) -> dict | None:
    """Stamp untrusted provenance onto the ledger detail for an adversarial-classified tool
    (`taint.ADVERSARIAL_TOOLS`) — merged OVER the caller-supplied detail so a same-named key can
    never silently shadow it. Gated on `taint_tracking_on()`, same condition as the marker write:
    taint.py's own fail-closed invariant #1 is "all taint env unset => inert, zero behavior
    change," so with no PROXIMO_TAINT_* env set the ledger detail shape for every tool — adversarial
    or not — must stay byte-for-byte what it was before this module existed. Non-adversarial
    actions, or adversarial ones while tracking is off, pass `detail` through unchanged."""
    if not (is_adversarial(action) and taint_tracking_on()):
        return detail
    return {**(detail or {}), "untrusted": True, "content_trust": "adversarial"}


def _audited(action: str, target: str, fn: Callable[[], Any], *,
             mutation: bool = False, outcome: str = "ok", detail: dict | None = None) -> Any:
    """Run fn, then audit the REAL outcome. On exception, record the error and re-raise.

    `outcome` defaults to "ok" (synchronous completion). Async ops that only *start* a task pass
    outcome="submitted" so the ledger never claims an in-flight task is done.

    For mutation calls (mutation=True) the return is a SYMMETRIC envelope:
        {"status": <outcome>, "result": <raw fn() return>}
    where ``status`` equals the ``outcome`` recorded to the ledger — so a caller can uniformly
    read ``resp["status"]`` and it is always honest (never "ok" for an async/submitted op).

    Read calls (mutation=False) pass the raw fn() return through unchanged — no envelope.
    """
    audit = _ledger()
    # Containment gate (mutations only): an out-of-band trip (PROXIMO_CONTAIN_TRIP_PATH) refuses
    # every mutation, fail-closed, and records the blocked attempt to the same tamper-evident chain.
    # Checked BEFORE fn() so the mutation never fires; reads and the dry-run _plan() path are not
    # gated. There is no tool to clear the trip — re-arm is out-of-band, exactly like arm/disarm.
    # This is the SAME primitive (enforce_containment) that manual-audit-path tools call directly
    # (pve_agent_exec; ct_exec/ct_psql before their auto-undo snapshot) — one source of truth.
    #
    # Gate order — RATE moved to AFTER consent (envelope.py "Seam order" note): FORBID is a cheap
    # deny-list check with no budget cost, so it stays an early hard wall before consent; RATE is
    # the only gate that SPENDS shared budget, so it waits until consent has cleared. This closes
    # a real hole: an agent that repeatedly plans+confirms actions CONSENT refuses would otherwise
    # burn the whole box's rate budget on doomed attempts, denying the operator's own approved
    # mutations for the rest of the window. Same order at all 5 mutation seams.
    if mutation:
        enforce_containment(action, target, audit, detail=detail)
        enforce_scope(action, target, audit, detail=detail)
        enforce_lease(action, target, audit, detail=detail)
        enforce_envelope_forbid(action, target, audit, detail=detail)
        enforce_consent(action, target, audit, detail=detail)
        enforce_envelope_rate(action, target, audit, detail=detail)
    # TAINT (Component 2, taint.py): adversarial-classified tools carry guest/external-authored
    # bytes back to the calling agent. Set the sticky marker BEFORE fn() runs — so a call that
    # RAISES still taints (an error body can carry attacker-shaped content too), and so the
    # marker is in place before the ledger write below. A marker-WRITE failure must never crash
    # the tool call itself: is_tainted() already fails closed (any non-FileNotFound OSError/
    # ValueError -> tainted) on a broken/inaccessible marker dir, so swallowing here doesn't
    # weaken the invariant — it just keeps a filesystem hiccup from taking down an unrelated tool
    # call. Inert (no-op) unless taint_tracking_on() — default surface unchanged.
    if is_adversarial(action) and taint_tracking_on():
        try:
            mark_tainted(os.path.dirname(audit.path), action)
        except Exception as e:  # noqa: BLE001 — any marker-write failure must fail CLOSED (below)
            # FAIL-CLOSED: a taint-tracking deployment that cannot WRITE the marker must not run fn()
            # and hand back adversarial output untracked. A co-located attacker can force this branch
            # by planting a symlink at `.proximo-taint`/`.lock` (mark_tainted refuses symlinks with
            # OSError); a transient FS error would otherwise silently un-taint the session, because
            # is_tainted() then sees no marker (FileNotFoundError -> clean) and later mutations run
            # ungated. Record and refuse rather than serve untracked bytes. (The earlier "swallow —
            # is_tainted fails closed anyway" reasoning was wrong: it only holds when the marker DIR
            # is broken in a way is_tainted also trips, not when the write simply never lands.)
            audit.record(action, target=target, mutation=mutation, outcome="blocked:taint_mark_failed",
                         detail=_untrusted_detail(action, {**(detail or {}), "error": type(e).__name__}),
                         remote=ledger_remote())
            raise ProximoError(
                f"taint tracking is enabled but the taint marker could not be written for {action!r} "
                "— refusing to return untrusted output untracked (fail-closed)"
            ) from e
    # L16 NOTE (inherent): there is a narrow window between fn() completing and the outcome
    # audit.record() below. On process death (SIGKILL/OOM/power loss) in that window, the
    # mutation runs but only a "planned" ledger entry exists — the outcome is not recorded.
    # Compensating control: audit.record() calls fsync (audit.py line ~367) so in-process
    # crashes are covered; the Proxmox task log is the authoritative record for async ops.
    # PROVE is tamper-evident, not tamper-proof against OS-level death.
    # A full fix (pre-record an "executing" entry) is a deliberate design change, not a one-liner:
    # it introduces a new ledger outcome state that ripples into audit_verify chain-verification,
    # the ledger test suite, and any callers that inspect outcome values — deferred intentionally.
    try:
        result = fn()
    except Exception as e:
        audit.record(action, target=target, mutation=mutation, outcome="error",
                     detail=_untrusted_detail(action, {**(detail or {}), "error": type(e).__name__}),
                     remote=ledger_remote())
        raise
    audit.record(action, target=target, mutation=mutation, outcome=outcome,
                 detail=_untrusted_detail(action, detail), remote=ledger_remote())
    if mutation:
        return {"status": outcome, "result": fence_output(action, result)}
    return fence_output(action, result)


def _record_plan(plan: Plan) -> None:
    """Write the previewed plan (incl. the live state it was based on) to the tamper-evident ledger,
    with outcome="planned". This is the PLAN->PROVE weld: a verified chain shows the exact preview."""
    audit = _ledger()
    audit.record(
        plan.action, target=plan.target, mutation=True, outcome="planned",
        detail={"change": plan.change, "risk": plan.risk, "risk_reasons": plan.risk_reasons,
                "blast_radius": plan.blast_radius, "current": plan.current,
                "affected": plan.affected, "complete": plan.complete}, remote=ledger_remote())


def _plan(action: str, target: str, build: Callable[[], Plan]) -> Plan:
    """Build a plan and record it — MANDATORY before any mutation (no plan, no mutation).

    Called on BOTH paths: the dry-run (confirm=False) returns it; the execute path (confirm=True)
    runs it first so every mutation is preceded by a recorded "planned" entry — a one-shot confirm
    cannot bypass the preview. If building the plan fails (e.g. plan_power's live read raises),
    audit the failed probe and re-raise; never mutate without a recorded plan.
    """
    audit = _ledger()
    try:
        plan = build()
    except Exception as e:
        audit.record(action, target=target, mutation=True, outcome="error",
                     detail={"error": type(e).__name__, "phase": "planning"}, remote=ledger_remote())
        raise
    # The server tool name + target are AUTHORITATIVE for the ledger: stamp them onto the plan so the
    # "planned" entry pairs with the later "submitted"/"ok" entry under ONE action AND ONE target
    # (PROVE coherence) — a plan_* helper's internal label can never drift the audit trail (and shared
    # helpers like plan_create, used by both pve_create_container and pve_create_vm, record under the
    # right tool each time). 2026-07-10 audit L15: the node-lifecycle plane's factory target differed
    # from the wrapper's, so the planned and executed ledger entries carried mismatched targets.
    plan.action = action
    plan.target = target
    _record_plan(plan)
    # CONSENT: thread this plan's content id to the mutation seams and reset the per-operation
    # satisfied flag, so enforce_consent can require a per-plan out-of-band human grant. No-op
    # (inert) unless PROXIMO_CONSENT_DIR is set — zero behavior change for existing deployments.
    set_pending_consent(consent_id_for(plan))
    # RATE wall (envelope.py): reset the per-operation reservation flag for this FRESH mutation, so
    # a multi-seam op (ct_exec: its own body, _auto_undo, _audited) reserves exactly ONE rate-budget
    # slot rather than one per seam. Mirrors set_pending_consent's per-operation reset immediately
    # above.
    begin_operation()
    return plan


def _wait_task(api: ApiBackend, upid: str, node: str | None = None,
               timeout: int = 120, interval: int = 2) -> dict:
    """Poll a Proxmox task to completion. Snapshot ops are async; the auto-undo path must wait for
    the snapshot to actually finish before mutating. Raises if the task fails or times out."""
    deadline = time.monotonic() + timeout
    while True:
        st = api.task_status(upid, node)
        if st.get("status") == "stopped":
            # Strict: only an explicit "OK" passes. A stopped task that reports no exitstatus is
            # treated as failure (fail-closed), not silently assumed successful.
            exit_ = st.get("exitstatus")
            if exit_ != "OK":
                raise ProximoError(f"task {upid} did not finish OK: {exit_!r}")
            return st
        if time.monotonic() >= deadline:
            raise ProximoError(f"task {upid} timed out after {timeout}s")
        time.sleep(interval)


def _auto_undo(action: str, target: str, api: ApiBackend, vmid: str,
               detail: dict, kind: str = "lxc", node: str | None = None) -> dict:
    """Take a labeled undo snapshot and WAIT for it. On success returns the undo-point dict; on
    failure returns an {"status": "blocked:undo_unavailable"} dict (and audits it) — the caller MUST NOT
    mutate when unavailable (fail-closed: no net, no risky act)."""
    audit = _ledger()
    # Defense-in-depth: ct_exec/ct_psql already gate before calling _auto_undo, so this is a no-op
    # on those paths — but api.snapshot_create() is a REAL mutation, so any future caller that
    # forgets its own gate is still covered here.
    enforce_containment(action, target, audit, detail=detail)
    enforce_scope(action, target, audit, detail=detail)
    enforce_lease(action, target, audit, detail=detail)
    enforce_envelope_forbid(action, target, audit, detail=detail)
    enforce_consent(action, target, audit, detail=detail)
    enforce_envelope_rate(action, target, audit, detail=detail)
    snapname = undo_snapname()
    try:
        upid = api.snapshot_create(vmid, snapname, kind=kind, node=node,
                                   description="proximo auto-undo before mutation")
        _wait_task(api, upid, node=node)
    except Exception as e:
        audit.record(action, target=target, mutation=True, outcome="blocked:undo_unavailable",
                     detail={**detail, "error": type(e).__name__}, remote=ledger_remote())
        return {
            "status": "blocked:undo_unavailable",
            "message": ("Requested an undo snapshot but it could not be created/completed (the "
                        "container's storage may not support snapshots). Command NOT run "
                        "(fail-closed). Re-run without snapshot=True to proceed unprotected."),
            "error": type(e).__name__,
        }
    audit.record(action, target=target, mutation=True, outcome="undo_point",
                 detail={"snapshot": snapname, "task": upid}, remote=ledger_remote())
    return {"snapshot": snapname, "task": upid,
            "revert": f"pve_rollback vmid={vmid} snapname={snapname}",
            "note": ("undo points are NOT auto-pruned — they accumulate and consume storage; "
                     "delete with pve_snapshot_delete when no longer needed.")}


def _blocked(action: str, target: str, outcome: str, message: str, detail: dict | None = None,
            *, mutation: bool = True) -> dict:
    """Shared body for the four 'refuse + audit' helpers below."""
    audit = _ledger()
    audit.record(action, target=target, mutation=mutation, outcome=outcome,
                 detail=detail, remote=ledger_remote())
    return {"status": outcome, "message": message}


def _blocked_allowlist(action: str, target: str, detail: dict | None = None,
                       *, mutation: bool = True) -> dict:
    """Refuse + audit a container op whose CTID isn't on the allowlist (fail-closed), as a clean dict
    — checked at the server layer BEFORE any snapshot/exec, so a forbidden CTID never gets touched.
    `mutation` must reflect the GATED tool's true class so blocked reads don't ledger as mutations."""
    return _blocked(action, target, "blocked:allowlist",
                    f"CTID {target} is not permitted by the allowlist (fail-closed).",
                    detail, mutation=mutation)


def _exec_disabled(action: str, target: str, detail: dict | None = None,
                   *, mutation: bool = True) -> dict:
    """In-container exec is off by default (safe). Refuse + audit; explain how to opt in.
    `mutation` must reflect the GATED tool's true class so blocked reads don't ledger as mutations."""
    return _blocked(action, target, "blocked:exec_disabled",
                    ("In-container exec is disabled (safe default: API-only). It grants near-root on the "
                     "PVE host; enable deliberately with PROXIMO_ENABLE_EXEC=1."),
                    detail, mutation=mutation)


def _agent_disabled(action: str, target: str, detail: dict | None = None,
                    *, mutation: bool = True) -> dict:
    """qemu-agent ops are off by default. Refuse + audit; explain how to opt in.
    `mutation` must reflect the GATED tool's true class so blocked reads don't ledger as mutations."""
    return _blocked(action, target, "blocked:agent_disabled",
                    ("qemu-agent ops are disabled (safe default: API-only). "
                     "Enable with PROXIMO_ENABLE_AGENT=1 and set PROXIMO_AGENT_ALLOWLIST."),
                    detail, mutation=mutation)


def _blocked_agent_allowlist(action: str, target: str, detail: dict | None = None,
                              *, mutation: bool = True) -> dict:
    """Refuse + audit a qemu-agent op whose VMID isn't on the allowlist (fail-closed).
    `mutation` must reflect the GATED tool's true class so blocked reads don't ledger as mutations."""
    return _blocked(action, target, "blocked:allowlist",
                    f"Guest {target} is not permitted by the agent allowlist (fail-closed).",
                    detail, mutation=mutation)


def _agent_gate(cfg, action: str, vmid: str, *, mutation: bool) -> dict | None:
    """Shared qemu-agent gate: off-by-default, then allowlist (fail-closed), in order.
    Returns the blocked-response dict (already recorded to the ledger) if refused, or
    None to proceed. `mutation` must reflect the GATED tool's true class so blocked reads
    don't ledger as mutations."""
    if not cfg.enable_agent:
        return _agent_disabled(action, f"qemu/{vmid}", mutation=mutation)
    if not cfg.agent_permitted(vmid):
        return _blocked_agent_allowlist(action, f"qemu/{vmid}", mutation=mutation)
    return None


# --- In-container exec (ssh -> pct) — MUTATION-CAPABLE, confirm-gated ---

@tool()
def ct_exec(
    ctid: Annotated[str, Field(description="Numeric CTID of the target LXC container (allowlist-scoped).")],
    command: Annotated[list[str], Field(description="Argv list to run inside the container (not a shell string).")],
    snapshot: Annotated[bool, Field(description="Take a fail-closed auto-undo snapshot before running.")] = False,
    confirm: Annotated[bool, Field(description="False (default) returns a dry-run PLAN; true executes.")] = False,
) -> dict:
    """Run a command inside an LXC (ssh -> pct exec). MUTATION-CAPABLE.

    Dry-run by default: without confirm=True you get a PLAN — the command plus a heuristic
    read-vs-write / destructive-pattern classification (advisory only) — recorded to the ledger.
    Re-call with confirm=True to execute. Disabled unless PROXIMO_ENABLE_EXEC is set (safe default
    is API-only). Allowlist-scoped (fail-closed) and audited.

    snapshot=True (UNDO): take an auto-undo snapshot first and WAIT for it; if it can't be made
    (e.g. storage doesn't support snapshots) the command is NOT run (fail-closed). On success the
    result carries an `undo_point` you can revert with pve_rollback.
    """
    cfg, api, exec_, audit = _svc()
    # Audit completeness is the default; PROXIMO_LEDGER_REDACT records a command fingerprint instead
    # of the argv (which can carry secrets, e.g. `--password ...`) — see audit.py + README.
    detail = command_fingerprint(command) if cfg.redact_ledger else {"command": command}
    if not cfg.enable_exec:
        return _exec_disabled("ct_exec", str(ctid), detail)
    ctid = _check_vmid(ctid)  # L07: validate CTID format at server layer before allowlist gate
    if not cfg.ct_permitted(ctid):
        return _blocked_allowlist("ct_exec", str(ctid), detail)
    plan = _plan("ct_exec", str(ctid), lambda: plan_exec(ctid, command, redact=cfg.redact_ledger))
    if not confirm:
        return {"status": "plan", "auto_snapshot": snapshot, **plan.as_dict()}

    # Containment gate BEFORE the auto-undo snapshot (which fires outside _audited) — refuse the
    # WHOLE operation while contained, not just the exec half. Same primitives + same gate order
    # _audited uses (RATE after consent — see the order comment there / envelope.py).
    enforce_containment("ct_exec", str(ctid), audit, detail=detail)
    enforce_scope("ct_exec", str(ctid), audit, detail=detail)
    enforce_lease("ct_exec", str(ctid), audit, detail=detail)
    enforce_envelope_forbid("ct_exec", str(ctid), audit, detail=detail)
    enforce_consent("ct_exec", str(ctid), audit, detail=detail)
    enforce_envelope_rate("ct_exec", str(ctid), audit, detail=detail)

    undo_point = None
    if snapshot:
        undo = _auto_undo("ct_exec", str(ctid), api, ctid, detail)
        if undo.get("status") == "blocked:undo_unavailable":
            return undo  # fail-closed: command NOT run
        undo_point = undo

    def _do() -> dict:
        r = exec_.run(ctid, command)
        out = {"returncode": r.returncode, "stdout": r.stdout, "stderr": r.stderr}
        if undo_point:
            out["undo_point"] = undo_point
        return out

    return _audited("ct_exec", str(ctid), _do, mutation=True,
                    detail={**detail, "confirmed": True, "undo": bool(undo_point)})


@tool()
def ct_psql(
    ctid: Annotated[str, Field(description="Numeric CTID of the container running PostgreSQL (allowlist-scoped).")],
    sql: Annotated[str, Field(description="SQL to run via psql inside the container, as the database OS user.")],
    db: Annotated[str, Field(description="Target database name.")] = "postgres",
    snapshot: Annotated[bool, Field(description="Take a fail-closed auto-undo snapshot before running.")] = False,
    confirm: Annotated[bool, Field(description="False (default) returns a dry-run PLAN; true executes.")] = False,
) -> dict:
    """Run SQL via psql inside a container (as the db OS user). MUTATION-CAPABLE.

    Dry-run by default: without confirm=True you get a PLAN — the SQL plus a heuristic
    read/DML/DDL classification (advisory only) — recorded to the ledger. Re-call with
    confirm=True to execute.

    snapshot=True (UNDO): take an auto-undo snapshot first and WAIT for it; if it can't be made the
    SQL is NOT run (fail-closed). On success the result carries an `undo_point` (revert via pve_rollback).
    """
    cfg, api, exec_, audit = _svc()
    # Audit completeness is the default; PROXIMO_LEDGER_REDACT records a fingerprint instead of
    # the body (which can carry secrets/PII) — see audit.py + README.
    detail = {"db": db, **(sql_fingerprint(sql) if cfg.redact_ledger else {"sql": sql})}
    if not cfg.enable_exec:
        return _exec_disabled("ct_psql", str(ctid), detail)
    ctid = _check_vmid(ctid)  # L07: validate CTID format at server layer before allowlist gate
    if not cfg.ct_permitted(ctid):
        return _blocked_allowlist("ct_psql", str(ctid), detail)
    plan = _plan("ct_psql", str(ctid), lambda: plan_psql(ctid, sql, db=db, redact=cfg.redact_ledger))
    if not confirm:
        return {"status": "plan", "auto_snapshot": snapshot, **plan.as_dict()}

    # Containment gate BEFORE the auto-undo snapshot (which fires outside _audited) — refuse the
    # WHOLE operation while contained, not just the exec half. Same primitives + same gate order
    # _audited uses (RATE after consent — see the order comment there / envelope.py).
    enforce_containment("ct_psql", str(ctid), audit, detail=detail)
    enforce_scope("ct_psql", str(ctid), audit, detail=detail)
    enforce_lease("ct_psql", str(ctid), audit, detail=detail)
    enforce_envelope_forbid("ct_psql", str(ctid), audit, detail=detail)
    enforce_consent("ct_psql", str(ctid), audit, detail=detail)
    enforce_envelope_rate("ct_psql", str(ctid), audit, detail=detail)

    undo_point = None
    if snapshot:
        undo = _auto_undo("ct_psql", str(ctid), api, ctid, detail)
        if undo.get("status") == "blocked:undo_unavailable":
            return undo  # fail-closed: SQL NOT run
        undo_point = undo

    def _do() -> dict:
        r = exec_.psql(ctid, sql, db=db)
        out = {"returncode": r.returncode, "stdout": r.stdout, "stderr": r.stderr}
        if undo_point:
            out["undo_point"] = undo_point
        return out

    return _audited("ct_psql", str(ctid), _do, mutation=True,
                    detail={**detail, "confirmed": True, "undo": bool(undo_point)})


def _anchor_moved_hint(prev_entries: int | None, cur_entries: int) -> str:
    """Explain a head that has moved past the off-box anchor pin, using pinned vs live entry counts
    so a routine forward-grow reads as benign stale-pin lag and a shrink reads as a real
    truncation/wipe alarm. The pin is intentionally NOT auto-advanced here (anti-poisoning; see
    audit_anchor.py) — re-pin deliberately once the ledger is confirmed intact."""
    base = (
        "the live ledger head has moved past the off-box anchor pin; the pin was NOT auto-advanced "
        "(advancing it on a possibly-tampered ledger would poison the anchor). "
    )
    if prev_entries is None:
        return base + "Re-pin the anchor deliberately once you've confirmed the ledger is intact."
    if cur_entries > prev_entries:
        return base + (
            f"The ledger has MORE entries than the pin ({cur_entries} > {prev_entries}) — "
            "consistent with legitimate forward growth; re-pin when you've confirmed the new "
            "entries are genuine. (A forged tail-append also grows the count, so confirm via the "
            "chain, not the count alone.)"
        )
    if cur_entries < prev_entries:
        return base + (
            f"The ledger has FEWER entries than the pin ({cur_entries} < {prev_entries}) — a "
            "TRUNCATION or WIPE signal. INVESTIGATE the ledger and the sink."
        )
    return base + (
        f"Same entry count ({cur_entries}) but a different head — the tail was rewritten or forged. "
        "INVESTIGATE."
    )


    # THIS Proximo's one PROVE ledger chain, which has no remote box to target. It is the
    # sole intentionally-bare tool; every other tool (incl. the ct_* exec tools) is @tool().
@mcp.tool()
def audit_verify(
    expected_head: Annotated[
        str | None,
        Field(
            description="64-char hex head() value pinned off-box; verifying against it also catches "
            "tail truncation, a forged tail-append, or a full ledger replacement. Omit to fall "
            "back to PROXIMO_AUDIT_EXPECTED_HEAD."
        ),
    ] = None,
) -> dict:
    """Verify the tamper-evident audit ledger's hash chain — PROVE the log is intact.

    Pass `expected_head` (the head() value you pinned off-box) to also catch tail
    truncation, a forged tail-append, or a full file replacement — a forward walk
    alone can't see those. Falls back to PROXIMO_AUDIT_EXPECTED_HEAD when omitted.
    """
    cfg, _, _, audit = _svc()
    pin = expected_head if expected_head is not None else cfg.expected_head
    if pin is not None:
        # Normalize a copy-pasted head (case-insensitive hexdigest; strip stray spaces/newline) the
        # same way config does — a blank/whitespace value becomes "unpinned", not a caller error.
        pin = pin.strip().lower() or None
    if pin is not None and not looks_like_head(pin):
        # A genuinely malformed pin is a CALLER error, not tamper — raise clearly instead of
        # letting it fall through to a "head mismatch" that cries wolf.
        raise ProximoError(
            f"invalid expected_head: {pin!r} (must be a 64-char hex head() value)"
        )
    v = audit.verify(expected_head=pin)
    # When nothing is pinned, the forward walk can't see tail truncation / forged append / wipe —
    # nudge the operator to anchor the head off-box (the strong guarantee), so the feature isn't
    # silently unused. No nudge once a pin is in effect.
    hint = None if pin is not None else (
        "not pinned against tail attacks: set PROXIMO_AUDIT_EXPECTED_HEAD (or pass expected_head=) "
        "to the current 'head' value, stored off-box, to detect tail truncation / forged append / "
        "full wipe — the off-box anchor is the strong guarantee."
    )
    # A pinned "head mismatch" with the chain otherwise intact is byte-identical whether it's a tail
    # attack or a keyed-default upgrade that rotated the head. If a rotation archive sits beside the
    # ledger, say so — the stderr migration warning is often swallowed by MCP stdio clients.
    rotation_hint = None
    if not v.ok and v.broken_at is None and pin is not None:
        archive = find_rotation_archive(audit.path)
        if archive:
            rotation_hint = (
                "a keyed-default migration archive sits beside this ledger "
                f"({os.path.basename(archive)!r}). If you upgraded Proximo since you pinned, this "
                "'head mismatch' is the expected migration head-rotation — re-pin "
                "PROXIMO_AUDIT_EXPECTED_HEAD to the 'head' value above. If you did NOT just upgrade, "
                "treat this as a genuine tail-attack signal and investigate."
            )
    # Off-box anchor: if a sink is configured, keep it in step with the ledger — but SAFELY.
    # getattr keeps the tests' SimpleNamespace cfgs (no anchor_sink attr) working.
    #
    # ANTI-POISONING INVARIANT (see audit_anchor.py): the on-demand export advances the off-box pin
    # ONLY on a first run (no pin yet) or when the live head is UNCHANGED. It NEVER re-pins to a
    # head that has MOVED — otherwise a verify that just detected a truncation/wipe would overwrite
    # the good pin with the tampered head and hide the attack after the next restart. A moved head
    # is surfaced as anchor_hint (count-directional) instead; advancing the pin past it is the
    # operator's deliberate act.
    #
    # FAIL-CLOSED: a sink read or publish failure is NOT swallowed into a green verify — a
    # configured anchor that can't be reached is suspicious, so we refuse the call rather than let
    # the pin go silently stale.
    anchor = getattr(cfg, "anchor_sink", None)
    anchor_name = None
    anchor_last_export = None
    anchor_hint = None
    if anchor is not None:
        anchor_name = anchor.name
        try:
            prev = anchor.last_pin()
        except AnchorError as e:
            raise ProximoError(
                f"off-box audit anchor is unreachable: {e}. Refusing the verify (fail-closed; a "
                "verify that can't consult its own tamper anchor is not a clean check)."
            ) from e
        head_now = audit.head()
        prev_head = prev["head"] if prev else None
        if prev_head is None or prev_head == head_now:
            # First run (establish the pin) or head unchanged (idempotent re-pin): safe to publish.
            ts = datetime.now(UTC).isoformat()
            try:
                anchor.publish(head_now, ts, cfg.node, audit.path, entries=v.entries)
            except AnchorError as e:
                raise ProximoError(
                    f"off-box audit anchor export failed: {e}. The anchor pin could not be updated "
                    "— fix the sink and retry (fail-closed; the verify result is withheld so a "
                    "stale off-box pin is not mistaken for a clean check)."
                ) from e
            anchor_last_export = ts
        else:
            # Head MOVED from the pinned head: do NOT auto-advance (anti-poisoning). Explain which
            # way it moved via the pinned vs live entry count so a forward-grow reads as benign lag
            # and a shrink reads as a real truncation/wipe alarm.
            anchor_hint = _anchor_moved_hint(prev.get("entries"), v.entries)
    return {
        "ok": v.ok,
        "entries": v.entries,
        "broken_at_line": v.broken_at,
        "reason": v.reason,
        "head": audit.head(),
        "expected_head": pin,
        "keyed": audit.keyed,
        "hint": hint,
        "rotation_hint": rotation_hint,
        "anchor_sink": anchor_name,
        "anchor_last_export": anchor_last_export,
        "anchor_hint": anchor_hint,
    }


# ---------------------------------------------------------------------------
# qemu-agent plane (Wave 3) — in-guest ops via the QEMU Guest Agent
# ---------------------------------------------------------------------------

# Pace the exec-status poll loop so it never busy-waits the PVE API (mirrors _wait_task's sleep).
_AGENT_POLL_INTERVAL = 1.0


@tool()
def pve_agent_exec(
    vmid: Annotated[str, Field(description="Numeric VMID of the target QEMU guest (allowlist-scoped).")],
    command: Annotated[list[str], Field(description="Argv list to run in the guest via the qemu-agent.")],
    node: Annotated[str | None, Field(description="PVE node the guest runs on; omit to resolve automatically.")] = None,
    timeout: Annotated[int, Field(description="Seconds to poll for exit before returning status='running'.")] = 30,
    confirm: Annotated[bool, Field(description="False (default) returns a dry-run PLAN; true executes.")] = False,
) -> dict:
    """MUTATION: run a command inside a guest via the qemu-agent (async, polls for result).

    Dry-run by default: without confirm=True you get a PLAN recorded to the ledger.
    Re-call with confirm=True to execute.

    Requires PROXIMO_ENABLE_AGENT=1 and the VMID in PROXIMO_AGENT_ALLOWLIST.
    The command runs INSIDE the guest OS — no undo primitive on this plane.

    Returns status="ok" only when the agent reports the process exited.
    Returns status="running" with pid when the poll deadline is reached before exit.
    """
    cfg, api, _, audit = _svc()
    blocked = _agent_gate(cfg, "pve_agent_exec", vmid, mutation=True)
    if blocked:
        return blocked

    # Ledger redaction parity with ct_exec: a guest exec argv can carry a secret (e.g. `mysql -pPW`).
    # When PROXIMO_LEDGER_REDACT is set, store a fingerprint instead of the argv — in BOTH the plan's
    # change line (via redact=) and the execute-path audit detail.
    detail = command_fingerprint(command) if cfg.redact_ledger else {"command": command}
    plan = _plan("pve_agent_exec", f"qemu/{vmid}",
                 lambda: plan_agent_exec(vmid, command, node, redact=cfg.redact_ledger))
    if not confirm:
        return {"status": "plan", **plan.as_dict()}

    # Containment gate: this tool has a manual audit path (below) that never runs through
    # _audited(), so it must call the same primitives directly, in the SAME order (RATE after
    # consent — see the order comment in _audited / envelope.py), BEFORE the real guest-OS
    # mutation (api.agent_exec) fires — outside the try/except so a refusal here is never
    # re-caught and re-recorded as outcome="error".
    enforce_containment("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)
    enforce_scope("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)
    enforce_lease("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)
    enforce_envelope_forbid("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)
    enforce_consent("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)
    enforce_envelope_rate("pve_agent_exec", f"qemu/{vmid}", audit, detail=detail)

    # TAINT: pve_agent_exec is a manual-audit-path tool (never runs through _audited(), see the
    # comment above) but IS adversarial-classified — the guest OS controls out-data/err-data.
    # Same construction as _audited: mark BEFORE the real guest exec fires (so a call that raises
    # still taints) and FAIL-CLOSED if the marker can't be written — refuse rather than run the
    # guest exec and return its output untracked (a planted symlink on the marker dir/lock, or a
    # transient FS error, would otherwise silently un-taint the session).
    if taint_tracking_on():
        try:
            mark_tainted(os.path.dirname(audit.path), "pve_agent_exec")
        except Exception as e:  # noqa: BLE001 — any marker-write failure must fail CLOSED (below)
            audit.record("pve_agent_exec", target=f"qemu/{vmid}", mutation=True,
                         outcome="blocked:taint_mark_failed",
                         detail=_untrusted_detail("pve_agent_exec", {"error": type(e).__name__}),
                         remote=ledger_remote())
            raise ProximoError(
                "taint tracking is enabled but the taint marker could not be written for "
                "'pve_agent_exec' — refusing to return untrusted output untracked (fail-closed)"
            ) from e

    # Execute: POST exec, then poll exec-status until exited or deadline.
    # Manual audit path so we can record honest outcome ("ok" vs "running").
    try:
        exec_result = api.agent_exec(vmid, node, command)
        pid = exec_result.get("pid")
        if pid is None:
            raise ValueError("agent exec returned no pid")  # noqa: TRY301

        # VERIFIED live (PVE 9.2): exec-status returns exited/exitcode/out-data/err-data.
        deadline = time.monotonic() + timeout
        while True:
            status = api.agent_exec_status(vmid, node, pid)
            # 'exited' arrives as a JSON bool; accept int 1 too defensively, and NEVER treat a
            # falsy/missing value as completion (that would fake an "ok" for a still-running cmd).
            if status.get("exited") in (True, 1):
                # Process completed — honest "ok" outcome. out-data/err-data are plain text (not base64).
                out_data = status.get("out-data", "")
                err_data = status.get("err-data", "")
                result = {
                    "pid": pid,
                    "exitcode": status.get("exitcode"),
                    "out-data": out_data,
                    "err-data": err_data,
                }
                audit.record("pve_agent_exec", target=f"qemu/{vmid}", mutation=True, outcome="ok",
                             detail=_untrusted_detail("pve_agent_exec",
                                                      {**detail, "confirmed": True, "pid": pid}),
                             remote=ledger_remote())
                # Fence ONLY the `result` field (the guest-controlled out-data/err-data), keeping the
                # top-level `status` intact — same symmetric-envelope contract _audited honors for
                # ct_exec/ct_psql. Fencing the whole {status,result} dict would bury `status` inside
                # the JSON string and break `resp["status"]`. fence_output is a no-op unless FENCE is on.
                return {"status": "ok", "result": fence_output("pve_agent_exec", result)}
            if time.monotonic() >= deadline:
                # Timeout BEFORE exit observed — honest "running" outcome, never "ok". This branch
                # carries NO guest output (the command hasn't produced out-data yet) — only status,
                # pid, and a Proximo-authored message — so there is nothing adversarial to fence.
                audit.record("pve_agent_exec", target=f"qemu/{vmid}", mutation=True,
                             outcome="running",
                             detail=_untrusted_detail(
                                 "pve_agent_exec",
                                 {**detail, "confirmed": True, "pid": pid, "timeout": timeout}),
                             remote=ledger_remote())
                return {
                    "status": "running", "pid": pid,
                    "message": f"command is still running (pid={pid}) — did not exit within {timeout}s; "
                               "poll pve_agent_info with command='exec-status' and the returned pid."}
            time.sleep(_AGENT_POLL_INTERVAL)  # pace polls — do not hammer the PVE API
    except Exception as e:
        audit.record("pve_agent_exec", target=f"qemu/{vmid}", mutation=True, outcome="error",
                     detail=_untrusted_detail("pve_agent_exec",
                                              {"error": type(e).__name__, "confirmed": True}),
                     remote=ledger_remote())
        raise


# --- PROXIMO_SURFACES — opt-in registration scoping (context hygiene + surface reduction). ---
# Unset/empty => all tools, zero behavior change. Set (e.g. "pve,exec") => tools of unpicked
# planes are REMOVED from the registry before serving, so they never reach the client's context
# at all — a structural gate, not a runtime refusal. Applied in main() AFTER load_env_file()
# (registration happens at import, before the env file is read — pruning post-load is what makes
# a file-set PROXIMO_SURFACES actually work; same footgun CONSENT hit in 0.13). audit_verify is
# always kept: PROVE is never scopeable away.
SURFACES: dict[str, tuple[str, ...]] = {
    "pve": ("pve_",),   # Proxmox VE (includes the pve_agent_* qemu-agent edge)
    "pbs": ("pbs_",),   # Proxmox Backup Server
    "pmg": ("pmg_",),   # Proxmox Mail Gateway
    "pdm": ("pdm_",),   # Proxmox Datacenter Manager (reads + governed fleet control: power/snapshot/migrate)
    "exec": ("ct_",),   # in-container exec/psql/logs/diagnose (ssh -> pct)
}
_ALWAYS_REGISTERED = frozenset({"audit_verify"})


def surface_keep(names: Iterable[str], spec: str | None) -> set[str]:
    """Pure filter: which tool names stay registered under a PROXIMO_SURFACES spec.
    None/blank => everything (inert). Unknown surface name => ValueError — a typo must
    refuse startup, never silently serve a different surface than the operator believes."""
    names = set(names)
    if spec is None or not spec.strip():
        return names
    picked = [t.strip().lower() for t in spec.split(",") if t.strip()]
    unknown = sorted(set(picked) - set(SURFACES))
    if unknown:
        raise ValueError(
            f"PROXIMO_SURFACES: unknown surface(s) {unknown} — valid: {sorted(SURFACES)} "
            "(refusing to start rather than serve a surface you didn't pick)")
    prefixes = tuple(p for t in picked for p in SURFACES[t])
    return {n for n in names if n.startswith(prefixes) or n in _ALWAYS_REGISTERED}


_TRUEISH = frozenset({"1", "true", "yes", "on"})


def configured_surfaces() -> set[str]:
    """Which planes are actually configured on this box — detected, not declared.

    A plane counts as configured when its env base URL is present OR a target of that
    kind exists in PROXIMO_TARGETS. `exec` counts only when PROXIMO_ENABLE_EXEC is on.
    This is what lets a PVE+PBS-only box auto-serve just those planes' tools — no flag.
    """
    found: set[str] = set()
    for plane, env in (("pve", "PROXIMO_API_BASE_URL"), ("pbs", "PROXIMO_PBS_BASE_URL"),
                       ("pmg", "PROXIMO_PMG_BASE_URL"), ("pdm", "PROXIMO_PDM_BASE_URL")):
        if os.environ.get(env, "").strip():
            found.add(plane)
    try:  # a target of any kind configures that plane; a broken registry must not crash startup
        for fields in load_registry().values():
            kind = fields.get("kind")
            if kind in SURFACES:
                found.add(kind)
    except Exception:  # noqa: S110 — broken registry surfaces elsewhere; detection must not crash startup
        pass
    if os.environ.get("PROXIMO_ENABLE_EXEC", "").strip().lower() in _TRUEISH:
        found.add("exec")
    return found


def _prune_registry(server_mcp, keep: set[str], reason: str) -> None:
    registry = server_mcp._tool_manager._tools
    total = len(registry)
    for name in [n for n in registry if n not in keep]:
        server_mcp.remove_tool(name)
    print(f"proximo: {reason} — {len(keep)} of {total} tools registered", file=sys.stderr)


def _apply_surfaces(server_mcp=mcp) -> None:
    """Scope the live registry to the planes in use. ValueError propagates to main().

    Precedence: (1) explicit PROXIMO_SURFACES wins verbatim — including `all` to force the
    full surface; (2) otherwise auto-scope to the *configured* planes (default-on; disable
    with PROXIMO_AUTOSCOPE=off); (3) if nothing is detectable, serve everything (never
    surprise an operator with an empty server when config is ambiguous)."""
    spec = os.environ.get("PROXIMO_SURFACES")
    if spec and spec.strip():
        if spec.strip().lower() == "all":   # explicit escape hatch: serve the full surface
            return
        registry = server_mcp._tool_manager._tools
        _prune_registry(server_mcp, surface_keep(registry.keys(), spec), f"PROXIMO_SURFACES={spec.strip()}")
        return
    if os.environ.get("PROXIMO_AUTOSCOPE", "").strip().lower() in ("off", "0", "false", "no"):
        return
    planes = configured_surfaces()
    if not (planes - {"exec"}):   # no data plane detected → ambiguous, serve all (touch nothing)
        return
    registry = server_mcp._tool_manager._tools
    keep = surface_keep(registry.keys(), ",".join(sorted(planes)))
    if len(keep) < len(registry):   # only announce/prune when it actually narrows
        _prune_registry(server_mcp, keep, f"auto-scoped to configured planes ({','.join(sorted(planes))})")


def main() -> None:
    # Source ~/.config/proximo/proximo.env FIRST (before doctor or any from_env) so a PROXIMO_* var
    # set in the documented file actually reaches the stdio server — otherwise it is silently ignored,
    # which is fail-dangerous for a security gate like PROXIMO_CONSENT_DIR. Real/inline env still wins.
    load_env_file()
    try:
        _apply_surfaces()
    except ValueError as e:
        print(f"proximo: {e}", file=sys.stderr)
        raise SystemExit(1) from None
    # `proximo doctor` — verify your token/config (read-only preflight) BEFORE wiring Proximo into
    # an AI client. Prints what THIS token can and cannot do; never starts the server.
    if len(sys.argv) > 1 and sys.argv[1] == "doctor":
        import argparse
        import json
        parser = argparse.ArgumentParser(prog="proximo doctor", add_help=False)
        parser.add_argument("--target", default=None,
                            help="Named target from PROXIMO_TARGETS registry to probe.")
        args = parser.parse_args(sys.argv[2:])
        try:
            result = pve_doctor(proximo_target=args.target)
        except Exception as e:  # config/token/connectivity problem — give a plain message, not a trace
            print(f"proximo doctor: {e}", file=sys.stderr)
            raise SystemExit(1) from None
        # Credential-free by construction: `result` is doctor_check's advisory report — version,
        # capability lists, and config POSTURE (node, base_url, TLS bools, CA path, allowlist
        # counts). The token secret / PMG password are read only to build the auth header and are
        # never serialized into this report (verified: the secret appears 0x in the output). The
        # CodeQL py/clear-text-logging-sensitive-data flag here is a taint over-approximation
        # through the shared config object, not a real disclosure.
        print(json.dumps(result, indent=2))
        return
    # `proximo mint` — print-only onboarding recipe: create → write → grant → wire → verify.
    # Prints the exact runbook for a least-privilege credential per product; makes NO API call,
    # never handles a secret, and never starts the server. Hands off to `proximo doctor`.
    if len(sys.argv) > 1 and sys.argv[1] == "mint":
        import argparse
        import json

        from proximo.mint import PRODUCTS, build_recipe, render_text
        parser = argparse.ArgumentParser(prog="proximo mint")
        parser.add_argument("--product", default="pve",
                            help=f"one of: {', '.join(PRODUCTS)} (default: pve)")
        parser.add_argument("--user", default=None,
                            help="service user (default: proximo@<product-realm>)")
        parser.add_argument("--token-name", default="mcp",
                            help="token name (default: mcp; unused for pmg)")
        parser.add_argument("--token-file", default=None,
                            help="credential file (default: ~/.config/proximo/<product>.token)")
        parser.add_argument("--write", action="store_true",
                            help="print the scoped WRITE grant instead of the read-only default")
        parser.add_argument("--json", action="store_true",
                            help="emit the recipe as structured JSON (mirrors doctor)")
        args = parser.parse_args(sys.argv[2:])
        try:
            recipe = build_recipe(product=args.product, user=args.user,
                                  token_name=args.token_name, token_file=args.token_file,
                                  write=args.write)
        except ValueError as e:
            print(f"proximo mint: {e}", file=sys.stderr)
            raise SystemExit(2) from None
        print(json.dumps(recipe, indent=2) if args.json else render_text(recipe))
        return
    # `proximo hello` — the print-only agent front door: the six-move welcome, sharp
    # edges first, the ask last. Makes NO API call, sends nothing, never starts the
    # server; --sign only PRINTS the gh command an agent would run by its own hand.
    if len(sys.argv) > 1 and sys.argv[1] == "hello":
        import argparse
        import json

        from proximo.hello import build_greeting
        from proximo.hello import render_text as render_hello
        parser = argparse.ArgumentParser(prog="proximo hello")
        parser.add_argument("--sign", default=None, metavar="NOTE",
                            help="print (never run) the gh command that would post NOTE"
                                 " to the Agent Guestbook")
        parser.add_argument("--json", action="store_true",
                            help="emit the greeting as structured JSON (mirrors doctor/mint)")
        args = parser.parse_args(sys.argv[2:])
        if args.sign is not None and not args.sign.strip():
            print("proximo hello: --sign needs a non-empty note", file=sys.stderr)
            raise SystemExit(2)
        greeting = build_greeting(sign=args.sign)
        print(json.dumps(greeting, indent=2) if args.json else render_hello(greeting))
        return
    print(BANNER, file=sys.stderr)
    mcp.run()


# --- Re-exports: every tool moved to proximo.tools.* is re-imported here by name so that
# (a) importing proximo.server still registers every tool with FastMCP as a side effect
#     (the exact count is machine-checked by tests/test_tool_count.py, not asserted in prose here),
# and (b) the existing `server.<tool_name>` surface (direct-call tests, CLI, introspection
# sweeps that do `getattr(server, name)`) keeps working unchanged. ---
from proximo import prompts as _prompts  # noqa: E402,F401  # safe-runbook MCP prompts (registration side effect)
from proximo.tools.pbs import (  # noqa: E402,F401
    pbs_datastore_create,
    pbs_datastore_delete,
    pbs_datastore_get,
    pbs_datastore_status,
    pbs_datastore_update,
    pbs_datastores_list,
    pbs_gc_start,
    pbs_gc_status,
    pbs_group_change_owner,
    pbs_jobs_list,
    pbs_namespace_create,
    pbs_namespace_delete,
    pbs_namespaces_list,
    pbs_prune,
    pbs_remote_create,
    pbs_remote_delete,
    pbs_remote_get,
    pbs_remote_update,
    pbs_remotes_list,
    pbs_snapshot_delete,
    pbs_snapshot_notes_set,
    pbs_snapshot_protected_set,
    pbs_snapshots_list,
    pbs_tasks_list,
    pbs_traffic_control_delete,
    pbs_traffic_control_upsert,
    pbs_traffic_controls_list,
    pbs_verify_start,
)
from proximo.tools.pdm import (  # noqa: E402,F401
    pdm_acl_list,
    pdm_node_status,
    pdm_pbs_datastores_list,
    pdm_pbs_remote_status,
    pdm_pbs_snapshots_list,
    pdm_ping,
    pdm_pve_cluster_status,
    pdm_pve_lxc_config,
    pdm_pve_lxc_list,
    pdm_pve_node_list,
    pdm_pve_qemu_config,
    pdm_pve_qemu_list,
    pdm_pve_resources,
    pdm_remote_config_get,
    pdm_remote_version,
    pdm_remotes_list,
    pdm_resources_list,
    pdm_resources_status,
    pdm_roles_list,
    pdm_tasks_list,
    pdm_users_list,
    pdm_version,
)
from proximo.tools.pdm_fleet import (  # noqa: E402,F401
    pdm_pve_lxc_migrate,
    pdm_pve_lxc_power,
    pdm_pve_lxc_remote_migrate,
    pdm_pve_lxc_snapshot_create,
    pdm_pve_lxc_snapshot_delete,
    pdm_pve_lxc_snapshot_rollback,
    pdm_pve_qemu_migrate,
    pdm_pve_qemu_power,
    pdm_pve_qemu_remote_migrate,
    pdm_pve_qemu_snapshot_create,
    pdm_pve_qemu_snapshot_delete,
    pdm_pve_qemu_snapshot_rollback,
)
from proximo.tools.pmg_mail import (  # noqa: E402,F401
    pmg_action_objects_list,
    pmg_backup_create,
    pmg_doctor,
    pmg_domain_create,
    pmg_domain_delete,
    pmg_domains_list,
    pmg_mynetworks_add,
    pmg_mynetworks_remove,
    pmg_node_rrddata,
    pmg_node_status,
    pmg_node_syslog,
    pmg_postfix_flush,
    pmg_postfix_qshape,
    pmg_quarantine_action,
    pmg_quarantine_attachment,
    pmg_quarantine_blocklist_add,
    pmg_quarantine_blocklist_list,
    pmg_quarantine_blocklist_remove,
    pmg_quarantine_spam,
    pmg_quarantine_spamstatus,
    pmg_quarantine_spamusers,
    pmg_quarantine_virus,
    pmg_quarantine_virusstatus,
    pmg_quarantine_welcomelist_add,
    pmg_quarantine_welcomelist_list,
    pmg_quarantine_welcomelist_remove,
    pmg_relay_config,
    pmg_ruledb_digest,
    pmg_ruledb_rule_actions_list,
    pmg_ruledb_rule_from_list,
    pmg_ruledb_rule_get,
    pmg_ruledb_rule_to_list,
    pmg_ruledb_rule_what_list,
    pmg_ruledb_rule_when_list,
    pmg_ruledb_rules_list,
    pmg_service_control,
    pmg_service_status,
    pmg_spam_config,
    pmg_spam_config_update,
    pmg_statistics_domains,
    pmg_statistics_mail,
    pmg_statistics_mailcount,
    pmg_statistics_receiver,
    pmg_statistics_recent,
    pmg_statistics_sender,
    pmg_statistics_spamscores,
    pmg_statistics_virus,
    pmg_tasks_list,
    pmg_tracker_detail,
    pmg_tracker_list,
    pmg_transport_create,
    pmg_transport_delete,
    pmg_what_group_get,
    pmg_what_group_objects,
    pmg_what_groups_list,
    pmg_when_group_get,
    pmg_when_group_objects,
    pmg_when_groups_list,
    pmg_who_group_get,
    pmg_who_group_objects,
    pmg_who_groups_list,
)
from proximo.tools.pmg_rules import (  # noqa: E402,F401
    pmg_action_bcc_create,
    pmg_action_bcc_update,
    pmg_action_delete,
    pmg_action_disclaimer_create,
    pmg_action_disclaimer_update,
    pmg_action_field_create,
    pmg_action_field_update,
    pmg_action_notification_create,
    pmg_action_notification_update,
    pmg_action_removeattachments_create,
    pmg_action_removeattachments_update,
    pmg_ruledb_rule_action_attach,
    pmg_ruledb_rule_action_detach,
    pmg_ruledb_rule_create,
    pmg_ruledb_rule_delete,
    pmg_ruledb_rule_from_attach,
    pmg_ruledb_rule_from_detach,
    pmg_ruledb_rule_to_attach,
    pmg_ruledb_rule_to_detach,
    pmg_ruledb_rule_update,
    pmg_ruledb_rule_what_attach,
    pmg_ruledb_rule_what_detach,
    pmg_ruledb_rule_when_attach,
    pmg_ruledb_rule_when_detach,
    pmg_what_group_create,
    pmg_what_group_delete,
    pmg_what_group_update,
    pmg_what_object_add,
    pmg_what_object_delete,
    pmg_what_object_update,
    pmg_when_group_create,
    pmg_when_group_delete,
    pmg_when_group_update,
    pmg_when_object_add,
    pmg_when_object_delete,
    pmg_when_object_update,
    pmg_who_group_create,
    pmg_who_group_delete,
    pmg_who_group_update,
    pmg_who_object_add,
    pmg_who_object_delete,
    pmg_who_object_update,
)
from proximo.tools.pve_access import (  # noqa: E402,F401
    pve_acl_list,
    pve_acl_modify,
    pve_acl_prune,
    pve_group_create,
    pve_group_delete,
    pve_group_get,
    pve_group_update,
    pve_groups_list,
    pve_overbroad_grants,
    pve_realm_create,
    pve_realm_delete,
    pve_realm_get,
    pve_realm_update,
    pve_realms_list,
    pve_role_create,
    pve_role_delete,
    pve_role_update,
    pve_roles_list,
    pve_tfa_delete,
    pve_tfa_get,
    pve_tfa_list,
    pve_token_create,
    pve_token_revoke,
    pve_tokens_list,
    pve_user_create,
    pve_user_delete,
    pve_user_get,
    pve_user_update,
    pve_users_list,
)
from proximo.tools.pve_agent import (  # noqa: E402,F401
    pve_agent_file_read,
    pve_agent_file_write,
    pve_agent_fs,
    pve_agent_info,
    pve_agent_set_password,
)
from proximo.tools.pve_backup import (  # noqa: E402,F401
    pbs_job_create,
    pbs_job_delete,
    pbs_job_run,
    pbs_job_update,
    pbs_realm_sync,
    pve_backup,
    pve_backup_delete,
    pve_backup_freshness,
    pve_backup_job_create,
    pve_backup_job_delete,
    pve_backup_job_list,
    pve_backup_job_update,
    pve_backup_list,
    pve_replication_create,
    pve_replication_delete,
    pve_replication_update,
    pve_restore,
)
from proximo.tools.pve_certs import (  # noqa: E402,F401
    pve_acme_account_create,
    pve_acme_account_delete,
    pve_acme_account_update,
    pve_acme_cert_order,
    pve_acme_cert_renew,
    pve_acme_cert_revoke,
    pve_acme_plugin_create,
    pve_acme_plugin_delete,
    pve_acme_plugin_update,
    pve_node_acme_domains_set,
)
from proximo.tools.pve_cluster import (  # noqa: E402,F401
    pve_cluster_resources,
    pve_cluster_status,
    pve_guest_migrate,
    pve_ha_groups_list,
    pve_ha_resource_add,
    pve_ha_resource_remove,
    pve_ha_resources_list,
    pve_ha_rule_create,
    pve_ha_rule_delete,
    pve_ha_rule_update,
    pve_ha_rules_list,
    pve_pool_create,
    pve_pool_delete,
    pve_pool_get,
    pve_pool_update,
    pve_pools_list,
    pve_storage_config_get,
    pve_storage_config_list,
    pve_storage_create,
    pve_storage_delete,
    pve_storage_update,
    pve_task_log,
    pve_task_stop,
    pve_task_wait,
    pve_tasks_list,
)
from proximo.tools.pve_firewall import (  # noqa: E402,F401
    pve_firewall_alias_create,
    pve_firewall_alias_delete,
    pve_firewall_alias_list,
    pve_firewall_alias_update,
    pve_firewall_ipset_create,
    pve_firewall_ipset_delete,
    pve_firewall_ipset_entry_add,
    pve_firewall_ipset_entry_remove,
    pve_firewall_options_get,
    pve_firewall_options_set,
    pve_firewall_rule_add,
    pve_firewall_rule_remove,
    pve_firewall_rule_update,
    pve_firewall_rules_list,
    pve_firewall_security_group_create,
    pve_firewall_security_group_delete,
    pve_firewall_set_enabled,
    pve_ipset_list,
    pve_security_groups_list,
)
from proximo.tools.pve_guest import (  # noqa: E402,F401
    ct_diagnose,
    ct_logs,
    pve_clone,
    pve_cloudinit_get,
    pve_cloudinit_set,
    pve_create_container,
    pve_create_vm,
    pve_delete_guest,
    pve_diagnose,
    pve_disk_move,
    pve_disk_resize,
    pve_doctor,
    pve_guest_config_get,
    pve_guest_config_revert,
    pve_guest_config_set,
    pve_guest_power,
    pve_guest_status,
    pve_list_guests,
    pve_node_status,
    pve_rollback,
    pve_snapshot_create,
    pve_snapshot_delete,
    pve_snapshot_list,
    pve_storage_content,
    pve_storage_content_delete,
    pve_storage_download,
    pve_storage_status,
    pve_task_status,
    pve_template_convert,
)
from proximo.tools.pve_network import (  # noqa: E402,F401
    pve_network_apply,
    pve_network_iface_create,
    pve_network_iface_update,
    pve_network_list,
    pve_sdn_apply,
    pve_sdn_subnet_create,
    pve_sdn_subnet_delete,
    pve_sdn_subnet_list,
    pve_sdn_subnet_update,
    pve_sdn_vnet_create,
    pve_sdn_vnet_delete,
    pve_sdn_vnet_update,
    pve_sdn_vnets_list,
    pve_sdn_zone_create,
    pve_sdn_zone_delete,
    pve_sdn_zone_update,
    pve_sdn_zones_list,
)
from proximo.tools.pve_node import (  # noqa: E402,F401
    pve_node_cert_delete,
    pve_node_cert_upload,
    pve_node_disk_initgpt,
    pve_node_disk_smart,
    pve_node_disk_wipe,
    pve_node_disks_list,
    pve_node_dns_set,
    pve_node_hosts_get,
    pve_node_hosts_set,
    pve_node_migrateall,
    pve_node_startall,
    pve_node_stopall,
    pve_node_storage_backend_create,
    pve_node_storage_backend_delete,
    pve_node_storage_backend_list,
    pve_node_time_get,
    pve_node_time_set,
)
from proximo.tools.pve_observability import (  # noqa: E402,F401
    pve_hardware_list,
    pve_mapping_pci_create,
    pve_mapping_pci_delete,
    pve_mapping_pci_list,
    pve_mapping_pci_update,
    pve_mapping_usb_create,
    pve_mapping_usb_delete,
    pve_mapping_usb_list,
    pve_mapping_usb_update,
    pve_metrics_server_delete,
    pve_metrics_server_list,
    pve_metrics_server_set,
    pve_node_certificates,
    pve_node_dns,
    pve_node_journal,
    pve_node_rrddata,
    pve_node_service_control,
    pve_node_service_status,
    pve_node_services_list,
    pve_node_subscription,
    pve_node_syslog,
    pve_notification_endpoint_create,
    pve_notification_endpoint_delete,
    pve_notification_endpoint_list,
    pve_notification_endpoint_update,
    pve_notification_matcher_delete,
    pve_notification_matcher_set,
    pve_notification_test,
)

if __name__ == "__main__":
    main()
