#!/usr/bin/env python3
"""
HITL payload and response schema for interactive path approval (ADR-0007).
The client-side permission gate collects outside-project ``PathRequest``
objects, pauses the run with an ``interrupt`` carrying
:func:`build_permission_payload`, and resolves the user's per-path
allow/deny decision with :func:`resolve_decisions`. The interrupt mechanism
itself is the shared LangGraph one (ADR-0046); this module owns only the
permission-specific payload, response schema, and decision mapping.
See ``devdocs/system/mcp-permissions.md``.
File: klea_utils/mcp/permission_hitl.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import logging
from collections.abc import Sequence
from typing import Any, Literal
from pydantic import BaseModel, Field, ValidationError
from klea_utils.mcp.path_detect import PathRequest, truncate_middle
logger = logging.getLogger(__name__)
#: Default question shown with a permission ask.
DEFAULT_QUESTION = (
"The agent wants to access paths outside the project directory. "
"Allow or deny each one."
)
#: A per-path decision: allow for this dispatch only, allow for the session
#: (persisted in ``allowed_dirs``), deny, or dismiss the request as "not a
#: path" (a heuristic false positive; the call proceeds for this dispatch and
#: nothing is remembered).
Decision = Literal["once", "session", "deny", "not_a_path"]
[docs]
class PathDecision(BaseModel):
"""A user's decision for one requested path.
``key`` matches a request's :attr:`~klea_utils.mcp.path_detect.PathRequest.approval_key`
(the directory for an ``outside`` request, the file for a ``sensitive``
one); ``decision`` is ``once`` (this dispatch), ``session`` (also
persisted for the thread), ``deny``, or ``not_a_path`` (dismiss a guessed
false positive for this dispatch; the call proceeds).
"""
key: str
decision: Decision = "deny"
[docs]
class PermissionResponse(BaseModel):
"""Resume payload for a permission ask (ADR-0007).
``action="cancel"`` denies every request. ``action="answer"`` carries one
:class:`PathDecision` per requested directory; a requested directory with
no decision is denied (fail closed).
"""
action: Literal["answer", "cancel"] = "answer"
decisions: list[PathDecision] = Field(default_factory=list)
[docs]
class PermissionResolution(BaseModel):
"""Resolved decisions for one permission ask.
Fields hold approval keys (a directory for an ``outside`` request, a file
for a ``sensitive`` one). :attr:`allowed_now` is allowed for this
dispatch only; :attr:`allowed_session` is also persisted for the thread;
:attr:`denied` is refused (the gate turns each into a non-halting denial);
:attr:`not_a_path` is dismissed for **this dispatch only** -- the request
was a heuristic false positive (the user is not approving it), so the call
proceeds without gating it, and no memory is kept.
"""
allowed_now: list[str] = Field(default_factory=list)
allowed_session: list[str] = Field(default_factory=list)
denied: list[str] = Field(default_factory=list)
not_a_path: list[str] = Field(default_factory=list)
[docs]
def effective_keys(self) -> list[str]:
"""Return the approval keys allowed for the current dispatch."""
return [*self.allowed_now, *self.allowed_session]
[docs]
def build_permission_payload(
requests: Sequence[PathRequest], *, question: str = ""
) -> dict[str, Any]:
"""Build the interrupt payload for a list of path requests.
The payload is a plain mapping (the stream/interrupt convention): ``kind``
is ``"permission"``, ``question`` is the prompt text, and ``requests`` is
the list of serialized :class:`~klea_utils.mcp.path_detect.PathRequest`
objects for the client to render. Each request carries the full ``path``
and ``directory`` for decisions plus bounded ``path_display``,
``directory_display``, ``raw_preview`` and ``argument_preview`` fields for
the UI; the payload's ``raw`` value is capped to ``raw_preview``.
:param requests: The outside-path requests to ask about.
:param question: Custom prompt text; defaults to :data:`DEFAULT_QUESTION`.
:returns: The interrupt payload dict.
"""
logger.debug(
f"Building permission payload\n{len(requests) = }\n"
f"{[request.approval_key for request in requests] = }"
)
payload_requests: list[dict[str, Any]] = []
for request in requests:
item = request.model_dump()
covered = request.covered_paths or [request.path]
item.update(
raw=request.raw_preview(),
raw_preview=request.raw_preview(),
raw_length=len(request.raw),
path_display=request.path_display(),
directory_display=request.directory_display(),
covered_paths=covered,
covered_paths_display=[truncate_middle(path) for path in covered],
)
payload_requests.append(item)
return {
"kind": "permission",
"question": question or DEFAULT_QUESTION,
"requests": payload_requests,
}
def _coerce_response(response: Any) -> PermissionResponse:
"""Return *response* as a :class:`PermissionResponse`, fail-closed.
Accepts an already-validated model, a raw mapping (as ``chat_core`` builds
from the client answer), or anything else; an invalid response becomes an
empty answer, which denies every request.
"""
if isinstance(response, PermissionResponse):
return response
if isinstance(response, dict):
try:
return PermissionResponse.model_validate(response)
except ValidationError:
logger.warning("Invalid permission response; denying all requests")
return PermissionResponse()
[docs]
def resolve_decisions(
requests: Sequence[PathRequest], response: Any
) -> PermissionResolution:
"""Resolve a resume value into allow-now / allow-session / deny key sets.
Uses each request's :attr:`~klea_utils.mcp.path_detect.PathRequest.approval_key`
(the directory for an ``outside`` request, the file for a ``sensitive``
one). A cancel, an invalid response, or a request with no matching
decision denies that request (fail closed).
:param requests: The requests that were asked about.
:param response: The value returned by the interrupt (a
:class:`PermissionResponse`, a raw mapping, or ``None``).
:returns: The :class:`PermissionResolution`.
"""
parsed = _coerce_response(response)
requested = [request.approval_key for request in requests]
if parsed.action == "cancel":
logger.info(f"Permission ask cancelled; denying all\n{requested = }")
return PermissionResolution(denied=list(requested))
by_key = {decision.key: decision.decision for decision in parsed.decisions}
resolution = PermissionResolution()
for key in requested:
decision = by_key.get(key, "deny")
if decision == "session":
resolution.allowed_session.append(key)
elif decision == "once":
resolution.allowed_now.append(key)
elif decision == "not_a_path":
resolution.not_a_path.append(key)
else:
resolution.denied.append(key)
logger.debug(f"Resolved permission ask\n{resolution = }")
return resolution