obsidian-wiki/scripts/manage_mcp_service.py

#!/usr/bin/env python3
"""Manage the user-level HTTP MCP service for obsidian-wiki source files."""

from __future__ import annotations

import argparse
import json
import os
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import time
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Sequence
from urllib.parse import urlsplit, urlunsplit

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))

from obsidian_wiki.config import load_config, resolve_vault_path


SERVICE_NAME = "obsidian-wiki-mcp.service"
ENV_RELATIVE_PATH = Path(".runtime") / "obsidian-wiki-mcp.env"
MCP_CLIENT_NAME = "wiki-file-assets"
DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 8765
DEFAULT_PATH = "/mcp"
DEFAULT_INGEST_UPLOAD_PATH = "/mcp-ingest/uploads"
PUBLIC_BASE_URL_ENV = "OBSIDIAN_MCP_PUBLIC_BASE_URL"
PUBLIC_UPLOAD_URL_ENV = "OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL"
ALLOW_SOURCE_FILES_ENV = "OBSIDIAN_MCP_ALLOW_SOURCE_FILES"
ALLOWED_HOSTS_ENV = "OBSIDIAN_MCP_ALLOWED_HOSTS"
ALLOWED_ORIGINS_ENV = "OBSIDIAN_MCP_ALLOWED_ORIGINS"
ARCHIVE_SOURCES_ENV = "WIKI_ARCHIVE_SOURCES"
DEFAULT_TOKEN_BYTES = 32
DEFAULT_UV_CACHE_DIR = "/tmp/uv-cache"
DEFAULT_OPENCODE_PATH = "opencode"
DEFAULT_INGEST_OPENCODE_MODEL = "openai/gpt-5.5"
DEFAULT_INGEST_OPENCODE_VARIANT = "high"
DEFAULT_QUERY_OPENCODE_MODEL = "openai/gpt-5.5"
DEFAULT_QUERY_OPENCODE_VARIANT = "minimal"
DEFAULT_INGEST_JOB_TIMEOUT_SECONDS = 30 * 60
DEFAULT_QUERY_JOB_TIMEOUT_SECONDS = 5 * 60
DEFAULT_INGEST_MAX_REQUEST_BYTES = 0
DEFAULT_SERVICE_MEMORY_MAX = "2G"
DEFAULT_RUNTIME_CLEANUP_DAYS = 30
TOOL_FILE_ASSETS = "file-assets"
TOOL_INGEST = "ingest"
TOOL_QUERY = "query"
AVAILABLE_TOOL_GROUPS = (TOOL_FILE_ASSETS, TOOL_INGEST, TOOL_QUERY)
DEFAULT_ENABLED_TOOLS = (TOOL_FILE_ASSETS,)
TERMINAL_JOB_STATES = {"succeeded", "failed", "cancelled"}
PROXY_ENV_NAMES = ("http_proxy", "https_proxy", "all_proxy", "no_proxy", "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY")


def normalize_tool_name(name: str) -> str:
    normalized = name.strip().lower().replace("_", "-")
    aliases = {
        "file": TOOL_FILE_ASSETS,
        "files": TOOL_FILE_ASSETS,
        "file-assets": TOOL_FILE_ASSETS,
        "source-file": TOOL_FILE_ASSETS,
        "source-files": TOOL_FILE_ASSETS,
        "ingest": TOOL_INGEST,
        "upload": TOOL_INGEST,
        "uploads": TOOL_INGEST,
        "opencode-ingest": TOOL_INGEST,
        "query": TOOL_QUERY,
        "wiki-query": TOOL_QUERY,
        "ask": TOOL_QUERY,
        "question": TOOL_QUERY,
        "questions": TOOL_QUERY,
    }
    if normalized not in aliases:
        available = ", ".join(AVAILABLE_TOOL_GROUPS)
        raise ValueError(f"unknown MCP tool group: {name}. Available: {available}")
    return aliases[normalized]


def parse_bool(value: str | None) -> bool:
    return str(value or "").strip().lower() in {"1", "true", "yes", "on", "enabled"}


def project_archive_sources(project_root: Path, *, default: bool = False) -> bool:
    value = load_config(project_root).get(ARCHIVE_SOURCES_ENV)
    return default if value in (None, "") else parse_bool(value)


def parse_config_list(value: str | None) -> tuple[str, ...]:
    return tuple(item.strip() for item in str(value or "").split(",") if item.strip())


def normalize_enabled_tools(tools: Sequence[str]) -> tuple[str, ...]:
    enabled = {normalize_tool_name(tool) for tool in tools if str(tool).strip()}
    return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in enabled)


def enabled_tools_from_config(values: dict[str, str]) -> tuple[str, ...]:
    raw_tools = values.get("OBSIDIAN_MCP_ENABLED_TOOLS", "").strip()
    if raw_tools:
        if raw_tools.lower() in {"none", "off", "disabled"}:
            return ()
        return normalize_enabled_tools(raw_tools.split(","))

    enabled = set(DEFAULT_ENABLED_TOOLS)
    if parse_bool(values.get("OBSIDIAN_MCP_ENABLE_INGEST")):
        enabled.add(TOOL_INGEST)
    return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in enabled)


def format_enabled_tools(enabled_tools: Sequence[str]) -> str:
    return ",".join(normalize_enabled_tools(enabled_tools))


def update_ingest_tool(enabled_tools: Sequence[str], enable_ingest: bool | None) -> tuple[str, ...]:
    return update_tool_group(enabled_tools, TOOL_INGEST, enable_ingest)


def update_query_tool(enabled_tools: Sequence[str], enable_query: bool | None) -> tuple[str, ...]:
    return update_tool_group(enabled_tools, TOOL_QUERY, enable_query)


def update_tool_group(enabled_tools: Sequence[str], tool: str, enabled: bool | None) -> tuple[str, ...]:
    current = set(normalize_enabled_tools(enabled_tools))
    tool_name = normalize_tool_name(tool)
    if enabled is True:
        current.add(tool_name)
    elif enabled is False:
        current.discard(tool_name)
    return tuple(tool for tool in AVAILABLE_TOOL_GROUPS if tool in current)


@dataclass(frozen=True)
class ServicePaths:
    project_root: Path
    unit_path: Path
    env_path: Path


@dataclass(frozen=True)
class OpenCodeModelSettings:
    ingest_model: str = DEFAULT_INGEST_OPENCODE_MODEL
    ingest_variant: str = DEFAULT_INGEST_OPENCODE_VARIANT
    query_model: str = DEFAULT_QUERY_OPENCODE_MODEL
    query_variant: str = DEFAULT_QUERY_OPENCODE_VARIANT


@dataclass(frozen=True)
class ServiceConfig:
    token: str
    host: str = DEFAULT_HOST
    port: int = DEFAULT_PORT
    path: str = DEFAULT_PATH
    public_base_url: str = ""
    public_upload_url: str = ""
    enabled_tools: tuple[str, ...] = DEFAULT_ENABLED_TOOLS
    allow_source_files: bool = False
    archive_sources: bool = False
    allowed_hosts: tuple[str, ...] = ()
    allowed_origins: tuple[str, ...] = ()
    ingest_job_timeout_seconds: int = DEFAULT_INGEST_JOB_TIMEOUT_SECONDS
    query_job_timeout_seconds: int = DEFAULT_QUERY_JOB_TIMEOUT_SECONDS
    ingest_max_request_bytes: int = DEFAULT_INGEST_MAX_REQUEST_BYTES
    opencode_path: str = ""
    model_settings: OpenCodeModelSettings = OpenCodeModelSettings()


def generate_token(num_bytes: int = DEFAULT_TOKEN_BYTES) -> str:
    return secrets.token_urlsafe(num_bytes)


def systemd_quote(value: str | Path) -> str:
    return shlex.quote(str(value))


def find_uv_path() -> Path:
    configured = os.environ.get("UV_PATH", "").strip()
    discovered = configured or shutil.which("uv")
    if not discovered:
        raise RuntimeError("Could not find uv on PATH. Install uv or set UV_PATH to its absolute path.")
    return Path(discovered).expanduser().resolve()


def resolve_opencode_path(configured: str | Path | None = None) -> str:
    value = str(configured or "").strip()
    if value:
        configured_path = Path(value).expanduser()
        if configured_path.is_absolute() or configured_path.parent != Path("."):
            return str(configured_path.resolve())
        discovered = shutil.which(value)
        return str(Path(discovered).resolve()) if discovered else value
    discovered = shutil.which(DEFAULT_OPENCODE_PATH)
    return str(Path(discovered).resolve()) if discovered else DEFAULT_OPENCODE_PATH


def resolve_service_paths(project_root: Path, config_home: Path | None = None) -> ServicePaths:
    root = project_root.expanduser().resolve()
    base_config = config_home or Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
    unit_path = base_config.expanduser() / "systemd" / "user" / SERVICE_NAME
    return ServicePaths(project_root=root, unit_path=unit_path, env_path=root / ENV_RELATIVE_PATH)


def resolve_vault_for_service(project_root: Path, vault_path: Path | str | None = None) -> Path:
    if vault_path is not None:
        return Path(vault_path).expanduser().resolve()
    configured = resolve_vault_path(None, start=project_root)
    if configured:
        return Path(configured).expanduser().resolve()
    return project_root.expanduser().resolve()


def render_service_unit(
    *,
    project_root: Path,
    vault_path: Path,
    env_file: Path,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    path: str = DEFAULT_PATH,
    uv_cache_dir: str = DEFAULT_UV_CACHE_DIR,
    uv_path: Path | None = None,
    memory_max: str = DEFAULT_SERVICE_MEMORY_MAX,
) -> str:
    resolved_uv_path = uv_path or find_uv_path()
    command = " ".join(
        [
            systemd_quote(resolved_uv_path),
            "run",
            "--no-sync",
            "python",
            "-m",
            "obsidian_wiki.mcp_server",
            "--vault",
            systemd_quote(vault_path),
            "--transport",
            "streamable-http",
        ]
    )
    return (
        "[Unit]\n"
        "Description=obsidian-wiki HTTP MCP server\n"
        "After=network-online.target\n"
        "Wants=network-online.target\n"
        "\n"
        "[Service]\n"
        "Type=simple\n"
        f"WorkingDirectory={systemd_quote(project_root)}\n"
        f"EnvironmentFile={systemd_quote(env_file)}\n"
        f"Environment=UV_CACHE_DIR={systemd_quote(uv_cache_dir)}\n"
        f"PassEnvironment={' '.join(PROXY_ENV_NAMES)}\n"
        f"ExecStart={command}\n"
        "Restart=on-failure\n"
        "RestartSec=5\n"
        f"MemoryMax={memory_max}\n"
        "\n"
        "[Install]\n"
        "WantedBy=default.target\n"
    )


def prepare_runtime(
    project_root: Path,
    *,
    uv_path: Path | None = None,
    uv_cache_dir: str = DEFAULT_UV_CACHE_DIR,
) -> None:
    resolved_uv_path = uv_path or find_uv_path()
    env = {**os.environ, "UV_CACHE_DIR": uv_cache_dir}
    subprocess.run(
        [
            str(resolved_uv_path),
            "run",
            "--extra",
            "mcp",
            "python",
            "-c",
            "import mcp.server.fastmcp",
        ],
        cwd=project_root,
        env=env,
        text=True,
        check=True,
    )


def write_token_env_file(
    env_path: Path,
    token: str,
    *,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    path: str = DEFAULT_PATH,
    public_base_url: str | None = None,
    public_upload_url: str | None = None,
    enabled_tools: Sequence[str] | None = None,
    enable_ingest: bool | None = None,
    enable_query: bool | None = None,
    allow_source_files: bool | None = None,
    archive_sources: bool = False,
    allowed_hosts: Sequence[str] | None = None,
    allowed_origins: Sequence[str] | None = None,
    ingest_job_timeout_seconds: int = DEFAULT_INGEST_JOB_TIMEOUT_SECONDS,
    query_job_timeout_seconds: int = DEFAULT_QUERY_JOB_TIMEOUT_SECONDS,
    ingest_max_request_bytes: int = DEFAULT_INGEST_MAX_REQUEST_BYTES,
    opencode_path: str | Path | None = None,
    model_settings: OpenCodeModelSettings | None = None,
) -> None:
    tools = normalize_enabled_tools(DEFAULT_ENABLED_TOOLS if enabled_tools is None else enabled_tools)
    tools = update_ingest_tool(tools, enable_ingest)
    tools = update_query_tool(tools, enable_query)
    source_files_allowed = bool(allow_source_files)
    archive_sources_enabled = bool(archive_sources)
    models = model_settings or OpenCodeModelSettings()
    normalized_public_base_url = normalize_public_base_url(public_base_url)
    normalized_public_upload_url = normalize_public_upload_url(public_upload_url)
    effective_allowed_hosts = tuple(
        derive_allowed_hosts(
            [host],
            public_base_url=normalized_public_base_url,
            public_upload_url=normalized_public_upload_url,
        )
        if allowed_hosts is None
        else allowed_hosts
    )
    effective_allowed_origins = tuple(
        derive_allowed_origins(normalized_public_base_url, normalized_public_upload_url)
        if allowed_origins is None
        else allowed_origins
    )
    env_path.parent.mkdir(parents=True, exist_ok=True)
    env_path.parent.chmod(0o700)
    fd = os.open(env_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as env_file:
        env_file.write(f"OBSIDIAN_MCP_TOKEN={token}\n")
        env_file.write(f"OBSIDIAN_MCP_HOST={host}\n")
        env_file.write(f"OBSIDIAN_MCP_PORT={port}\n")
        env_file.write(f"OBSIDIAN_MCP_PATH={path}\n")
        if normalized_public_base_url:
            env_file.write(f"{PUBLIC_BASE_URL_ENV}={normalized_public_base_url}\n")
        if normalized_public_upload_url:
            env_file.write(f"{PUBLIC_UPLOAD_URL_ENV}={normalized_public_upload_url}\n")
        env_file.write(f"{ALLOW_SOURCE_FILES_ENV}={1 if source_files_allowed else 0}\n")
        env_file.write(f"{ARCHIVE_SOURCES_ENV}={1 if archive_sources_enabled else 0}\n")
        env_file.write(f"{ALLOWED_HOSTS_ENV}={','.join(effective_allowed_hosts)}\n")
        env_file.write(f"{ALLOWED_ORIGINS_ENV}={','.join(effective_allowed_origins)}\n")
        env_file.write(f"OBSIDIAN_MCP_INGEST_JOB_TIMEOUT_SECONDS={max(1, int(ingest_job_timeout_seconds))}\n")
        env_file.write(f"OBSIDIAN_MCP_QUERY_JOB_TIMEOUT_SECONDS={max(1, int(query_job_timeout_seconds))}\n")
        env_file.write(f"OBSIDIAN_MCP_INGEST_MAX_REQUEST_BYTES={max(0, int(ingest_max_request_bytes))}\n")
        env_file.write(f"OBSIDIAN_MCP_ENABLED_TOOLS={format_enabled_tools(tools)}\n")
        env_file.write(f"OBSIDIAN_MCP_ENABLE_INGEST={1 if TOOL_INGEST in tools else 0}\n")
        env_file.write(f"OBSIDIAN_MCP_INGEST_OPENCODE_MODEL={models.ingest_model}\n")
        env_file.write(f"OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT={models.ingest_variant}\n")
        env_file.write(f"OBSIDIAN_MCP_QUERY_OPENCODE_MODEL={models.query_model}\n")
        env_file.write(f"OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT={models.query_variant}\n")
        if opencode_path:
            env_file.write(f"OBSIDIAN_MCP_OPENCODE_PATH={opencode_path}\n")
            if TOOL_INGEST in tools:
                env_file.write(f"OBSIDIAN_MCP_INGEST_OPENCODE_PATH={opencode_path}\n")
    env_path.chmod(0o600)


def read_service_env_file(env_path: Path) -> dict[str, str]:
    if not env_path.exists():
        raise RuntimeError(
            f"Missing MCP service env file: {env_path}. "
            "Run `scripts/obsidian-wiki-ops.py mcp start` to create it."
        )

    values: dict[str, str] = {}
    for raw_line in env_path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        key, separator, value = line.partition("=")
        if separator:
            values[key.strip()] = value.strip()
    return values


def model_settings_from_config(values: dict[str, str]) -> OpenCodeModelSettings:
    return OpenCodeModelSettings(
        ingest_model=values.get("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL", DEFAULT_INGEST_OPENCODE_MODEL).strip()
        or DEFAULT_INGEST_OPENCODE_MODEL,
        ingest_variant=values.get("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT", DEFAULT_INGEST_OPENCODE_VARIANT).strip()
        or DEFAULT_INGEST_OPENCODE_VARIANT,
        query_model=values.get("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL", DEFAULT_QUERY_OPENCODE_MODEL).strip()
        or DEFAULT_QUERY_OPENCODE_MODEL,
        query_variant=values.get("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT", DEFAULT_QUERY_OPENCODE_VARIANT).strip()
        or DEFAULT_QUERY_OPENCODE_VARIANT,
    )


def saved_service_config(project_root: Path) -> ServiceConfig:
    paths = resolve_service_paths(project_root)
    values = read_service_env_file(paths.env_path)
    token = values.get("OBSIDIAN_MCP_TOKEN", "").strip()
    if not token:
        raise RuntimeError(f"Missing OBSIDIAN_MCP_TOKEN in {paths.env_path}")
    host = values.get("OBSIDIAN_MCP_HOST") or DEFAULT_HOST
    port = int(values.get("OBSIDIAN_MCP_PORT") or DEFAULT_PORT)
    path = values.get("OBSIDIAN_MCP_PATH") or DEFAULT_PATH
    public_base_url = normalize_public_base_url(values.get(PUBLIC_BASE_URL_ENV))
    public_upload_url = normalize_public_upload_url(values.get(PUBLIC_UPLOAD_URL_ENV))
    enabled_tools = enabled_tools_from_config(values)
    if ALLOW_SOURCE_FILES_ENV in values:
        allow_source_files = parse_bool(values[ALLOW_SOURCE_FILES_ENV])
    else:
        allow_source_files = TOOL_FILE_ASSETS in enabled_tools
    archive_sources = parse_bool(values.get(ARCHIVE_SOURCES_ENV))
    allowed_hosts = parse_config_list(values.get(ALLOWED_HOSTS_ENV))
    if not allowed_hosts:
        allowed_hosts = derive_allowed_hosts(
            [host, *detect_remote_hosts()],
            public_base_url=public_base_url,
            public_upload_url=public_upload_url,
        )
    allowed_origins = parse_config_list(values.get(ALLOWED_ORIGINS_ENV))
    if ALLOWED_ORIGINS_ENV not in values:
        allowed_origins = derive_allowed_origins(public_base_url, public_upload_url)
    opencode_path = (
        values.get("OBSIDIAN_MCP_OPENCODE_PATH", "").strip()
        or values.get("OBSIDIAN_MCP_INGEST_OPENCODE_PATH", "").strip()
    )
    return ServiceConfig(
        token=token,
        host=host,
        port=port,
        path=path,
        public_base_url=public_base_url,
        public_upload_url=public_upload_url,
        enabled_tools=enabled_tools,
        allow_source_files=allow_source_files,
        archive_sources=archive_sources,
        allowed_hosts=allowed_hosts,
        allowed_origins=allowed_origins,
        ingest_job_timeout_seconds=max(
            1,
            int(values.get("OBSIDIAN_MCP_INGEST_JOB_TIMEOUT_SECONDS") or DEFAULT_INGEST_JOB_TIMEOUT_SECONDS),
        ),
        query_job_timeout_seconds=max(
            1,
            int(values.get("OBSIDIAN_MCP_QUERY_JOB_TIMEOUT_SECONDS") or DEFAULT_QUERY_JOB_TIMEOUT_SECONDS),
        ),
        ingest_max_request_bytes=max(
            0,
            int(values.get("OBSIDIAN_MCP_INGEST_MAX_REQUEST_BYTES") or DEFAULT_INGEST_MAX_REQUEST_BYTES),
        ),
        opencode_path=opencode_path,
        model_settings=model_settings_from_config(values),
    )


def write_service_unit(
    paths: ServicePaths,
    *,
    vault_path: Path,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
    path: str = DEFAULT_PATH,
    uv_path: Path | None = None,
) -> None:
    paths.unit_path.parent.mkdir(parents=True, exist_ok=True)
    unit = render_service_unit(
        project_root=paths.project_root,
        vault_path=vault_path,
        env_file=paths.env_path,
        host=host,
        port=port,
        path=path,
        uv_path=uv_path,
    )
    paths.unit_path.write_text(unit, encoding="utf-8")


def detect_remote_hosts() -> list[str]:
    hosts: list[str] = []
    try:
        result = subprocess.run(["hostname", "-I"], capture_output=True, text=True, check=False)
    except OSError:
        result = None
    if result and result.returncode == 0:
        hosts.extend(result.stdout.split())

    try:
        for info in socket.getaddrinfo(socket.gethostname(), None):
            hosts.append(info[4][0])
    except OSError:
        pass

    cleaned: list[str] = []
    for host in hosts:
        normalized = host.strip()
        if not normalized or normalized.startswith("127.") or normalized == "::1" or normalized.startswith("fe80:"):
            continue
        if normalized not in cleaned:
            cleaned.append(normalized)
    return cleaned


def public_base_url_from_hosts(hosts: Sequence[str], *, port: int = DEFAULT_PORT, path: str = DEFAULT_PATH) -> str:
    host = hosts[0] if hosts else ""
    return f"http://{_url_host(host)}:{port}{path}" if host else ""


def public_hosts_for_bind(bind_host: str, detected_hosts: Sequence[str]) -> tuple[str, ...]:
    normalized = str(bind_host).strip()
    if normalized.strip("[]") in {"0.0.0.0", "::", ""}:
        return tuple(detected_hosts)
    return (normalized,)


def _url_host(host: str) -> str:
    value = str(host).strip()
    if ":" in value and not value.startswith("["):
        return f"[{value}]"
    return value


def normalize_public_base_url(public_base_url: str | None) -> str:
    value = str(public_base_url or "").strip().rstrip("/")
    if not value:
        return ""
    parsed = urlsplit(value)
    if not parsed.scheme or not parsed.netloc:
        return ""
    return urlunsplit((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", ""))


def upload_url_from_public_base(
    public_base_url: str,
    *,
    fallback_host: str,
    port: int = DEFAULT_PORT,
    public_upload_url: str | None = None,
) -> str:
    explicit_upload_url = normalize_public_upload_url(public_upload_url)
    if explicit_upload_url:
        return explicit_upload_url
    parsed = urlsplit(public_base_url)
    if parsed.scheme and parsed.netloc:
        return urlunsplit((parsed.scheme, parsed.netloc, DEFAULT_INGEST_UPLOAD_PATH, "", ""))
    return f"http://{_url_host(fallback_host)}:{port}{DEFAULT_INGEST_UPLOAD_PATH}"


def normalize_public_upload_url(public_upload_url: str | None) -> str:
    value = str(public_upload_url or "").strip().rstrip("/")
    if not value:
        return ""
    parsed = urlsplit(value)
    if not parsed.scheme or not parsed.netloc:
        return ""
    path = parsed.path.rstrip("/") or DEFAULT_INGEST_UPLOAD_PATH
    return urlunsplit((parsed.scheme, parsed.netloc, path, "", ""))


def derive_allowed_hosts(
    hosts: Sequence[str],
    *,
    public_base_url: str | None = None,
    public_upload_url: str | None = None,
) -> tuple[str, ...]:
    patterns = {"localhost", "localhost:*", "127.0.0.1", "127.0.0.1:*", "[::1]", "[::1]:*"}
    candidates = [str(host).strip() for host in hosts]
    for url in (public_base_url, public_upload_url):
        parsed = urlsplit(str(url or ""))
        if parsed.hostname:
            candidates.append(parsed.hostname)
    for candidate in candidates:
        value = candidate.strip().strip("[]")
        if not value or value in {"0.0.0.0", "::"}:
            continue
        if ":" in value:
            patterns.update({f"[{value}]", f"[{value}]:*"})
        else:
            patterns.update({value, f"{value}:*"})
    return tuple(sorted(patterns))


def derive_allowed_origins(*urls: str | None) -> tuple[str, ...]:
    origins = set()
    for url in urls:
        parsed = urlsplit(str(url or ""))
        if parsed.scheme and parsed.netloc:
            origins.add(urlunsplit((parsed.scheme, parsed.netloc, "", "", "")))
    return tuple(sorted(origins))


def merge_allowlist(existing: Sequence[str], required: Sequence[str]) -> tuple[str, ...]:
    return tuple(sorted(set(existing) | set(required)))


def render_remote_config(
    *,
    hosts: Sequence[str],
    token: str,
    port: int = DEFAULT_PORT,
    path: str = DEFAULT_PATH,
    public_base_url: str | None = None,
    public_upload_url: str | None = None,
    config_path: Path | None = None,
    enabled_tools: Sequence[str] | None = None,
) -> str:
    host = hosts[0] if hosts else "<server-ip>"
    detected_urls = [f"http://{_url_host(candidate)}:{port}{path}" for candidate in hosts]
    fallback_url = detected_urls[0] if detected_urls else f"http://{_url_host(host)}:{port}{path}"
    public_url = normalize_public_base_url(public_base_url)
    url = public_url or fallback_url
    upload_url = upload_url_from_public_base(
        url,
        fallback_host=host,
        port=port,
        public_upload_url=public_upload_url,
    )
    alternatives = "\n".join(f"- {candidate}" for candidate in detected_urls if candidate != url)
    alternatives_section = f"\nOther detected addresses:\n{alternatives}\n" if alternatives else ""
    config_section = f"\nConfig file:\n{config_path}\n" if config_path is not None else ""
    enabled_section = ""
    if enabled_tools is not None:
        enabled_section = f"\nEnabled tool groups:\n{format_enabled_tools(enabled_tools) or 'none'}\n"
    ingest_upload_section = ""
    if enabled_tools is not None and TOOL_INGEST in normalize_enabled_tools(enabled_tools):
        detected_upload_urls = [
            f"http://{_url_host(candidate)}:{port}{DEFAULT_INGEST_UPLOAD_PATH}" for candidate in hosts[1:]
        ]
        upload_alternatives = "\n".join(
            f"- {candidate}" for candidate in detected_upload_urls if candidate != upload_url
        )
        upload_alternatives_section = (
            f"\nOther ingest upload addresses:\n{upload_alternatives}\n" if upload_alternatives else ""
        )
        ingest_upload_section = (
            "\n"
            "Ingest file upload endpoint:\n"
            f"{upload_url}\n"
            f"{upload_alternatives_section}"
            "Remote agents should upload with create_ingest_upload_ticket and X-Obsidian-Wiki-Upload-Token.\n"
            "Bearer upload is for local/admin use. POST multipart/form-data with one or more `files` fields.\n"
        )
    query_section = ""
    if enabled_tools is not None and TOOL_QUERY in normalize_enabled_tools(enabled_tools):
        query_section = (
            "\n"
            "Wiki query tools:\n"
            "Use ask_wiki for ordinary wiki, business facts, or hybrid questions.\n"
            "For long-running jobs, use start_wiki_query, get_wiki_query_status, "
            "get_wiki_query_result, cancel_wiki_query, and check_query_opencode_readiness.\n"
        )
    return (
        "MCP service is running.\n"
        "\n"
        "Use this URL from another machine:\n"
        f"{url}\n"
        f"{alternatives_section}"
        f"{config_section}"
        f"{enabled_section}"
        f"{ingest_upload_section}"
        f"{query_section}"
        "\n"
        "Use this header:\n"
        f"Authorization: Bearer {token}\n"
        "\n"
        "Example MCP client config:\n"
        "{\n"
        '  "mcpServers": {\n'
        f'    "{MCP_CLIENT_NAME}": {{\n'
        '      "type": "streamable-http",\n'
        f'      "url": "{url}",\n'
        '      "headers": {\n'
        f'        "Authorization": "Bearer {token}"\n'
        "      }\n"
        "    }\n"
        "  }\n"
        "}\n"
    )


def show_config(project_root: Path) -> str:
    paths = resolve_service_paths(project_root)
    config = saved_service_config(project_root)
    return render_remote_config(
        hosts=detect_remote_hosts(),
        token=config.token,
        port=config.port,
        path=config.path,
        public_base_url=config.public_base_url,
        public_upload_url=config.public_upload_url,
        config_path=paths.env_path,
        enabled_tools=config.enabled_tools,
    )


def run_systemctl(args: Sequence[str], *, check: bool = True) -> subprocess.CompletedProcess[str]:
    return subprocess.run(["systemctl", "--user", *args], text=True, check=check)


def import_proxy_environment() -> None:
    names = [name for name in PROXY_ENV_NAMES if os.environ.get(name)]
    if names:
        run_systemctl(["import-environment", *names], check=False)


def restart_service_if_installed(paths: ServicePaths) -> bool:
    if not paths.unit_path.exists():
        return False
    run_systemctl(["restart", SERVICE_NAME])
    return True


def write_service_config(env_path: Path, config: ServiceConfig) -> None:
    write_token_env_file(
        env_path,
        config.token,
        host=config.host,
        port=config.port,
        path=config.path,
        public_base_url=config.public_base_url,
        public_upload_url=config.public_upload_url,
        enabled_tools=config.enabled_tools,
        allow_source_files=config.allow_source_files,
        archive_sources=config.archive_sources,
        allowed_hosts=config.allowed_hosts,
        allowed_origins=config.allowed_origins,
        ingest_job_timeout_seconds=config.ingest_job_timeout_seconds,
        query_job_timeout_seconds=config.query_job_timeout_seconds,
        ingest_max_request_bytes=config.ingest_max_request_bytes,
        opencode_path=config.opencode_path,
        model_settings=config.model_settings,
    )


def new_token(
    *,
    project_root: Path,
    host: str | None = None,
    port: int | None = None,
    path: str | None = None,
    public_base_url: str | None = None,
    public_upload_url: str | None = None,
    token_bytes: int = DEFAULT_TOKEN_BYTES,
) -> str:
    paths = resolve_service_paths(project_root)
    try:
        saved = saved_service_config(project_root)
        has_saved_config = True
    except RuntimeError:
        saved = ServiceConfig(token="")
        has_saved_config = False
    hosts = detect_remote_hosts()
    effective_host = saved.host if host is None else host
    effective_port = saved.port if port is None else port
    effective_path = saved.path if path is None else path
    address_overridden = any(value is not None for value in (host, port, path))
    if public_base_url is not None:
        effective_public_base_url = normalize_public_base_url(public_base_url)
    elif has_saved_config and not address_overridden:
        effective_public_base_url = saved.public_base_url
    else:
        effective_public_base_url = public_base_url_from_hosts(
            public_hosts_for_bind(effective_host, hosts),
            port=effective_port,
            path=effective_path,
        )
    effective_public_upload_url = (
        saved.public_upload_url
        if public_upload_url is None
        else normalize_public_upload_url(public_upload_url)
    )
    endpoint_overridden = any(
        value is not None
        for value in (host, port, path, public_base_url, public_upload_url)
    )
    if has_saved_config and not endpoint_overridden:
        allowed_hosts = saved.allowed_hosts
        allowed_origins = saved.allowed_origins
    else:
        allowed_hosts = merge_allowlist(
            saved.allowed_hosts if has_saved_config else (),
            derive_allowed_hosts(
                [effective_host, *hosts],
                public_base_url=effective_public_base_url,
                public_upload_url=effective_public_upload_url,
            ),
        )
        allowed_origins = merge_allowlist(
            saved.allowed_origins if has_saved_config else (),
            derive_allowed_origins(effective_public_base_url, effective_public_upload_url),
        )
    archive_sources = project_archive_sources(
        paths.project_root,
        default=saved.archive_sources,
    )
    config = replace(
        saved,
        token=generate_token(token_bytes),
        host=effective_host,
        port=effective_port,
        path=effective_path,
        public_base_url=effective_public_base_url,
        public_upload_url=effective_public_upload_url,
        archive_sources=archive_sources,
        allowed_hosts=allowed_hosts,
        allowed_origins=allowed_origins,
    )
    write_service_config(paths.env_path, config)
    restart_service_if_installed(paths)
    return render_remote_config(
        hosts=hosts,
        token=config.token,
        port=config.port,
        path=config.path,
        public_base_url=config.public_base_url,
        public_upload_url=config.public_upload_url,
        config_path=paths.env_path,
        enabled_tools=config.enabled_tools,
    )


def start_service(
    *,
    project_root: Path,
    vault_path: Path | str | None = None,
    host: str | None = None,
    port: int | None = None,
    path: str | None = None,
    public_base_url: str | None = None,
    public_upload_url: str | None = None,
    token_bytes: int = DEFAULT_TOKEN_BYTES,
    enable_ingest: bool | None = None,
    enable_query: bool | None = None,
    allow_source_files: bool | None = None,
) -> str:
    paths = resolve_service_paths(project_root)
    try:
        saved = saved_service_config(project_root)
        has_saved_config = True
    except RuntimeError:
        if paths.env_path.exists():
            raise
        saved = ServiceConfig(token=generate_token(token_bytes))
        has_saved_config = False
    effective_host = saved.host if host is None else host
    effective_port = saved.port if port is None else port
    effective_path = saved.path if path is None else path
    address_overridden = any(value is not None for value in (host, port, path))
    enabled_tools = update_ingest_tool(saved.enabled_tools, enable_ingest)
    enabled_tools = update_query_tool(enabled_tools, enable_query)
    opencode_path = saved.opencode_path
    if ({TOOL_INGEST, TOOL_QUERY} & set(enabled_tools)) and not opencode_path:
        opencode_path = resolve_opencode_path()
    hosts = detect_remote_hosts()
    if public_base_url is not None:
        effective_public_base_url = normalize_public_base_url(public_base_url)
    elif has_saved_config and not address_overridden:
        effective_public_base_url = saved.public_base_url
    else:
        effective_public_base_url = public_base_url_from_hosts(
            public_hosts_for_bind(effective_host, hosts),
            port=effective_port,
            path=effective_path,
        )
    effective_public_upload_url = (
        saved.public_upload_url
        if public_upload_url is None
        else normalize_public_upload_url(public_upload_url)
    )
    endpoint_overridden = any(
        value is not None
        for value in (host, port, path, public_base_url, public_upload_url)
    )
    if has_saved_config and not endpoint_overridden:
        allowed_hosts = saved.allowed_hosts
        allowed_origins = saved.allowed_origins
    else:
        allowed_hosts = merge_allowlist(
            saved.allowed_hosts if has_saved_config else (),
            derive_allowed_hosts(
                [effective_host, *hosts],
                public_base_url=effective_public_base_url,
                public_upload_url=effective_public_upload_url,
            ),
        )
        allowed_origins = merge_allowlist(
            saved.allowed_origins if has_saved_config else (),
            derive_allowed_origins(effective_public_base_url, effective_public_upload_url),
        )
    archive_sources = project_archive_sources(
        paths.project_root,
        default=saved.archive_sources,
    )
    config = replace(
        saved,
        host=effective_host,
        port=effective_port,
        path=effective_path,
        public_base_url=effective_public_base_url,
        public_upload_url=effective_public_upload_url,
        enabled_tools=enabled_tools,
        opencode_path=opencode_path,
        allow_source_files=(
            saved.allow_source_files if allow_source_files is None else allow_source_files
        ),
        archive_sources=archive_sources,
        allowed_hosts=allowed_hosts,
        allowed_origins=allowed_origins,
    )
    write_service_config(paths.env_path, config)
    uv_path = find_uv_path()
    prepare_runtime(paths.project_root, uv_path=uv_path)
    resolved_vault = resolve_vault_for_service(paths.project_root, vault_path=vault_path)
    write_service_unit(
        paths,
        vault_path=resolved_vault,
        host=config.host,
        port=config.port,
        path=config.path,
        uv_path=uv_path,
    )

    if {TOOL_INGEST, TOOL_QUERY} & set(config.enabled_tools):
        import_proxy_environment()
    run_systemctl(["daemon-reload"])
    run_systemctl(["enable", SERVICE_NAME])
    run_systemctl(["restart", SERVICE_NAME])
    return render_remote_config(
        hosts=hosts,
        token=config.token,
        port=config.port,
        path=config.path,
        public_base_url=config.public_base_url,
        public_upload_url=config.public_upload_url,
        config_path=paths.env_path,
        enabled_tools=config.enabled_tools,
    )


def render_tool_status(
    *,
    config_path: Path,
    enabled_tools: Sequence[str],
    restarted: bool = False,
) -> str:
    enabled = set(normalize_enabled_tools(enabled_tools))
    lines = [
        "MCP tool configuration.",
        "",
        f"Config file: {config_path}",
        f"Enabled tool groups: {format_enabled_tools(enabled_tools) or 'none'}",
        "",
        "Available tool groups:",
    ]
    for tool in AVAILABLE_TOOL_GROUPS:
        state = "enabled" if tool in enabled else "disabled"
        lines.append(f"- {tool}: {state}")
    lines.append("")
    lines.append(f"Service restarted: {'yes' if restarted else 'no'}")
    return "\n".join(lines) + "\n"


def ensure_service_config(project_root: Path) -> tuple[ServicePaths, ServiceConfig]:
    paths = resolve_service_paths(project_root)
    try:
        config = saved_service_config(project_root)
    except RuntimeError:
        hosts = detect_remote_hosts()
        public_base_url = public_base_url_from_hosts(
            hosts,
            port=DEFAULT_PORT,
            path=DEFAULT_PATH,
        )
        config = ServiceConfig(
            token=generate_token(),
            public_base_url=public_base_url,
            allowed_hosts=derive_allowed_hosts(
                [DEFAULT_HOST, *hosts],
                public_base_url=public_base_url,
            ),
            allowed_origins=derive_allowed_origins(public_base_url),
        )
        write_service_config(paths.env_path, config)
    return paths, config


def list_tools(project_root: Path) -> str:
    paths, config = ensure_service_config(project_root)
    return render_tool_status(config_path=paths.env_path, enabled_tools=config.enabled_tools)


def set_tool_enabled(project_root: Path, tool: str, enabled: bool) -> str:
    tool_name = normalize_tool_name(tool)
    paths, config = ensure_service_config(project_root)
    current = set(config.enabled_tools)
    if enabled:
        current.add(tool_name)
    else:
        current.discard(tool_name)
    updated_tools = tuple(item for item in AVAILABLE_TOOL_GROUPS if item in current)
    opencode_path = config.opencode_path
    if ({TOOL_INGEST, TOOL_QUERY} & set(updated_tools)) and not opencode_path:
        opencode_path = resolve_opencode_path()
    config = replace(
        config,
        enabled_tools=updated_tools,
        opencode_path=opencode_path,
    )
    write_service_config(paths.env_path, config)
    if paths.unit_path.exists() and {TOOL_INGEST, TOOL_QUERY} & set(updated_tools):
        import_proxy_environment()
    restarted = restart_service_if_installed(paths)
    return render_tool_status(config_path=paths.env_path, enabled_tools=updated_tools, restarted=restarted)


def stop_service() -> None:
    run_systemctl(["stop", SERVICE_NAME])


def uninstall_service(project_root: Path) -> None:
    paths = resolve_service_paths(project_root)
    run_systemctl(["disable", "--now", SERVICE_NAME], check=False)
    if paths.unit_path.exists():
        paths.unit_path.unlink()
    if paths.env_path.exists():
        paths.env_path.unlink()
    try:
        paths.env_path.parent.rmdir()
    except OSError:
        pass
    run_systemctl(["daemon-reload"], check=False)
    run_systemctl(["reset-failed", SERVICE_NAME], check=False)


def service_status() -> None:
    run_systemctl(["status", SERVICE_NAME], check=False)


def cleanup_runtime(project_root: Path, *, max_age_days: int = DEFAULT_RUNTIME_CLEANUP_DAYS) -> str:
    if max_age_days <= 0:
        raise RuntimeError("max_age_days must be positive")
    root = project_root.expanduser().resolve()
    cutoff = time.time() - (max_age_days * 24 * 60 * 60)
    ingest_root = root / ".runtime" / "mcp-ingest"
    query_root = root / ".runtime" / "mcp-query"
    active_upload_ids = _active_ingest_upload_ids(ingest_root / "jobs")
    result = {
        "expired_upload_tickets_deleted": _delete_expired_upload_tickets(ingest_root / "upload-tickets"),
        "old_ingest_uploads_deleted": _delete_old_ingest_uploads(
            ingest_root / "uploads",
            cutoff=cutoff,
            active_upload_ids=active_upload_ids,
        ),
        "old_ingest_jobs_deleted": _delete_old_terminal_jobs(ingest_root / "jobs", cutoff=cutoff),
        "old_query_jobs_deleted": _delete_old_terminal_jobs(query_root / "jobs", cutoff=cutoff),
    }
    return "\n".join(
        [
            "MCP runtime cleanup.",
            "",
            f"Project root: {root}",
            f"Max age: {max_age_days} days",
            f"Expired upload tickets deleted: {result['expired_upload_tickets_deleted']}",
            f"Old ingest uploads deleted: {result['old_ingest_uploads_deleted']}",
            f"Old ingest jobs/logs deleted: {result['old_ingest_jobs_deleted']}",
            f"Old query jobs/logs deleted: {result['old_query_jobs_deleted']}",
        ]
    ) + "\n"


def _active_ingest_upload_ids(jobs_root: Path) -> set[str]:
    upload_ids: set[str] = set()
    for job in _iter_job_records(jobs_root):
        if job.get("state") in TERMINAL_JOB_STATES:
            continue
        upload_ids.update(str(upload_id) for upload_id in job.get("upload_ids", []))
    return upload_ids


def _delete_expired_upload_tickets(ticket_root: Path) -> int:
    if not ticket_root.exists():
        return 0
    deleted = 0
    now = time.time()
    for ticket_path in ticket_root.glob("*.json"):
        meta = _read_json(ticket_path)
        if meta is None:
            continue
        try:
            expired = now > float(meta.get("expires_epoch", 0))
        except (TypeError, ValueError):
            expired = False
        if not expired:
            continue
        ticket_path.unlink(missing_ok=True)
        deleted += 1
    return deleted


def _delete_old_ingest_uploads(upload_root: Path, *, cutoff: float, active_upload_ids: set[str]) -> int:
    if not upload_root.exists():
        return 0
    deleted = 0
    for meta_path in upload_root.glob("*/*/upload.json"):
        upload_dir = meta_path.parent
        meta = _read_json(meta_path)
        if meta is None:
            continue
        if str(meta.get("upload_id", "")) in active_upload_ids:
            continue
        if _tree_mtime(upload_dir) >= cutoff:
            continue
        shutil.rmtree(upload_dir, ignore_errors=True)
        deleted += 1
    return deleted


def _delete_old_terminal_jobs(jobs_root: Path, *, cutoff: float) -> int:
    if not jobs_root.exists():
        return 0
    deleted = 0
    for job in _iter_job_records(jobs_root):
        if job.get("state") not in TERMINAL_JOB_STATES:
            continue
        job_id = str(job.get("job_id", ""))
        job_dir = jobs_root / job_id
        if not job_dir.exists() or _tree_mtime(job_dir) >= cutoff:
            continue
        shutil.rmtree(job_dir, ignore_errors=True)
        deleted += 1
    return deleted


def _iter_job_records(jobs_root: Path) -> list[dict[str, object]]:
    jobs: list[dict[str, object]] = []
    if not jobs_root.exists():
        return jobs
    for job_path in jobs_root.glob("*/job.json"):
        record = _read_json(job_path)
        if record is not None:
            jobs.append(record)
    return jobs


def _read_json(path: Path) -> dict[str, object] | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def _tree_mtime(path: Path) -> float:
    if not path.exists():
        return time.time()
    latest = path.stat().st_mtime
    for child in path.rglob("*"):
        try:
            latest = max(latest, child.stat().st_mtime)
        except OSError:
            continue
    return latest


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Manage the HTTP obsidian-wiki MCP user service.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    new_token_parser = subparsers.add_parser("new-token", help="Generate a token and print remote MCP client configuration.")
    new_token_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
    new_token_parser.add_argument("--host")
    new_token_parser.add_argument("--port", type=int)
    new_token_parser.add_argument("--path")
    new_token_parser.add_argument("--public-base-url")
    new_token_parser.add_argument("--public-upload-url")
    new_token_parser.add_argument("--token-bytes", default=DEFAULT_TOKEN_BYTES, type=int)

    start_parser = subparsers.add_parser("start", help="Install or update the user service and print remote MCP config.")
    start_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
    start_parser.add_argument("--vault", dest="vault_path", type=Path, help="Vault path to serve. Defaults to OBSIDIAN_VAULT_PATH config.")
    start_parser.add_argument("--host")
    start_parser.add_argument("--port", type=int)
    start_parser.add_argument("--path")
    start_parser.add_argument("--public-base-url")
    start_parser.add_argument("--public-upload-url")
    start_parser.add_argument("--token-bytes", default=DEFAULT_TOKEN_BYTES, type=int)
    ingest_group = start_parser.add_mutually_exclusive_group()
    ingest_group.add_argument(
        "--enable-ingest",
        dest="enable_ingest",
        action="store_true",
        default=None,
        help="Expose write-capable upload and local OpenCode ingest MCP tools.",
    )
    ingest_group.add_argument(
        "--disable-ingest",
        dest="enable_ingest",
        action="store_false",
        help="Disable write-capable upload and local OpenCode ingest MCP tools.",
    )
    query_group = start_parser.add_mutually_exclusive_group()
    query_group.add_argument(
        "--enable-query",
        dest="enable_query",
        action="store_true",
        default=None,
        help="Expose local OpenCode wiki-query MCP tools.",
    )
    query_group.add_argument(
        "--disable-query",
        dest="enable_query",
        action="store_false",
        help="Disable local OpenCode wiki-query MCP tools.",
    )
    source_files_group = start_parser.add_mutually_exclusive_group()
    source_files_group.add_argument(
        "--allow-source-files",
        dest="allow_source_files",
        action="store_true",
        default=None,
        help="Allow file-assets tools to return source file content.",
    )
    source_files_group.add_argument(
        "--deny-source-files",
        dest="allow_source_files",
        action="store_false",
        help="Deny source file content while retaining file-assets metadata tools.",
    )

    subparsers.add_parser("stop", help="Stop the user service without deleting service files.")

    uninstall_parser = subparsers.add_parser("uninstall", help="Stop and remove the managed user service and token file.")
    uninstall_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)

    show_config_parser = subparsers.add_parser("show-config", help="Print the saved token and remote MCP client configuration.")
    show_config_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)

    subparsers.add_parser("status", help="Show systemd status for the managed service.")

    cleanup_parser = subparsers.add_parser("cleanup", help="Prune expired tickets and old MCP runtime uploads/jobs/logs.")
    cleanup_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)
    cleanup_parser.add_argument("--max-age-days", default=DEFAULT_RUNTIME_CLEANUP_DAYS, type=int)

    tools_parser = subparsers.add_parser("tools", help="List or update enabled MCP tool groups in the service config.")
    tools_subparsers = tools_parser.add_subparsers(dest="tools_command", required=True)

    tools_list_parser = tools_subparsers.add_parser("list", help="List configured MCP tool groups.")
    tools_list_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)

    tools_enable_parser = tools_subparsers.add_parser("enable", help="Enable an MCP tool group.")
    tools_enable_parser.add_argument("tool", choices=AVAILABLE_TOOL_GROUPS)
    tools_enable_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)

    tools_disable_parser = tools_subparsers.add_parser("disable", help="Disable an MCP tool group.")
    tools_disable_parser.add_argument("tool", choices=AVAILABLE_TOOL_GROUPS)
    tools_disable_parser.add_argument("--project-root", default=Path(__file__).resolve().parent.parent, type=Path)

    args = parser.parse_args(argv)

    if args.command == "new-token":
        print(
            new_token(
                project_root=args.project_root,
                host=args.host,
                port=args.port,
                path=args.path,
                public_base_url=args.public_base_url,
                public_upload_url=args.public_upload_url,
                token_bytes=args.token_bytes,
            ),
            end="",
        )
        return 0
    if args.command == "start":
        print(
            start_service(
                project_root=args.project_root,
                vault_path=args.vault_path,
                host=args.host,
                port=args.port,
                path=args.path,
                public_base_url=args.public_base_url,
                public_upload_url=args.public_upload_url,
                token_bytes=args.token_bytes,
                enable_ingest=args.enable_ingest,
                enable_query=args.enable_query,
                allow_source_files=args.allow_source_files,
            ),
            end="",
        )
        return 0
    if args.command == "stop":
        stop_service()
        print(f"Stopped {SERVICE_NAME}.")
        return 0
    if args.command == "uninstall":
        uninstall_service(args.project_root)
        print(f"Uninstalled {SERVICE_NAME}.")
        return 0
    if args.command == "show-config":
        print(show_config(args.project_root), end="")
        return 0
    if args.command == "status":
        service_status()
        return 0
    if args.command == "cleanup":
        print(cleanup_runtime(args.project_root, max_age_days=args.max_age_days), end="")
        return 0
    if args.command == "tools":
        if args.tools_command == "list":
            print(list_tools(args.project_root), end="")
            return 0
        if args.tools_command == "enable":
            print(set_tool_enabled(args.project_root, args.tool, True), end="")
            return 0
        if args.tools_command == "disable":
            print(set_tool_enabled(args.project_root, args.tool, False), end="")
            return 0
    raise AssertionError(f"unhandled command: {args.command}")


if __name__ == "__main__":
    raise SystemExit(main())