scripts.claude_governance_gate

Reference › scripts › claude_governance_gate

The Claude Code PreToolUse hook shim that implements the ADR-0032 admission-control gate. Claude Code runs it before each tool call. It uses the gate policy in gate_policy and the ToolInterceptor in tool_interceptor. Its main carries @traces for spec 025 (REQ-025-098, -099, -100 and -102).

Invocation: main takes no command-line options. It reads one hook payload from standard input and writes the hook decision to standard output, always returning 0, because allow and block are both expressed in the output. ORCHESTRATA_CLAUDE_GATE_LOG, when set, names an audit log to which each decision is appended (_append_audit_log). The failure policy and the timeout come from environment-based settings (_parsed_failure_policy, _parsed_timeout_seconds, _load_declaration).

The flow: _read_call parses the payload and the tool input (_parse_payload, _coerce_tool_input). A call that the gate must not intercept, such as the gate's own tooling, is passed on without evaluation (_emit_self_exclusion). Otherwise _evaluate runs the interceptor on a worker thread (_InterceptorRun) under the timeout, and _classify_worker turns the result into a GateOutcome. A decision to deny gives _emit_deny, and a decision that rewrites the tool input gives _emit_allow_with_update. An outcome that could not be determined, from an error or timeout, is resolved by the declared failure policy (_resolve_and_emit, _emit_undetermined_error).

At a glance

Class diagram

Dependency diagram

API

Classes

_InterceptorRun

Bases: threading.Thread · Decorators: traces(spec='025', reqs=['REQ-025-099']) · Line 265

Runs one ToolInterceptor.intercept() call to completion on a dedicated thread, so :func:_evaluate can bound it with a join timeout (ADR-0032 #6) without a signal-based alarm (main-thread…

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/claude_governance_gate.py

"""Claude Code ``PreToolUse`` hook shim — the ADR-0032 admission-control gate.

Wired via ``orchestrata_ultimate.drivers.claude_settings.write_governance_settings``,
which points a Claude Code ``.claude/settings.json`` ``PreToolUse`` hook
(matcher ``"*"``) at ``uv run --project <repo_root> python
<repo_root>/scripts/claude_governance_gate.py``. Claude Code invokes that
command once per tool call, piping a JSON payload
(``tool_name``/``tool_input``/``session_id``/...) on stdin and reading this
script's stdout for the hook decision.

This shim implements the admission-control contract of ADR-0032 (which
amends ADR-0002) over the REAL, EXISTING
:class:`orchestrata_ultimate.core.adapters.claude.tool_interceptor.ToolInterceptor`
— no new policy engine, no redesign of that module:

1. **Self-exclusion first** (ADR-0032 #5). A call matching
   :data:`orchestrata_ultimate.core.adapters.claude.gate_policy.SELF_EXCLUSION_SET`
   short-circuits straight to allow, before the interceptor is even
   constructed, so a bug in the governance path can never wedge the one
   path that would repair it.
2. **Bounded evaluation** (ADR-0032 #6). The interceptor call runs in a
   worker thread joined with the declared timeout — ``signal``-based
   alarms only work on the main thread, so a bounded wait is used instead.
3. **Outcome, not exception handler, drives the decision** (ADR-0032 #4).
   Malformed stdin, an interceptor exception, and a timeout are three
   distinct :class:`~orchestrata_ultimate.core.adapters.claude.gate_policy.GateOutcome`
   values — ``UNDETERMINED_MALFORMED``/``UNDETERMINED_ENGINE_ERROR``/
   ``UNDETERMINED_TIMEOUT`` — each logged distinctly to stderr and resolved
   through :func:`orchestrata_ultimate.core.adapters.claude.gate_policy.resolve`
   against this shim's declared
   :class:`~orchestrata_ultimate.core.adapters.claude.gate_policy.GateDeclaration`.
   There is no ``return 0`` fallthrough anywhere in this module.
4. **MODIFY is wired** (ADR-0032 #1). When the interceptor computes a
   mutation, this shim emits it via the documented hook ``updatedInput``
   key — never silently dropped. A ``MODIFY`` verdict with no mutation
   payload is itself treated as an engine error (never a silent bare
   allow).

This shim's own declaration defaults to **fail-closed** (ADR-0002), the
default :class:`~...gate_policy.GateDeclaration` posture. Set
``ORCHESTRATA_CLAUDE_GATE_FAILURE_POLICY=fail_open`` (with a non-empty
``ORCHESTRATA_CLAUDE_GATE_FAIL_OPEN_REASON``) to declare fail-open instead —
this is a reviewable, explicit override, never an implicit default.

Output protocol (Claude Code ``PreToolUse`` hook contract):
    allow (no mutation) -> exit 0, no stdout.
    allow (with mutation) -> exit 0, stdout is
             ``{"hookSpecificOutput": {"hookEventName": "PreToolUse",
             "permissionDecision": "allow",
             "permissionDecisionReason": "<reason>",
             "updatedInput": {...}}}``
    block -> exit 0, stdout is
             ``{"hookSpecificOutput": {"hookEventName": "PreToolUse",
             "permissionDecision": "deny",
             "permissionDecisionReason": "<reason>"}}``

Every non-ALLOW/BLOCK-by-verdict path is logged to stderr, never to stdout
(stdout is reserved for the hook JSON).

Optional audit trail: when ``ORCHESTRATA_CLAUDE_GATE_LOG`` is set, one JSON
line per decision (tool_name, decision, reason, ts) is appended there.
"""

from __future__ import annotations

import asyncio
import json
import os
import sys
import threading
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, TextIO

from orchestrata_ultimate.core.adapters.claude.gate_policy import (
    FailurePolicy,
    GateDeclaration,
    GateOutcome,
    matches_self_exclusion,
    resolve,
)
from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.core.adapters.claude.tool_interceptor import (
        InterceptorResult,
        ToolInterceptor,
    )

# ADR-0032 #6: this gate's own declared timeout, overridable per-invocation
# (never per-call — the declaration is data, read once per process).
_DEFAULT_TIMEOUT_SECONDS = 5.0
_GATE_ID = "claude_governance_gate"

# ADR-0032 #4: the two outcomes that are "undetermined because evaluation
# itself failed", as opposed to UNDETERMINED_MALFORMED (resolved earlier, in
# _read_call, before an interceptor is ever constructed).
_UNDETERMINED_ERROR_OUTCOMES = frozenset(
    {GateOutcome.UNDETERMINED_TIMEOUT, GateOutcome.UNDETERMINED_ENGINE_ERROR}
)


@traces(spec="025", reqs=["REQ-025-101"])
def _parsed_failure_policy(env: Mapping[str, str]) -> FailurePolicy:
    """Read ``ORCHESTRATA_CLAUDE_GATE_FAILURE_POLICY``; an unset or invalid
    value falls back to the ADR-0002 default (fail-closed), never fail-open.
    """
    policy_raw = env.get(
        "ORCHESTRATA_CLAUDE_GATE_FAILURE_POLICY", FailurePolicy.FAIL_CLOSED.value
    )
    try:
        return FailurePolicy(policy_raw)
    except ValueError:
        return FailurePolicy.FAIL_CLOSED


@traces(spec="025", reqs=["REQ-025-101"])
def _parsed_timeout_seconds(env: Mapping[str, str]) -> float:
    """Read ``ORCHESTRATA_CLAUDE_GATE_TIMEOUT_SECONDS``; an unset or
    non-positive/non-numeric value falls back to the declared default.
    """
    timeout_raw = env.get(
        "ORCHESTRATA_CLAUDE_GATE_TIMEOUT_SECONDS", str(_DEFAULT_TIMEOUT_SECONDS)
    )
    try:
        timeout_seconds = float(timeout_raw)
        if timeout_seconds <= 0:
            raise ValueError("timeout must be > 0")
    except ValueError:
        return _DEFAULT_TIMEOUT_SECONDS
    return timeout_seconds


@traces(spec="025", reqs=["REQ-025-101"])
def _load_declaration(env: Mapping[str, str]) -> GateDeclaration:
    """Build this shim's :class:`GateDeclaration` from its environment.

    The declaration is read as data (ADR-0032 #3). A misconfigured
    combination (e.g. FAIL_OPEN with no reason) must never crash the gate
    before it can even resolve an outcome -- that would be a worse fail-open
    than the one the reason requirement is meant to guard against, so it
    falls back to FAIL_CLOSED (ADR-0002's default posture) instead.

    Args:
        env: Environment mapping.

    Returns:
        This gate's declaration for the current invocation.
    """
    failure_policy = _parsed_failure_policy(env)
    timeout_seconds = _parsed_timeout_seconds(env)
    fail_open_reason = env.get("ORCHESTRATA_CLAUDE_GATE_FAIL_OPEN_REASON", "")

    try:
        return GateDeclaration(
            gate_id=_GATE_ID,
            timeout_seconds=timeout_seconds,
            failure_policy=failure_policy,
            fail_open_reason=fail_open_reason,
        )
    except ValueError as exc:
        print(
            f"claude_governance_gate: invalid gate declaration ({exc}) -- "
            "falling back to FAIL_CLOSED (ADR-0002 default)",
            file=sys.stderr,
        )
        return GateDeclaration(gate_id=_GATE_ID, timeout_seconds=timeout_seconds)


@traces(spec="025", reqs=["REQ-025-102"])
def _append_audit_log(
    log_path: str | None, tool_name: str, decision: str, reason: str
) -> None:
    """Append one JSONL audit line, if ``log_path`` is configured.

    Never raises — a broken audit sink must not affect the gate decision.

    Args:
        log_path: Destination path, or ``None``/empty to skip.
        tool_name: The gated tool's name.
        decision: ``"allow"`` or ``"block"``.
        reason: Human-readable reason for the decision.
    """
    if not log_path:
        return
    entry = {
        "tool_name": tool_name,
        "decision": decision,
        "reason": reason,
        "ts": time.time(),
    }
    try:
        with open(log_path, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(entry) + "\n")
    except OSError as exc:
        print(f"claude_governance_gate: audit log write failed: {exc}", file=sys.stderr)


@traces(spec="025", reqs=["REQ-025-100"])
def _emit_deny(reason: str) -> None:
    """Print the Claude Code deny-decision JSON envelope to stdout."""
    print(
        json.dumps(
            {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": reason,
                }
            }
        )
    )


def _emit_allow_with_update(reason: str, updated_input: dict[str, Any]) -> None:
    """Print the Claude Code allow-with-mutation JSON envelope to stdout.

    ADR-0032 #1: this is how a computed ``MODIFY`` result actually reaches
    the tool call — never a silent drop.
    """
    print(
        json.dumps(
            {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "allow",
                    "permissionDecisionReason": reason,
                    "updatedInput": updated_input,
                }
            }
        )
    )


@traces(spec="025", reqs=["REQ-025-099"])
def _outcome_for_decision_value(decision_value: str) -> GateOutcome:
    """Map a raw ``InterceptorDecision`` string value to a :class:`GateOutcome`.

    Takes the *value*, not the enum, so this module never needs
    ``InterceptorDecision`` imported at call sites outside the worker
    thread's own deferred import — that import stays inside the try/except
    that turns an import failure into ``UNDETERMINED_ENGINE_ERROR`` rather
    than crashing the whole shim (ADR-0032 #4).

    Args:
        decision_value: ``InterceptorResult.decision.value``.

    Returns:
        The corresponding :class:`GateOutcome`.
    """
    if decision_value == "allow":
        return GateOutcome.ALLOW
    if decision_value == "block":
        return GateOutcome.BLOCK
    if decision_value == "modify":
        return GateOutcome.MODIFY
    # "delay" is declared on InterceptorDecision but no ToolInterceptor code
    # path ever produces it today; treat any unrecognised/unimplemented
    # decision conservatively as an engine error rather than silently
    # allowing it through.
    return GateOutcome.UNDETERMINED_ENGINE_ERROR


@traces(spec="025", reqs=["REQ-025-099"])
class _InterceptorRun(threading.Thread):
    """Runs one ``ToolInterceptor.intercept()`` call to completion on a
    dedicated thread, so :func:`_evaluate` can bound it with a join timeout
    (ADR-0032 #6) without a ``signal``-based alarm (main-thread only, and
    Claude Code hooks are not guaranteed to run on the main thread of a
    reusable interpreter).

    Daemonized: if the call never returns (the timeout path), the thread is
    abandoned rather than blocking process exit.
    """

    def __init__(
        self,
        tool_name: str,
        tool_input: dict[str, Any],
        session_id: str | None,
        interceptor_factory: type[ToolInterceptor] | None = None,
    ) -> None:
        super().__init__(daemon=True)
        self._tool_name = tool_name
        self._tool_input = tool_input
        self._session_id = session_id
        self._interceptor_factory = interceptor_factory
        self.result: InterceptorResult | None = None
        self.error: Exception | None = None

    def run(self) -> None:
        """Construct a fresh interceptor and run one ``intercept()`` call,
        capturing either a result or an exception (never re-raising)."""
        try:
            from orchestrata_ultimate.core.adapters.claude.tool_interceptor import (
                ToolInterceptor,
            )

            factory = self._interceptor_factory or ToolInterceptor
            interceptor = factory()
            self.result = asyncio.run(
                interceptor.intercept(
                    self._tool_name, self._tool_input, self._session_id
                )
            )
        except Exception as exc:  # noqa: BLE001 - captured, not swallowed
            self.error = exc


@traces(spec="025", reqs=["REQ-025-099"])
def _evaluate(
    tool_name: str,
    tool_input: dict[str, Any],
    session_id: str | None,
    timeout_seconds: float,
    interceptor_factory: type[ToolInterceptor] | None = None,
) -> tuple[GateOutcome, InterceptorResult | None, Exception | None]:
    """Run the interceptor bounded by ``timeout_seconds``; classify the result.

    Args:
        tool_name: The tool being called.
        tool_input: The tool's argument mapping.
        session_id: Optional session identifier.
        timeout_seconds: The declared timeout (ADR-0032 #6).
        interceptor_factory: Real ``ToolInterceptor`` constructor to use —
            injection seam for tests that need a pre-configured (e.g.
            transformer-registered) real interceptor. Defaults to the
            production ``ToolInterceptor`` with no engine/transformers
            attached.

    Returns:
        See :func:`_classify_worker`.
    """
    worker = _InterceptorRun(tool_name, tool_input, session_id, interceptor_factory)
    worker.start()
    worker.join(timeout_seconds)
    return _classify_worker(worker)


@traces(spec="025", reqs=["REQ-025-099"])
def _classify_worker(
    worker: _InterceptorRun,
) -> tuple[GateOutcome, InterceptorResult | None, Exception | None]:
    """Classify a joined ``_InterceptorRun`` into a gate-outcome triple.

    Exactly one of ``result``/``error`` is non-``None`` when the outcome is
    DETERMINED or ``UNDETERMINED_ENGINE_ERROR`` respectively; both are
    ``None`` for ``UNDETERMINED_TIMEOUT``.
    """
    if worker.is_alive():
        return GateOutcome.UNDETERMINED_TIMEOUT, None, None
    if worker.error is not None:
        return GateOutcome.UNDETERMINED_ENGINE_ERROR, None, worker.error

    result = worker.result
    if result is None:  # pragma: no cover - defensive; thread always sets one
        return (
            GateOutcome.UNDETERMINED_ENGINE_ERROR,
            None,
            RuntimeError(
                "interceptor thread finished with neither a result nor an error"
            ),
        )
    return _outcome_for_decision_value(result.decision.value), result, None


@traces(spec="025", reqs=["REQ-025-100", "REQ-025-101", "REQ-025-102"])
def _resolve_and_emit(
    outcome: GateOutcome,
    declaration: GateDeclaration,
    tool_name: str,
    log_path: str | None,
    detail: str,
) -> int:
    """Log, resolve, audit, and (if resolved to block) emit deny — for every
    outcome that is not a bare, mutation-free ALLOW.

    This is the one place ``resolve()`` is called from this module, and the
    one place an outcome's stderr line is written — ADR-0032 #4's "each
    recorded distinctly".
    """
    decision = resolve(outcome, declaration)
    print(
        f"claude_governance_gate: outcome={outcome.value} "
        f"policy={declaration.failure_policy.value} -> {decision}: {detail}",
        file=sys.stderr,
    )
    reason = f"{outcome.value}: {detail}"
    _append_audit_log(log_path, tool_name, decision, reason)
    if decision == "block":
        _emit_deny(reason)
    return 0


@traces(
    spec="025",
    reqs=[
        "REQ-025-098",
        "REQ-025-099",
        "REQ-025-100",
        "REQ-025-102",
    ],
)
def main(
    stdin: TextIO | None = None,
    env: Mapping[str, str] | None = None,
    interceptor_factory: type[ToolInterceptor] | None = None,
) -> int:
    """Read one hook payload from stdin, gate it, emit the hook decision.

    See the module docstring for the full ADR-0032 contract. ``stdin``/
    ``env`` default to ``sys.stdin``/``os.environ``; ``interceptor_factory``
    is a real-``ToolInterceptor``-constructor injection seam for tests
    (e.g. a transformer- or deny-list-preconfigured real interceptor) —
    production callers never pass it. Always returns 0: allow and block are
    both expressed via stdout content, per the hook protocol.
    """
    active_stdin: TextIO = stdin if stdin is not None else sys.stdin
    env = env if env is not None else os.environ
    log_path = env.get("ORCHESTRATA_CLAUDE_GATE_LOG")
    declaration = _load_declaration(env)

    call = _read_call(active_stdin, declaration, log_path)
    if isinstance(call, int):
        return call
    tool_name, tool_input, session_id = call

    if matches_self_exclusion(tool_name, tool_input):
        return _emit_self_exclusion(tool_name, log_path)

    outcome, result, error = _evaluate(
        tool_name,
        tool_input,
        session_id,
        declaration.timeout_seconds,
        interceptor_factory,
    )
    if outcome in _UNDETERMINED_ERROR_OUTCOMES:
        return _emit_undetermined_error(
            outcome, declaration, tool_name, log_path, error
        )

    assert result is not None  # every remaining outcome carries a result
    return _emit_determined(outcome, result, declaration, tool_name, log_path)


@traces(spec="025", reqs=["REQ-025-098"])
def _parse_payload(raw: str) -> dict[str, Any]:
    """Parse the hook's stdin payload; raise on anything malformed.

    Raises:
        json.JSONDecodeError: ``raw`` is not valid JSON.
        ValueError: ``raw`` parsed to something other than a JSON object.
    """
    payload = json.loads(raw)
    if not isinstance(payload, dict):
        raise ValueError("stdin JSON was not an object")
    return payload


@traces(spec="025", reqs=["REQ-025-098", "REQ-025-101"])
def _coerce_tool_input(
    tool_name: str,
    tool_input_raw: Any,
    declaration: GateDeclaration,
    log_path: str | None,
) -> dict[str, Any] | int:
    """Validate ``payload["tool_input"]`` is a mapping, or resolve a
    malformed value straight to an exit code.

    A non-dict ``tool_input`` (a string, list, number, ...) must never reach
    ``matches_self_exclusion``/the interceptor unvalidated -- both assume a
    mapping. Resolving this as ``UNDETERMINED_MALFORMED`` here (rather than
    letting an AttributeError/TypeError escape ``main()`` uncaught) is what
    keeps a malformed payload fail-closed instead of crashing the whole
    process into an unintended fail-open exit.

    Returns:
        The coerced ``dict`` (``{}`` for absent/``None``), or an
        already-resolved process exit code if ``tool_input_raw`` was present
        but not a JSON object.
    """
    if tool_input_raw is None:
        return {}
    if isinstance(tool_input_raw, dict):
        return tool_input_raw
    detail = f"tool_input was not a JSON object (got {type(tool_input_raw).__name__})"
    return _resolve_and_emit(
        GateOutcome.UNDETERMINED_MALFORMED,
        declaration,
        tool_name=tool_name if isinstance(tool_name, str) else "",
        log_path=log_path,
        detail=detail,
    )


@traces(spec="025", reqs=["REQ-025-098", "REQ-025-101"])
def _read_call(
    active_stdin: TextIO, declaration: GateDeclaration, log_path: str | None
) -> tuple[str, dict[str, Any], str | None] | int:
    """Read + parse the hook payload; extract the call, or resolve a
    malformed payload straight to an exit code.

    Returns:
        Either the ``(tool_name, tool_input, session_id)`` call, or an
        already-resolved process exit code if the payload was malformed.
    """
    try:
        payload = _parse_payload(active_stdin.read())
    except (json.JSONDecodeError, ValueError) as exc:
        return _resolve_and_emit(
            GateOutcome.UNDETERMINED_MALFORMED,
            declaration,
            tool_name="",
            log_path=log_path,
            detail=f"malformed stdin JSON: {exc}",
        )
    tool_name = payload.get("tool_name", "")
    tool_input = _coerce_tool_input(
        tool_name, payload.get("tool_input"), declaration, log_path
    )
    if isinstance(tool_input, int):
        return tool_input
    session_id = payload.get("session_id")
    return tool_name, tool_input, session_id


@traces(spec="025", reqs=["REQ-025-101"])
def _emit_undetermined_error(
    outcome: GateOutcome,
    declaration: GateDeclaration,
    tool_name: str,
    log_path: str | None,
    error: Exception | None,
) -> int:
    """Emit + resolve a timeout or engine-error outcome (never MALFORMED,
    which is resolved earlier in :func:`_read_call`)."""
    detail = (
        f"evaluation exceeded {declaration.timeout_seconds}s timeout"
        if outcome is GateOutcome.UNDETERMINED_TIMEOUT
        else f"interceptor error: {error}"
    )
    return _resolve_and_emit(outcome, declaration, tool_name, log_path, detail)


@traces(spec="025", reqs=["REQ-025-102"])
def _emit_self_exclusion(tool_name: str, log_path: str | None) -> int:
    """ADR-0032 #5: log + audit a self-exclusion short-circuit; always allow."""
    print(
        f"claude_governance_gate: outcome=self-exclusion tool={tool_name!r} "
        "-> allow: matches SELF_EXCLUSION_SET",
        file=sys.stderr,
    )
    _append_audit_log(log_path, tool_name, "allow", "self-exclusion")
    return 0


@traces(spec="025", reqs=["REQ-025-100", "REQ-025-102"])
def _emit_determined(
    outcome: GateOutcome,
    result: InterceptorResult,
    declaration: GateDeclaration,
    tool_name: str,
    log_path: str | None,
) -> int:
    """Emit the hook decision for a DETERMINED outcome (ALLOW/BLOCK/MODIFY)."""
    if outcome is GateOutcome.BLOCK:
        decision = resolve(outcome, declaration)
        _append_audit_log(log_path, tool_name, decision, result.reason)
        _emit_deny(result.reason)
        return 0

    if outcome is GateOutcome.MODIFY:
        if result.modified_input is None:
            # ADR-0032 #1: a MODIFY verdict with no mutation payload is
            # itself a gate failure -- never silently drop it, never
            # silently fall through to a bare allow.
            return _resolve_and_emit(
                GateOutcome.UNDETERMINED_ENGINE_ERROR,
                declaration,
                tool_name,
                log_path,
                detail="MODIFY decision carried no modified_input",
            )
        decision = resolve(outcome, declaration)
        _append_audit_log(log_path, tool_name, decision, result.reason)
        _emit_allow_with_update(result.reason, result.modified_input)
        return 0

    # ALLOW
    decision = resolve(outcome, declaration)
    _append_audit_log(log_path, tool_name, decision, result.reason)
    return 0


if __name__ == "__main__":
    sys.exit(main())