intutic_clawde.gate

Pre-execution tool gate for agent frameworks without a shipped Intutic harness.

Intutic ships PreToolUse hooks for the harnesses in the sync-daemon's gate registry (packages/shared-types/src/gateKind.ts). LangGraph — and most DIY Python agent loops — are not among them. This subpackage is the missing adapter, written against Intutic's own published gate contract:

• it reads the same ~/.intutic/hooks/policy-snapshot.rules artifact the sync-daemon compiles for every shipped harness (snapshot),
• it applies SOP rules authored in the product, including the WHERE argument clause (soprules, a mirror of the control plane's matchSopRule),
• it checks container image provenance on deploy commands (imagecheck),
• and it consults POST /api/v1/hook-gate on the control plane (client), reporting every decision via /api/v1/hook-events.

Framework integration is deliberately framework-agnostic: the @guard decorator wraps any callable tool, guard_tools wraps a list of LangChain/LangGraph-style tool objects by duck typing (no langchain import), and CrewAI / AutoGen tools — plain callables — work through the same decorator. intutic_headers returns the headers a ChatOpenAI/OpenAI client pointed at the Intutic proxy should send for trace attribution.

On deny the gate raises IntuticGateRefusal, whose message follows the same [Intutic Governance] BLOCKED: ... python-raise contract as the Open WebUI filter, so anything that already recognises that family recognises this one. A hold rule raises its subclass IntuticGateHold ([Intutic Governance] HELD: ...), carrying the hold's id.

Classes

Gate

class Gate(cfg: GateConfig, client: Optional[GateClient] = None)

The pre-execution gate. guard() evaluates one tool call against the local tiers (SSO group policy, policy snapshot, MCP servers, image integrity) and, with a GateClient, the control-plane tiers (SOP rules, hook gate), and raises IntuticGateRefusal to refuse it. Without a client only the local tiers run.

Methods

Gate.guard()

def guard(tool_name: str, tool_input: dict) -> None

Raise IntuticGateRefusal if this call must not run. Return to allow.

Under a chaos run's plan, the call is reported to the run's events file whatever the tiers decide, and a call they allowed can still be refused by the fault that is due (CHAOS_FAULT, or IntuticGateHold for a held one). A plan that is set but unreadable refuses every call: an experiment run without the faults it asked for reports a resilience it never measured.

Gate.policy()

def policy() -> dict

Load the image allowlist.

A missing or malformed allowlist is NOT treated as "allow everything". An empty policy with require_digest still refuses every unpinned image, which is the safe direction; a policy we cannot read at all raises, and the caller turns that into a block.

Gate.snapshot()

def snapshot() -> snapshot.Snapshot

Gate.sop_rules()

def sop_rules() -> list

Active SOP rules, fetched once per process.

Deliberately not refreshed on a timer the way policy.ts does: a rule set that changes mid-run makes a verdict depend on when the fetch landed. Construct a new Gate (or clear _sop_rules) to refresh.

GateClient

class GateClient(
    base_url: Optional[str] = None,
    api_key: str = '',
    workspace_id: str = '',
    session_id: str = '',
    harness: str = 'langgraph',
    fail_closed: bool = True,
    timeout: float = 2.0,
    transport: Optional[Transport] = None,
    get_transport: Optional[GetTransport] = None,
)

Control-plane client for the pre-execution gate.

base_url is the control plane's address: INTUTIC_CONTROL_PLANE_URL, then Intutic's hosted control plane. Without an api_key the client is local_only. harness defaults to "langgraph" and is sent as harnessType on every gate call and event, so incidents attribute to the right adapter. Override it when wrapping a different framework. fail_closed blocks a call when the hook gate gives no verdict (the control plane cannot be reached, answers without one, or answers GOVERNANCE_UNAVAILABLE); False lets such a call go ahead. timeout is how long one control-plane request may take, in seconds. transport and get_transport replace the HTTP POST and GET calls, for tests.

Attributes

Name	Type	Default	Description
local_only	bool	—	True when there is no API key. Such a client cannot authenticate, so it sends nothing: the control-plane tiers (SOP register, hook gate) are skipped, events and holds are not reported, and the local tiers decide.

Methods

GateClient.approved_bypasses()

def approved_bypasses() -> Optional[list]

The workspace's approved bypasses (GET /api/v1/decisions/approved-bypasses), or None when they cannot be read. Never raises: a hold that cannot be checked for an approval stays held.

GateClient.emit()

def emit(
    event: str,
    tool_name: str,
    reason: str = '',
    tool_input: Optional[dict] = None,
    incident_id: Optional[str] = None,
    file_path: Optional[str] = None,
    gate_decision_id: Optional[str] = None,
) -> bool

Fire-and-forget telemetry. Never raises.

gate_decision_id is the hook gate's decisionId for the call this event reports: the control plane then files the decision's incident once and does not export it to SIEM a second time.

A tool_blocked event creates a governance_incidents row whose description embeds our reason verbatim:

[Hook Gate] Tool "shell" blocked by local SOP enforcement. Reason: <reason> Harness: langgraph

That string is what appears on the incidents dashboard.

GateClient.from_env()

@classmethod
def from_env(
    session_id: Optional[str] = None,
    harness: str = 'langgraph',
    fail_closed: bool = True,
) -> 'GateClient'

Build a client from the environment, falling back to CLI credentials.

Reading ~/.intutic/credentials.json means the gate works with whatever workspace intutic login already bound, instead of duplicating the key. With no key in either place the client is local_only.

GateClient.hook_gate()

def hook_gate(tool_name: str, tool_input: dict) -> GateResponse

Synchronous allow/deny for one tool call.

Worth being precise about: the endpoint matches ~19 DLP regexes against the serialised toolInput, plus SOP rules. It is defence in depth on top of the local tiers, not a replacement for them.

When there is no verdict (the control plane cannot be reached, answers non-2xx or without a boolean allowed, or answers GOVERNANCE_UNAVAILABLE because its own checks could not run), the verdict is decided by fail_closed: True (the default) reports BLOCK, because a client with no verdict has nothing to defer to; False reports allow. A 400 INVALID_REQUEST is a BLOCK whatever fail_closed says. A shadow-mode workspace's {allowed: true, degraded: true} is an allow.

A local_only client makes no call and answers allow with code LOCAL_ONLY, whatever fail_closed says: with no key there is no control-plane tier to fail, only local ones, which have already run.

GateClient.judge_finalize()

def judge_finalize(
    full_content: str,
    monitored_model: str,
    personal_sops: Optional[list] = None,
) -> dict

Grade generated text against the workspace's active SOP.

POST /api/v1/judge/finalize. Note that plain /api/v1/judge does NOT exist — it is a JWT-bypass prefix in PUBLIC_ROUTES, not a handler, and returns 404.

The endpoint interpolates the active SOP's markdown verbatim into the judge's system prompt, which is what makes it usable as a general "does this text contradict this document" check. The SOP must be VALIDATED and active; only one is ever loaded (newest by updatedAt).

Returns {verdict, triggered, personalTriggered, correctionSummary, independent}. verdict is 'PASS' | 'TRIGGERED' | 'REVIEW' from the control plane, or 'UNAVAILABLE' — set by the control plane on its own failures (it returns HTTP 503, which raises in _post and lands in the except below) and by this method on transport errors. Branch on verdict, and treat only 'PASS' as clean:

• 'TRIGGERED': a violation was found.
• 'REVIEW': the response was held for human review. The typed judge was unsure, or did not answer in time (the judge GPU was busy). No org verdict exists yet: triggered is False because nothing was decided, not because the response is clean. A fail-closed caller must treat REVIEW as not clean. The control plane records the interim verdict and the reviewer's ruling on its side.
• 'UNAVAILABLE' (or any other value): NO grading happened.

This cannot block. lib/monitorModel.ts is explicit: no LLM verdict blocks a request; the KILL sites in the proxy are deterministic. The caller must branch on the verdict itself.

History worth keeping: the endpoint used to fail OPEN — every internal failure (LiteLLM down, 30s timeout, unparsable verdict) returned HTTP 200 {triggered: false}, indistinguishable from a clean pass, and this client's except-branch only caught transport errors. The control plane now 503s with verdict:'UNAVAILABLE' on every failure mode, which routes them all through the except below — so a fail-closed caller finally covers what it always claimed to.

GateClient.record_hold()

def record_hold(hold: Dict[str, Any]) -> bool

Record one hold in the review queue (POST /api/v1/decisions), the way the hook gates' records reach it. Never raises; False when the control plane did not accept it, so the refusal can say there is nothing to approve yet.

GateConfig

@dataclass
class GateConfig(
    repo_root: str = '.',
    workspace_id: str = '',
    allowlist_path: str = '.intutic/image-allowlist.json',
    enforce: bool = True,
    use_hook_gate: bool = True,
    use_sop_rules: bool = True,
)

How a Gate evaluates calls. Every field has a default.

Attributes

Name	Type	Default	Description
allowlist_path	str	'.intutic/image-allowlist.json'	The image allowlist, relative to repo_root unless absolute.
enforce	bool	True	False turns Gate.guard() into a no-op.
repo_root	str	'.'	Where relative manifest and allowlist paths resolve from.
use_hook_gate	bool	True	Call the control plane's hook gate (needs a client).
use_sop_rules	bool	True	Fetch and apply the workspace's SOP rules (needs a client).
workspace_id	str	''	Compared with the workspace the policy snapshot was compiled for, and scopes the SOP-rule fetch.

Methods

GateConfig.allowlist_abs()

def allowlist_abs() -> str

GateResponse

@dataclass
class GateResponse(
    allowed: bool,
    reason: str = '',
    incident_id: Optional[str] = None,
    reached: bool = True,
    code: Optional[str] = None,
    decision_id: Optional[str] = None,
)

Attributes

Name	Type	Default	Description
allowed	bool	—
code	Optional[str]	None
decision_id	Optional[str]	None
incident_id	Optional[str]	None
reached	bool	True
reason	str	''

ImageVerdict

@dataclass
class Verdict(
    ok: bool,
    code: str = '',
    detail: str = '',
    images: list[ImageRef] = list(),
)

Attributes

Name	Type	Default	Description
code	str	''
detail	str	''
images	list[ImageRef]	field(default_factory=list)
ok	bool	—
reason	str	—	The single line an operator sees on the dashboard.

Defined as intutic_clawde.gate.imagecheck.Verdict.

IntuticGateHold

class IntuticGateHold(reason: str, hold_id: Optional[str])

Bases: IntuticGateRefusal

Raised when a hold rule stopped the call to ask a person first (code HELD). A subclass, so a caller that stops on every refusal stops on this too; one that tells the user about holds catches it first.

hold_id names the hold in the review queue, and is None when the hold could not be recorded (no control plane to record it in), in which case there is nothing to approve yet. The message starts [Intutic Governance] HELD:, as the hook gates print a hold, and says who can approve it and when a retry passes. See hold.py.

IntuticGateRefusal

class IntuticGateRefusal(reason: str, code: str, incident_id: Optional[str] = None)

Bases: ClawdeError

Raised when a tool call must not run.

The python-raise contract: the exception's message begins [Intutic Governance] BLOCKED: — the same family the Open WebUI filter raises — so harnesses and log scrapers that already recognise that prefix recognise this refusal too. The structured fields (reason, code, one of GATE_REFUSAL_CODES, and incident_id) carry the machine-readable version.

Snapshot

@dataclass
class Snapshot(
    rules: list[Rule] = list(),
    digest: str = 'none',
    state: str = 'absent',
    workspace_id: str = '',
    generated_at: str = '',
    age_days: int = 0,
    dropped_rules: int = 0,
    sso_groups: Optional[sso.SsoGroupRecord] = None,
    mcp_registry: Optional[dict] = None,
    mcp_allowlist: Optional[dict] = None,
    opaque: bool = False,
)

Attributes

Name	Type	Default	Description
age_days	int	0
digest	str	'none'
dropped_rules	int	0
generated_at	str	''
health_message	str	—
mcp_allowlist	Optional[dict]	None
mcp_registry	Optional[dict]	None
opaque	bool	False
rules	list[Rule]	field(default_factory=list)
sso_groups	Optional[sso.SsoGroupRecord]	None
state	str	'absent'
workspace_id	str	''

SopRegister

@dataclass
class SopRegister(rules: list, opaque: bool)

The register as GET /api/v1/sop/rules answers it: the rules, and whether the workspace's intervention mode is OPAQUE (interventionMode; absent from a control plane that predates it, which reads as not OPAQUE). The gate uses the mode when there is no policy snapshot to read it from.

Attributes

Name	Type	Default	Description
opaque	bool	—
rules	list	—

SopRule

@dataclass(frozen=True)
class SopRule(
    id: str,
    tool_pattern: str,
    action: str,
    reason: str,
    arg_pattern: str | None = None,
)

Attributes

Name	Type	Default	Description
action	str	—
arg_pattern	str | None	None
id	str	—
reason	str	—
tool_pattern	str	—

Methods

SopRule.matches()

def matches(tool_name: str, tool_input_json: str) -> bool

Mirror of matchSopRule, including its silent-skip on bad regex.

Functions

active()

def active() -> Optional[Gate]

The process-wide gate install set, or None.

canonical_json()

def canonical_json(value: Any) -> str

JSON with object keys sorted at every level, so key order never changes a hash — byte for byte what the TypeScript gates' canonicalJson writes.

classify()

def classify(tool_name: str, tool_input) -> list[str]

Classify one tool call into the abstract actions it performs.

Returns prefixed tokens (action:deploy, ...) in the order the Rust emits them. Order matters: the ordering detectors read position in the sequence.

fetch_register()

def fetch_register(
    base_url: str,
    api_key: str,
    workspace_id: str,
    timeout: float = 3.0,
) -> SopRegister | None

fetch_rules with the workspace's mode; None if the register did not answer.

first_match()

def first_match(
    rules: list[SopRule],
    tool_name: str,
    tool_input: dict,
) -> SopRule | None

The rule that decides this call: the most severe one matching it (block, then require_approval, then warn), the first of that severity in control-plane order. None when none matches.

guard()

def guard(
    func: Optional[Callable] = None,
    *,
    name: Optional[str] = None,
    gate: Optional[Gate] = None,
) -> Callable

Decorator: evaluate the gate before the tool body runs.

Usage:

@guard
def shell(command: str) -> str: ...

@guard(name="shell")           # explicit tool name
def run_command(command: str) -> str: ...

On deny the wrapper raises IntuticGateRefusal and the body never runs; the deny has already been reported via hook-events by the gate. gate overrides the process-wide instance registered with install(); without either, calling the wrapped tool raises RuntimeError rather than silently running unguarded.

guard_tools()

def guard_tools(tools: Iterable[Any], *, gate: Optional[Gate] = None) -> list

Wrap a list of tools (LangChain/LangGraph/smolagents tool objects, or plain callables) so each is gated before execution.

Duck-typed on purpose — no langchain/smolagents import, no hard dependency:

• an object with a callable .func (StructuredTool, @tool-created Tool) gets its .func wrapped in place;
• otherwise an object with ._run (LangChain BaseTool subclass) gets ._run wrapped in place (set via object.setattr, so pydantic models that restrict attribute assignment still work);
• otherwise an object with .forward (smolagents Tool subclass — Tool.__call__ dispatches to self.forward(...)) gets .forward wrapped in place, the same way. Added when the smolagents adapter (adapters/smolagents.py) was built: without this branch, a smolagents Tool instance is still callable (it has __call__), so it fell into the plain-callable branch below and got REPLACED by a bare wrapper function — silently dropping the .name/.inputs/ schema attributes ToolCallingAgent needs to expose it to the model. The .forward branch keeps the original Tool object (and its schema) intact, only swapping its execution method, matching the ._run branch's contract exactly;
• a plain callable is returned wrapped by @guard — which is also how CrewAI and AutoGen tools are guarded, directly with the decorator.

Already-guarded entries pass through untouched, so calling this twice does not double-gate. Returns the same objects (mutated in place) in order.

hold_key()

def hold_key(tool_name: str, tool_input: Any) -> tuple[str, str]

The bypass key's call half: (tool name normalised, hash of its arguments).

hold_message()

def hold_message(rule_reason: str, rule_id: str, outcome: HoldOutcome) -> str

What the agent is told — the one place this package words a hold. Says who can approve, and that a retry passes only with the bypass setting on, because "retry after approval" alone is false under the default settings. An empty rule_id names no rule: an OPAQUE workspace's hold (opaque.py).

install()

def install(gate: Gate) -> None

Set the process-wide gate that @guard and guard_tools use.

intutic_headers()

def intutic_headers(
    session_id: Optional[str] = None,
    workspace_id: Optional[str] = None,
    harness: str = 'langgraph',
) -> Dict[str, str]

Default headers for a ChatOpenAI/OpenAI client pointed at the Intutic proxy.

x-session-id matters: the proxy defaults an unset one to the literal "unknown", which merges every run into a single dashboard session. x-intutic-harness attributes traces to this adapter — the proxy honours it for trace attribution (unknown headers are ignored by older proxies, so it is always safe to send).

Usage:

ChatOpenAI(base_url=..., default_headers=intutic_headers(session_id=run_id))

is_deploy()

def is_deploy(tool_name: str, tool_input) -> bool

True when this call would deploy — the trigger for the image check.

is_test()

def is_test(tool_name: str, tool_input) -> bool

True when this call runs a test suite.

load_snapshot()

def load_snapshot(workspace_id: str = '', path: str | None = None) -> Snapshot

Read the policy snapshot the sync daemon compiled: from path, else INTUTIC_SNAPSHOT_RULES, else ~/.intutic/hooks/policy-snapshot.rules.

A missing file gives an empty snapshot whose state is "absent". A snapshot whose digest does not match its rules, or that was compiled for a workspace other than a non-empty workspace_id, is "invalid": its rules and MCP records are dropped.

parse_register()

def parse_register(payload: object) -> SopRegister

parse_rules, and the interventionMode beside them.

parse_rules()

def parse_rules(payload: object) -> list[SopRule]

Accept the rule list in whichever envelope the control plane used.

/api/v1/sop/rules returns {rules:[...]}, while sibling list endpoints variously return {items:[...]} (camelCase) and {data:[...]} (snake_case). Reading the wrong key yields zero rules and a tier that enforces nothing while looking healthy — the worst failure a gate can have — so all three are accepted rather than assumed.

supports_arg_patterns()

def supports_arg_patterns(rules: list[SopRule]) -> bool

True when the control plane emitted at least one usable argPattern.

A control plane predating intutic-enterprise#14 cannot produce one, so this doubles as a deployment probe — surface it rather than letting the tier be silently inert.

touches_infra()

def touches_infra(path: str) -> bool

True when a written path is infrastructure-as-code.

Constants

ARGUMENTS_SIZE_LIMIT

ARGUMENTS_SIZE_LIMIT = 1024 * 1024

The largest tool arguments, in UTF-8 bytes of compact JSON, the gate evaluates.

COMMAND_SIZE_LIMIT

COMMAND_SIZE_LIMIT = 256 * 1024

The largest shell command, in UTF-8 bytes, the gate evaluates.

GATE_REFUSAL_CODES

GATE_REFUSAL_CODES = ...

Every code an IntuticGateRefusal can carry: the tier that refused, or for the image-integrity tier the specific failure. Held to packages/shared-types/fixtures/refusal-codes.json (gate) by a test, as the TypeScript gate's list and the gate SDK reference are.

READ_ONLY_TOOLS

READ_ONLY_TOOLS = frozenset({'read_file', 'list_files', 'read', 'cat', 'view'})