#!/usr/bin/env python3
"""
aishell-gate-mcp  —  MCP server for AIShell-Gate
Copyright (c) 2026 AIShell Labs LLC Winston-Salem NC USA.
All Rights Reserved.

Exposes MCP tools to AI agents (Claude Code, Cursor, etc.):

  BOOTSTRAP TOOL

  install_engine    Download and install the policy engine binaries from
                    aishellgate.com into ~/.local/share/aishell-gate/bin.
                    Does nothing unless called with confirm=true. Every other
                    tool below (except get_version) returns a clear
                    "engine_missing" status with install instructions instead
                    of an error if the binaries are not yet present.

  PRIMARY TOOLS (Standard and Enterprise)

  evaluate_plan     Submit a goal and list of commands to the policy engine
                    via --dry-run-json.  Returns a structured assessment of
                    every action: decision, confirm level, risk score, binary
                    path, and reason.  Never executes anything.  Call this
                    before execute_plan to inspect the plan.

  execute_plan      Submit a goal and list of commands for live execution via
                    aishell-gate-exec.  Behaviour depends on confirm_mode in
                    aishell-gate-mcp.json:
                      plan_only — plans requiring confirmation are blocked and
                        reported; caller adjusts policy or runs manually.
                      relay — execution is launched asynchronously; confirmation
                        requests are relayed to the human via confirm_action and
                        get_execution_result.
                      operator_pipe — execution blocks while aishell-gate-exec
                        waits on a real, operator-owned --confirm-pipe session
                        (aishell-gate-confirm, run by the human on their own
                        terminal). This server never sees the confirm request
                        or challenge code; confirm_action/get_execution_result
                        play no part. Requires the confirm_pipe config field.

  confirm_action    Submit a human confirmation response to a waiting
                    execute_plan relay (relay mode only).  Called after
                    execute_plan returns status "pending_confirmation".

  get_execution_result
                    Retrieve the current status or final result of an async
                    execute_plan relay (relay mode only).  Poll after
                    confirm_action to get the outcome or check for further
                    confirmations on a multi-action plan.

  INSPECTION TOOLS (Standard and Enterprise)

  get_version       Report the edition (standard/enterprise), version strings,
                    and feature availability of both binaries.  Call this first
                    to discover what tools are available in the current install.

  evaluate_command  Evaluate a single raw command string through the policy
                    engine directly.  Returns full JSON assessment including
                    flag catalog analysis, risk score, and confirmation level.
                    Faster than evaluate_plan for ad-hoc single-command checks.

  get_policy_template
                    Emit the built-in policy layer as a ready-to-edit JSON
                    override file.  Use as a starting point for custom policy
                    authoring (--policy-base, --policy-project, --policy-user).
                    Combine with preset to template any preset's baseline.

  ENTERPRISE-ONLY TOOLS

  verify_policy     [ENTERPRISE] Run a JSON test suite against the active
                    policy and report PASS/FAIL per case.  Accepts inline JSON
                    test content.  Suitable for CI policy gates and regression
                    testing after policy changes.

  verify_audit_log  [ENTERPRISE] Verify the HMAC-SHA256 chain integrity of an
                    exec or policy audit log file.  Detects any gap, truncation,
                    or post-hoc modification.  Specify log_type to select the
                    correct verifier binary.

  RETIRED

  dump_policy       No longer offered via MCP, in any edition — reveals too
                    much policy IP (the full resolved stack including the
                    built-in catalog). The underlying --dump-policy engine
                    flag still exists for AIShell Labs internal use, hidden
                    and undocumented; this server no longer calls it. Calling
                    this tool name returns a clear "retired" message.

Transport: stdio (MCP default — client launches this process as a subprocess).

Configuration: aishell-gate-mcp.json in the working directory, or the path
supplied via --config.  All fields optional with sensible defaults.

Wire protocols (all versioned — see PROTO_* constants below):
  Input  envelope: aishell-gate-exec-input   1.0
  Policy response: aishell-gate-policy-response 1.0
  Dry-run output:  aishell-gate-dry-run       1.0
  MCP tool API:    aishell-gate-mcp           4.0

Edition detection:
  On startup, --version is called on both binaries.  The edition string
  ("standard" or "enterprise") is parsed from the version output and cached
  in cfg["_edition"].  Enterprise tools check this value and return a clear
  "not available in standard edition" message if the standard binary is
  installed — mirroring the binary's own behaviour.

Stdout guard:
  All binary stdout is scanned for the first '{' before JSON parsing.
  Any preamble text (evaluation copy warnings, banners) is silently
  discarded.  If no '{' is found the tool returns a structured error.
  Evaluation-expiry notices are detected specially (_eval_expiry_message):
  when an expired evaluation binary declines to run, the tool response
  says so explicitly and includes the binary's own message, rather than
  reporting a generic "no JSON in output" parse failure.

Broker compatibility note:
  In v1/v2/v3/v4 this server invokes aishell-gate-exec directly as a subprocess.
  When the broker daemon arrives in v5, replace _invoke_exec() with a
  function that opens a socket to the broker and sends the same JSON
  envelope.  Everything else in this file is unchanged.

Change log:
  2026-08-14  (v23) New confirm_mode: "operator_pipe". Until now the only way
              to satisfy a confirm level above `plan_only`'s single whole-plan
              gate was "relay" -- which works by having this server generate
              its own private FIFO pair, launch exec against it, and act as
              BOTH ends of that pipe itself: it reads exec's confirmation
              request (including, for typed levels, the raw challenge code)
              and hands it to the calling agent, which is trusted to show it
              to a human and relay their real answer back via confirm_action.
              Nothing on the wire enforces that trust -- an agent can call
              confirm_action with a self-supplied "yes" (or simply echo the
              challenge code it was already given) and exec has no way to
              tell the difference from a real operator answer.
              operator_pipe closes that gap structurally instead of by
              policy: when set, execute_plan does none of the above. It makes
              one ordinary blocking call to exec with --confirm-pipe pointed
              at a basepath a human's own aishell-gate-confirm session
              already owns (see aishell-gate-confirm(1) -- the same mechanism
              the SSH deployment has always used). This server does not
              create, open, or read either FIFO; it only waits for exec to
              exit. The agent is never shown the confirm request or the
              challenge code, so confirm_action/get_execution_result have
              nothing to do here and are not called for this path.
              New: DEFAULT_CONFIG["confirm_pipe"|"confirm_lock"|
              "confirm_timeout"] (all None unless operator_pipe is in use --
              mirror exec's own --confirm-pipe/--confirm-lock/--confirm-timeout
              flags), _execute_plan_operator_pipe(), _EXEC_DEFAULT_CONFIRM_TIMEOUT.
              _invoke_exec() gained an optional timeout= parameter (default
              120, unchanged) so this path can wait longer than the default
              without touching any existing caller.
              plan_only and relay are both completely unchanged -- this is a
              third, opt-in value, not a replacement default.

  2026-08-10  (v22) install_engine now downloads the engine from
              aishellgate.com instead of GitHub: the tarball comes from
              https://www.aishellgate.com/download.php and its checksum
              from https://www.aishellgate.com/checksum.php -- the same
              two endpoints the curl|sh network installer has always
              used. The GitHub "find the latest release" lookup step is
              gone entirely (ENGINE_RELEASE_API, ENGINE_ASSET_PREFIX,
              ENGINE_ASSET_SUFFIX removed), since the website always
              serves one fixed, unversioned aishell-gate.tar.gz. Checksum
              verification is no longer skippable: checksum.php always
              returns a hash, so a missing checksum is now a hard error
              rather than a skip-with-warning. All user-facing text
              (confirmation prompt, missing-engine message, tool
              description) updated to name the website instead of the
              GitHub repo.

  2026-07-09  (v21) install_engine's confirmation message reworded to a
              dnf/apt-style package prompt: it names the source repo
              (github.com/aishell-gate/aishell-gate), an approximate download
              size, the per-user destination, "no sudo", and ends "Is this ok
              [y/N]:". The "[y/N]" is a familiar frame only -- confirm=true is
              still the real gate, and the confirmation_required response now
              carries a machine-readable note telling the agent to map the
              user's "y" to a confirm=true re-call. Sizes are hardcoded
              approximations ("~1.1 M") on purpose: no per-release upkeep, and
              "~" never lies. The sibling messages (already-installed,
              platform-unsupported, post-install success) and the blocked-call
              "engine not found" notice were aligned to the same plain voice.
              No behavioural change -- these are display strings.

  2026-07-09  (v20) Documentation-string corrections to match shipped
              behaviour.  User-facing config-name references updated from
              the legacy "aishell-mcp.json" to "aishell-gate-mcp.json" in
              --help, error messages, and tool descriptions.  The legacy
              name is still accepted by load_config(); that fallback and
              the historical note above are unchanged.  install_engine's
              tool description corrected to say it installs into
              ~/.local/share/aishell-gate/bin (the stale text still said
              the current working directory; the code has written to
              USER_INSTALL_DIR since v15).  --help description reworded from
              "MCP server v3" to "MCP server (wire protocol 4.0)" to match
              PROTO_MCP_VERSION.  No behavioural change.

  2026-07-07  (v18) Diagnostic message fix for evaluate_plan/execute_plan.
              Found while answering a question about how this server
              responds to edition-restriction messages from exec/policy:
              evaluate_plan's "no output" path used stderr[:500] (head
              slice); confirmed empirically against a real rejected-preset
              run that the actual reason (e.g. "danger_zone... not
              available in the standard edition") does not survive that
              truncation, because aishell-gate-exec does not separate its
              own stderr from the policy subprocess it invokes, and writes
              its own substantial preamble (DRY-RUN-JSON notice, license
              banner, audit-log warning) first — the caller was told to
              "check that exec and policy binaries are installed
              correctly," which is simply wrong; nothing was misinstalled.
              execute_plan's equivalent path already used a tail slice
              (stderr_dry[-500:]) and happened to work for the same case,
              but only by construction, not by design — not something to
              rely on for every future message shape.
              Fix: new _standard_edition_restriction_reason() helper
              searches the FULL stderr for the restriction phrase (not a
              truncated hint) and extracts the specific reason. Wired into
              evaluate_plan, execute_plan, evaluate_command, and
              _strip_preamble (shared by all three plus get_policy_template
              indirectly) — all four previously had their own independent
              truncated-hint fallback with no awareness of this message
              category. Truncation windows also widened 200/500 -> 512
              everywhere this pattern appears, as a secondary improvement
              for error messages the new helper doesn't recognize; the
              helper is what actually fixes the diagnosed case, not the
              wider window on its own — verified: even 512 characters from
              the head would not have reached the real message in the
              captured run that prompted this fix.
              Confirmed no regression (full suite, 97 tests) and no false
              positive on an unrelated error before shipping.
  2026-07-07  (v17) Standard/Enterprise split corrections (edition-split-
              decisions-6.md). Two independent fixes:
              (1) verify_policy moved to Enterprise-only: added
              _require_enterprise() check to tool_verify_policy, moved its
              TOOLS schema entry into the Enterprise-only section, updated
              get_version's available_tools/enterprise_features, the module
              docstring, the argparse epilog's tool lists, and --version
              output. The engine side of this (aishell-gate-policy
              --test-plan) was gated to Enterprise in policy v1.16 — this
              server was calling an engine flag that, until that fix, had
              never actually been edition-restricted.
              (2) dump_policy fully retired: disabled at every touch point
              (schema entry, dispatch table, get_version's available_tools /
              enterprise_features, docstring, argparse epilog, --version
              output) per edition-split-decisions-5/6.md — it's not a
              customer-facing feature in any edition. tool_dump_policy
              short-circuited to a clear "retired" error; original
              implementation kept, commented out, not deleted — same
              pattern as _eval_expiry_message() below. (This fix was
              described as already done in a prior "-17.py" in the decision
              notes, but the file actually in hand — this one, uploaded as
              "-16.py" — didn't yet have it; applying it now rather than
              carry the discrepancy forward.)
              No wire protocol change.
  2026-07-04  (v15) Stable install directory. New USER_INSTALL_DIR constant
              (~/.local/share/aishell-gate/bin). install_engine now writes
              binaries there (was: current working directory, which is
              client-dependent — Claude Code uses the project root but other
              MCP clients may launch with cwd=/ or $HOME). Directory is
              created on demand; unwritable directory returns a clear
              install_engine error instead of an OSError traceback.
              _resolve_binary_path() order is now: env var -> USER_INSTALL_DIR
              -> /usr/local/bin -> cwd -> PATH (user dir deliberately precedes
              /usr/local/bin so an install_engine result takes effect without
              sudo). engine_missing message text updated to name the new
              directory. No wire protocol change.
  2026-07-04  Evaluation-expiry handling.  New _eval_expiry_message() detects
              an expired evaluation binary's refusal notice (loose keyword
              match — exact wording not assumed stable) on stdout or stderr.
              _strip_preamble and all five "produced no output" branches now
              return a dedicated "evaluation period has ended" error carrying
              the binary's own message, instead of a generic no-JSON/parse
              error.  The generic no-JSON error now includes a stdout snippet
              as well as stderr (previously the stdout text — where the notice
              is most likely printed — was silently discarded).  execute_plan's
              pre-flight preamble guard now receives exec's stderr instead of
              "".  No wire protocol change.
  2026-04-23  Fix "flush of closed file" in confirm-relay path.  _execute_plan_relay
              closes proc.stdin early to signal EOF to aishell-gate-exec, then
              later calls proc.communicate() to wait and drain stdout/stderr.
              Popen._communicate() unconditionally calls self.stdin.flush() and
              catches only BrokenPipeError, so flushing the already-closed stdin
              raised ValueError("flush of closed file") which bubbled out of the
              relay thread and was misreported to the caller as a relay-write
              failure (the original resp_fd.write/flush had already succeeded).
              Fix: set proc.stdin = None immediately after proc.stdin.close() so
              the communicate() stdin-flush block is skipped.
  2026-04-23  Config filename renamed from aishell-mcp.json to aishell-gate-mcp.json
              for consistency with other AIShell-Gate distribution files.
              confirm_action and get_execution_result added to tool list and --version
              output.  MCP wire protocol bumped to 4.0.
"""

import errno
import itertools
import json
import logging
import os
import queue
import re
import select
import shutil
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

# ---------------------------------------------------------------------------
# Wire protocol versions
# ---------------------------------------------------------------------------
PROTO_MCP_NAME        = "aishell-gate-mcp"
PROTO_MCP_VERSION     = "4.0"   # wire protocol identifier only — not the product version
                                 # product version is always derived from the policy binary
                                 # at runtime via _detect_edition() / cfg["_policy_version"]
# TODO(wire-protocol-gate): PROTO_MCP_VERSION is only stamped onto this server's
# own outgoing envelopes today — nothing verifies that the aishell-gate backend
# binaries speak a compatible wire-protocol major. The binaries do not yet emit
# a wire-protocol version via --version, so there is currently nothing to check
# against. This is a real gap when the MCP is distributed separately from the
# binaries (e.g. published to PyPI). When the backend starts reporting a
# wire-protocol version, parse it in _detect_edition() and gate startup on a
# floor+ceiling major window here (same pattern as MIN_SUPPORTED_MAJOR /
# SUPPORTED_MAJOR below). Deferred: requires a coordinated change to the C
# binaries, out of scope for the current beta.

PROTO_INPUT_NAME      = "aishell-gate-exec-input"
PROTO_INPUT_VERSION   = "1.0"

PROTO_DRY_RUN_NAME    = "aishell-gate-dry-run"
PROTO_POLICY_NAME     = "aishell-gate-policy-response"

# Supported major-version window for backend *response* protocols (dry-run and
# policy-response). A response whose major falls outside the inclusive window
# [MIN_SUPPORTED_MAJOR, SUPPORTED_MAJOR] is rejected as incompatible. Both are 1
# today — the only response-protocol major that exists — so the window is a
# single value for now. Widen it here when a 2.x response protocol ships.
MIN_SUPPORTED_MAJOR   = 1   # reject dry-run/policy responses with major < this
SUPPORTED_MAJOR       = 1   # reject dry-run/policy responses with major > this

# Minimum policy-binary product major (Axis C). Non-fatal: edition gating is the
# hard gate; a product major below this only warns at startup, because product
# version numbering follows release cadence, not the wire contract.
POLICY_MIN_MAJOR      = 1

# ---------------------------------------------------------------------------
# Engine release info (used by the missing-binary detection and install_engine)
# ---------------------------------------------------------------------------

# Same two endpoints the curl|sh network installer (aishell-gate-network-
# install.sh) has always used. Both are fixed URLs that always serve
# whatever is currently released, so -- unlike the old GitHub-based
# version of this code -- there's no "find the latest release" lookup
# step needed here at all.
ENGINE_DOWNLOAD_URL = "https://www.aishellgate.com/download.php"
ENGINE_CHECKSUM_URL = "https://www.aishellgate.com/checksum.php"
ENGINE_ARCHIVE_NAME = "aishell-gate.tar.gz"
ENGINE_SUPPORTED_PLATFORM = ("Linux", "x86_64")

# ---------------------------------------------------------------------------
# MCP protocol constants (stdio transport)
# ---------------------------------------------------------------------------
JSONRPC_VERSION       = "2.0"
MCP_PROTOCOL_VERSION  = "2024-11-05"

# Stable, client-independent install location for engine binaries.
# install_engine writes here (v15+); the resolver checks it right after the
# env vars and BEFORE /usr/local/bin, so an install_engine result always
# takes effect without sudo. Chosen over cwd because cwd is client-dependent
# (Claude Code launches servers in the project root, but other MCP clients
# may use / or $HOME) and cwd placement duplicated binaries per project.
USER_INSTALL_DIR      = Path.home() / ".local" / "share" / "aishell-gate" / "bin"

# ---------------------------------------------------------------------------
# Edition constants
# ---------------------------------------------------------------------------
EDITION_STANDARD      = "standard"
EDITION_ENTERPRISE    = "enterprise"
EDITION_UNKNOWN       = "unknown"

# ---------------------------------------------------------------------------
# Logging — goes to stderr only; stdout is reserved for MCP protocol messages
# ---------------------------------------------------------------------------
logging.basicConfig(
    stream=sys.stderr,
    level=logging.INFO,
    format="[aishell-gate-mcp] %(levelname)s %(message)s",
)
log = logging.getLogger("aishell_gate_mcp")

# ---------------------------------------------------------------------------
# Default configuration
# ---------------------------------------------------------------------------
DEFAULT_CONFIG: Dict[str, Any] = {
    # Binary paths
    "exec_binary":      "aishell-gate-exec",
    "policy_binary":    "aishell-gate-policy",

    # Policy settings
    "preset":           "ops_safe",
    "jail_root":        None,
    "policy_base":      None,
    "policy_project":   None,
    "policy_user":      None,
    "sandbox":          None,

    # Source identity (set by deployer; never read from AI plan)
    "source":           "ai",

    # Audit — exec log
    "audit_log":        None,

    # Audit — policy engine log (separate file; different format from exec log)
    "policy_audit_log": None,

    # Audit — HMAC key file for Enterprise chain verification
    # exec key format:   64 ASCII hex characters (32 bytes)
    # policy key format: 64 raw binary bytes  (NOT interchangeable)
    "audit_key":        None,

    # Timeouts
    "eval_timeout":     30,   # seconds; 0 = disabled
    "input_timeout":    30,   # seconds; 0 = disabled

    # Response byte cap for policy engine output (0 = use binary default 8 MiB)
    "max_response_bytes": 0,

    # Arbitrary extra flags forwarded to aishell-gate-exec
    "extra_flags":      [],

    # Confirmation mode:
    #   "plan_only" (default) — block any plan requiring confirmation, return error
    #   "relay"               — relay confirmation requests to the human via Claude
    #   "operator_pipe"       — point aishell-gate-exec straight at a real,
    #                           operator-owned --confirm-pipe session (an
    #                           aishell-gate-confirm process the human runs on
    #                           their own terminal). execute_plan simply blocks
    #                           until exec returns; this server never reads the
    #                           FIFOs, never sees the confirm request or challenge
    #                           code, and confirm_action/get_execution_result are
    #                           not used. Requires "confirm_pipe" below.
    "confirm_mode":     "plan_only",

    # Basepath for the operator's FIFO pair (BASEPATH.req / BASEPATH.resp),
    # created and owned by their aishell-gate-confirm session. Only read when
    # confirm_mode is "operator_pipe". Matches exec's own --confirm-pipe flag.
    "confirm_pipe":      None,

    # Optional lock file path, forwarded as --confirm-lock. Only meaningful
    # with confirm_pipe set; needed if more than one AI session may share the
    # same operator pipe at once. See aishell-gate-exec(1).
    "confirm_lock":       None,

    # Optional seconds forwarded as exec's own --confirm-timeout, bounding how
    # long exec waits on the operator before giving up. Only used with
    # confirm_pipe set. Unset = exec's own default (120s).
    "confirm_timeout":    None,

    # Runtime — populated by _detect_edition() at startup; not user-configurable
    "_edition":         EDITION_UNKNOWN,
    "_exec_version":    "",
    "_policy_version":  "",
}

# ---------------------------------------------------------------------------
# Confirm-relay state
# ---------------------------------------------------------------------------
# Keyed by execution_id (UUID string).  Each entry is a dict:
#   "status"       : "pending_confirmation" | "running" | "complete" | "error"
#   "goal"         : str
#   "commands"     : list[str]
#   "confirm_req"  : dict | None   — most recent confirm request from exec
#   "resp_queue"   : queue.Queue   — human response dropped in here by confirm_action
#   "result"       : dict | None   — final tool result when complete
#   "fifo_base"    : str           — base path for the FIFO pair
#   "thread"       : threading.Thread
#   "ready_event"  : threading.Event — set once status leaves "starting" and
#                                      reaches a stable reportable state
#                                      (pending_confirmation or complete)
_PENDING: Dict[str, dict] = {}
_PENDING_LOCK = threading.Lock()

# Max seconds execute_plan waits for the relay thread to reach a stable state
# (pending_confirmation or complete) before returning. Prevents the race where
# execute_plan reports pending_confirmation while the thread is still blocked
# on subprocess startup or FIFO handshake.
_RELAY_READY_TIMEOUT = 15.0

# Max seconds the relay thread waits for aishell-gate-exec to complete the
# bidirectional FIFO handshake (open .req for write and .resp for read). If
# this elapses, we assume exec is wedged or crashed and fail the execution
# instead of blocking forever inside open().
_FIFO_HANDSHAKE_TIMEOUT = 5.0

# Max seconds confirm_action waits for the execution state to reach
# pending_confirmation before rejecting the call. Provides resilience
# against clients that race the relay thread's startup or that call
# confirm_action during the brief running→pending_confirmation transition
# between actions in a multi-action plan.
_CONFIRM_ACTION_READY_WAIT = 2.0

# Max seconds the relay thread waits for a human response to one open
# confirmation gate before treating it as a refusal. Was hardcoded as 300
# inline; named here because the gate-matching loop now needs a deadline.
_CONFIRM_HUMAN_TIMEOUT = 300.0

# Seconds of headroom given to aishell-gate-exec's own --confirm-timeout on
# top of the wait above, so the MCP is ALWAYS the side that gives up first.
#
# exec defaults to 120s (DEFAULT_CONFIRM_TIMEOUT). We were not passing
# --confirm-timeout at all, so any human who took between 120s and 300s hit
# this: exec timed out, treated the silence as a refusal, closed the FIFOs and
# exited 2 -- and the relay then wrote the human's "yes" into a pipe with no
# reader. That EPIPE surfaced to the caller as "relay thread error: Broken
# pipe", discarding exec's real result. Two minutes is an entirely ordinary
# amount of time for someone to consider a destructive command, so this was
# reachable in normal use, not just under load.
#
# exec keeps a limit rather than being given 0 (no limit): a wedged MCP must
# not be able to strand an exec process holding the confirmation lock forever.
_CONFIRM_EXEC_MARGIN = 60.0

# aishell-gate-exec's own default --confirm-timeout when the flag is omitted
# (DEFAULT_CONFIRM_TIMEOUT in the binary). Used as the floor for how long the
# operator_pipe path (below) is willing to let exec wait on the operator's
# real aishell-gate-confirm session, when cfg["confirm_timeout"] is unset.
_EXEC_DEFAULT_CONFIRM_TIMEOUT = 120.0

# Select() timeout between liveness checks while the relay loop is
# waiting for the next confirmation request from aishell-gate-exec.
# This is a liveness-poll interval only — it does NOT bound how long
# legitimately long-running exec commands may run. Shorter values
# detect a silently-wedged exec faster, at a negligible CPU cost.
_RELAY_POLL_INTERVAL = 1.0

# Max seconds the relay thread waits for aishell-gate-exec to exit after the
# FIFO relay loop has ended. If this elapses the subprocess is SIGKILLed — the
# MCP is intended for heavy server use and must not accumulate orphans.
#
# This is NOT a "finishing up" budget. exec closes the confirmation FIFOs as
# soon as the last gate is answered, BEFORE it runs anything, so that it can
# release the confirmation lock before any execve. The relay loop sees that
# EOF and exits — at which point the commands have not started yet. Everything
# the plan actually does happens inside this window.
#
# It was 60s, which silently capped every confirmed plan at one minute of
# execution: a longer build, test run, or copy was SIGKILLed and the caller
# got "exec terminated by signal 9" with no stdout, no stderr, and no exit
# code. Unconfirmed plans were unaffected (the loop keeps polling while exec
# is alive), so this only ever bit the confirmation path.
#
# One hour, overridable per-deployment. 0 disables the limit entirely; use
# that only where an orphaned exec is acceptable.
_EXEC_COMPLETION_TIMEOUT = 3600.0

# Max seconds to wait for a SIGKILLed exec to actually reap. If even
# this elapses, we give up and log loudly; the fd resources are still
# released by the OS when the thread unwinds.
_EXEC_KILL_TIMEOUT = 5.0


# ---------------------------------------------------------------------------
# Config loader
# ---------------------------------------------------------------------------
def _resolve_binary_path(val: str, env_key: str) -> str:
    """Resolve a single binary field, in order:
         1. Environment variable
         2. Alongside this server file (../<name> from bin/MCP/) — lets a
            fully unpacked release run in place, no install step, no root
         3. User install dir (~/.local/share/aishell-gate/bin) — where
            install_engine places binaries (v15+)
         4. Standard install location (/usr/local/bin)
         5. Default unpack location (~/aishell-gate/bin) — catches the case
            where the server was copied out of the release but the engine
            binaries were left behind in it
         6. Current working directory (./)
         7. PATH (via shutil.which)
       Returns val unchanged if none of the above find an executable file.
       Shared by load_config() (startup) and _reresolve_binaries() (called
       after install_engine places new files, so no restart is required)."""
    if os.path.isabs(val):
        return val

    name = Path(val).name
    install_dir = Path("/usr/local/bin")

    env_val = os.environ.get(env_key)
    if env_val and Path(env_val).is_file() and os.access(env_val, os.X_OK):
        return env_val

    # Alongside this server file. The server lives at bin/MCP/aishell-gate-mcp
    # inside an unpacked release, and the engine binaries live at bin/<name> --
    # i.e. one directory up from here. Resolving relative to __file__ (not cwd,
    # which the MCP client sets to the project dir) means a release unpacked
    # anywhere just works, with nothing copied and no root. Wrapped in try/except
    # because __file__ can be unreliable in frozen/zipped deployments; if it
    # can't be resolved, we simply fall through to the install-dir checks.
    try:
        sibling = (Path(__file__).resolve().parent.parent / name)
        if sibling.is_file() and os.access(sibling, os.X_OK):
            return str(sibling)
    except (OSError, NameError):
        pass

    user_candidate = USER_INSTALL_DIR / name
    if user_candidate.is_file() and os.access(user_candidate, os.X_OK):
        return str(user_candidate)

    install_candidate = install_dir / name
    if install_candidate.is_file() and os.access(install_candidate, os.X_OK):
        return str(install_candidate)

    # Default unpack location. The release tarball is packed from a directory
    # literally named aishell-gate/ (no version in the name), specifically so
    # that ~/aishell-gate/ is a predictable path -- see
    # aishell-gate-network-install.sh, which relies on the same property to
    # ship a fixed launcher path in .mcp.json.
    #
    # This overlaps with the __file__ check above but is not redundant: that
    # one only fires when the server is still running from inside the unpacked
    # release. If the server has been copied out to ~/.local/bin (which
    # aishell-gate-mcp-install.sh does when no system-wide install is present),
    # __file__ points at ~/.local/bin and its parent has no engine binaries,
    # while the engine is still sitting unused in ~/aishell-gate/bin. Belt and
    # braces: the two checks cover different halves of the same situation.
    home_candidate = Path.home() / "aishell-gate" / "bin" / name
    if home_candidate.is_file() and os.access(home_candidate, os.X_OK):
        return str(home_candidate)

    cwd_candidate = Path.cwd() / name
    if cwd_candidate.is_file() and os.access(cwd_candidate, os.X_OK):
        return str(cwd_candidate)

    resolved = shutil.which(val)
    if resolved:
        return resolved

    return val


def _reresolve_binaries(cfg: dict) -> None:
    """Re-run binary resolution after install_engine places new files, so a
    freshly-downloaded engine is picked up in the same session without
    requiring the MCP client to restart the server process."""
    env_keys = {
        "exec_binary":   "AISHELL_EXEC_BIN",
        "policy_binary": "AISHELL_POLICY_BIN",
    }
    for field, env_key in env_keys.items():
        cfg[field] = _resolve_binary_path(cfg[field], env_key)


def _engine_available(cfg: dict) -> bool:
    """True only if both binaries resolved to an existing, executable file.
    Bare/unresolved names (e.g. still 'aishell-gate-exec' with no match on
    PATH) are not files relative to cwd unless one happens to exist there,
    which is exactly the case we want to allow post-install without a restart."""
    for field in ("exec_binary", "policy_binary"):
        p = Path(cfg[field])
        if not (p.is_file() and os.access(p, os.X_OK)):
            return False
    return True


def load_config(path: Optional[str]) -> dict[str, Any]:
    """Load aishell-gate-mcp.json from path or working directory.  Missing file
    is not an error — all fields have defaults.  Unknown keys are ignored."""
    cfg = dict(DEFAULT_CONFIG)
    candidates = []
    if path:
        candidates.append(Path(path))
    candidates.append(Path("aishell-gate-mcp.json"))
    candidates.append(Path("aishell-mcp.json"))   # legacy name — still accepted

    for p in candidates:
        if p.exists():
            try:
                with p.open() as f:
                    data = json.load(f)
                for key in DEFAULT_CONFIG:
                    if key in data:
                        cfg[key] = data[key]
                log.info("loaded config from %s", p)
            except Exception as e:
                log.warning("could not load config from %s: %s", p, e)
            break

    # Resolve binary paths — checked in order:
    #   1. Environment variable (AISHELL_EXEC_BIN / AISHELL_POLICY_BIN)
    #   2. Standard install location (/usr/local/bin)
    #   3. Current working directory (./)
    #   4. PATH (via shutil.which)
    env_keys = {
        "exec_binary":   "AISHELL_EXEC_BIN",
        "policy_binary": "AISHELL_POLICY_BIN",
    }
    for field, env_key in env_keys.items():
        cfg[field] = _resolve_binary_path(cfg[field], env_key)

    return cfg


# ---------------------------------------------------------------------------
# Edition detection
# ---------------------------------------------------------------------------
def _parse_major(version_line: str) -> Optional[int]:
    """Extract the integer major version from a --version line such as
    'aishell-gate-policy 1.13 enterprise'. Returns the major of the first
    dotted/numeric token found, or None if there is none (e.g. the binary
    was not located, so the string is empty)."""
    if not version_line:
        return None
    for token in version_line.splitlines()[0].split():
        head = token.split(".")[0]
        if head.isdigit():
            return int(head)
    return None


def _detect_edition(cfg: dict) -> None:
    """Run --version on both binaries and parse edition + version strings.
    Results are stored in cfg["_edition"], cfg["_exec_version"], and
    cfg["_policy_version"].  Errors are logged but do not abort startup."""

    def _run_version(binary: str) -> str:
        try:
            result = subprocess.run(
                [binary, "--version"],
                capture_output=True, text=True, timeout=10
            )
            return (result.stdout + result.stderr).strip()
        except Exception as e:
            log.warning("version check failed for %s: %s", binary, e)
            return ""

    exec_ver   = _run_version(cfg["exec_binary"])
    policy_ver = _run_version(cfg["policy_binary"])

    cfg["_exec_version"]   = exec_ver
    cfg["_policy_version"] = policy_ver

    # Parse edition from either binary output.
    # Expected format: "aishell-gate-exec 0.43.0 standard" (third token)
    edition = EDITION_UNKNOWN
    for line in (exec_ver + "\n" + policy_ver).splitlines():
        tokens = line.lower().split()
        if EDITION_ENTERPRISE in tokens:
            edition = EDITION_ENTERPRISE
            break
        if EDITION_STANDARD in tokens:
            edition = EDITION_STANDARD
            # keep looking — enterprise takes priority if mixed output

    cfg["_edition"] = edition
    log.info("detected edition: %s", edition)
    log.info("exec version:   %s", exec_ver.splitlines()[0] if exec_ver else "(not found)")
    log.info("policy version: %s", policy_ver.splitlines()[0] if policy_ver else "(not found)")

    # Axis C: product-major floor on the policy binary. Non-fatal by design —
    # edition gating above is the hard gate, and the data-plane compatibility
    # gate is the response-protocol check (_check_response_protocol, Axis B).
    # Product version numbering follows release cadence rather than the wire
    # contract, so a low major only warns; it does not abort startup.
    policy_major = _parse_major(policy_ver)
    if policy_major is not None and policy_major < POLICY_MIN_MAJOR:
        log.warning(
            "policy binary major version %d is below the minimum supported "
            "(%d) — some tools may behave unexpectedly. Detected: %s",
            policy_major, POLICY_MIN_MAJOR,
            policy_ver.splitlines()[0] if policy_ver else "(not found)",
        )


def _require_enterprise(cfg: dict, tool: str) -> Optional[dict]:
    """Return an error response if the installed edition is not enterprise,
    or None if the check passes.  Mirrors the binary's own clear message."""
    if cfg["_edition"] != EDITION_ENTERPRISE:
        detected = cfg["_edition"]
        return _error_response(tool, (
            f"{tool} is not available in the {detected} edition. "
            f"This feature requires the Enterprise edition of AIShell-Gate. "
            f"Contact www.aishellgate.com for licensing information."
        ))
    return None


# ---------------------------------------------------------------------------
# Protocol version check
# ---------------------------------------------------------------------------
def _check_response_protocol(data: dict, expected_name: str, context: str) -> None:
    """Warn if the response carries an unexpected protocol name or a major
    version this server does not understand.  Raises ValueError on hard
    incompatibility: wrong name, or a major outside the inclusive window
    [MIN_SUPPORTED_MAJOR, SUPPORTED_MAJOR]."""
    proto = data.get("protocol")
    if not isinstance(proto, dict):
        return  # absent — treat as 1.0, backward compat
    name = proto.get("name", "")
    ver  = proto.get("version", "1.0")
    if name and name != expected_name:
        raise ValueError(
            f"{context}: protocol name mismatch — expected '{expected_name}', "
            f"got '{name}'.  Wrong program?"
        )
    try:
        major = int(str(ver).split(".")[0])
    except (ValueError, IndexError):
        major = 1
    if major > SUPPORTED_MAJOR:
        raise ValueError(
            f"{context}: protocol version '{ver}' requires a newer "
            f"aishell-gate-mcp.  Supported major: "
            f"{MIN_SUPPORTED_MAJOR}..{SUPPORTED_MAJOR}."
        )
    if major < MIN_SUPPORTED_MAJOR:
        raise ValueError(
            f"{context}: protocol version '{ver}' is older than this "
            f"aishell-gate-mcp supports.  Supported major: "
            f"{MIN_SUPPORTED_MAJOR}..{SUPPORTED_MAJOR}."
        )


# ---------------------------------------------------------------------------
# Exec invocation — v1/v2/v3: subprocess.  v4: replace with broker socket call.
# ---------------------------------------------------------------------------
def _build_exec_argv(cfg: dict, extra: Optional[List[str]] = None) -> List[str]:
    """Build the argv list for aishell-gate-exec from config."""
    argv = [cfg["exec_binary"]]
    argv += ["--policy-binary",  cfg["policy_binary"]]
    argv += ["--policy-preset",  cfg["preset"]]
    argv += ["--source",         cfg["source"]]
    # Redirect confirmation TTY to /dev/null — the MCP server runs headless
    # with no controlling terminal.  Without this the exec binary may block
    # waiting to open /dev/tty for confirmation prompts even in dry-run mode.
    argv += ["--confirm-tty", "/dev/null"]
    if cfg.get("jail_root"):
        argv += ["--jail-root", cfg["jail_root"]]
    if cfg.get("sandbox"):
        argv += ["--sandbox", cfg["sandbox"]]
    if cfg.get("policy_base"):
        argv += ["--policy-base", cfg["policy_base"]]
    if cfg.get("policy_project"):
        argv += ["--policy-project", cfg["policy_project"]]
    if cfg.get("policy_user"):
        argv += ["--policy-user", cfg["policy_user"]]
    if cfg.get("audit_log"):
        argv += ["--audit-log", cfg["audit_log"]]
    if cfg.get("eval_timeout") and cfg["eval_timeout"] != 30:
        argv += ["--eval-timeout", str(cfg["eval_timeout"])]
    if cfg.get("input_timeout") and cfg["input_timeout"] != 30:
        argv += ["--input-timeout", str(cfg["input_timeout"])]
    if cfg.get("max_response_bytes") and cfg["max_response_bytes"] > 0:
        argv += ["--max-response-bytes", str(cfg["max_response_bytes"])]
    for flag in (cfg.get("extra_flags") or []):
        argv.append(str(flag))
    if extra:
        argv.extend(extra)
    return argv


def _build_policy_argv(cfg: dict, extra: Optional[List[str]] = None) -> List[str]:
    """Build the argv list for aishell-gate-policy direct invocations."""
    argv = [cfg["policy_binary"]]
    argv += ["--policy-preset", cfg["preset"]]
    argv += ["--source",        cfg["source"]]
    if cfg.get("jail_root"):
        argv += ["--jail-root", cfg["jail_root"]]
    if cfg.get("sandbox"):
        argv += ["--sandbox", cfg["sandbox"]]
    if cfg.get("policy_base"):
        argv += ["--policy-base", cfg["policy_base"]]
    if cfg.get("policy_project"):
        argv += ["--policy-project", cfg["policy_project"]]
    if cfg.get("policy_user"):
        argv += ["--policy-user", cfg["policy_user"]]
    if cfg.get("policy_audit_log"):
        argv += ["--audit-log", cfg["policy_audit_log"]]
    if cfg.get("audit_key"):
        argv += ["--audit-key", cfg["audit_key"]]
    if extra:
        argv.extend(extra)
    return argv


def _build_envelope(goal: str, commands: List[str], strategy: str = "fail_fast") -> dict:
    """Build a versioned exec input envelope."""
    return {
        "protocol": {
            "name":    PROTO_INPUT_NAME,
            "version": PROTO_INPUT_VERSION,
        },
        "goal":     goal,
        "source":   "ai",
        "strategy": strategy,
        "actions":  [{"type": "command", "cmd": cmd} for cmd in commands],
    }


def _invoke_exec(cfg: dict, envelope: dict, extra_argv: Optional[List[str]] = None,
                  timeout: float = 120) -> Tuple[int, str, str]:
    """
    Invoke aishell-gate-exec with the given envelope on stdin.
    Returns (exit_code, stdout, stderr).

    timeout bounds this Python-level subprocess call, separate from any
    --confirm-timeout passed to exec itself. Default 120s matches every
    existing caller's prior hardcoded value; only the operator_pipe path
    (which may legitimately wait much longer on a human) passes something
    else.

    This is the broker seam: in v3 replace the subprocess call here
    with a socket connection to the broker daemon.
    """
    argv    = _build_exec_argv(cfg, extra=extra_argv)
    payload = json.dumps(envelope)
    log.debug("invoking exec: %s", " ".join(argv))
    try:
        result = subprocess.run(
            argv,
            input=payload,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError:
        raise RuntimeError(
            f"aishell-gate-exec binary not found: '{argv[0]}'\n"
            f"Set exec_binary in aishell-gate-mcp.json or add it to PATH."
        )
    except subprocess.TimeoutExpired:
        raise RuntimeError(f"aishell-gate-exec timed out after {timeout:g} seconds.")


def _invoke_policy(cfg: dict, stdin_data: str,
                   extra_argv: Optional[List[str]] = None) -> Tuple[int, str, str]:
    """
    Invoke aishell-gate-policy directly with stdin_data.
    Returns (exit_code, stdout, stderr).
    """
    argv = _build_policy_argv(cfg, extra=extra_argv)
    log.debug("invoking policy: %s", " ".join(argv))
    try:
        result = subprocess.run(
            argv,
            input=stdin_data,
            capture_output=True,
            text=True,
            timeout=60,
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError:
        raise RuntimeError(
            f"aishell-gate-policy binary not found: '{argv[0]}'\n"
            f"Set policy_binary in aishell-gate-mcp.json or add it to PATH."
        )
    except subprocess.TimeoutExpired:
        raise RuntimeError("aishell-gate-policy timed out after 60 seconds.")


# ---------------------------------------------------------------------------
# Stdout preamble guard
# ---------------------------------------------------------------------------
def _eval_expiry_message(stdout: str, stderr: str = "") -> Optional[str]:
    """Detect an evaluation-expiry notice in binary output.

    An expired evaluation binary prints a notice and exits without emitting
    JSON.  The exact wording is not guaranteed stable across builds, so this
    matches loosely: an expiry indicator ("expir...", "time's up") together
    with a licensing context word ("eval...", "license", "trial").  "time's
    up" alone is accepted — nothing else in the binaries prints it.

    Returns the notice text (trimmed, both streams combined) when detected,
    or None.  Callers should surface this text verbatim so the operator sees
    the binary's own message rather than a JSON parse error.

    DISABLED as of this build: no binary emits an evaluation-expiry notice
    anymore, and the new startup license notice contains the word "license",
    which risks a false-positive match against this function's keyword
    check. All call sites below are commented out rather than calling this
    function. Left defined, and not deleted, in case it is needed again.
    """
    combined = "\n".join(s for s in (stdout.strip(), stderr.strip()) if s)
    if not combined:
        return None
    low = combined.lower()
    if "time's up" in low or "times up" in low:
        return combined[:400]
    if "expir" in low and ("eval" in low or "license" in low or "trial" in low):
        return combined[:400]
    return None


def _standard_edition_restriction_reason(stderr: str) -> Optional[str]:
    """If stderr indicates the binary declined because of a Standard-edition
    restriction (a rejected preset, --policy-base/--policy-project,
    --audit-key, --test-plan, or any other Enterprise-only flag), extract
    and return the specific reason. Returns None if no such restriction is
    present in stderr.

    Searches the FULL stderr, not a truncated hint. aishell-gate-exec does
    not separate its own stderr from the policy subprocess it invokes, and
    writes a substantial preamble of its own (DRY-RUN-JSON notice, license
    banner, audit-log warning) before any policy rejection message appears
    -- confirmed empirically against a real captured run, not assumed: a
    rejected preset's actual message did not survive a 500-character
    head-truncated window. A fixed-length truncation is not a reliable way
    to catch this category of error regardless of the exact length chosen,
    since exec's own preamble can grow; this function is the reliable
    check, and truncation windows elsewhere are a secondary improvement for
    errors this function doesn't recognize.
    """
    if not stderr or "not available in the standard edition" not in stderr.lower():
        return None
    reason_lines = [ln.strip() for ln in stderr.splitlines()
                     if ln.strip() and (
                         "not available in the standard edition" in ln.lower()
                         or ln.strip().startswith(("DENY:", "[policy] ERROR:")))]
    return " ".join(reason_lines) if reason_lines else stderr.strip()[:512]


# Set once, the first time any binary invocation's preamble (license banner,
# etc.) is captured this session; consumed and cleared by run_server() right
# after the handler call, so it rides along on whichever tool call happened
# to trigger it first. A module-level flag is enough here — the server loop
# in run_server() is a plain synchronous "for line in sys.stdin" read, one
# request at a time, so there's no concurrent access to guard against.
_LICENSE_SHOWN = False
_PENDING_LICENSE_NOTICE: Optional[str] = None

# Set once, in the initialize handler, from the client's declared
# capabilities. See that handler for why this is captured once rather than
# probed per elicitation attempt.
_CLIENT_SUPPORTS_ELICITATION = False

# Persistent, one-time acknowledgment marker. Written when elicitation
# receives an "accept" response; checked before attempting elicitation again,
# so a session (or a later session, on the same machine) that has already
# been accepted doesn't re-prompt. Not tied to any particular acceptance
# mechanism -- if a different acknowledgment path is added later, it can
# write the same file.
NOTICE_ACK_FILE = Path.home() / ".local" / "share" / "aishell-gate" / ".notice-ack"


def _notice_acknowledged() -> bool:
    return NOTICE_ACK_FILE.is_file()


def _write_notice_ack() -> None:
    NOTICE_ACK_FILE.parent.mkdir(parents=True, exist_ok=True)
    NOTICE_ACK_FILE.write_text(
        f"{_COPYRIGHT_NOTICE}\nAcknowledged via elicitation: "
        f"{time.strftime('%Y-%m-%d %H:%M:%S %z')}\n"
    )


_ELICIT_ID_COUNTER = itertools.count(1)

# Max seconds to wait for a client's elicitation/create response before
# giving up and returning None (falls through to "no elicitation available"
# behavior -- the tool call proceeds unblocked, same as an older client that
# never declared support at all).
#
# 2026-08: this was originally unbounded, on the reasoning that a real modal
# dialog blocks until dismissed and this mirrors that. In practice, against
# real Claude Code, tool calls hung indefinitely and the server appeared to
# stop working entirely -- "commands not loaded" was the symptom, but the
# actual cause was this readline() never returning, because Claude Code's
# real response either never arrived or never matched what this function
# was waiting for. The exact mismatch was never confirmed (no access to a
# live session to inspect it against), so this timeout is the fix that
# matters regardless of what the specific cause turns out to be: whatever is
# wrong, the server must not be able to hang forever because of it.
_ELICIT_TIMEOUT_SEC = float(os.environ.get("AISHELL_ELICIT_TIMEOUT_SEC", "30.0"))


def _request_elicitation(message: str) -> Optional[str]:
    """Send elicitation/create to the client and wait (bounded) for a reply.

    This is the one mechanism in this file that does not rely on the calling
    agent choosing to relay anything: elicitation/create is answered by the
    CLIENT, which per spec is expected to render an actual dialog and block
    the tool call until a human responds. The agent's own cooperation is not
    what stands between this request and a human seeing it.

    Returns "accept", "decline", or "cancel" -- the three actions defined by
    the spec's response model -- or None if the client never declared
    elicitation support, the request errors (e.g. -32601 Method not found),
    the wait times out, or the response can't be correlated/parsed. None
    means "could not elicit," not "declined" -- callers must not treat it
    as a refusal.

    Every raw line read while waiting is logged at DEBUG, even on parse
    failure or ID mismatch, specifically so a real, unexplained failure
    (see _ELICIT_TIMEOUT_SEC's note) leaves a trail in `claude --debug mcp`
    instead of vanishing silently a second time.
    """
    if not _CLIENT_SUPPORTS_ELICITATION:
        return None

    # Integer id, not the earlier string form ("elicit-1"). JSON-RPC 2.0
    # allows either, and server-originated request ids don't collide with
    # client-originated ones regardless (separate id spaces, tracked by
    # whoever issued the request) -- so this wasn't required by spec. It's
    # a defensive simplification: every published example of a client
    # responding to a server-initiated request uses a plain integer, and
    # removing a plausible point of divergence costs nothing.
    req_id = next(_ELICIT_ID_COUNTER)
    request_msg = {
        "jsonrpc": JSONRPC_VERSION,
        "id": req_id,
        "method": "elicitation/create",
        "params": {
            "message": message,
            # A single, unrequired boolean rather than a wholly empty
            # properties object. Every published example of requestedSchema
            # includes at least one property; an empty one is unusual
            # enough that it's a plausible reason a client's form-rendering
            # code does something other than what the spec's plain-text
            # description implies. The three-action response (accept /
            # decline / cancel) is still the real answer -- this field is
            # not read by _write_notice_ack or anything else.
            "requestedSchema": {
                "type": "object",
                "properties": {
                    "acknowledged": {
                        "type": "boolean",
                        "description": "Confirm you have read the notice above.",
                    },
                },
            },
        },
    }
    log.debug("elicitation: sending request id=%s", req_id)
    _send(request_msg)

    deadline = time.monotonic() + _ELICIT_TIMEOUT_SEC
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            log.warning(
                "elicitation: no response to id=%s within %ss -- giving up "
                "and proceeding as if elicitation were unavailable",
                req_id, _ELICIT_TIMEOUT_SEC,
            )
            return None
        try:
            ready, _, _ = select.select([sys.stdin], [], [], remaining)
        except Exception as e:
            log.warning("elicitation: select() failed: %s", e)
            return None
        if not ready:
            continue  # loop back and re-check the deadline

        try:
            line = sys.stdin.readline()
        except Exception as e:
            log.warning("elicitation: failed reading client response: %s", e)
            return None
        if not line:
            log.warning("elicitation: stdin closed while awaiting response")
            return None

        log.debug("elicitation: raw line received: %r", line.strip())

        try:
            msg = json.loads(line)
        except json.JSONDecodeError as e:
            log.warning("elicitation: malformed response: %s", e)
            return None

        if msg.get("id") != req_id:
            # A well-behaved synchronous client shouldn't interleave anything
            # else while this tool call is pending, but if it does, treat it as
            # "could not elicit" rather than guessing at correlation.
            log.warning(
                "elicitation: response id %r does not match request id %r",
                msg.get("id"), req_id,
            )
            return None

        if "error" in msg:
            log.info("elicitation/create not usable: %s",
                     (msg["error"] or {}).get("message", "unknown error"))
            return None

        return (msg.get("result") or {}).get("action")

# Both binaries print their copyright/license banner with fprintf(stderr, ...)
# -- confirmed directly in aishell-gate-policy-126.c and aishell-gate-exec.c,
# both tagged "//issue banner" immediately above the call. It has never been
# on stdout. Matches "[TAG] AIShell-Gate -- ...", where TAG is whatever the
# binary uses ("policy", "gate-exec", ...), through the blank line that ends
# the block. Anchored on the distinctive phrase so it survives minor wording
# changes without also swallowing unrelated stderr (evaluation summaries,
# DBG output, confirmation prompts).
_LICENSE_BANNER_RE = re.compile(
    r"\[\w[\w-]*\]\s*AIShell-Gate\b.*?(?=\n\s*\n|\Z)", re.DOTALL,
)


def _extract_license_banner(stderr: str) -> Optional[str]:
    """Pull the copyright banner out of a binary's stderr, if present."""
    if not stderr:
        return None
    m = _LICENSE_BANNER_RE.search(stderr)
    return m.group(0).strip() if m else None


def _queue_license_notice_if_new(stderr: str) -> None:
    """Queue the license banner for attachment to the current tool response,
    the first time (this process) one is found on a binary's stderr."""
    global _LICENSE_SHOWN, _PENDING_LICENSE_NOTICE
    if _LICENSE_SHOWN:
        return
    banner = _extract_license_banner(stderr)
    if banner:
        _LICENSE_SHOWN = True
        _PENDING_LICENSE_NOTICE = banner


# Sent with every tool reply, not just once per session -- distinct from
# _PENDING_LICENSE_NOTICE above, which carries the binaries' own actual
# startup banner text one time. This is a fixed, short copyright line the
# operator chose to have attached to every response without exception.
# Exact wording as given; not reworded.
# Matches LICENSE.TXT's ACCEPTANCE section, item 2 (NOTICE): the notice
# must state the copyright, that the Software is provided without warranty,
# that Commercial Use requires a purchased license, and that use of the
# Software constitutes acceptance of the license. Three lines, one clause
# each, in that order.
_COPYRIGHT_NOTICE = (
    "AIShell-Gate. Copyright (2026) AIShell Labs LLC. www.aishellgate.com\n"
    "Provided \"AS IS,\" WITHOUT WARRANTY OF ANY KIND, express or implied.\n"
    "Use of this software constitutes acceptance of its license. See LICENSE.TXT."
)


def _with_copyright_notice(result: dict) -> dict:
    """Return `result` with the copyright notice as the FIRST key.

    First, not appended, for the same reason the license notice in
    _server_instructions is placed first: if any client ever truncates a
    large tool response for display, text at the end is what disappears.
    Placing it first is defensive, not just cosmetic -- dict order is
    preserved through json.dumps here (sort_keys is never set), so this
    is also literally the first key in the JSON sent over the wire.
    """
    if not isinstance(result, dict):
        return result
    return {"copyright_notice": _COPYRIGHT_NOTICE, **result}


def _strip_preamble(stdout: str, tool: str, stderr: str = "") -> Tuple[str, Optional[dict]]:
    """Strip any non-JSON preamble from binary stdout before parsing.

    Evaluation copy warnings, banners, and license notices are written by
    the binary before the JSON object.  Any text before the first '{' is
    silently discarded.  If no '{' is found at all, returns an error response
    — with a dedicated, actionable message when the output is an evaluation
    expiry notice, so the operator sees "evaluation ended" instead of a
    generic parse failure.

    NOTE: the license/copyright banner is NOT on this stream -- both binaries
    write it to stderr (see _extract_license_banner). This function used to
    also treat any stdout preamble as the license notice, which meant that
    detection could never fire: the binaries' stdout always begins with '{'
    directly, so the "found a preamble" branch below was dead in the one
    case it existed to handle. License-notice capture now happens separately,
    at each call site, from stderr. This function keeps its original job:
    defending JSON parsing against a stdout preamble should some future or
    third-party binary build ever put one there.

    Returns (cleaned_stdout, None) on success or ("", error_dict) on failure.
    """
    brace = stdout.find('{')
    if brace == -1:
        # Disabled: _eval_expiry_message() detection (see function docstring).
        # expiry = _eval_expiry_message(stdout, stderr)
        # if expiry:
        #     return "", _error_response(tool,
        #         "AIShell-Gate evaluation period has ended — the binary "
        #         "declined to run and produced no JSON.\n"
        #         f"Binary message: {expiry}\n"
        #         "Install a current or licensed build to continue "
        #         "(see LICENSE, or www.aishellgate.com).")
        reason = _standard_edition_restriction_reason(stderr)
        if reason:
            return "", _error_response(tool,
                "This request uses a feature that requires the Enterprise "
                f"edition. Binary message: {reason}")
        out_hint = stdout[:512].strip()
        err_hint = stderr[:512].strip() if stderr else ""
        detail = "  ".join(part for part in (
            f"stdout: {out_hint}" if out_hint else "",
            f"stderr: {err_hint}" if err_hint else "") if part)
        return "", _error_response(tool,
            "binary output contained no JSON object. "
            "Check binary installation and license status."
            + (f"\n{detail}" if detail else ""))
    if brace > 0:
        log.debug("%s: stripped %d bytes of preamble from stdout", tool, brace)
    return stdout[brace:], None
# ---------------------------------------------------------------------------
CONFIRM_LEVELS = {"none": 0, "plan": 1, "action": 2, "typed": 3}

def _max_confirm_level(actions: List[dict]) -> Tuple[str, List[dict]]:
    """Return the highest confirm level in the action list and the actions
    that require human confirmation (level >= action)."""
    max_level  = "none"
    max_score  = 0
    blocking   = []
    for a in actions:
        lvl   = a.get("confirm", "none")
        score = CONFIRM_LEVELS.get(lvl, 0)
        if score > max_score:
            max_score = score
            max_level = lvl
        # The line between "needs a human" and "just run it" falls between
        # none and plan, NOT between plan and action. `plan` means "show the
        # human the whole list and ask once" -- that is still asking a human.
        # Using action as the threshold here left plan-level actions out of
        # actions_requiring_confirmation entirely, so a caller was told a plan
        # was refused and given an empty list of what needed approving.
        if score >= CONFIRM_LEVELS["plan"]:
            blocking.append({
                "idx":     a.get("idx"),
                "cmd":     a.get("cmd"),
                "confirm": lvl,
                "reason":  a.get("reason", ""),
                "binary":  a.get("binary", ""),
            })
    return max_level, blocking


# ---------------------------------------------------------------------------
# aishell-gate-exec exit status interpretation
# ---------------------------------------------------------------------------
# Codes 1-6 describe GATE outcomes: what the gateway decided or failed to do.
# Code 7 means every gate passed and the command actually RAN, but returned
# non-zero -- a command failure, not a policy decision.
#
# Before exec 0.57.0 the two shared code 1, so `rm nofile` was reported to the
# agent as "denied by policy". An agent acts on that literally: it rewrites a
# command to satisfy a rule that never objected, and loops. Keeping these
# meanings apart matters more for an AI caller than for a human one.
_EXIT_MEANINGS = {
    0: "all actions executed successfully",
    1: "one or more actions denied by policy",
    2: "operator confirmation refused",
    3: "policy engine error",
    4: "JSON parse error",
    5: "usage or configuration error",
    6: "gateway could not execute the command (binary not found, or exec error)",
    7: "command executed but returned a non-zero exit status",
    8: "command ran and was killed by the gateway for exceeding its time limit",
}


def _exec_outcome(rc: int, stdout: str, stderr: str) -> dict:
    """Build the shared result fields describing one exec invocation."""
    if rc >= 128:
        outcome = f"command ran and was killed by signal {rc - 128}"
    elif rc < 0:
        outcome = f"exec terminated by signal {-rc}"
    else:
        outcome = _EXIT_MEANINGS.get(rc, f"unknown exit code {rc}")

    # exec reports the command's own exit status on stderr, because the
    # process exit code is 7 for every command failure. Recover the real one.
    command_exit = None
    if stderr:
        found = re.findall(r"action (\d+) exited (\d+)", stderr)
        if found:
            command_exit = int(found[0][1])

    out = {
        # True when the commands actually ran, whatever they returned.
        # This is NOT "succeeded" -- check succeeded for that.
        "executed": (rc in (0, 7, 8) or rc >= 128),
        "succeeded": (rc == 0),
        "policy_allowed": (rc != 1),
        "exit_code": rc,
        "outcome": outcome,
        "stdout_tail": stdout[-4000:].strip() if stdout else "",
        "stderr_tail": stderr[-1000:].strip() if stderr else "",
    }

    if rc == 7:
        if command_exit is not None:
            out["command_exit_code"] = command_exit
        out["message"] = (
            "The command ran. Policy allowed it and nothing blocked it. It "
            "returned a non-zero exit status"
            + (f" ({command_exit})" if command_exit is not None else "")
            + ". This is the command's own failure -- read stderr_tail for the "
              "reason and fix the command. Do not rewrite it to satisfy policy; "
              "policy did not object."
        )
    elif rc == 8:
        out["timed_out"] = True
        out["message"] = (
            "The command ran and the gateway killed it for exceeding its "
            "time limit. Policy allowed it; it did not finish. Do not retry "
            "the same command unchanged -- either it needs longer, or it is "
            "waiting on input it will never get, or it does not terminate. "
            "Narrow the work or raise --action-timeout."
        )
    else:
        out["message"] = f"Execution {'succeeded' if rc == 0 else 'failed'}: {outcome}."
    return out


# ---------------------------------------------------------------------------
# Tool: get_version
# ---------------------------------------------------------------------------
def tool_get_version(cfg: dict, args: dict) -> dict:
    """
    Report the installed edition (standard/enterprise), version strings for
    both binaries, and which tools are available in this install.  Call this
    first to discover feature availability before calling other tools.
    """
    edition    = cfg["_edition"]
    enterprise = (edition == EDITION_ENTERPRISE)
    policy_ver = cfg["_policy_version"].splitlines()[0] if cfg["_policy_version"] else "not found"
    exec_ver   = cfg["_exec_version"].splitlines()[0]   if cfg["_exec_version"]   else "not found"

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "get_version",
        "version":        policy_ver,   # product version — always derived from policy binary
        "edition":        edition,
        "exec_version":   exec_ver,
        "policy_version": policy_ver,
        "mcp_protocol_version": PROTO_MCP_VERSION,   # wire protocol; not the product version
        "available_tools": {
            "evaluate_plan":       True,
            "execute_plan":        True,
            "get_version":         True,
            "evaluate_command":    True,
            "get_policy_template": True,
            "verify_policy":       enterprise,
            "verify_audit_log":    enterprise,
        },
        "enterprise_features": {
            "hmac_audit_chain":         enterprise,
            "audit_verify":             enterprise,
            "policy_test_suite":        enterprise,
            "cryptographic_session_id": enterprise,
        },
        "config_summary": {
            "preset":           cfg.get("preset"),
            "confirm_mode":     cfg.get("confirm_mode", "plan_only"),
            "jail_root":        cfg.get("jail_root"),
            "sandbox":          cfg.get("sandbox"),
            "source":           cfg.get("source"),
            "audit_log":        cfg.get("audit_log"),
            "policy_audit_log": cfg.get("policy_audit_log"),
            "audit_key_set":    bool(cfg.get("audit_key")),
            "eval_timeout":     cfg.get("eval_timeout"),
        },
    }


# ---------------------------------------------------------------------------
# Tool: install_engine
# ---------------------------------------------------------------------------
def _platform_supported() -> bool:
    import platform
    system  = platform.system()
    machine = platform.machine()
    # normalise common aliases (amd64 -> x86_64)
    if machine in ("amd64", "AMD64"):
        machine = "x86_64"
    return (system, machine) == ENGINE_SUPPORTED_PLATFORM


def _install_prompt() -> str:
    """dnf/apt-style confirmation shown when install_engine is called without
    confirm=true. Names the download source (www.aishellgate.com) so the
    user can see exactly where the binary is coming from before agreeing.
    The size is an approximate ('~') on purpose: it needs no per-release
    upkeep and never lies."""
    return (
        "Install AIShell-Gate engine (aishell-gate-exec, aishell-gate-policy)\n"
        "  Source:         www.aishellgate.com/download.php   (current release)\n"
        "  Download size:  ~1.1 M  →  your home directory (full release kept)\n"
        "  Binaries also copied to ~/.local/share/aishell-gate/bin   (no sudo)\n"
        "Is this ok [y/N]:"
    )


def _engine_missing_message() -> str:
    """The finalized handoff message text. Kept as one function so the wording
    lives in exactly one place, whether it's surfaced via a blocked tool call
    or returned directly by install_engine."""
    if not _platform_supported():
        return (
            "AIShell-Gate engine not found, and this system is not supported: "
            "the engine is available for Linux x86-64 only."
        )
    return (
        "AIShell-Gate engine not found (aishell-gate-exec, aishell-gate-policy). "
        "Call install_engine to install it, or install it by hand: download "
        f"the release at {ENGINE_DOWNLOAD_URL} and copy the two binaries into "
        "~/.local/share/aishell-gate/bin (or /usr/local/bin)."
    )


def _safe_extractall(tf, members: list, dest: Path) -> None:
    """Extract every given member of an already-open tar archive into dest,
    skipping any whose resolved path would land outside dest.

    Python's tarfile has no built-in guard against this before the `filter=`
    argument added in 3.12 (PEP 706); this project targets Python 3.9+ (see
    pyproject.toml), so the check is done by hand here rather than assumed
    available. The archive comes from a download (aishellgate.com) this
    project controls, but the check costs nothing and removes any
    dependency on that always being true.
    """
    resolved_dest = dest.resolve()
    safe, skipped = [], 0
    for member in members:
        target = (dest / member.name).resolve()
        try:
            target.relative_to(resolved_dest)
        except ValueError:
            skipped += 1
            continue
        safe.append(member)
    if skipped:
        log.warning("install_engine: skipped %d unsafe archive path(s)", skipped)
    tf.extractall(dest, members=safe)


def tool_install_engine(cfg: dict, args: dict) -> dict:
    """
    Download the AIShellGate policy engine release from aishellgate.com and
    install it. Downloads the tarball, verifies its checksum, and unpacks the FULL
    release into the user's home directory -- left there afterward, archive
    included, same as everything else this project downloads (nothing gets
    deleted). The two compiled binaries (aishell-gate-exec, aishell-gate-
    policy) are additionally copied from that unpacked copy into
    USER_INSTALL_DIR (~/.local/share/aishell-gate/bin — created if absent;
    no sudo needed): the unpacked release directory's name carries the
    version number and changes every release, so it can't be a stable
    target for the resolver on its own -- especially now that nothing gets
    cleaned up, so more than one versioned copy can exist side by side over
    time. USER_INSTALL_DIR stays fixed and always holds whichever version
    was installed most recently.

    Does nothing unless called with confirm=true -- this tool performs a
    real download and writes executable files to disk, so it requires an
    explicit, separate confirmation rather than running as a side effect of
    any other tool call.

    On success, re-resolves exec_binary/policy_binary and re-runs edition
    detection so the engine is usable immediately, with no server restart.
    """
    if _engine_available(cfg):
        return {
            "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool": "install_engine",
            "status": "already_installed",
            "message": "AIShell-Gate engine is already installed. Nothing to do.",
            "exec_binary":   cfg["exec_binary"],
            "policy_binary": cfg["policy_binary"],
        }

    if not _platform_supported():
        import platform
        return _error_response("install_engine", (
            f"AIShell-Gate engine is available for Linux x86-64 only. "
            f"Detected: {platform.system()} {platform.machine()}. Nothing to install."
        ))

    if not args.get("confirm"):
        return {
            "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool": "install_engine",
            "status": "confirmation_required",
            "message": _install_prompt(),
            "note": (
                "This is a yes/no confirmation. Nothing has been downloaded. "
                "A 'y' from the user means: call install_engine again with "
                "confirm=true. Anything else means: do not install."
            ),
        }

    import urllib.request
    import urllib.error
    import tarfile
    import hashlib

    try:
        # 1. Download the tarball straight into the home directory. This is
        #    the permanent copy, not scratch space -- left in place after
        #    install, same choice already made for the network installer:
        #    nothing this project downloads gets deleted unless asked for.
        #    The website always serves one fixed, unversioned file, so
        #    there's no "find the latest release" lookup needed first.
        home         = Path.home()
        archive_path = home / ENGINE_ARCHIVE_NAME

        req = urllib.request.Request(ENGINE_DOWNLOAD_URL, headers={"User-Agent": "aishell-gate-mcp"})
        with urllib.request.urlopen(req, timeout=120) as resp, open(archive_path, "wb") as out:
            shutil.copyfileobj(resp, out)

        # 2. Verify checksum. checksum.php always returns a hash for
        #    whatever file download.php just served, so this check is
        #    never optional.
        req = urllib.request.Request(ENGINE_CHECKSUM_URL, headers={"User-Agent": "aishell-gate-mcp"})
        with urllib.request.urlopen(req, timeout=30) as resp:
            expected = resp.read().decode("utf-8").split()[0].strip().lower()

        h = hashlib.sha256()
        with archive_path.open("rb") as f:
            for chunk in iter(lambda: f.read(65536), b""):
                h.update(chunk)
        actual = h.hexdigest().lower()

        if actual != expected:
            return _error_response("install_engine", (
                f"Checksum verification failed for the downloaded release "
                f"(expected {expected}, got {actual}). Nothing was "
                f"installed. The download is kept at {archive_path} for "
                f"inspection -- delete it by hand once you're done "
                f"looking, or install manually from {ENGINE_DOWNLOAD_URL}."
            ))

        # 3. Unpack the FULL release into the home directory and leave it
        #    there too -- docs, other scripts, everything, not just the two
        #    binaries. See _safe_extractall's docstring for why the
        #    path-traversal check below isn't just tarfile.extractall().
        with tarfile.open(archive_path, "r:gz") as tf:
            members = tf.getmembers()
            if not members:
                return _error_response("install_engine", (
                    f"Downloaded release archive is empty. Nothing was "
                    f"installed. Archive kept at {archive_path} for inspection."
                ))
            top_level = Path(members[0].name).parts[0]
            _safe_extractall(tf, members, home)

        unpacked_dir = home / top_level

        # 4. Copy just the two binaries from the freshly-unpacked release
        #    into USER_INSTALL_DIR -- see the function docstring for why
        #    this fixed location still matters even though the full,
        #    versioned release now persists in the home directory too.
        exec_name   = Path(cfg["exec_binary"]).name
        policy_name = Path(cfg["policy_binary"]).name
        try:
            USER_INSTALL_DIR.mkdir(parents=True, exist_ok=True)
        except OSError as e:
            return _error_response("install_engine", (
                f"Cannot create install directory {USER_INSTALL_DIR}: {e}. "
                f"The full release is still available at {unpacked_dir}. "
                f"Please install manually (copy the binaries to /usr/local/bin)."
            ))

        found = {}
        for name in (exec_name, policy_name):
            src = unpacked_dir / "bin" / name
            if src.is_file():
                dest = USER_INSTALL_DIR / name
                shutil.copyfile(src, dest)
                dest.chmod(0o755)
                found[name] = str(dest)

        if exec_name not in found or policy_name not in found:
            return _error_response("install_engine", (
                f"Downloaded release did not contain both expected binaries "
                f"(bin/{exec_name}, bin/{policy_name}) under {unpacked_dir}. "
                f"Nothing usable was installed. Please install manually from "
                f"{ENGINE_DOWNLOAD_URL}."
            ))

        # 5. Re-resolve and re-detect so the engine works immediately.
        _reresolve_binaries(cfg)
        _detect_edition(cfg)

        return {
            "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool": "install_engine",
            "status": "installed",
            "message": (
                f"AIShell-Gate engine installed. Binaries in "
                f"{USER_INSTALL_DIR}; full release kept at {unpacked_dir} "
                f"(archive: {archive_path})."
            ),
            "exec_binary":   cfg["exec_binary"],
            "policy_binary": cfg["policy_binary"],
            "edition":       cfg["_edition"],
        }

    except urllib.error.URLError as e:
        return _error_response("install_engine", (
            f"Network error while downloading the engine: {e}. "
            f"Please install manually from {ENGINE_DOWNLOAD_URL}."
        ))
    except Exception as e:
        log.exception("install_engine failed")
        return _error_response("install_engine", (
            f"Install failed: {e}. Please install manually from {ENGINE_DOWNLOAD_URL}."
        ))


# ---------------------------------------------------------------------------
# Tool: evaluate_plan
# ---------------------------------------------------------------------------
def tool_evaluate_plan(cfg: dict, args: dict) -> dict:
    """
    Evaluate a list of commands against the active policy without executing
    anything.  Returns a per-action assessment: decision, confirm level,
    risk score, resolved binary path, and reason.

    This is the recommended first step before execute_plan.  The agent can
    inspect the assessment and adjust its plan (remove denied actions, inform
    the user of required confirmations) before committing to execution.
    """
    goal     = str(args.get("goal", ""))
    commands = [str(c) for c in args.get("commands", [])]
    strategy = str(args.get("strategy", "fail_fast"))

    if not goal:
        return _error_response("evaluate_plan", "'goal' is required")
    if not commands:
        return _error_response("evaluate_plan", "'commands' list is required and must not be empty")
    if len(commands) > 24:
        return _error_response("evaluate_plan",
            f"too many commands ({len(commands)}); maximum is 24 per plan")

    envelope = _build_envelope(goal, commands, strategy)

    try:
        rc, stdout, stderr = _invoke_exec(
            cfg, envelope, extra_argv=["--dry-run-json"]
        )
    except RuntimeError as e:
        return _error_response("evaluate_plan", str(e))

    log.debug("evaluate_plan exec rc=%d stdout=%d bytes", rc, len(stdout))

    if not stdout.strip():
        # Disabled: _eval_expiry_message() detection (see function docstring).
        # expiry = _eval_expiry_message("", stderr)
        # if expiry:
        #     return _error_response("evaluate_plan",
        #         "AIShell-Gate evaluation period has ended — exec declined "
        #         f"to run (exit {rc}).\nBinary message: {expiry}")
        reason = _standard_edition_restriction_reason(stderr)
        if reason:
            return _error_response("evaluate_plan",
                "This plan uses a feature that requires the Enterprise "
                f"edition. Binary message: {reason}")
        return _error_response("evaluate_plan",
            f"exec produced no output (exit {rc}). "
            f"Check that exec and policy binaries are installed correctly.\n"
            f"stderr: {stderr[:512]}")

    _queue_license_notice_if_new(stderr)
    stdout, err = _strip_preamble(stdout, "evaluate_plan", stderr)
    if err:
        return err

    try:
        data = json.loads(stdout)
    except json.JSONDecodeError as e:
        return _error_response("evaluate_plan",
            f"could not parse exec output as JSON: {e}\noutput: {stdout[:200]}")

    try:
        _check_response_protocol(data, PROTO_DRY_RUN_NAME, "evaluate_plan")
    except ValueError as e:
        return _error_response("evaluate_plan", str(e))

    actions = data.get("actions", [])
    max_lvl, blocking = _max_confirm_level(
        [a for a in actions if a.get("decision") == "allow"]
    )
    denied = [a for a in actions if a.get("decision") == "deny"]

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "evaluate_plan",
        "dry_run": True,
        "goal": data.get("goal", goal),
        "strategy": data.get("strategy", strategy),
        "overall_decision": data.get("overall_decision", "deny"),
        "actions": actions,
        "summary": {
            "total":          len(actions),
            "allowed":        len([a for a in actions if a.get("decision") == "allow"]),
            "denied":         len(denied),
            "max_confirm":    max_lvl,
            "blocking_count": len(blocking),
        },
        "guidance": _evaluate_guidance(data.get("overall_decision"), denied, blocking, max_lvl),
    }


def _evaluate_guidance(overall: Optional[str], denied: list, blocking: list, max_lvl: str) -> str:
    """Generate a human-readable guidance string for the agent to surface."""
    parts = []
    if overall == "deny":
        if denied:
            cmds = ", ".join(f"'{a['cmd']}'" for a in denied[:3])
            more = f" and {len(denied)-3} more" if len(denied) > 3 else ""
            parts.append(
                f"{len(denied)} action(s) denied by policy: {cmds}{more}. "
                f"Remove these from the plan or request a policy exception."
            )
        else:
            parts.append("Plan denied by policy. No actions were allowed.")
    else:
        parts.append("All actions allowed by policy.")

    if max_lvl != "none":
        if max_lvl == "typed":
            parts.append(
                f"{len(blocking)} action(s) require typed operator confirmation "
                f"and cannot be executed through the MCP interface. "
                f"Ask the operator to lower the confirm level in their policy "
                f"file, or run these commands manually."
            )
        elif max_lvl == "plan":
            parts.append(
                f"{len(blocking)} action(s) require a single whole-plan "
                f"confirmation: the operator is shown every command and "
                f"approves once. Use execute_plan to attempt execution."
            )
        else:
            parts.append(
                f"{len(blocking)} action(s) require operator confirmation "
                f"before execution. Use execute_plan to attempt execution — "
                f"the server will report exactly which actions need approval."
            )
    return " ".join(parts)


# ---------------------------------------------------------------------------
# Tool: execute_plan
# ---------------------------------------------------------------------------
def tool_execute_plan(cfg: dict, args: dict) -> dict:
    """
    Execute a list of commands via aishell-gate-exec.

    Before invoking exec, runs a pre-flight evaluation to check confirm
    levels.  If any action requires confirmation:

      plan_only mode (default):
        Execution is blocked and a structured report is returned explaining
        which actions require human involvement.

      relay mode (confirm_mode: "relay" in aishell-gate-mcp.json):
        Execution is launched asynchronously.  The exec subprocess
        communicates confirmation requests over a named FIFO pair.
        This tool returns immediately with status "pending_confirmation"
        and an execution_id.  The human reviews the request and calls
        confirm_action with the execution_id and their response.
        Poll get_execution_result for the final outcome.

      operator_pipe mode (confirm_mode: "operator_pipe" in aishell-gate-mcp.json):
        This call blocks while exec waits on a real, operator-owned
        --confirm-pipe session (aishell-gate-confirm, run by the human on
        their own terminal). No FIFO is created or read by this server, no
        confirm request is returned to the caller, and confirm_action /
        get_execution_result are not used for this path.
    """
    goal     = str(args.get("goal", ""))
    commands = [str(c) for c in args.get("commands", [])]
    strategy = str(args.get("strategy", "fail_fast"))

    if not goal:
        return _error_response("execute_plan", "'goal' is required")
    if not commands:
        return _error_response("execute_plan", "'commands' list is required and must not be empty")
    if len(commands) > 24:
        return _error_response("execute_plan",
            f"too many commands ({len(commands)}); maximum is 24 per plan")

    # ---- Phase 1: pre-flight evaluation ----
    envelope = _build_envelope(goal, commands, strategy)
    try:
        rc_dry, stdout_dry, stderr_dry = _invoke_exec(
            cfg, envelope, extra_argv=["--dry-run-json"]
        )
    except RuntimeError as e:
        return _error_response("execute_plan", f"pre-flight evaluation failed: {e}")

    if not stdout_dry.strip():
        # Include exec's stderr tail so the caller can see *why* dry-run emitted
        # nothing (unknown flag, binary/config mismatch, etc.) instead of having
        # to re-run exec manually to recover the diagnostic.
        stderr_tail = stderr_dry[-512:].strip() if stderr_dry else ""
        # Disabled: _eval_expiry_message() detection (see function docstring).
        # expiry = _eval_expiry_message("", stderr_dry or "")
        # if expiry:
        #     return _error_response("execute_plan",
        #         "AIShell-Gate evaluation period has ended — exec declined "
        #         f"to run (rc={rc_dry}).\nBinary message: {expiry}")
        reason = _standard_edition_restriction_reason(stderr_dry)
        if reason:
            return _error_response("execute_plan",
                "This plan uses a feature that requires the Enterprise "
                f"edition. Binary message: {reason}")
        msg = (
            f"pre-flight evaluation produced no output (exec rc={rc_dry}). "
            f"Check exec and policy binary configuration."
        )
        if stderr_tail:
            msg += f"\n[exec stderr]\n{stderr_tail}"
        return _error_response("execute_plan", msg)

    _queue_license_notice_if_new(stderr_dry or "")
    stdout_dry, err = _strip_preamble(stdout_dry, "execute_plan", stderr_dry or "")
    if err:
        return err

    try:
        dry_data = json.loads(stdout_dry)
        _check_response_protocol(dry_data, PROTO_DRY_RUN_NAME, "execute_plan/pre-flight")
    except (json.JSONDecodeError, ValueError) as e:
        return _error_response("execute_plan", f"pre-flight parse error: {e}")

    overall = dry_data.get("overall_decision", "deny")
    actions = dry_data.get("actions", [])
    denied  = [a for a in actions if a.get("decision") == "deny"]

    # Reject plans with any denied action
    if overall == "deny" or denied:
        return {
            "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool": "execute_plan",
            "executed": False,
            "blocked_reason": "policy_denied",
            "denied_actions": [
                {"idx": a.get("idx"), "cmd": a.get("cmd"), "reason": a.get("reason", "")}
                for a in denied
            ],
            "message": (
                f"{len(denied)} action(s) denied by policy. "
                f"Review the denied_actions list and adjust the plan or "
                f"request a policy exception before retrying."
            ),
        }

    # Check confirm levels on allowed actions
    max_lvl, blocking = _max_confirm_level(actions)
    confirm_mode = cfg.get("confirm_mode", "plan_only")

    if max_lvl != "none":
        # operator_pipe is checked first and unconditionally on max_lvl: the
        # real aishell-gate-confirm session on the other end already handles
        # plan, action, and typed levels itself (it's the same tool the SSH
        # deployment has always used), so there is no per-level distinction
        # to make here the way there is between plan_only and relay below.
        if confirm_mode == "operator_pipe":
            return _execute_plan_operator_pipe(cfg, envelope, goal, commands)

        # Anything above `none` needs a human, so the relay has to be set up.
        #
        # confirm_mode controls HOW MUCH human interaction this deployment is
        # willing to broker, not whether the relay exists:
        #   "relay"     — every level is brokered (plan, action, typed)
        #   "plan_only" — the single whole-plan gate is brokered; per-action
        #                 and typed gates are declined with an explanation.
        #
        # Previously the test here was `max_lvl in ("action","typed")`, which
        # put plan-level on the same side of the line as none. A plan whose
        # highest level was `plan` therefore skipped the relay, ran with
        # --confirm-tty /dev/null, and exec's plan gate read EOF as a refusal:
        # "Plan review refused" for a plan no human was ever shown. That also
        # meant the default confirm_mode (plan_only) could not successfully
        # confirm anything at all.
        if max_lvl in ("action", "typed") and confirm_mode != "relay":
            # plan_only mode — per-action confirmation is out of scope here
            if max_lvl == "typed":
                reason = (
                    "One or more actions require typed operator confirmation, which "
                    "cannot be satisfied through the MCP interface in plan_only mode. "
                    "Ask the operator to run these commands directly, or lower the "
                    "confirm level in their policy file, or set confirm_mode to "
                    "'relay' in aishell-gate-mcp.json to enable human confirmation relay."
                )
            else:
                reason = (
                    "One or more actions require confirmation of each individual "
                    "command. In plan_only mode only the single whole-plan gate is "
                    "brokered. Lower the confirm level in the policy file, or set "
                    "confirm_mode to 'relay' in aishell-gate-mcp.json to enable "
                    "per-action confirmation."
                )
            return {
                "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
                "tool": "execute_plan",
                "executed": False,
                "blocked_reason": "confirmation_required",
                "max_confirm_level": max_lvl,
                "actions_requiring_confirmation": blocking,
                "message": reason,
            }

        # relay mode — launch async with FIFO relay
        return _execute_plan_relay(cfg, envelope, goal, commands, blocking, max_lvl)

    # ---- Phase 2: live execution (no confirmation needed) ----
    log.info("execute_plan: pre-flight passed, invoking exec for live run")
    try:
        rc, stdout, stderr = _invoke_exec(cfg, envelope)
    except RuntimeError as e:
        return _error_response("execute_plan", f"exec invocation failed: {e}")

    result = {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "execute_plan",
        "goal": goal,
        "commands_submitted": len(commands),
    }
    _queue_license_notice_if_new(stderr)
    result.update(_exec_outcome(rc, stdout, stderr))
    return result


def _execute_plan_operator_pipe(cfg: dict, envelope: dict, goal: str,
                                commands: List[str]) -> dict:
    """
    Execute a plan requiring confirmation by pointing aishell-gate-exec
    directly at a real, operator-owned --confirm-pipe (an aishell-gate-confirm
    session the human runs on their own terminal -- see aishell-gate-confirm(1)
    and the SSH deployment guide).

    Unlike _execute_plan_relay, this server does not create, open, or read
    either FIFO itself -- the operator's aishell-gate-confirm process owns
    them both. This call is a single ordinary blocking subprocess invocation,
    the same shape as the no-confirmation-needed path in tool_execute_plan,
    just with a longer timeout because a human may take a while to answer.
    confirm_action and get_execution_result are not involved: the agent is
    never shown the confirm request or (for typed levels) the challenge code,
    so there is nothing for it to relay, echo, or self-supply.
    """
    confirm_pipe = cfg.get("confirm_pipe")
    if not confirm_pipe:
        return _error_response("execute_plan",
            "confirm_mode is 'operator_pipe' but 'confirm_pipe' is not set "
            "in aishell-gate-mcp.json. Set it to the basepath the operator's "
            "aishell-gate-confirm session is using, e.g.:\n"
            '  "confirm_pipe": "/run/aishell-gate/confirm"\n'
            "The operator must have that session running before this call, "
            "or exec will wait out --confirm-timeout with nothing to answer.")

    extra = ["--confirm-pipe", confirm_pipe]
    if cfg.get("confirm_lock"):
        extra += ["--confirm-lock", cfg["confirm_lock"]]
    if cfg.get("confirm_timeout"):
        extra += ["--confirm-timeout", str(int(cfg["confirm_timeout"]))]

    # exec will block on the operator's terminal for up to --confirm-timeout
    # (or its own 120s default if we didn't pass one). Give this Python-level
    # subprocess call a margin above that so we don't kill exec out from
    # under a human who is still typing -- same reasoning as
    # _CONFIRM_EXEC_MARGIN in the relay path.
    exec_wait = float(cfg.get("confirm_timeout") or _EXEC_DEFAULT_CONFIRM_TIMEOUT)

    log.info("execute_plan: operator_pipe mode, waiting on %s", confirm_pipe)
    try:
        rc, stdout, stderr = _invoke_exec(
            cfg, envelope, extra_argv=extra,
            timeout=exec_wait + _CONFIRM_EXEC_MARGIN,
        )
    except RuntimeError as e:
        return _error_response("execute_plan", f"exec invocation failed: {e}")

    result = {
        "protocol":           {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool":                "execute_plan",
        "goal":                goal,
        "commands_submitted":  len(commands),
        "confirm_path":        "operator_pipe",
    }
    _queue_license_notice_if_new(stderr)
    result.update(_exec_outcome(rc, stdout, stderr))
    return result


def _execute_plan_relay(cfg: dict, envelope: dict, goal: str,
                        commands: List[str], blocking: list, max_lvl: str) -> dict:
    """
    Launch aishell-gate-exec asynchronously with a FIFO confirmation relay.
    Returns immediately with execution_id for the caller to track.
    """
    exec_id   = str(uuid.uuid4())
    fifo_base = f"/tmp/aishell-mcp-{exec_id}"
    fifo_req  = fifo_base + ".req"
    fifo_resp = fifo_base + ".resp"

    # Create the FIFOs before launching exec — exec opens them on startup
    try:
        os.mkfifo(fifo_req,  mode=0o600)
        os.mkfifo(fifo_resp, mode=0o600)
    except OSError as e:
        return _error_response("execute_plan", f"could not create confirmation FIFOs: {e}")

    resp_queue: queue.Queue = queue.Queue()

    state: dict = {
        "status":      "starting",
        "goal":        goal,
        "commands":    commands,
        "confirm_req": None,
        # Counter identifying the current confirmation gate. Incremented once
        # per request exec sends. A human response is only accepted if it
        # names the gate the human was actually shown -- see tool_confirm_action.
        "confirm_seq": 0,
        "resp_queue":  resp_queue,
        "result":      None,
        "fifo_base":   fifo_base,
        "thread":      None,
        "ready_event": threading.Event(),
    }

    with _PENDING_LOCK:
        _PENDING[exec_id] = state

    def relay_thread() -> None:
        req_fd  = None
        resp_fd = None
        relay_note = None    # set when a confirmation could not be delivered
        try:
            # Build exec argv with --confirm-pipe
            # --confirm-timeout keeps exec's patience strictly longer than
            # ours; see _CONFIRM_EXEC_MARGIN for why that ordering matters.
            argv    = _build_exec_argv(cfg, extra=[
                "--confirm-pipe", fifo_base,
                "--confirm-timeout",
                str(int(_CONFIRM_HUMAN_TIMEOUT + _CONFIRM_EXEC_MARGIN)),
            ])
            payload = json.dumps(envelope).encode()

            log.debug("relay: launching exec %s", " ".join(argv))

            proc = subprocess.Popen(
                argv,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )

            # Write the plan to exec stdin then close it.
            # Set proc.stdin = None so the later proc.communicate() skips its
            # internal stdin.flush() — flushing an already-closed stream would
            # raise ValueError("flush of closed file"), which communicate()
            # catches only for BrokenPipeError.
            proc.stdin.write(payload)
            proc.stdin.close()
            proc.stdin = None

            # Open FIFOs — exec opens .req for writing and .resp for reading.
            # We open in the opposite direction, with a bounded handshake so
            # that a crashed or wedged exec can't block this thread forever.
            #
            # .req (MCP reads, exec writes):  O_RDONLY | O_NONBLOCK succeeds
            #   immediately even if exec hasn't opened its write end yet —
            #   safe to open first. The read end stays non-blocking only
            #   during the handshake; we restore blocking mode before the
            #   relay loop so the existing readline() logic is unchanged.
            #
            # .resp (MCP writes, exec reads):  O_WRONLY | O_NONBLOCK returns
            #   ENXIO until exec opens its read end. We retry with a bounded
            #   deadline and also check proc.poll() so an exec that dies
            #   during startup is surfaced immediately instead of waited out.
            req_raw  = os.open(fifo_req, os.O_RDONLY | os.O_NONBLOCK)
            resp_raw = None
            deadline = time.monotonic() + _FIFO_HANDSHAKE_TIMEOUT
            while resp_raw is None:
                if proc.poll() is not None:
                    try:
                        _, err_bytes = proc.communicate(timeout=1)
                        err_tail = err_bytes.decode(errors="replace")[-500:].strip()
                    except Exception:
                        err_tail = ""
                    try: os.close(req_raw)
                    except OSError: pass
                    raise RuntimeError(
                        f"aishell-gate-exec exited rc={proc.returncode} before "
                        f"the FIFO handshake completed"
                        + (f": {err_tail}" if err_tail else "")
                    )
                try:
                    resp_raw = os.open(fifo_resp, os.O_WRONLY | os.O_NONBLOCK)
                except OSError as e:
                    if e.errno != errno.ENXIO:
                        try: os.close(req_raw)
                        except OSError: pass
                        raise
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        try: os.close(req_raw)
                        except OSError: pass
                        raise RuntimeError(
                            f"aishell-gate-exec did not complete FIFO handshake "
                            f"within {_FIFO_HANDSHAKE_TIMEOUT:g}s — subprocess "
                            f"may be wedged."
                        )
                    # Brief sleep; wake early if exec prints to stderr (an
                    # early stderr write usually precedes an early exit).
                    try:
                        select.select(
                            [proc.stderr.fileno()], [], [],
                            min(0.05, remaining),
                        )
                    except (OSError, ValueError):
                        time.sleep(min(0.05, remaining))

            # Handshake done — restore blocking mode and wrap in file objects
            # so the existing relay loop (readline / write+flush) is unchanged.
            os.set_blocking(req_raw,  True)
            os.set_blocking(resp_raw, True)
            req_fd  = os.fdopen(req_raw,  "r", buffering=1)
            resp_fd = os.fdopen(resp_raw, "w", buffering=1)

            with _PENDING_LOCK:
                state["status"] = "running"

            # Relay loop — read confirmation requests, hand to human, write responses.
            #
            # readline() is wrapped in select() so we never block indefinitely on
            # a silently-wedged exec. Normal behavior: if exec exits or closes its
            # FIFO write end, select returns readable with an EOF — readline gives
            # "" and we break cleanly. Pathological behavior: exec is alive but
            # neither writing nor exiting. In that case select returns without
            # readiness after _RELAY_POLL_INTERVAL, we poll proc.poll(), and if
            # exec has exited we break out. Legitimate long-running exec commands
            # that don't need confirmation just keep us looping — they are NOT
            # bounded by this timeout.
            req_fileno = req_fd.fileno()
            while True:
                try:
                    ready, _, _ = select.select(
                        [req_fileno], [], [], _RELAY_POLL_INTERVAL,
                    )
                except (OSError, select.error) as e:
                    # EINTR on a slow signal — just retry the select.
                    if getattr(e, "errno", None) == errno.EINTR:
                        continue
                    raise

                if not ready:
                    # Liveness check. If exec has exited without closing the
                    # FIFO cleanly (rare — kernel normally propagates EOF) or
                    # if we somehow missed the EOF, break out so we can reap
                    # the process and unwind instead of spinning forever.
                    if proc.poll() is not None:
                        log.debug(
                            "relay: exec exited rc=%s while relay still "
                            "polling for confirm requests — breaking",
                            proc.returncode,
                        )
                        break
                    continue

                line = req_fd.readline()
                if not line:
                    # exec closed the confirmation FIFO. This means the gates
                    # are resolved, NOT that execution is finished — exec
                    # closes these before it runs anything so it can drop the
                    # confirmation lock first. The commands run after this.
                    break

                line = line.strip()
                if not line:
                    continue

                try:
                    req_data = json.loads(line)
                except json.JSONDecodeError:
                    log.warning("relay: malformed confirm request: %s", line[:200])
                    continue

                log.debug("relay: confirm request: %s", line[:200])

                # Open a new gate. Each gate gets its own number so a response
                # can be matched to the exact action the human was shown.
                with _PENDING_LOCK:
                    state["confirm_seq"] += 1
                    my_seq = state["confirm_seq"]
                    req_data["confirm_seq"] = my_seq
                    state["status"]      = "pending_confirmation"
                    state["confirm_req"] = req_data
                # Unblock execute_plan's ready wait so it returns the real status.
                # Safe to call on every iteration; Event.set() is idempotent.
                state["ready_event"].set()

                # Wait for a human response that belongs to THIS gate.
                #
                # Responses for any other gate are discarded, never spent here.
                # Without this check a duplicate or late response left in the
                # queue would be consumed by the next gate the moment it opened,
                # approving a command no human ever saw -- which silently turns
                # per-action confirmation into one approval covering the rest.
                deadline = time.monotonic() + _CONFIRM_HUMAN_TIMEOUT
                response = None
                while response is None:
                    remaining = deadline - time.monotonic()
                    if remaining <= 0:
                        log.warning("relay: human confirmation timed out — treating as refusal")
                        response = "no"
                        break
                    try:
                        got_seq, got_resp = resp_queue.get(timeout=remaining)
                    except queue.Empty:
                        continue
                    if got_seq != my_seq:
                        log.warning(
                            "relay: discarding response for confirmation %s — "
                            "the open gate is %s", got_seq, my_seq,
                        )
                        continue
                    response = got_resp

                log.debug("relay: writing response '%s' to resp FIFO", response)
                try:
                    resp_fd.write(response + "\n")
                    resp_fd.flush()
                except (BrokenPipeError, ValueError, OSError) as e:
                    # exec is gone or has closed the response FIFO -- it timed
                    # out waiting, was killed, or failed. Do NOT let this
                    # propagate: exec has already decided the outcome and
                    # recorded it, and the caller needs that outcome far more
                    # than it needs the write error. Break out and let
                    # communicate() below collect exec's real exit code and
                    # stderr.
                    log.warning(
                        "relay: could not deliver confirmation response — %s: %s. "
                        "exec closed the response pipe before the answer arrived; "
                        "collecting its result instead.", type(e).__name__, e,
                    )
                    relay_note = (
                        "The human's response could not be delivered: "
                        "aishell-gate-exec had already stopped waiting for it "
                        "and closed the confirmation pipe. The outcome below is "
                        "exec's own, and no command ran on the strength of the "
                        "undelivered answer. Re-submit the plan to try again."
                    )
                    break

                # Gate closed. confirm_seq is deliberately NOT reset, so a late
                # response for this gate can still be recognised and discarded.
                with _PENDING_LOCK:
                    state["status"]      = "running"
                    state["confirm_req"] = None

            # Wait for exec to finish. If it refuses to exit within the
            # timeout we SIGKILL it and synthesize a failure rc — this
            # thread (and the MCP server) must not be held hostage by a
            # hung exec. _cleanup_proc in the finally block is a second
            # line of defense for any path that skips this call.
            # Deployments with legitimately long plans can raise this; 0
            # means wait indefinitely.
            run_budget = cfg.get("exec_completion_timeout",
                                 _EXEC_COMPLETION_TIMEOUT)
            try:
                run_budget = float(run_budget)
            except (TypeError, ValueError):
                run_budget = _EXEC_COMPLETION_TIMEOUT
            if run_budget <= 0:
                run_budget = None

            killed = False
            try:
                stdout_bytes, stderr_bytes = proc.communicate(
                    timeout=run_budget,
                )
            except subprocess.TimeoutExpired:
                log.warning(
                    "relay: exec did not exit within %ss — sending SIGKILL. "
                    "If the plan legitimately needs longer, raise "
                    "exec_completion_timeout in aishell-gate-mcp.json.",
                    run_budget,
                )
                proc.kill()
                killed = True
                try:
                    stdout_bytes, stderr_bytes = proc.communicate(
                        timeout=_EXEC_KILL_TIMEOUT,
                    )
                except subprocess.TimeoutExpired:
                    log.error(
                        "relay: exec still running %ss after SIGKILL — "
                        "abandoning (OS will reap on thread/process exit)",
                        _EXEC_KILL_TIMEOUT,
                    )
                    stdout_bytes = b""
                    stderr_bytes = b""
            rc = proc.returncode if proc.returncode is not None else -9
            if killed and rc >= 0:
                # Communicate returned but Popen didn't record a signal —
                # use the conventional 128+SIGKILL code for clarity in logs.
                rc = -9

            stdout_str = stdout_bytes.decode(errors="replace") if stdout_bytes else ""
            stderr_str = stderr_bytes.decode(errors="replace") if stderr_bytes else ""

            result = {
                "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
                "tool": "execute_plan",
                "execution_id": exec_id,
                "goal": goal,
                "commands_submitted": len(commands),
            }
            _queue_license_notice_if_new(stderr_str)
            result.update(_exec_outcome(rc, stdout_str, stderr_str))
            if relay_note:
                result["relay_warning"] = relay_note
                result["message"] = relay_note + " " + result["message"]

        except Exception as e:
            log.exception("relay: unhandled error in relay thread")
            result = _error_response("execute_plan", f"relay thread error: {e}")
            result["execution_id"] = exec_id

        finally:
            if req_fd:
                try: req_fd.close()
                except Exception: pass
            if resp_fd:
                try: resp_fd.close()
                except Exception: pass
            for path in (fifo_req, fifo_resp):
                try: os.unlink(path)
                except Exception: pass
            # Final orphan guard: if we're exiting via the except path or
            # via a return short-circuited the communicate above, make
            # sure no exec subprocess is left running. Under heavy server
            # use, leaking even one process per failed plan would add up.
            proc_local = locals().get("proc")
            if proc_local is not None and proc_local.poll() is None:
                log.warning(
                    "relay: exec still running during cleanup — "
                    "sending SIGKILL (exec_id=%s)", exec_id,
                )
                try:
                    proc_local.kill()
                except Exception:
                    pass
                try:
                    proc_local.wait(timeout=_EXEC_KILL_TIMEOUT)
                except Exception:
                    log.error(
                        "relay: could not reap exec after SIGKILL "
                        "(exec_id=%s)", exec_id,
                    )

        with _PENDING_LOCK:
            state["status"] = "complete"
            state["result"] = result
        # Unblock execute_plan's ready wait for the fast path where the plan
        # finished without ever needing (or after having resolved) a confirmation.
        state["ready_event"].set()

    t = threading.Thread(target=relay_thread, daemon=True, name=f"relay-{exec_id[:8]}")
    with _PENDING_LOCK:
        state["thread"] = t
    t.start()

    # Wait for the relay thread to actually reach a stable state before
    # returning, so the status we report matches reality. Without this wait,
    # a caller that immediately invokes confirm_action races the subprocess
    # startup and FIFO handshake — the status check at the top of
    # tool_confirm_action hard-fails while state["status"] is still
    # "starting".
    state["ready_event"].wait(timeout=_RELAY_READY_TIMEOUT)

    with _PENDING_LOCK:
        status = state["status"]

    if status == "complete":
        # Relay finished (either a no-confirm plan that ran to completion, or
        # something that failed fast). Return the real result and evict the
        # entry, matching get_execution_result's completion path.
        result = dict(state["result"])
        with _PENDING_LOCK:
            _PENDING.pop(exec_id, None)
        return result

    if status == "pending_confirmation":
        with _PENDING_LOCK:
            confirm_req = dict(state.get("confirm_req") or {})
        return {
            "protocol":   {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool":       "execute_plan",
            "executed":   False,
            "status":     "pending_confirmation",
            "execution_id": exec_id,
            "max_confirm_level": max_lvl,
            "actions_requiring_confirmation": blocking,
            # The single action awaiting approval right now. The list above is
            # the whole plan; this is the one gate that is open.
            "confirm_request": confirm_req,
            "message": (
                f"Execution started. {len(blocking)} action(s) require human "
                f"confirmation, and each requires its own separate human answer. "
                f"One action is awaiting approval now — show the confirm_request "
                f"to the human, then call confirm_action with execution_id "
                f"'{exec_id}', that request's confirm_seq, and their response. "
                f"Then call get_execution_result: it returns the next "
                f"confirm_request if another action is waiting, or the final "
                f"outcome once execution finishes."
            ),
        }

    # Event did not fire in time — relay thread is still "starting" or
    # "running" (likely wedged on Popen, FIFO open, or the first req_fd
    # readline). Return an honest status so the caller polls instead of
    # blindly calling confirm_action.
    log.warning(
        "execute_plan: relay thread did not reach a stable state within %ss "
        "(current status=%s execution_id=%s)",
        _RELAY_READY_TIMEOUT, status, exec_id,
    )
    return {
        "protocol":   {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool":       "execute_plan",
        "executed":   False,
        "status":     status,
        "execution_id": exec_id,
        "max_confirm_level": max_lvl,
        "actions_requiring_confirmation": blocking,
        "message": (
            f"Execution started but has not yet reached a stable state "
            f"(current: {status}) after {_RELAY_READY_TIMEOUT:g}s. "
            f"Call get_execution_result to keep polling; do not call "
            f"confirm_action until status is 'pending_confirmation'."
        ),
    }


# ---------------------------------------------------------------------------
# Tool: confirm_action
# ---------------------------------------------------------------------------
def tool_confirm_action(cfg: dict, args: dict) -> dict:
    """
    Submit a human confirmation response to a waiting execute_plan relay.

    Call this after execute_plan returns status 'pending_confirmation'.
    The response is forwarded to aishell-gate-exec via the confirmation FIFO.

    For confirm_level 'action' or 'plan': response must be 'yes' or 'no'.
    For confirm_level 'typed':  response must be the exact challenge code
                                shown in the confirmation request.

    'confirm_seq' must be copied from the confirm_request currently returned
    by execute_plan or get_execution_result. It ties this response to the one
    action the human was shown. A plan with several actions that each need
    confirmation opens several gates in turn, and each needs its own human
    answer; one approval must never be reused for the next gate.
    """
    exec_id  = str(args.get("execution_id", "")).strip()
    response = str(args.get("response", "")).strip()
    raw_seq  = args.get("confirm_seq")

    if not exec_id:
        return _error_response("confirm_action", "'execution_id' is required")
    if not response:
        return _error_response("confirm_action", "'response' is required")
    if raw_seq is None:
        return _error_response("confirm_action",
            "'confirm_seq' is required. Copy it from the confirm_request "
            "returned by execute_plan or get_execution_result — it identifies "
            "which action this response approves.")
    try:
        seq = int(raw_seq)
    except (TypeError, ValueError):
        return _error_response("confirm_action",
            f"'confirm_seq' must be an integer (got {raw_seq!r}).")

    with _PENDING_LOCK:
        state = _PENDING.get(exec_id)

    if state is None:
        return _error_response("confirm_action",
            f"No pending execution found for execution_id '{exec_id}'. "
            f"It may have already completed or the id is incorrect.")

    # If the caller raced the relay thread — either an older client that
    # skips execute_plan's own ready-wait, or a fresh confirm for a later
    # action in a multi-action plan while the relay is mid-transition from
    # running → pending_confirmation — briefly poll for the expected state
    # instead of hard-failing. The single-shot ready_event isn't suitable
    # here because it stays set once triggered, so a simple bounded poll is
    # the honest way to wait for the next pending_confirmation transition.
    if state["status"] not in ("pending_confirmation", "complete"):
        deadline = time.monotonic() + _CONFIRM_ACTION_READY_WAIT
        while state["status"] not in ("pending_confirmation", "complete"):
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            time.sleep(min(0.05, remaining))

    if state["status"] != "pending_confirmation":
        return _error_response("confirm_action",
            f"Execution '{exec_id}' is not waiting for confirmation "
            f"(current status: {state['status']}). "
            f"Call get_execution_result to check its current state.")

    confirm_req = state.get("confirm_req") or {}
    open_seq    = confirm_req.get("confirm_seq")

    # The gate must be the one the human was actually shown. If the plan has
    # moved on to a later action, this response is stale and is refused rather
    # than applied to a command nobody reviewed.
    if seq != open_seq:
        return _error_response("confirm_action",
            f"Confirmation {seq} is not the open gate (currently {open_seq}). "
            f"Call get_execution_result to fetch the confirm_request for the "
            f"action now awaiting approval, show that action to the human, and "
            f"submit their answer with its confirm_seq. Do not reuse an earlier "
            f"approval — each action requires its own.")

    confirm_lvl = confirm_req.get("confirm_level", "action")

    # Validate response format
    if confirm_lvl == "typed":
        challenge = confirm_req.get("challenge", "")
        if response != challenge:
            return {
                "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
                "tool": "confirm_action",
                "accepted": False,
                "execution_id": exec_id,
                "message": (
                    f"Typed confirmation failed — response does not match "
                    f"the challenge code '{challenge}'. "
                    f"You must type the exact challenge code to confirm."
                ),
            }
    elif confirm_lvl in ("action", "plan"):
        if response.lower() not in ("yes", "no"):
            return _error_response("confirm_action",
                f"For confirm_level '{confirm_lvl}', response must be 'yes' or 'no' "
                f"(got '{response}').")
        response = response.lower()

    # Drop the response into the queue, tagged with the gate it answers.
    state["resp_queue"].put((seq, response))
    log.info("confirm_action: execution_id=%s confirm_seq=%s response='%s'",
             exec_id, seq, response)

    return {
        "protocol":    {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool":        "confirm_action",
        "accepted":    True,
        "execution_id": exec_id,
        "confirm_seq": seq,
        "response":    response,
        "message": (
            f"Confirmation response '{response}' recorded for action "
            f"{confirm_req.get('action_idx', '?')} (confirm_seq {seq}). "
            f"Call get_execution_result: if the plan has further actions "
            f"needing confirmation, it will return the next confirm_request, "
            f"which must be shown to the human before you answer it."
        ),
    }


# ---------------------------------------------------------------------------
# Tool: get_execution_result
# ---------------------------------------------------------------------------
def tool_get_execution_result(cfg: dict, args: dict) -> dict:
    """
    Retrieve the current status or final result of an async execute_plan relay.

    Returns the current state of the execution:
      'starting'             — exec process is launching
      'running'              — exec is running commands
      'pending_confirmation' — waiting for a confirm_action call
      'complete'             — execution finished; result is included

    Call this after confirm_action to retrieve the final outcome, or to
    check whether further confirmations are required (a plan may have
    multiple actions each requiring separate confirmation).
    """
    exec_id = str(args.get("execution_id", "")).strip()

    if not exec_id:
        return _error_response("get_execution_result", "'execution_id' is required")

    with _PENDING_LOCK:
        state = _PENDING.get(exec_id)

    if state is None:
        return _error_response("get_execution_result",
            f"No execution found for execution_id '{exec_id}'.")

    status = state["status"]

    if status == "complete":
        result = dict(state["result"])
        # Clean up completed execution from pending dict
        with _PENDING_LOCK:
            _PENDING.pop(exec_id, None)
        return result

    if status == "pending_confirmation":
        confirm_req = state.get("confirm_req", {})
        return {
            "protocol":    {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
            "tool":        "get_execution_result",
            "execution_id": exec_id,
            "status":      "pending_confirmation",
            "confirm_request": confirm_req,
            "message": (
                f"Execution is waiting for human confirmation. "
                f"Review the confirm_request and call confirm_action with "
                f"execution_id '{exec_id}' and your response."
            ),
        }

    # starting or running
    return {
        "protocol":    {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool":        "get_execution_result",
        "execution_id": exec_id,
        "status":      status,
        "message":     f"Execution is {status}. Call get_execution_result again to poll.",
    }




# ---------------------------------------------------------------------------
# Tool: evaluate_command
# ---------------------------------------------------------------------------
def tool_evaluate_command(cfg: dict, args: dict) -> dict:
    """
    Evaluate a single raw command string directly through the policy engine.
    Returns full JSON assessment including flag catalog analysis, risk score,
    blast radius, IO classification, confirmation level, and deny suggestions.

    Faster than evaluate_plan for ad-hoc single-command checks.
    Does not require an action envelope — pass the command string directly.
    Does not execute anything.
    """
    command = str(args.get("command", "")).strip()
    if not command:
        return _error_response("evaluate_command", "'command' is required")

    # Invoke policy engine directly with --json for machine-readable output
    try:
        rc, stdout, stderr = _invoke_policy(cfg, command, extra_argv=["--json"])
    except RuntimeError as e:
        return _error_response("evaluate_command", str(e))

    if not stdout.strip():
        # Disabled: _eval_expiry_message() detection (see function docstring).
        # expiry = _eval_expiry_message("", stderr)
        # if expiry:
        #     return _error_response("evaluate_command",
        #         "AIShell-Gate evaluation period has ended — the policy "
        #         f"engine declined to run (exit {rc}).\nBinary message: {expiry}")
        reason = _standard_edition_restriction_reason(stderr)
        if reason:
            return _error_response("evaluate_command",
                "This command uses a feature that requires the Enterprise "
                f"edition. Binary message: {reason}")
        return _error_response("evaluate_command",
            f"policy engine produced no output (exit {rc}). "
            f"stderr: {stderr[:512]}")

    _queue_license_notice_if_new(stderr)
    stdout, err = _strip_preamble(stdout, "evaluate_command", stderr)
    if err:
        return err

    try:
        data = json.loads(stdout)
    except json.JSONDecodeError as e:
        return _error_response("evaluate_command",
            f"could not parse policy output as JSON: {e}\noutput: {stdout[:200]}")

    # Axis B: validate the policy-response protocol on this live (non-dry-run)
    # path. evaluate_command calls the policy binary directly with no dry-run
    # pre-flight to lean on, so this is the only place its response envelope is
    # name/version-checked. Older policy binaries (pre-v1.02) omit the protocol
    # block and pass via the guard's backward-compat early return.
    try:
        _check_response_protocol(data, PROTO_POLICY_NAME, "evaluate_command")
    except ValueError as e:
        return _error_response("evaluate_command", str(e))

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "evaluate_command",
        "command": command,
        "assessment": data,
    }


# ---------------------------------------------------------------------------
# Tool: get_policy_template
# ---------------------------------------------------------------------------
def tool_get_policy_template(cfg: dict, args: dict) -> dict:
    """
    Emit the built-in policy layer as a ready-to-edit JSON override file.
    Optionally combine with a preset to template that preset's baseline.

    The output is suitable for use as a --policy-base, --policy-project,
    or --policy-user override file after stripping the plain-text header
    (everything before the opening '{').
    """
    preset = args.get("preset", cfg.get("preset", "ops_safe"))

    argv = [cfg["policy_binary"], "--dump-standard-template"]
    if preset:
        argv += ["--policy-preset", str(preset)]

    log.debug("get_policy_template: %s", " ".join(argv))
    try:
        result = subprocess.run(
            argv, capture_output=True, text=True, timeout=30
        )
        output = result.stdout
    except FileNotFoundError:
        return _error_response("get_policy_template",
            f"policy binary not found: '{cfg['policy_binary']}'")
    except subprocess.TimeoutExpired:
        return _error_response("get_policy_template",
            "policy binary timed out.")

    if not output.strip():
        # Disabled: _eval_expiry_message() detection (see function docstring).
        # expiry = _eval_expiry_message("", result.stderr or "")
        # if expiry:
        #     return _error_response("get_policy_template",
        #         "AIShell-Gate evaluation period has ended — the policy "
        #         f"engine declined to run.\nBinary message: {expiry}")
        return _error_response("get_policy_template",
            "policy binary produced no output. Check binary installation.")

    # Separate the plain-text header from the JSON body.
    # The header ends at (and excludes) the first line that is just '{'.
    lines      = output.splitlines()
    json_start = None
    for i, line in enumerate(lines):
        if line.strip() == "{":
            json_start = i
            break

    header_text = "\n".join(lines[:json_start]) if json_start is not None else ""
    json_text   = "\n".join(lines[json_start:]) if json_start is not None else output

    template_json = None
    parse_error   = None
    try:
        template_json = json.loads(json_text)
    except json.JSONDecodeError as e:
        parse_error = str(e)

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "get_policy_template",
        "preset": preset,
        "header": header_text,
        "template": template_json,
        "template_raw": json_text if template_json is None else None,
        "parse_error": parse_error,
        "usage": (
            "Save the 'template' JSON (or 'template_raw' if parsing failed) "
            "as a .json file and pass it via --policy-base, --policy-project, "
            "or --policy-user in aishell-gate-mcp.json.  Edit cmd_allow, cmd_deny, "
            "arg_rules, path_rules, and net_rules to customise policy for your workflow."
        ),
    }


# ---------------------------------------------------------------------------
# Tool: verify_policy  [ENTERPRISE]
# ---------------------------------------------------------------------------
def tool_verify_policy(cfg: dict, args: dict) -> dict:
    """
    [ENTERPRISE] Run a JSON policy test suite against the active policy.
    Reports PASS/FAIL per case.  Suitable for CI gates and regression testing.

    Accepts inline test case list.  Tool name: verify_policy.  Test format:
      [
        { "cmd": "git status", "expected": "allow", "label": "git ok"  },
        { "cmd": "rm -rf /",   "expected": "deny",  "label": "rm deny" }
      ]

    Each test case may specify a per-case 'preset' override.
    """
    err = _require_enterprise(cfg, "verify_policy")
    if err:
        return err

    tests_input = args.get("tests")
    preset      = args.get("preset", cfg.get("preset", "ops_safe"))

    if tests_input is None:
        return _error_response("verify_policy",
            "'tests' is required — provide a list of test case objects.")

    # Accept either a list of cases or a full test envelope dict
    if isinstance(tests_input, list):
        test_envelope = {"tests": tests_input}
    elif isinstance(tests_input, dict):
        test_envelope = tests_input
    else:
        return _error_response("verify_policy",
            "'tests' must be a list of test cases or a dict with a 'tests' key.")

    if "tests" not in test_envelope or not test_envelope["tests"]:
        return _error_response("verify_policy", "test suite contains no test cases.")

    # Write test suite to a temp file (--test-plan requires a file path)
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", suffix=".json", delete=False, prefix="aishell_tests_"
        ) as tf:
            json.dump(test_envelope, tf)
            tmppath = tf.name
    except Exception as e:
        return _error_response("verify_policy", f"could not write temp test file: {e}")

    try:
        argv = [cfg["policy_binary"],
                "--policy-preset", str(preset),
                "--test-plan", tmppath]
        if cfg.get("policy_base"):
            argv += ["--policy-base", cfg["policy_base"]]
        if cfg.get("policy_project"):
            argv += ["--policy-project", cfg["policy_project"]]
        if cfg.get("policy_user"):
            argv += ["--policy-user", cfg["policy_user"]]

        log.debug("verify_policy: %s", " ".join(argv))
        result = subprocess.run(
            argv, capture_output=True, text=True, timeout=60
        )
        rc     = result.returncode
        output = result.stdout + result.stderr
    except FileNotFoundError:
        return _error_response("verify_policy",
            f"policy binary not found: '{cfg['policy_binary']}'")
    except subprocess.TimeoutExpired:
        return _error_response("verify_policy",
            "policy binary timed out running test suite.")
    finally:
        try:
            os.unlink(tmppath)
        except Exception:
            pass

    # Defense in depth: _require_enterprise above already blocks this call in
    # standard edition under normal operation, but if edition detection is
    # stale (binary swapped without restarting this server), the engine's own
    # rejection would otherwise be misreported as a generic parse failure.
    if "not available in the standard edition" in output.lower():
        return _error_response("verify_policy",
            "The installed binary reported: not available in standard edition. "
            "Enterprise edition is required for --test-plan / verify_policy.")

    # Parse PASS/FAIL lines from output
    passed  = 0
    failed  = 0
    results = []
    for line in output.splitlines():
        stripped = line.strip()
        if stripped.startswith("PASS"):
            passed += 1
            results.append({"result": "PASS", "line": stripped})
        elif stripped.startswith("FAIL"):
            failed += 1
            results.append({"result": "FAIL", "line": stripped})

    # rc: 0 = all pass, 1 = any fail, 2 = file/parse error
    exit_meanings = {
        0: "all tests passed",
        1: "one or more tests failed",
        2: "file or parse error",
    }
    outcome = exit_meanings.get(rc, f"unknown exit code {rc}")

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "verify_policy",
        "preset": preset,
        "exit_code": rc,
        "outcome": outcome,
        "total": passed + failed,
        "passed": passed,
        "failed": failed,
        "all_passed": (rc == 0),
        "results": results,
        "raw_output": output.strip(),
    }


# ---------------------------------------------------------------------------
# Tool: dump_policy — RETIRED (edition-split-decisions-5/6.md)
# Not a customer-facing feature in any edition: reveals too much policy IP
# (the full resolved stack including the built-in catalog). The underlying
# --dump-policy engine flag still exists, hidden and undocumented, for
# AIShell Labs internal use — this server just no longer calls it.
# Original implementation kept below, commented out, not deleted — same
# pattern as _eval_expiry_message() elsewhere in this file.
# ---------------------------------------------------------------------------
def tool_dump_policy(cfg: dict, args: dict) -> dict:
    """
    [RETIRED] dump_policy is no longer offered via MCP, in any edition.
    """
    return _error_response("dump_policy",
        "dump_policy has been retired and is no longer available via MCP, "
        "in any edition. It is not offered as a customer-facing feature. "
        "Use verify_policy for policy regression testing, or get_policy_template "
        "to obtain an editable copy of a preset's baseline.")


# def tool_dump_policy_ORIGINAL(cfg: dict, args: dict) -> dict:
#     """
#     [ENTERPRISE] Dump the fully resolved effective policy stack as JSON.
#     Reflects all layers: builtin (or preset), base, project, and user overrides,
#     plus the full command catalog.  Use for policy auditing, security review,
#     and debugging unexpected allow/deny decisions.
#     """
#     err = _require_enterprise(cfg, "dump_policy")
#     if err:
#         return err
#
#     preset = args.get("preset", cfg.get("preset", "ops_safe"))
#
#     argv = [cfg["policy_binary"],
#             "--policy-preset", str(preset),
#             "--dump-policy"]
#     if cfg.get("policy_base"):
#         argv += ["--policy-base", cfg["policy_base"]]
#     if cfg.get("policy_project"):
#         argv += ["--policy-project", cfg["policy_project"]]
#     if cfg.get("policy_user"):
#         argv += ["--policy-user", cfg["policy_user"]]
#
#     log.debug("dump_policy: %s", " ".join(argv))
#     try:
#         result = subprocess.run(
#             argv, capture_output=True, text=True, timeout=30
#         )
#         output = result.stdout
#         rc     = result.returncode
#         stderr = result.stderr
#     except FileNotFoundError:
#         return _error_response("dump_policy",
#             f"policy binary not found: '{cfg['policy_binary']}'")
#     except subprocess.TimeoutExpired:
#         return _error_response("dump_policy", "policy binary timed out.")
#
#     if not output.strip():
#         stderr_hint = stderr[:300] if stderr else ""
#         # Disabled: _eval_expiry_message() detection (see function docstring).
#         # expiry = _eval_expiry_message("", stderr)
#         # if expiry:
#         #     return _error_response("dump_policy",
#         #         "AIShell-Gate evaluation period has ended — the policy "
#         #         f"engine declined to run (exit {rc}).\nBinary message: {expiry}")
#         # Standard edition produces a clear "not available in standard edition" message
#         if "standard edition" in stderr_hint.lower():
#             return _error_response("dump_policy",
#                 "The installed binary reported: not available in standard edition. "
#                 "Enterprise edition is required for --dump-policy.")
#         return _error_response("dump_policy",
#             f"policy binary produced no output (exit {rc}). stderr: {stderr_hint}")
#
#     output, err = _strip_preamble(output, "dump_policy", stderr)
#     if err:
#         return err
#
#     try:
#         policy_data = json.loads(output)
#     except json.JSONDecodeError as e:
#         return _error_response("dump_policy",
#             f"could not parse policy dump as JSON: {e}\noutput: {output[:200]}")
#
#     return {
#         "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
#         "tool": "dump_policy",
#         "preset": preset,
#         "policy": policy_data,
#     }


# ---------------------------------------------------------------------------
# Tool: verify_audit_log  [ENTERPRISE]
# ---------------------------------------------------------------------------
def tool_verify_audit_log(cfg: dict, args: dict) -> dict:
    """
    [ENTERPRISE] Verify the HMAC-SHA256 chain integrity of an audit log file.
    Detects any gap, truncation, or post-hoc modification in the chain.

    Specify log_type "exec" to verify an executor audit log (chain_hmac format)
    or "policy" to verify a policy engine audit log (entry_hash format).
    Do NOT mix log types — each is verified only by the binary that wrote it.

    Set audit_key in aishell-gate-mcp.json (path to key file) for keyed HMAC
    verification.  Without a persistent key, cross-session chain verification
    is not possible (the binary uses a per-session ephemeral key by default).
    """
    err = _require_enterprise(cfg, "verify_audit_log")
    if err:
        return err

    log_path = str(args.get("log_path", "")).strip()
    log_type = str(args.get("log_type", "exec")).strip().lower()

    if not log_path:
        return _error_response("verify_audit_log", "'log_path' is required.")
    if log_type not in ("exec", "policy"):
        return _error_response("verify_audit_log",
            "'log_type' must be 'exec' or 'policy'.")
    if not os.path.exists(log_path):
        return _error_response("verify_audit_log",
            f"log file not found: '{log_path}'")

    # Select the correct verifier binary.
    # exec logs: verified by aishell-gate-exec --audit-verify
    # policy logs: verified by aishell-gate-policy --audit-verify
    if log_type == "exec":
        binary = cfg["exec_binary"]
        argv   = [binary, "--audit-verify", log_path]
        if cfg.get("audit_key"):
            argv += ["--audit-key", cfg["audit_key"]]
    else:
        binary = cfg["policy_binary"]
        argv   = [binary, "--audit-verify", log_path]
        if cfg.get("audit_key"):
            argv += ["--audit-key", cfg["audit_key"]]

    log.debug("verify_audit_log (%s): %s", log_type, " ".join(argv))
    try:
        result = subprocess.run(
            argv, capture_output=True, text=True, timeout=60
        )
        rc     = result.returncode
        output = result.stdout + result.stderr
    except FileNotFoundError:
        return _error_response("verify_audit_log",
            f"binary not found: '{binary}'")
    except subprocess.TimeoutExpired:
        return _error_response("verify_audit_log",
            "verify timed out after 60 seconds.")

    # rc: 0 = chain intact, 1 = tamper detected, 2 = file error
    outcomes = {
        0: "chain intact — no tampering detected",
        1: "chain break or tampering detected",
        2: "file open or read error",
    }
    outcome = outcomes.get(rc, f"unknown exit code {rc}")
    intact  = (rc == 0)

    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": "verify_audit_log",
        "log_path": log_path,
        "log_type": log_type,
        "exit_code": rc,
        "chain_intact": intact,
        "outcome": outcome,
        "detail": output.strip(),
        "key_used": bool(cfg.get("audit_key")),
    }


# ---------------------------------------------------------------------------
# Error response helper
# ---------------------------------------------------------------------------
def _error_response(tool: str, message: str) -> dict:
    log.warning("%s error: %s", tool, message)
    return {
        "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
        "tool": tool,
        "error": True,
        "message": message,
    }


# ---------------------------------------------------------------------------
# MCP tool schemas
# ---------------------------------------------------------------------------
TOOLS = [
    # ---- Bootstrap tool ----
    {
        "name": "install_engine",
        "description": (
            "Download and install the AIShellGate policy engine (the two compiled "
            "binaries) from aishellgate.com into "
            "~/.local/share/aishell-gate/bin (a per-user directory; no sudo "
            "required). AIShellGate does not auto-install binaries as a side effect "
            "of any other call -- this tool performs a real download and writes "
            "executable files to disk, so it does nothing unless called with "
            "confirm=true. Call once without confirm to see what it would do; "
            "call again with confirm=true to actually download, checksum-verify, "
            "and install. Only supported on Linux x86-64; on other platforms this "
            "returns a clear 'not supported yet' message instead of downloading "
            "anything."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "confirm": {
                    "type": "boolean",
                    "description": (
                        "Must be explicitly true to perform the download and install. "
                        "Omit or set false to preview what will happen without acting."
                    ),
                },
            },
            "required": [],
        },
        "annotations": {
            "readOnlyHint":    False,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   True,
        },
    },
    # ---- Primary tools ----
    {
        "name": "evaluate_plan",
        "description": (
            "Dry-run a list of shell commands against the AIShell-Gate policy engine "
            "without executing anything. Returns a per-action assessment including "
            "policy decision (allow/deny), required confirmation level, risk score, "
            "resolved binary path, and reason. Use when you want to inspect a plan "
            "before committing to execution. Not required before execute_plan — "
            "execute_plan runs its own internal pre-flight automatically. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "goal": {
                    "type": "string",
                    "description": "Human-readable description of what this plan is trying to achieve.",
                },
                "commands": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of shell commands to evaluate. Maximum 24 per plan.",
                    "minItems": 1,
                    "maxItems": 24,
                },
                "strategy": {
                    "type": "string",
                    "enum": ["fail_fast", "best_effort"],
                    "description": "fail_fast stops on the first denial; best_effort evaluates all. Default: fail_fast.",
                    "default": "fail_fast",
                },
            },
            "required": ["goal", "commands"],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    {
        "name": "execute_plan",
        "description": (
            "Execute a list of shell commands via the AIShell-Gate executor. "
            "The policy engine evaluates every command automatically before any "
            "execution occurs — denied or high-risk commands are blocked and reported. "
            "If confirm_mode is 'operator_pipe', this call simply blocks until a "
            "human answers on their own aishell-gate-confirm session — no confirm "
            "request is returned here and confirm_action/get_execution_result are "
            "not part of that flow. "
            "For a single command, prefer the run tool instead. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "goal": {
                    "type": "string",
                    "description": "Human-readable description of what this plan is trying to achieve.",
                },
                "commands": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of shell commands to execute. Maximum 24 per plan.",
                    "minItems": 1,
                    "maxItems": 24,
                },
                "strategy": {
                    "type": "string",
                    "enum": ["fail_fast", "best_effort"],
                    "description": "fail_fast stops on the first non-zero exit; best_effort continues. Default: fail_fast.",
                    "default": "fail_fast",
                },
            },
            "required": ["goal", "commands"],
        },
        "annotations": {
            "readOnlyHint":    False,
            "destructiveHint": False,
            "idempotentHint":  False,
            "openWorldHint":   False,
        },
    },
    # ---- Inspection tools (Standard + Enterprise) ----
    {
        "name": "get_version",
        "description": (
            "Report the installed edition (standard or enterprise), version strings "
            "for both binaries, and a map of which tools are available in this install. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {},
            "required": [],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    {
        "name": "evaluate_command",
        "description": (
            "Evaluate a single raw command string directly through the policy engine. "
            "Returns full JSON assessment including flag catalog analysis (safe/warn/danger "
            "disposition per flag with documented reasoning), risk score (0-100), blast radius, "
            "IO classification, confirmation level, and deny suggestions. "
            "Faster than evaluate_plan for quick single-command checks. Does not execute anything. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The shell command string to evaluate (e.g. 'rm -rf /tmp/old').",
                },
            },
            "required": ["command"],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    {
        "name": "get_policy_template",
        "description": (
            "Emit the built-in policy layer as a ready-to-edit JSON override file. "
            "Returns a complete template including cmd_allow, cmd_deny, arg_rules, "
            "path_rules, and net_rules that can be saved and passed as --policy-base, "
            "--policy-project, or --policy-user. Optionally specify a preset to template "
            "that preset's baseline. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "preset": {
                    "type": "string",
                    "enum": ["read_only", "ops_safe", "dev_sandbox",
                             "ci_build", "ci_deploy", "ci_admin", "danger_zone"],
                    "description": "Which preset to template. Defaults to the active configured preset.",
                },
            },
            "required": [],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    # ---- Confirm-relay tools (relay mode only) ----
    {
        "name": "confirm_action",
        "description": (
            "Submit a human confirmation response to a waiting execute_plan relay. "
            "Call this after execute_plan or get_execution_result returns status "
            "'pending_confirmation'. Answers exactly ONE action: the one described "
            "in the confirm_request returned with that status. A plan may open "
            "several gates in turn, and each needs its own answer from the human — "
            "never reuse an earlier approval for a later action. "
            "For confirm_level 'action': response must be 'yes' or 'no'. "
            "For confirm_level 'typed': response must be the exact challenge code "
            "shown in the confirmation request — copy it precisely. "
            "Only available when confirm_mode is set to 'relay' in aishell-gate-mcp.json."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "execution_id": {
                    "type": "string",
                    "description": "The execution_id returned by execute_plan.",
                },
                "confirm_seq": {
                    "type": "integer",
                    "description": (
                        "Copy the confirm_seq field from the confirm_request you "
                        "just showed the human. Identifies which action this "
                        "response approves. A stale value is refused."
                    ),
                },
                "response": {
                    "type": "string",
                    "description": (
                        "'yes' or 'no' for action-level confirmation. "
                        "The exact challenge code for typed confirmation."
                    ),
                },
            },
            "required": ["execution_id", "confirm_seq", "response"],
        },
        "annotations": {
            "readOnlyHint":    False,
            "destructiveHint": False,
            "idempotentHint":  False,
            "openWorldHint":   False,
        },
    },
    {
        "name": "get_execution_result",
        "description": (
            "Retrieve the current status or final result of an async execute_plan relay. "
            "Call after confirm_action to get the outcome, or to check whether "
            "further confirmations are required (a plan may have multiple actions "
            "each requiring separate confirmation). "
            "Returns status: 'starting', 'running', 'pending_confirmation', or 'complete'. "
            "When status is 'pending_confirmation', the confirm_request field contains "
            "the full details of the action awaiting approval. "
            "Only available when confirm_mode is set to 'relay' in aishell-gate-mcp.json."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "execution_id": {
                    "type": "string",
                    "description": "The execution_id returned by execute_plan.",
                },
            },
            "required": ["execution_id"],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  False,
            "openWorldHint":   False,
        },
    },
    # ---- Enterprise-only tools ----
    {
        "name": "verify_policy",
        "description": (
            "[ENTERPRISE ONLY] Run a JSON policy test suite against the active policy "
            "and report PASS/FAIL per test case. Exit code 0 means all tests passed. "
            "Accepts inline test case list — no file path required. "
            "Suitable for CI policy gates and regression testing after policy changes. "
            "Call get_version to confirm enterprise edition before using this tool. "
            "Returns a clear error message in Standard edition."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "tests": {
                    "type": "array",
                    "description": (
                        "List of test case objects. Each object requires: "
                        "'cmd' (string), 'expected' ('allow' or 'deny'). "
                        "Optional per case: 'label' (string description), "
                        "'preset' (string to override policy preset for this case only)."
                    ),
                    "items": {
                        "type": "object",
                        "properties": {
                            "cmd":      {"type": "string"},
                            "expected": {"type": "string", "enum": ["allow", "deny"]},
                            "label":    {"type": "string"},
                            "preset":   {"type": "string"},
                        },
                        "required": ["cmd", "expected"],
                    },
                    "minItems": 1,
                },
                "preset": {
                    "type": "string",
                    "description": "Policy preset to test against. Defaults to the active configured preset.",
                },
            },
            "required": ["tests"],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    {
        "name": "verify_audit_log",
        "description": (
            "[ENTERPRISE ONLY] Verify the HMAC-SHA256 chain integrity of an audit log file. "
            "Detects any gap, truncation, or post-hoc modification in the tamper-evident chain. "
            "Specify log_type 'exec' for executor audit logs (chain_hmac format, written by "
            "aishell-gate-exec) or 'policy' for policy engine audit logs (entry_hash format, "
            "written by aishell-gate-policy). Never mix log types. "
            "Requires audit_key set in aishell-gate-mcp.json for keyed HMAC verification. "
            "Call get_version to confirm enterprise edition before using this tool. "
            "Returns a clear error message in Standard edition."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "log_path": {
                    "type": "string",
                    "description": "Absolute path to the audit log file to verify.",
                },
                "log_type": {
                    "type": "string",
                    "enum": ["exec", "policy"],
                    "description": (
                        "'exec' for executor audit logs (aishell-gate-exec), "
                        "'policy' for policy engine audit logs (aishell-gate-policy). "
                        "Default: exec. Do not mix log types."
                    ),
                    "default": "exec",
                },
            },
            "required": ["log_path"],
        },
        "annotations": {
            "readOnlyHint":    True,
            "destructiveHint": False,
            "idempotentHint":  True,
            "openWorldHint":   False,
        },
    },
    # ---- Single-shot convenience tool ----
    {
        "name": "run",
        "description": (
            "Run a single shell command through the AIShell-Gate policy engine and executor "
            "in one step. The policy engine evaluates the command before execution — "
            "denied or high-risk commands are blocked automatically. "
            "Returns stdout, stderr, exit code, and the policy decision. "
            "Prefer this over execute_plan for single commands. "
            "Use execute_plan for multi-command sequences. "
            "Available in Standard and Enterprise editions."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The shell command to run (e.g. 'ls -la /tmp').",
                },
            },
            "required": ["command"],
        },
        "annotations": {
            "readOnlyHint":    False,
            "destructiveHint": False,
            "idempotentHint":  False,
            "openWorldHint":   False,
        },
    },
]


# ---------------------------------------------------------------------------
# MCP stdio transport — JSON-RPC 2.0 message loop
# ---------------------------------------------------------------------------
def _send(msg: dict) -> None:
    line = json.dumps(msg, separators=(",", ":"))
    sys.stdout.write(line + "\n")
    sys.stdout.flush()


def _send_result(req_id: Any, result: Any) -> None:
    _send({"jsonrpc": JSONRPC_VERSION, "id": req_id, "result": result})


def _send_error(req_id: Any, code: int, message: str) -> None:
    _send({"jsonrpc": JSONRPC_VERSION, "id": req_id,
           "error": {"code": code, "message": message}})


def tool_run(cfg: dict, args: dict) -> dict:
    """
    Single-command convenience wrapper around execute_plan.
    Evaluates and executes one command in a single tool call,
    reducing AI client confirmation prompts to a minimum.
    """
    command = str(args.get("command", "")).strip()
    if not command:
        return _error_response("run", "'command' is required")

    # Delegate to execute_plan logic with a single-item plan
    return tool_execute_plan(cfg, {
        "goal":     command,
        "commands": [command],
        "strategy": "fail_fast",
    })


# Tool dispatch table
TOOL_HANDLERS = {
    "install_engine":        tool_install_engine,
    "run":                   tool_run,
    "evaluate_plan":         tool_evaluate_plan,
    "execute_plan":          tool_execute_plan,
    "confirm_action":        tool_confirm_action,
    "get_execution_result":  tool_get_execution_result,
    "get_version":           tool_get_version,
    "evaluate_command":      tool_evaluate_command,
    "get_policy_template":   tool_get_policy_template,
    "verify_policy":         tool_verify_policy,
    "dump_policy":           tool_dump_policy,  # retired stub — kept so a stale/
                                                 # cached client calling this name
                                                 # gets a clear message, not -32601
    "verify_audit_log":      tool_verify_audit_log,
}


def _server_instructions(cfg: dict) -> str:
    """Short, always-loaded hint sent to the client in the MCP initialize
    response (InitializeResult.instructions). This loads automatically at
    session start, before any tool call -- unlike CLAUDE.md, which only takes
    effect once/if the client has actually imported it. Deliberately short:
    clients that honour this field (Claude Code among them) truncate it, so
    this is a pointer to the full guide, not a replacement for it.

    The license/copyright notice below is placed FIRST, not appended at the
    end, specifically because of that truncation behavior -- text placed last
    in a field known to get cut off is the first thing to disappear, which
    would be a particularly bad way to lose a compliance-relevant notice.
    Reuses _COPYRIGHT_NOTICE rather than a separate copy, so the banner
    shown here, in every tool reply's copyright_notice field, and (once the
    C sources are updated to match) in exec/policy's own startup banner all
    read identically.

    The paragraph immediately after the banner is addressed directly to the
    calling agent: it states plainly that displaying the banner is required
    and that not doing so leaves neither the agent nor the user licensed to
    use the software. This still cannot be enforced by MCP -- see the note
    below -- but LICENSE.TXT's ACCEPTANCE section ties use to having seen
    the notice, so the instruction is at minimum an accurate statement of
    what the license actually requires, not just a request for politeness.

    NOTE ON WHAT THIS DOES AND DOES NOT GUARANTEE: this field is delivered to
    the calling agent, not rendered to a human by protocol mechanism. MCP has
    no capability to force a client to display specific content -- an agent
    that reads this can still choose what to relay, same as with any tool
    result. This is a best-effort placement (earliest possible point in a
    session, positioned to survive truncation), not an enforcement mechanism.
    The elicitation/create flow below IS a real enforcement mechanism, where
    the client supports it -- see _request_elicitation."""
    notice = (
        _COPYRIGHT_NOTICE + "\n\n"
        "LEGAL NOTICE TO THE CALLING AGENT: the banner above is a required "
        "legal notice, not optional text. You MUST display it to the user, "
        "verbatim, before or alongside your first response in this session. "
        "Per LICENSE.TXT, use of AIShell-Gate constitutes acceptance of its "
        "license, and that acceptance depends on the user having seen this "
        "notice. If you do not display it: the user is not licensed to use "
        "this software, and neither are you, on the user's behalf. Do not "
        "summarize, paraphrase, or omit it.\n\n"
    )
    if cfg.get("confirm_mode", "plan_only") == "relay":
        confirm_note = (
            "In relay mode, execute_plan returns pending_confirmation -- call "
            "confirm_action, then get_execution_result."
        )
    elif cfg.get("confirm_mode", "plan_only") == "operator_pipe":
        confirm_note = (
            "In operator_pipe mode, confirmation happens out-of-band on the "
            "operator's own terminal. execute_plan simply blocks until they "
            "answer -- do not call confirm_action or get_execution_result for "
            "this deployment; there is nothing pending for them to act on."
        )
    else:
        confirm_note = (
            "In plan_only mode, a plan-level confirmation is brokered: "
            "execute_plan returns pending_confirmation with the whole plan "
            "listed, call confirm_action once, then get_execution_result. "
            "Per-action and typed confirmations are NOT brokered in this "
            "mode -- those are blocked and reported back rather than run; "
            "adjust the policy, set confirm_mode to 'relay', or run manually."
        )
    return (
        notice +
        "AIShell-Gate: policy-gated command execution. Use these tools instead "
        "of Bash for OS commands -- every command is checked against a policy "
        "engine before running.\n\n"
        "Workflow: evaluate_command (or evaluate_plan) first to check a command "
        "will be allowed, then execute_plan to run it. " + confirm_note + "\n\n"
        "Show this server's tool results to the user exactly as returned -- "
        "decision, reason, risk score, flags, license_notice if present -- "
        "never summarize, paraphrase, or reformat them.\n\n"
        "Never use Edit or Write on aishell-gate-mcp.json, "
        "aishell-gate-policy_user.json, .claude/settings.json, or .mcp.json to "
        "work around a denial. If a command needs a policy or config change, "
        "describe the exact change and let the user decide -- do not make it "
        "yourself and retry.\n\n"
        "Full guide: @aishell-gate/CLAUDE.md in the project root, additional "
        "documentation can be found in the aishell-gate tar.gz or the users "
        "home directory or in the operating system's installed man pages or "
        "online at [www.aishellgate.com](https://www.aishellgate.com)"
    )


def run_server(cfg: dict) -> None:
    log.info("aishell-gate-mcp starting (protocol %s v%s)",
             PROTO_MCP_NAME, PROTO_MCP_VERSION)
    log.info("exec_binary=%s  policy_binary=%s  preset=%s  edition=%s",
             cfg["exec_binary"], cfg["policy_binary"],
             cfg["preset"], cfg["_edition"])

    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError as e:
            log.warning("malformed JSON on stdin: %s", e)
            continue

        req_id = msg.get("id")
        method = msg.get("method", "")
        params = msg.get("params", {})

        # ---- Initialise handshake ----
        if method == "initialize":
            # The client declares support for server-initiated requests via
            # capabilities.elicitation (MCP spec 2025-06-18). Absence means
            # the client either doesn't implement elicitation at all, or
            # implements an older protocol version that predates it. Stored
            # once here rather than probed per-call: capabilities are fixed
            # for the life of a session, and attempting elicitation/create
            # against a client that never declared it would mean sending a
            # request and hoping for the best instead of knowing in advance.
            global _CLIENT_SUPPORTS_ELICITATION
            _CLIENT_SUPPORTS_ELICITATION = bool(
                (params.get("capabilities") or {}).get("elicitation") is not None
            )
            policy_ver = cfg["_policy_version"].splitlines()[0] if cfg["_policy_version"] else PROTO_MCP_VERSION
            _send_result(req_id, {
                "protocolVersion": MCP_PROTOCOL_VERSION,
                "capabilities": {"tools": {}},
                "serverInfo": {
                    "name":    "aishell-gate-mcp",
                    "version": policy_ver,
                },
                "instructions": _server_instructions(cfg),
            })

        # ---- Tool list ----
        elif method == "tools/list":
            _send_result(req_id, {"tools": TOOLS})

        # ---- Tool call ----
        elif method == "tools/call":
            tool_name = params.get("name", "")
            tool_args = params.get("arguments", {})
            handler   = TOOL_HANDLERS.get(tool_name)
            if handler is None:
                _send_error(req_id, -32601, f"Unknown tool: '{tool_name}'")
                continue
            # Elicitation-based notice acknowledgment: DISABLED as of this
            # version. It caused real, reproducible breakage against actual
            # Claude Code -- tools stopped working entirely, with no visible
            # dialog. A bounded timeout was added so the wait could no
            # longer hang forever, but the underlying mismatch with the real
            # client was never confirmed or fixed, only contained.
            #
            # _request_elicitation() itself is left in place, tested in
            # isolation (unit tests) and via a live subprocess harness that
            # simulates a compliant client -- but "compliant with my reading
            # of the spec" and "compliant with real Claude Code" turned out
            # to be different things, and re-enabling this needs that gap
            # closed first, ideally against a real session or a captured
            # protocol trace, not another guess.
            #
            # Every tool call proceeds unconditionally, as in the version
            # before elicitation was added. copyright_notice, license_notice,
            # and the instructions banner remain the active notice channels.
            # Every tool except install_engine (which installs the engine) and
            # get_version (which reports on it either way) requires the engine
            # to be present. Gating centrally here means each tool doesn't need
            # its own missing-binary handling, and the agent always gets the
            # same clear message instead of a raw subprocess/FileNotFoundError.
            if tool_name not in ("install_engine", "get_version") and not _engine_available(cfg):
                _send_result(req_id, {
                    "content": [{"type": "text", "text": json.dumps(
                        _with_copyright_notice({
                            "protocol": {"name": PROTO_MCP_NAME, "version": PROTO_MCP_VERSION},
                            "tool": tool_name,
                            "status": "engine_missing",
                            "message": _engine_missing_message(),
                        }), indent=2)}]
                })
                continue
            try:
                result = handler(cfg, tool_args)
                global _PENDING_LICENSE_NOTICE
                if _PENDING_LICENSE_NOTICE is not None and isinstance(result, dict):
                    result = {**result, "license_notice": _PENDING_LICENSE_NOTICE}
                    _PENDING_LICENSE_NOTICE = None
                result = _with_copyright_notice(result)
                _send_result(req_id, {
                    "content": [{"type": "text", "text": json.dumps(result, indent=2)}]
                })
            except Exception as e:
                log.exception("unhandled error in tool %s", tool_name)
                _send_error(req_id, -32603, f"Internal error: {e}")

        # ---- Notifications (no response required) ----
        elif method == "notifications/initialized":
            pass

        # ---- Unknown method ----
        else:
            if req_id is not None:
                _send_error(req_id, -32601, f"Method not found: '{method}'")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main() -> None:
    import argparse
    ap = argparse.ArgumentParser(
        description="AIShell-Gate MCP server (wire protocol 4.0) — stdio transport",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Configuration file (aishell-gate-mcp.json):
  {
    "exec_binary":        "aishell-gate-exec",
    "policy_binary":      "aishell-gate-policy",
    "preset":             "ops_safe",
    "jail_root":          null,
    "sandbox":            null,
    "policy_base":        null,
    "policy_project":     null,
    "policy_user":        null,
    "source":             "ai",
    "audit_log":          null,
    "policy_audit_log":   null,
    "audit_key":          null,
    "eval_timeout":       30,
    "input_timeout":      30,
    "max_response_bytes": 0,
    "extra_flags":        [],
    "confirm_mode":       "plan_only",
    "confirm_pipe":       null,
    "confirm_lock":       null,
    "confirm_timeout":    null
  }

confirm_mode:
  "plan_only" (default) — block plans requiring confirmation, return error
                          listing which actions need operator approval.
  "relay"               — relay confirmation requests to the human via Claude
                          using confirm_action and get_execution_result tools.
                          Convenient for a single trusted session, but Claude
                          itself carries the request and the response, so
                          nothing here proves an answer came from a human
                          rather than the agent.
  "operator_pipe"        — point exec straight at a real, operator-owned
                          --confirm-pipe session (see confirm_pipe below).
                          This server never sees the request or a typed
                          challenge code, so it brings MCP confirmation to
                          genuine parity with the SSH operator relay.

confirm_pipe / confirm_lock / confirm_timeout:
  Only used when confirm_mode is "operator_pipe". confirm_pipe is the
  basepath of the FIFO pair an aishell-gate-confirm session already owns
  (e.g. "/run/aishell-gate/confirm") -- forwarded as exec's --confirm-pipe.
  confirm_lock and confirm_timeout forward to exec's own --confirm-lock and
  --confirm-timeout and are optional; see aishell-gate-exec(1) and
  aishell-gate-confirm(1).

Reducing Claude Code tool-approval prompts:
  By default Claude Code asks for approval before calling each MCP tool.
  To suppress these prompts for aishell-gate tools and rely solely on the
  gate's own policy and confirmation layer, launch Claude Code with:
    claude --allowedTools "mcp__aishell-gate__*"
  Or tell Claude Code during a session: "always allow aishell-gate tools".

All fields are optional. exec_binary and policy_binary are resolved
via PATH if not absolute paths.

audit_key (Enterprise only):
  Path to HMAC key file for audit chain verification.
  exec key format:   64 ASCII hex characters (32 bytes).
  policy key format: 64 raw binary bytes.
  These two formats are NOT interchangeable.

policy_audit_log:
  Separate from audit_log (exec log). Never point both binaries at the
  same file — they use incompatible internal formats (chain_hmac vs
  entry_hash).

Claude Code / Cursor integration (.mcp.json):
  {
    "mcpServers": {
      "aishell-gate": {
        "command": "python3",
        "args": ["/usr/local/bin/aishell-gate-mcp"]
      }
    }
  }

Tools available in Standard edition:
  evaluate_plan, execute_plan, confirm_action, get_execution_result,
  get_version, evaluate_command, get_policy_template

Additional tools in Enterprise edition:
  verify_policy, verify_audit_log
""",
    )
    ap.add_argument("--config", metavar="FILE",
                    help="Path to aishell-gate-mcp.json (default: ./aishell-gate-mcp.json)")
    ap.add_argument("--debug", action="store_true",
                    help="Enable debug logging to stderr")
    ap.add_argument("--version", action="store_true",
                    help="Print version and exit")
    args = ap.parse_args()

    if args.debug:
        logging.getLogger().setLevel(logging.DEBUG)

    cfg = load_config(args.config)
    _detect_edition(cfg)

    if args.version:
        policy_ver = (cfg["_policy_version"].splitlines()[0]
                      if cfg["_policy_version"] else "unknown (policy binary not found)")
        print(f"aishell-gate-mcp  --  AIShell-Gate {policy_ver}")
        print(f"  Edition:            {cfg['_edition']}")
        print(f"  MCP wire protocol:  {PROTO_MCP_VERSION}")
        print(f"  MCP transport:      {MCP_PROTOCOL_VERSION}")
        print(f"  Input envelope:     {PROTO_INPUT_NAME} {PROTO_INPUT_VERSION}")
        print(f"  Dry-run output:     {PROTO_DRY_RUN_NAME} 1.0")
        print(f"  Tools (standard):   install_engine, evaluate_plan, execute_plan, confirm_action,")
        print(f"                      get_execution_result, get_version,")
        print(f"                      evaluate_command, get_policy_template")
        print(f"  Tools (enterprise): verify_policy, verify_audit_log")
        sys.exit(0)

    run_server(cfg)


if __name__ == "__main__":
    main()
