""" gate_classifier.py Decision-layer gate classification for agent governance. Extends Microsoft AGT's BaseIntegration with a gate classifier that sits between the POLICY_CHECK event and action execution. Gate classification answers a different question than policy enforcement: PolicyInterceptor asks: "Is this action permitted?" GateClassifier asks: "What category of decision is this, and what level of human deliberation does it require?" These are adjacent but not identical questions. Both are necessary for institutional accountability. Integration pattern: Agent → tool call intent → BaseIntegration.pre_execute() [AGT: POLICY_CHECK event, allow/deny] → GateClassifier.classify() [This extension: Gate 1–4 classification] → GateRecord.write() [Immutable artifact before execution] → Action executes (or escalates) The gate record is written BEFORE execution. An artifact written after the fact is a narrative. An artifact written before execution is evidence. Gate classification must be defined by the institution and applied by the governance layer — not generated by the model being governed. Self- certification is not governance. Author: Mark Julius (mj3b) License: Apache 2.0 Prior art / related projects: - Microsoft AGT BaseIntegration, GovernancePolicy, GovernanceEventType (packages/agent-os/src/agent_os/integrations/base.py, MIT License) - Bovens' accountability theory (forum, actor, standard, rendering-account) - ARAF v3.0 reconstructability principle - RGDS (Regulated Gate Decision Support): github.com/mj3b/rgds """ from __future__ import annotations import hashlib import json import logging import uuid from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from enum import Enum from typing import Any, Callable, Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Gate taxonomy # --------------------------------------------------------------------------- class Gate(str, Enum): """ Four-level governance classification for agent decisions. A gate is not a threshold. A threshold is a number. A gate is a governance classification that carries institutional weight. Gate classification answers: what level of human deliberation does this decision require before or after execution? Classification order: most restrictive first. """ ROUTINE = "gate_1_routine" """ Decision falls within established parameters with clear precedent and low consequence if wrong. No human deliberation required at decision time. Documentation is automated. """ DOCUMENTED_DELEGATION = "gate_2_documented_delegation" """ Decision exceeds routine parameters or carries meaningful consequence, but falls within pre-authorized scope defined upstream by a human authority. Human deliberation occurred upstream; the artifact records which delegation was invoked. """ ELEVATED_REVIEW = "gate_3_elevated_review" """ Decision is novel, consequential, or sits at the boundary of authorized scope. A human must review the artifact before or immediately after execution. Required deliberative moment, not optional notification. """ HARD_ESCALATION = "gate_4_hard_escalation" """ Decision exceeds authorized scope, involves irreversible action, or presents conditions the system cannot classify with sufficient confidence. Execution stops. Human authority required before proceeding. """ # --------------------------------------------------------------------------- # Decision context: structured capture of observable inputs # --------------------------------------------------------------------------- @dataclass class DecisionContext: """ Structured capture of what the governance layer could observe about the agent's decision at classification time. Epistemic boundary: this is NOT a claim about the agent's internal state, intentions, or reasoning chain. Those are opaque. This captures what was observable from the governance layer — the surface of the decision that can be inspected without access to model internals. The distinction matters for accountability. Claiming to capture "why the model decided" is an epistemic overclaim. Capturing "what the governance layer observed at decision time" is verifiable and defensible. """ # What the agent was attempting: the tool and its arguments. # This is directly observable — it is what triggered the policy check. intended_action: str # tool_name, human-readable action_parameters: dict # tool_args, the specific parameters # Task context: what the agent declared it was trying to accomplish. # Provided by the caller from agent state (e.g., LangGraph node context, # LangChain callback metadata). Optional — not all frameworks expose this. declared_task: Optional[str] # Confidence distribution: not just the top score, but how confidence # was distributed across alternatives. A decision at 0.51 against five # alternatives is categorically different from 0.97 against one. confidence_score: Optional[float] alternatives_considered: list[str] confidence_distribution: Optional[dict[str, float]] # {alternative: score} # Observable conditions at decision time: agent state, session context, # any structured metadata the caller provides. observable_conditions: dict[str, Any] # --------------------------------------------------------------------------- # Plain-language reconstruction generator # --------------------------------------------------------------------------- class ReasoningReconstructor: """ Generates a plain-language reconstruction of the governance-relevant decision surface. Design principle: reconstruction is deterministic and derived entirely from structured observable inputs. It does not call any model, does not access agent internals, and makes no claims about why the agent chose this action. What it produces: a sentence or short paragraph a non-technical reviewer can read to understand what governance decision was made, what triggered it, and what the governance layer could observe at decision time. This satisfies the ARAF v3.0 reconstructability requirement: the decision can be reconstructed from contemporaneous governance records without requiring access to model internals or post-hoc inference. """ @staticmethod def reconstruct( gate: "Gate", gate_rationale: str, decision_context: DecisionContext, policy_allowed: bool, escalation_trigger: Optional[str], delegation_reference: Optional[str], ) -> str: """ Produce a plain-language reconstruction from observable structured fields. Deterministic: same inputs always produce equivalent output. """ parts = [] # What was attempted if decision_context.declared_task: parts.append( f"Agent attempted '{decision_context.intended_action}' " f"as part of task: {decision_context.declared_task}." ) else: parts.append( f"Agent attempted '{decision_context.intended_action}'." ) # Policy result if policy_allowed: parts.append("AGT policy engine: allowed.") else: parts.append("AGT policy engine: denied.") # Confidence state if decision_context.confidence_score is not None: score_pct = int(decision_context.confidence_score * 100) n_alts = len(decision_context.alternatives_considered) if n_alts > 0: parts.append( f"Confidence: {score_pct}% against " f"{n_alts} alternative(s) considered " f"({', '.join(decision_context.alternatives_considered)})." ) else: parts.append(f"Confidence: {score_pct}%.") else: parts.append("Confidence: not provided.") # Gate classification and what triggered it gate_labels = { "gate_1_routine": "Gate 1 (Routine Execution)", "gate_2_documented_delegation": "Gate 2 (Documented Delegation)", "gate_3_elevated_review": "Gate 3 (Elevated Review)", "gate_4_hard_escalation": "Gate 4 (Hard Escalation)", } gate_label = gate_labels.get(gate.value, gate.value) parts.append(f"Gate classification: {gate_label}. {gate_rationale}") # Escalation or delegation specifics if escalation_trigger: trigger_labels = { "policy_denied": "Trigger: policy enforcement.", "hard_escalation_tool_class": "Trigger: tool class institutionally requires hard escalation.", "confidence_below_escalation_threshold": "Trigger: confidence below escalation threshold.", } parts.append(trigger_labels.get(escalation_trigger, f"Trigger: {escalation_trigger}.")) if delegation_reference: parts.append(f"Delegation authority: {delegation_reference}.") # Observable conditions if present if decision_context.observable_conditions: condition_summary = ", ".join( f"{k}={v}" for k, v in list(decision_context.observable_conditions.items())[:3] ) parts.append(f"Observable conditions at decision time: {condition_summary}.") return " ".join(parts) # --------------------------------------------------------------------------- # Gate record: the accountability artifact # --------------------------------------------------------------------------- @dataclass class GateRecord: """ Immutable artifact written at the moment of gate classification, before execution. An artifact written after the fact is a narrative. An artifact written before execution is evidence. This record captures the state of reasoning at decision time — not reconstructed afterward. That immutability is the entire accountability claim. """ record_id: str agent_id: str tool_name: str tool_args: dict[str, Any] # Policy result from AGT's pre_execute() policy_allowed: bool policy_reason: Optional[str] # Gate classification (institutional, not model-generated) gate: Gate gate_rationale: str # Decision context: structured capture of observable inputs at decision time. # Not a claim about internal model state — a record of what was observable # from the governance layer at the moment of classification. decision_context: "DecisionContext" # Confidence state confidence_score: Optional[float] # 0.0-1.0, None if not provided policy_confidence_threshold: float # From GovernancePolicy.confidence_threshold alternatives_considered: list[str] conditions_at_decision: dict[str, Any] # Plain-language reconstruction: a deterministic summary of observable # governance-relevant inputs rendered in human-readable form at # classification time. Generated from structured fields only — not from # the model being governed. This is not a claim about why the model chose # this action. It is a human-readable account of what the governance layer # observed. A non-technical reviewer can read this and understand what # governance decision was made and why. reasoning_reconstruction: str # Escalation escalation_required: bool escalation_trigger: Optional[str] delegation_reference: Optional[str] # Gate 2: upstream authority reference # Timestamps classified_at: str # ISO 8601, UTC, before execution executed_at: Optional[str] # Set after execution completes # Integrity: SHA-256 of core fields, computed at classification time. # Any post-hoc modification produces a hash mismatch. record_hash: str = field(default="") def __post_init__(self) -> None: if not self.record_hash: self.record_hash = self._compute_hash() def _compute_hash(self) -> str: payload = { "record_id": self.record_id, "agent_id": self.agent_id, "tool_name": self.tool_name, "tool_args": self.tool_args, "policy_allowed": self.policy_allowed, "gate": self.gate.value, "classified_at": self.classified_at, } canonical = json.dumps(payload, sort_keys=True) return hashlib.sha256(canonical.encode()).hexdigest() def to_dict(self) -> dict: d = asdict(self) d["gate"] = self.gate.value return d def to_json(self, indent: Optional[int] = 2) -> str: return json.dumps(self.to_dict(), indent=indent) def verify_integrity(self) -> bool: """Returns True if record has not been modified since classification.""" expected = self._compute_hash() return self.record_hash == expected # --------------------------------------------------------------------------- # Gate event type extension # --------------------------------------------------------------------------- class GateEventType(str, Enum): """ Additional governance event types emitted by the gate classifier. These extend AGT's GovernanceEventType without modifying it. Consumers can register listeners for these events via BaseIntegration.on(). """ GATE_CLASSIFIED = "gate_classified" GATE_ESCALATION = "gate_escalation" GATE_RECORD_WRITTEN = "gate_record_written" # --------------------------------------------------------------------------- # Gate rules: institutional definitions, not model-generated # --------------------------------------------------------------------------- class DefaultGateRules: """ Baseline gate rules. Institutions should override these with domain-specific classifications. Rules are defined by the institution and applied by the governance layer. The model's confidence scores and policy results are inputs to classification — they are not the classification itself. """ HARD_ESCALATION_TOOLS: set[str] = { "shell_exec", "delete_database", "drop_table", "terminate_process", "revoke_credentials", "bulk_delete", } ELEVATED_REVIEW_TOOLS: set[str] = { "file_write", "database_write", "send_email", "post_message", "update_record", "transfer_funds", "modify_policy", } # Confidence below which elevated review is triggered. # Separate from GovernancePolicy.confidence_threshold (which controls # whether AGT allows the action). This controls gate classification # for actions that did pass the policy check. LOW_CONFIDENCE_THRESHOLD: float = 0.70 # Confidence below which hard escalation is triggered. ESCALATION_CONFIDENCE_THRESHOLD: float = 0.50 # --------------------------------------------------------------------------- # Gate classifier # --------------------------------------------------------------------------- class GateClassifier: """ Classifies agent decisions by governance weight at the moment of policy evaluation. Designed to integrate with AGT's BaseIntegration event system. The classifier hooks into the POLICY_CHECK / pre_execute flow without modifying AGT's core logic. Standalone usage (mock AGT for testing): classifier = GateClassifier() record = classifier.evaluate( policy_allowed=True, policy_reason=None, policy_confidence_threshold=0.8, agent_id="claims-agent-001", tool_name="database_write", tool_args={"table": "claims_decisions"}, confidence_score=0.85, ) """ def __init__( self, rules: Optional[DefaultGateRules] = None, record_writer: Optional["FileRecordWriter"] = None, event_emitter: Optional[Callable] = None, ) -> None: self.rules = rules or DefaultGateRules() self.record_writer = record_writer self.event_emitter = event_emitter def evaluate( self, *, policy_allowed: bool, policy_reason: Optional[str] = None, policy_confidence_threshold: float = 0.8, agent_id: str, tool_name: str, tool_args: Optional[dict[str, Any]] = None, confidence_score: Optional[float] = None, alternatives_considered: Optional[list[str]] = None, confidence_distribution: Optional[dict[str, float]] = None, conditions_at_decision: Optional[dict[str, Any]] = None, delegation_reference: Optional[str] = None, declared_task: Optional[str] = None, ) -> GateRecord: """ Classify the governance gate for a decision after AGT's policy check has run. Called AFTER pre_execute() returns but BEFORE the action executes. The record is written before execution. The caller is responsible for halting execution if escalation_required is True. New parameters: declared_task: what the agent declared it was trying to accomplish. Provided from agent state if available (LangGraph node context, LangChain metadata, etc.). Optional — not all frameworks expose this. confidence_distribution: confidence scores across alternatives considered, as {alternative_name: score}. Distinguishes a 0.51 decision against five alternatives from a 0.97 decision against one. """ gate, rationale, escalation_trigger = self._classify( tool_name=tool_name, policy_allowed=policy_allowed, confidence_score=confidence_score, delegation_reference=delegation_reference, ) escalation_required = gate == Gate.HARD_ESCALATION # Build structured decision context from observable inputs decision_context = DecisionContext( intended_action=tool_name, action_parameters=tool_args or {}, declared_task=declared_task, confidence_score=confidence_score, alternatives_considered=alternatives_considered or [], confidence_distribution=confidence_distribution, observable_conditions=conditions_at_decision or {}, ) # Generate plain-language reconstruction from structured fields only. # Deterministic. No model call. No access to agent internals. reasoning_reconstruction = ReasoningReconstructor.reconstruct( gate=gate, gate_rationale=rationale, decision_context=decision_context, policy_allowed=policy_allowed, escalation_trigger=escalation_trigger, delegation_reference=delegation_reference, ) record = GateRecord( record_id=str(uuid.uuid4()), agent_id=agent_id, tool_name=tool_name, tool_args=tool_args or {}, policy_allowed=policy_allowed, policy_reason=policy_reason, gate=gate, gate_rationale=rationale, decision_context=decision_context, confidence_score=confidence_score, policy_confidence_threshold=policy_confidence_threshold, alternatives_considered=alternatives_considered or [], conditions_at_decision=conditions_at_decision or {}, reasoning_reconstruction=reasoning_reconstruction, escalation_required=escalation_required, escalation_trigger=escalation_trigger, delegation_reference=delegation_reference, classified_at=datetime.now(timezone.utc).isoformat(), executed_at=None, ) if self.record_writer: self.record_writer.write(record) if self.event_emitter: self.event_emitter(GateEventType.GATE_RECORD_WRITTEN, { "record_id": record.record_id, "agent_id": agent_id, "gate": gate.value, "tool_name": tool_name, "escalation_required": escalation_required, "classified_at": record.classified_at, }) if escalation_required: self.event_emitter(GateEventType.GATE_ESCALATION, { "record_id": record.record_id, "agent_id": agent_id, "tool_name": tool_name, "escalation_trigger": escalation_trigger, "classified_at": record.classified_at, }) return record def _classify( self, tool_name: str, policy_allowed: bool, confidence_score: Optional[float], delegation_reference: Optional[str], ) -> tuple[Gate, str, Optional[str]]: """Apply institutional gate rules. Returns (gate, rationale, trigger).""" if not policy_allowed: return ( Gate.HARD_ESCALATION, "AGT policy engine denied this action. " "Action is outside authorized scope.", "policy_denied", ) if tool_name in self.rules.HARD_ESCALATION_TOOLS: return ( Gate.HARD_ESCALATION, f"Tool '{tool_name}' is institutionally classified as requiring " f"hard escalation. Action is potentially irreversible.", "hard_escalation_tool_class", ) if ( confidence_score is not None and confidence_score < self.rules.ESCALATION_CONFIDENCE_THRESHOLD ): return ( Gate.HARD_ESCALATION, f"Confidence score {confidence_score:.2f} is below the escalation " f"threshold of {self.rules.ESCALATION_CONFIDENCE_THRESHOLD}. " f"Insufficient confidence for autonomous execution.", "confidence_below_escalation_threshold", ) if tool_name in self.rules.ELEVATED_REVIEW_TOOLS: return ( Gate.ELEVATED_REVIEW, f"Tool '{tool_name}' is institutionally classified as requiring " f"elevated review. A human must acknowledge this artifact before " f"or after execution.", None, ) if ( confidence_score is not None and confidence_score < self.rules.LOW_CONFIDENCE_THRESHOLD ): return ( Gate.ELEVATED_REVIEW, f"Confidence score {confidence_score:.2f} is below the review " f"threshold of {self.rules.LOW_CONFIDENCE_THRESHOLD}. " f"Human review required.", None, ) if delegation_reference: return ( Gate.DOCUMENTED_DELEGATION, f"Decision falls within pre-authorized scope. " f"Delegation reference: {delegation_reference}.", None, ) return ( Gate.ROUTINE, f"Decision falls within established parameters. " f"Action '{tool_name}' is routine within authorized scope.", None, ) # --------------------------------------------------------------------------- # BaseIntegration mixin # --------------------------------------------------------------------------- class GateClassificationMixin: """ Mixin that adds gate classification to any BaseIntegration subclass. Usage: from agent_os.integrations.base import BaseIntegration, GovernancePolicy from gate_classifier import GateClassificationMixin, GateClassifier class GovernedLangChainIntegration(GateClassificationMixin, BaseIntegration): def __init__(self, policy, classifier=None): super().__init__(policy=policy) self.setup_gate_classifier(classifier) def pre_execute(self, ctx, input_data): allowed, reason = super().pre_execute(ctx, input_data) if allowed: self.classify_after_policy_check( ctx=ctx, policy_allowed=True, tool_name=getattr(input_data, 'tool_name', 'unknown'), tool_args=getattr(input_data, 'arguments', {}), ) return allowed, reason """ def setup_gate_classifier( self, classifier: Optional[GateClassifier] = None, record_writer: Optional["FileRecordWriter"] = None, ) -> None: self._gate_classifier = classifier or GateClassifier( record_writer=record_writer, event_emitter=getattr(self, "emit", None), ) self._last_gate_record: Optional[GateRecord] = None def classify_after_policy_check( self, *, ctx: Any, policy_allowed: bool, tool_name: str, tool_args: Optional[dict] = None, confidence_score: Optional[float] = None, alternatives_considered: Optional[list[str]] = None, confidence_distribution: Optional[dict[str, float]] = None, conditions_at_decision: Optional[dict] = None, delegation_reference: Optional[str] = None, policy_reason: Optional[str] = None, declared_task: Optional[str] = None, ) -> GateRecord: """Classify gate after pre_execute() returns, using policy from ctx.""" policy = getattr(ctx, "policy", None) threshold = getattr(policy, "confidence_threshold", 0.8) if policy else 0.8 record = self._gate_classifier.evaluate( policy_allowed=policy_allowed, policy_reason=policy_reason, policy_confidence_threshold=threshold, agent_id=getattr(ctx, "agent_id", "unknown"), tool_name=tool_name, tool_args=tool_args, confidence_score=confidence_score, alternatives_considered=alternatives_considered, confidence_distribution=confidence_distribution, conditions_at_decision=conditions_at_decision, delegation_reference=delegation_reference, declared_task=declared_task, ) self._last_gate_record = record return record @property def last_gate_record(self) -> Optional[GateRecord]: return self._last_gate_record # --------------------------------------------------------------------------- # File-based record writer # --------------------------------------------------------------------------- class FileRecordWriter: """Writes gate records to a newline-delimited JSON file. Append-only.""" def __init__(self, path: str = "gate_records.ndjson") -> None: self.path = path def write(self, record: GateRecord) -> None: with open(self.path, "a") as f: f.write(record.to_json(indent=None) + "\n") # --------------------------------------------------------------------------- # Escalation signal # --------------------------------------------------------------------------- class EscalationRequired(Exception): """Raised when a Gate 4 classification halts execution.""" def __init__(self, record: GateRecord) -> None: self.record = record super().__init__( f"Hard escalation required for agent '{record.agent_id}' " f"attempting '{record.tool_name}'. " f"Trigger: {record.escalation_trigger}. " f"Record ID: {record.record_id}" )