screen-locker/screen_locker/_mcp.py

"""Read-only MCP (Model Context Protocol) server for screen-locker.

Exposes the locker's *read-only* status surface as typed MCP tools, so an MCP
client (Claude Code and its subagents) can query today's/this week's workout
compliance, the shutdown projection, and the lock-decision trace without
shelling out to the ``screen-locker-status`` CLI or opening the Tk window.

Run via the dedicated venv that has the ``mcp`` extra installed::

    ~/.venvs/screen-locker-mcp/bin/python -m screen_locker._mcp

(see ``scripts/setup_mcp.sh`` and the repo-root ``.mcp.json``).

Safety invariants (do not break when adding tools):
  * **READ-ONLY.** There are no write/action tools here, by design. In
    particular there is deliberately **no** ``log_workout`` and no tool that
    mutates workout/compliance state — the project rule is that workouts are
    logged only from RunnerUp-verified TCX data, never from a caller's claim.
    Every tool below calls only side-effect-free leaf helpers that *read*
    on-disk state (``gather_status`` and the ``_compliance_state`` predicates),
    never a mutating CLI/daemon entry point.
  * **stdout is the JSON-RPC channel.** This module and every function a tool
    calls must never write to stdout. All logging is routed to STDERR below,
    and no tool calls ``screen_lock.py`` / ``status_view.main`` (which
    ``print`` / open Tk windows / ``sys.exit``).
  * **No secret ever leaves.** No tool returns the sync token
    (``~/.config/screen_locker/sync_token``) or any HMAC key; the predicates
    load only status/state and return booleans, and the status snapshot
    carries no secret material.
"""

from __future__ import annotations

from dataclasses import asdict
import logging
from pathlib import Path
import sys
from typing import Any

from mcp.server import MCPServer
from mcp.types import ToolAnnotations

from screen_locker._compliance_state import (
    has_logged_today,
    is_early_bird_pending,
    is_scheduled_skip_today,
    is_sick_day_today,
)
from screen_locker._constants import (
    EARLY_BIRD_PENDING_FILE,
)
from screen_locker._sick_tracker import load_history
from screen_locker._status_data import format_summary_line, gather_status
from screen_locker._status_projection import compliance_state_word

# Log to STDERR only — STDOUT carries the MCP JSON-RPC protocol frames, so a
# single stray stdout write would corrupt the stream and kill the session.
logging.basicConfig(
    level=logging.INFO,
    stream=sys.stderr,
    format="%(asctime)s [%(levelname)s] screen-locker-mcp: %(message)s",
)
logger = logging.getLogger(__name__)

mcp = MCPServer("screen-locker")

# Every tool here only reads local state: the workout log, the sick history
# and the clock. Saying so lets a client run them without asking -- and the
# day one of them stops being read-only, this annotation is the thing that
# has to change with it. (See the section banner below: "never mutate" was
# already the rule; this is the machine-readable half of it.)
_READS_ONLY = ToolAnnotations(
    read_only_hint=True,
    idempotent_hint=True,
    open_world_hint=False,
)

# The workout log lives beside the package, same default as
# ``_status_data._DEFAULT_LOG_FILE``; declared here to avoid importing a
# private name across modules.
_LOG_FILE = Path(__file__).resolve().parent / "log.json"


# ──────────────────────────────────────────────────────────────
# Read tools (state-only; never expose secrets, never mutate)
# ──────────────────────────────────────────────────────────────


@mcp.tool(title="Workout gate status", annotations=_READS_ONLY)
def get_status() -> dict[str, Any]:
    """Return the full read-only status snapshot.

    A JSON-friendly projection of ``gather_status()``: today's and this week's
    workout outcomes, the shutdown-time projection, the lock-decision trace,
    and the rolling sick-day / manual-workout budgets. Reads on-disk state
    only (no ADB, sudo, or network) and never mutates anything.
    """
    return asdict(gather_status())


@mcp.tool(title="One-line status summary", annotations=_READS_ONLY)
def get_summary() -> dict[str, Any]:
    """Return the one-line i3blocks summary plus the compliance-state word.

    ``summary_line`` is the same string the status bar shows
    (``format_summary_line``); ``compliance_state`` is one of ``ok`` / ``warn``
    / ``lock`` (``compliance_state_word``).
    """
    snapshot = gather_status()
    return {
        "summary_line": format_summary_line(snapshot),
        "compliance_state": compliance_state_word(snapshot),
    }


@mcp.tool(title="Why the screen is locked", annotations=_READS_ONLY)
def explain_lock() -> dict[str, Any]:
    """Return why the screen is or is not being locked right now.

    A JSON-friendly projection of the ``LockExplanation`` reconstruction of the
    lock-decision chain (which skip condition fired, the full ordered trace,
    and any pending auto-upgrade opportunity). Read-only: the heat-skip stage
    is reported as not-evaluated rather than triggering a live ``wttr.in`` call.
    """
    return asdict(gather_status().lock_explanation)


@mcp.tool(title="Lock-decision predicates", annotations=_READS_ONLY)
def get_flags() -> dict[str, bool]:
    """Return the individual boolean lock-decision predicates for today.

    Each predicate reads its own on-disk state file and degrades to ``False``
    when that file is missing or unreadable — none of them mutate state or
    expose secrets.
    """
    history = load_history()
    return {
        "has_logged_today": has_logged_today(_LOG_FILE),
        "is_scheduled_skip_today": is_scheduled_skip_today(),
        "is_early_bird_pending": is_early_bird_pending(EARLY_BIRD_PENDING_FILE),
        "is_sick_day_today": is_sick_day_today(history),
    }


def main() -> None:
    """Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs)."""
    logger.info("Starting screen-locker MCP server (python=%s)", sys.executable)
    mcp.run()  # pragma: no cover


if __name__ == "__main__":
    main()