Agent-Safe DevOps: Building Human-in-the-Loop Gates for Autonomous AI Coding Pipelines

A developer has published an architecture for "Human-in-the-Loop (HITL) Gates," a sidecar service that intercepts actions between autonomous AI coding agents and execution environments such as CI/CD runners, Terraform and Kubernetes. The gate service scores each action by probability and impact, then applies LOG, asynchronous REVIEW, or synchronous BLOCK policies, addressing failure modes including semantic hallucination, scope creep, dependency injection, secret leakage, infrastructure mutation and cascade failure. The design treats risk as contextual and cumulative, so repeated low-risk changes across many services can still trigger a gate.

Originally published on tamiz.pro https://tamiz.pro/insights/agent-safe-devops-human-in-the-loop-gates-autonomous-ai-pipelines . Autonomous AI coding agents — from code generation copilots to full-pipeline automation tools — are moving from demo to production. They can generate pull requests, modify infrastructure-as-code, update dependency manifests, and trigger deployments. The promise is radical velocity. The danger is radical blast radius. Without guardrails, a single hallucinated rm -rf / or a misconfigured IAM policy can cascade through an entire deployment pipeline. The engineering question is no longer can we automate this? but how do we automate this safely? This article dissects the architecture, policy models, and implementation patterns for Human-in-the-Loop HITL Gates — the checkpoint system that sits between an autonomous AI agent and the real world, ensuring that high-risk actions require human approval while low-risk actions flow through unimpeded. Before designing gates, you need a precise model of how autonomous agents fail. These aren't hypotheticals — they're drawn from documented incidents in AI-assisted development environments. | Category | Description | Example | Blast Radius | |---|---|---|---| | Semantic Hallucination | Agent generates plausible but incorrect code/config | Wrong S3 bucket ACL, incorrect Terraform resource | Medium–High | | Scope Creep | Agent modifies files outside its intended scope | Updates main.tf when asked to fix a test | High | | Dependency Injection | Agent introduces vulnerable or malicious dependencies | npm install of a typosquatted package | Critical | | Secret Leakage | Agent commits credentials or keys | Hardcoded API key in a new file | Critical | | Infrastructure Mutation | Agent deploys config that changes production behavior | Changes max connections in RDS | Critical | | Cascade Failure | A sequence of individually safe actions creates an unsafe state | 5 small config changes that together break auth | High | The key insight: risk is contextual and cumulative . A single change to a test file is trivial. The same change applied 50 times across 50 microservices, each subtly altering behavior, is a systemic risk. Gates must account for both individual action risk and accumulated pipeline risk. LOW IMPACT HIGH IMPACT ┌───────────────────────────────────── LOW PROB │ LOG no gate │ REVIEW async gate │ ABILITY │ e.g., fix typo │ e.g., config tweak │ ├───────────────────────────────────── HIGH PROB │ LOG + MONITOR │ BLOCK sync gate │ ABILITY │ e.g., test update │ e.g., prod deploy │ └───────────────────────────────────── The gate system maps every action to a cell in this matrix and applies the corresponding policy: LOG pass through, record , REVIEW async human approval with timeout , or BLOCK synchronous human approval required before proceeding . The gate system is a sidecar service that intercepts actions between an AI agent and the execution environment. It operates as a middleware layer in the CI/CD pipeline, not as a replacement for the pipeline itself. ┌──────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ │ │ │ │ │ │ AI Agent │────▶│ Gate Service │────▶│ Execution Env │ │ Copilot, │ │ Policy Engine, │ │ CI/CD Runner, │ │ Codex, │ │ Risk Scorer, │ │ Terraform, │ │ Custom │ │ Approval Flow │ │ K8s, Cloud │ │ │ │ │ │ │ └──────┬───────┘ └────────┬─────────┘ └────────┬─────────┘ │ │ │ │ ┌───────▼────────┐ │ │ │ Human Approver│ │ │ │ Slack, Email,│ │ │ │ Web Console │ │ │ └────────────────┘ │ │ │ │ └──────────────────────┴─────────────────────────┘ Audit Log Action Interceptor — Captures every action the agent wants to perform. In CI/CD, this hooks into pipeline steps. In infrastructure, it wraps Terraform/CloudFormation calls. Risk Scorer — Evaluates each action against a policy engine, producing a risk score 0–100 and a gate decision PASS, REVIEW, BLOCK . Approval Orchestrator — Manages the human-in-the-loop workflow: sends approval requests, handles timeouts, manages escalations. Audit Ledger — Immutable record of every action, decision, and approval. Critical for compliance and forensics. Rollback Coordinator — If a gated action is rejected post-execution e.g., during async review , orchestrates automatic rollback. Every intercepted action produces a GateDecision : // types/gate.ts export type GateAction = 'PASS' | 'REVIEW' | 'BLOCK' | 'ROLLBACK'; export interface GateDecision { actionId: string; riskScore: number; // 0-100 gateAction: GateAction; policyViolations: PolicyViolation ; reviewerRequired: boolean; timeoutMs: number; // For REVIEW actions createdAt: Date; metadata: Record<string, unknown ; } export interface PolicyViolation { ruleId: string; severity: 'info' | 'warning' | 'critical'; description: string; autoRemediable: boolean; } The distinction between REVIEW and BLOCK is critical: The risk scorer is the brain of the gate system. It must be fast sub-second latency for most decisions , accurate, and configurable. The scorer evaluates each action across multiple weighted dimensions: // risk/scorer.ts export interface RiskProfile { fileScope: FileScopeRisk; // Which files are touched? changeMagnitude: ChangeMagnitude; // How large is the diff? environment: EnvironmentRisk; // Which environment? dependencyImpact: DependencyRisk; // New packages, version changes? secretExposure: SecretRisk; // Potential credential leaks? infrastructureImpact: InfraRisk; // Terraform, K8s, IAM changes? cumulativeRisk: CumulativeRisk; // Pipeline-level accumulation } export interface FileScopeRisk { touchedFiles: string ; outOfScopeFiles: string ; // Files the agent shouldn't touch sensitiveFiles: string ; // e.g., prod configs, IAM policies } export interface ChangeMagnitude { linesAdded: number; linesDeleted: number; filesModified: number; isBinaryChange: boolean; } export interface EnvironmentRisk { targetEnvironment: 'dev' | 'staging' | 'production'; deploymentFrequency: number; // deploys per day historically criticality: 'low' | 'medium' | 'high' | 'critical'; } js // risk/scorer.ts continued const WEIGHTS = { fileScope: 0.20, changeMagnitude: 0.15, environment: 0.25, dependencyImpact: 0.15, secretExposure: 0.15, infrastructureImpact: 0.10, cumulativeRisk: 0.00, // applied as multiplier } as const; export function computeRiskScore profile: RiskProfile : number { let score = 0; // File scope: penalize out-of-scope and sensitive files score += WEIGHTS.fileScope profile.fileScope.outOfScopeFiles.length 15 + profile.fileScope.sensitiveFiles.length 25 ; // Change magnitude: exponential scaling for large diffs const magnitudeRaw = profile.changeMagnitude.linesAdded + profile.changeMagnitude.linesDeleted + profile.changeMagnitude.filesModified 10; score += WEIGHTS.changeMagnitude Math.min 100, magnitudeRaw 0.5 ; // Environment: production is inherently riskier const envMultipliers = { dev: 1, staging: 3, production: 10 }; score += WEIGHTS.environment envMultipliers profile.environment.targetEnvironment 8; // Dependency impact const depChanges = profile.dependencyImpact.newPackages.length; const versionBumps = profile.dependencyImpact.majorVersionBumps; score += WEIGHTS.dependencyImpact depChanges 20 + versionBumps 30 ; // Secret exposure if profile.secretExposure.detected { score += WEIGHTS.secretExposure 100; // Near-instant fail } // Infrastructure impact const infraChanges = profile.infrastructureImpact.terraformChanges; score += WEIGHTS.infrastructureImpact infraChanges 20; // Cumulative risk multiplier const cumulativeMultiplier = 1 + profile.cumulativeRisk.actionsThisPipeline / 10 ; score = Math.min 100, score cumulativeMultiplier ; return Math.round score ; } js // risk/thresholds.ts export const GATE THRESHOLDS = { PASS: { maxScore: 25, description: 'Auto-approve, log only' }, REVIEW: { maxScore: 60, description: 'Async human review with rollback option' }, BLOCK: { maxScore: 100, description: 'Synchronous human approval required' }, } as const; export function decideGate score: number : GateAction { if score <= GATE THRESHOLDS.PASS.maxScore return 'PASS'; if score <= GATE THRESHOLDS.REVIEW.maxScore return 'REVIEW'; return 'BLOCK'; } The thresholds are configurable per team, per repository, and per environment. A mature DevOps team might start strict everything BLOCKed and progressively relax as trust in the agent grows — a pattern we'll call Gate Decay . The gate service is a lightweight HTTP server that acts as a proxy. The AI agent submits actions to the gate service instead of directly to the execution environment. js // server/gate-service.ts import { FastifyInstance } from 'fastify'; import { computeRiskScore, decideGate } from '../risk/scorer'; import { PolicyEngine } from '../policy/engine'; import { ApprovalOrchestrator } from '../approval/orchestrator'; import { AuditLogger } from '../audit/logger'; import { RollbackCoordinator } from '../rollback/coordinator'; export interface GateRequest { actionId: string; agentId: string; action: { type: 'code-change' | 'deploy' | 'config-change' | 'dependency-update'; target: string; // e.g., 'https://github.com/org/repo' description: string; diff?: string; // Unified diff for code changes metadata: Record<string, unknown ; }; pipelineContext: { pipelineId: string; actionsSoFar: number; previousRiskScores: number ; environment: 'dev' | 'staging' | 'production'; }; } export function buildGateService fastify: FastifyInstance { const policyEngine = new PolicyEngine loadDefaultPolicies ; const orchestrator = new ApprovalOrchestrator ; const auditLogger = new AuditLogger ; const rollbackCoordinator = new RollbackCoordinator ; fastify.post '/gate/evaluate', async request, reply = { const req = request.body as GateRequest; // Step 1: Build risk profile const profile = policyEngine.buildRiskProfile req ; // Step 2: Compute score const score = computeRiskScore profile ; // Step 3: Check policy violations const violations = policyEngine.checkViolations req, profile ; // Step 4: Decide gate action let gateAction = decideGate score ; // Override: any critical violation forces BLOCK if violations.some v = v.severity === 'critical' { gateAction = 'BLOCK'; } // Step 5: Audit await auditLogger.log { actionId: req.actionId, agentId: req.agentId, score, gateAction, violations, timestamp: new Date , } ; // Step 6: Execute gate decision switch gateAction { case 'PASS': return reply.send { actionId: req.actionId, decision: 'APPROVED', riskScore: score, message: 'Action auto-approved. Logged for audit.', } ; case 'REVIEW': const reviewId = await orchestrator.requestAsyncReview { actionId: req.actionId, agentId: req.agentId, action: req.action, riskScore: score, violations, timeoutMs: 30 60 1000, // 30 minutes } ; return reply.send { actionId: req.actionId, decision: 'SUBMITTED FOR REVIEW', riskScore: score, reviewId, message: 'Action submitted for async review. Will execute if not rejected within timeout.', } ; case 'BLOCK': const approvalId = await orchestrator.requestSyncApproval { actionId: req.actionId, agentId: req.agentId, action: req.action, riskScore: score, violations, timeoutMs: 24 60 60 1000, // 24 hours } ; return reply.send { actionId: req.actionId, decision: 'PENDING APPROVAL', riskScore: score, approvalId, message: 'Action blocked. Awaiting human approval.', } ; } } ; // Webhook endpoint for human approvals fastify.post '/gate/approval/callback', async request, reply = { const { approvalId, decision, reviewerId } = request.body; const result = await orchestrator.processApproval { approvalId, decision, // 'approved' | 'rejected' reviewerId, } ; if decision === 'rejected' && result.originalAction.type === 'code-change' { await rollbackCoordinator.initiateRollback result.originalAction ; } return reply.send { status: 'processed' } ; } ; // Health and metrics fastify.get '/gate/health', async = { status: 'healthy', pendingApprovals: orchestrator.pendingCount , pendingReviews: orchestrator.reviewCount , uptime: process.uptime , } ; } On the AI agent side, the integration is a simple wrapper that routes all actions through the gate service: python // agent/safe-agent.ts import axios from 'axios'; export class GatedAgent { constructor private gateUrl: string, private agentId: string, private pipelineId: string, private environment: 'dev' | 'staging' | 'production' {} private riskHistory: number = ; async submitAction action: { type: string; target: string; description: string; diff?: string; metadata?: Record<string, unknown ; } : Promise<GateResponse { const actionId = crypto.randomUUID ; const response = await axios.post ${this.gateUrl}/gate/evaluate , { actionId, agentId: this.agentId, action, pipelineContext: { pipelineId: this.pipelineId, actionsSoFar: this.riskHistory.length, previousRiskScores: this.riskHistory, environment: this.environment, }, } ; const result = response.data; this.riskHistory.push result.riskScore ; switch result.decision { case 'APPROVED': // Execute the action directly return this.executeAction action ; case 'SUBMITTED FOR REVIEW': // Execute now, but monitor for rejection const executePromise = this.executeAction action ; this.monitorForRejection result.reviewId, action ; return executePromise; case 'PENDING APPROVAL': // Wait for human approval return this.waitForApproval result.approvalId, action ; } } private async waitForApproval approvalId: string, action: any : Promise<GateResponse { while true { await sleep 5000 ; // Poll every 5 seconds const status = await axios.get ${this.gateUrl}/gate/approval/${approvalId}/status ; if status.data.decision === 'approved' { return this.executeAction action ; } if status.data.decision === 'rejected' { return { decision: 'REJECTED', message: Action rejected by ${status.data.reviewerId} , }; } } } private async monitorForRejection reviewId: string, action: any : Promise<void { const start = Date.now ; const timeout = 30 60 1000; // 30 min while Date.now - start < timeout { await sleep 10000 ; const status = await axios.get ${this.gateUrl}/gate/review/${reviewId}/status ; if status.data.decision === 'rejected' { await this.rollback action ; break; } } } private async executeAction action: any : Promise<GateResponse { // Dispatch to actual execution: git push, terraform apply, etc. // Implementation depends on the execution environment return { decision: 'EXECUTED', actionId: action.id }; } private async rollback action: any : Promise<void { // Execute rollback logic console.log Rolling back action: ${action.description} ; } } function sleep ms: number { return new Promise resolve = setTimeout resolve, ms ; } Hardcoding gate rules in the scorer is brittle. The production approach is Policy-as-Code — declarative rules stored in version control, loaded at startup, and hot-reloadable. policies/production.yaml Gate policies for production environment version: "1.0" environment: production policies: - id: no-prod-config-changes description: "AI agents cannot modify production config files" condition: file patterns: - "config/prod/ " - " .prod.yml" - " .prod.yaml" action: BLOCK severity: critical reviewer group: "platform-team" - id: dependency-major-version-bump description: "Major version bumps require human review" condition: dependency change: type: "major" action: BLOCK severity: warning reviewer group: "security-team" - id: large-diff description: "Diffs over 500 lines require review" condition: change magnitude: lines total: { gt: 500 } action: REVIEW severity: info reviewer group: "team-leads" - id: iam-policy-change description: "Any IAM policy change is blocked" condition: file patterns: - "iam/ " - " .iam.json" action: BLOCK severity: critical reviewer group: "security-team" - id: terraform-destroy description: "Terraform destroy operations are always blocked" condition: command contains: - "terraform destroy" - "terraform apply -destroy" action: BLOCK severity: critical reviewer group: "infra-team" - id: cumulative-action-limit description: "More than 10 actions in one pipeline triggers review" condition: pipeline: max actions: 10 action: REVIEW severity: warning reviewer group: "team-leads" - id: test-file-auto-pass description: "Changes to test files are auto-approved" condition: file patterns: - " / .test. " - " / .spec. " - " / tests / " action: PASS severity: info reviewer group: null js // policy/engine.ts import { Policy, PolicyCondition } from './types'; export class PolicyEngine { private policies: Policy = ; constructor policies: Policy { this.policies = policies; } checkViolations request: GateRequest, profile: RiskProfile : PolicyViolation { const violations: PolicyViolation = ; for const policy of this.policies { if this.evaluateCondition policy.condition, request, profile { violations.push { ruleId: policy.id, severity: policy.severity, description: policy.description, autoRemediable: policy.autoRemediable ?? false, } ; } } return violations; } private evaluateCondition condition: PolicyCondition, request: GateRequest, profile: RiskProfile : boolean { // File pattern matching if condition.file patterns { const patterns = condition.file patterns; const touched = profile.fileScope.touchedFiles; return touched.some file = patterns.some p = matchGlob p, file ; } // Dependency change detection if condition.dependency change { const change = condition.dependency change; if change.type === 'major' { return profile.dependencyImpact.majorVersionBumps 0; } } // Change magnitude if condition.change magnitude { const mag = condition.change magnitude; const total = profile.changeMagnitude.linesAdded + profile.changeMagnitude.linesDeleted; if mag.lines total?.gt && total mag.lines total.gt return true; } // Command matching if condition.command contains { const commands = condition.command contains; const actionStr = JSON.stringify request.action ; return commands.some cmd = actionStr.includes cmd ; } // Pipeline-level conditions if condition.pipeline { const pipe = condition.pipeline; if pipe.max actions && request.pipelineContext.actionsSoFar = pipe.max actions { return true; } } return false; } } // Simple glob matcher use minimatch in production function matchGlob pattern: string, file: string : boolean { // Simplified — use a proper library in production if pattern.endsWith ' ' { return file.startsWith pattern.slice 0, -2 ; } if pattern.includes ' ' { const regex = new RegExp '^' + pattern.replace /\ /g, '. ' .replace /\?/g, '.' + '$' ; return regex.test file ; } return pattern === file; } In production, policies are stored in a Git repository. A file watcher or webhook detects changes and reloads the policy engine without restarting the service: js // policy/reloader.ts import { watch } from 'chokidar'; import { PolicyEngine } from './engine'; export class PolicyReloader { private engine: PolicyEngine; private watcher: import 'chokidar' .FSWatcher | null = null; constructor engine: PolicyEngine, policyPath: string { this.engine = engine; } async start policyDir: string { this.watcher = watch ${policyDir}/ .yaml , { persistent: true, } ; this.watcher.on 'change', async filePath = { console.log PolicyReloader Detected change: ${filePath} ; const newPolicies = await loadPoliciesFromDir policyDir ; this.engine.reload newPolicies ; console.log PolicyReloader Reloaded ${newPolicies.length} policies ; } ; } stop { this.watcher?.close ; } } The gate is only as good as the human interaction. A BLOCK that takes 48 hours to get approved is worse than no gate at all — it creates bottlenecks that engineers will work around. The orchestrator must support multiple channels: | Channel | Latency | Best For | Implementation | |---|---|---|---| | Slack/Teams | Minutes | REVIEW, low-risk BLOCK | Bot integration with buttons | | | Hours | Escalation, BLOCK for critical | SMTP + tracking | | Web Console | Real-time | Dashboard view, batch approval | React + WebSocket | | PagerDuty/Opsgenie | Minutes | Critical BLOCK during incidents | API integration | When a human receives an approval request, they need enough context to make a decision quickly: // approval/request.ts export interface ApprovalRequest { approvalId: string; agentId: string; agentName: string; // e.g., "CodeGen-Agent v2.1" actionDescription: string; // Human-readable summary riskScore: number; riskLevel: 'low' | 'medium' | 'high' | 'critical'; violations: PolicyViolation ; diffSummary: DiffSummary; context: { pipelineId: string; environment: string; branch: string; commitHash: string; relatedActions: ActionSummary ; // Previous actions in this pipeline }; recommendedAction: 'approve' | 'reject' | 'modify'; expiresAt: Date; } export interface DiffSummary { filesChanged: number; linesAdded: number; linesDeleted: number; keyChanges: string ; // Top 5 most important changes riskyPatterns: string ; // Detected risky code patterns } js // approval/slack-integration.ts import { WebClient } from '@slack/web-api'; export class SlackApprovalChannel { constructor private slackToken: string, private defaultChannel: string {} private client = new WebClient this.slackToken ; async sendApprovalRequest req: ApprovalRequest : Promise<string { const riskEmoji = { low: '🟢', medium: '🟡', high: '🟠', critical: '🔴', } req.riskLevel ; const blocks = { type: 'header', text: { type: 'plain text', text: ${riskEmoji} Approval Required: ${req.actionDescription} , }, }, { type: 'section', text: { type: 'mrkdwn', text: Agent: ${req.agentName} , Risk Score: ${req.riskScore}/100 ${req.riskLevel} , Environment: ${req.context.environment} , Pipeline: ${req.context.pipelineId} , Expires: ${req.expiresAt.toISOString } , .join '\n' , }, }, { type: 'section', text: { type: 'mrkdwn', text: Key Changes: ${req.diffSummary.keyChanges.map c = • ${c} .join '\n' } , }, }, req.violations.length 0 && { type: 'section', text: { type: 'mrkdwn', text: Policy Violations: ${req.violations.map v = ⚠️ ${v.description} .join '\n' } , }, }, { type: 'actions', elements: { type: 'button', text: { type: 'plain text', text: '✅ Approve' }, style: 'primary', value: approve:${req.approvalId} , }, { type: 'button', text: { type: 'plain text', text: '❌ Reject' }, style: 'danger', value: reject:${req.approvalId} , }, { type: 'button', text: { type: 'plain text', text: '👀 View Full Diff' }, url: req.context.diffUrl, }, , }, .filter Boolean ; const result = await this.client.chat.postMessage { channel: this.defaultChannel, blocks, thread ts: req.threadTs, // Thread related approvals together } ; return result.ts; } } T+0min → Slack notification to primary reviewer T+15min → Slack notification to backup reviewer T+30min → Email notification to team lead T+1h → PagerDuty alert for BLOCK actions on critical resources T+4h → Auto-reject configurable + incident ticket created This ensures that BLOCK actions don't silently stall pipelines indefinitely. Every gate decision must be logged immutably. This is your forensic record. // audit/logger.ts export interface AuditEntry { entryId: string; timestamp: Date; actionId: string; agentId: string; pipelineId: string; action: { type: string; target: string; description: string; diffHash: string; // SHA-256 of the diff for integrity }; riskAssessment: { score: number; level: 'low' | 'medium' | 'high' | 'critical'; dimensions: Record<string, number ; }; policyCheck: { policiesEvaluated: number; violations: PolicyViolation ; }; gateDecision: GateAction; humanInteraction?: { approvalId: string; reviewerId: string; reviewerName: string; decision: 'approved' | 'rejected'; decisionTimestamp: Date; comments: string; }; executionResult?: { status: 'executed' | 'failed' | 'rolled back'; timestamp: Date; rollbackId?: string; }; } Expose these metrics via Prometheus/OpenTelemetry: Counter: Total gate decisions by type agent gate decisions total{gate action="PASS"} 142 agent gate decisions total{gate action="REVIEW"} 37 agent gate decisions total{gate action="BLOCK"} 12 Histogram: Risk score distribution agent gate risk score bucket{le="25"} 142 agent gate risk score bucket{le="60"} 179 agent gate risk score bucket{le="100"} 191 Histogram: Time to approval for BLOCK actions agent gate approval duration seconds bucket{le="60"} 3 agent gate approval duration seconds bucket{le="300"} 8 agent gate approval duration seconds bucket{le="3600"} 11 Counter: Rejections and rollbacks agent gate rejections total 4 agent gate rollbacks total 2 Gauge: Pending approvals agent gate pending approvals 3 Counter: Policy violations by rule agent gate policy violations total{rule id="no-prod-config-changes"} 7 agent gate policy violations total{rule id="large-diff"} 23 Multiple AI agents may submit actions simultaneously. The gate service must handle this: js // gate/concurrency.ts import { Mutex } from 'async-mutex'; export class ConcurrencyAwareGate { private mutexes = new Map<string, Mutex ; async withPipelineLock pipelineId: string, fn: = Promise<any : Promise<any { let mutex = this.mutexes.get pipelineId ; if mutex { mutex = new Mutex ; this.mutexes.set pipelineId, mutex ; } return mutex.runExclusive async = { try { return await fn ; } finally { // Clean up mutex if no more actions in pipeline if this.isPipelineComplete pipelineId { this.mutexes.delete pipelineId ; } } } ; } } Actions must be idempotent. If the gate service crashes after logging but before executing, a retry should not double-execute: // gate/idempotency.ts export class IdempotencyStore { private store: Map<string, IdempotencyRecord = new Map ; async check actionId: string : Promise<'new' | 'in-progress' | 'completed' { const record = this.store.get actionId ; if record return 'new'; if record.status === 'completed' return 'completed'; return 'in-progress'; } async markInProgress actionId: string : Promise<void { this.store.set actionId, { status: 'in-progress', startedAt: new Date , } ; } async markCompleted actionId: string, result: any : Promise<void { this.store.set actionId, { status: 'completed', result, completedAt: new Date , } ; } } As agents prove themselves reliable, teams should gradually relax gates. This is Gate Decay — a deliberate, policy-driven reduction of gate strictness over time: policies/gate-decay.yaml Progressive trust model for a specific agent agent id: "codegen-agent-v2.1" trust level: "building" building → trusted → autonomous Phase 1: Building trust first 30 days gate config: code changes: REVIEW config changes: BLOCK deployments: BLOCK max daily actions: 50 Phase 2: Trusted after 30 days, if zero critical incidents gate config: code changes: PASS config changes: REVIEW deployments: BLOCK max daily actions: 200 Phase 3: Autonomous after 90 days, if zero critical incidents gate config: code changes: PASS config changes: PASS deployments: REVIEW max daily actions: 1000 Safety net: always enforced regardless of trust level always block: - iam-policy-change - terraform-destroy - secret-exposure The gate service must be treated as a critical infrastructure component: PASS -eligible actions, fail-open log the gap . For BLOCK -eligible actions, fail-closed never execute without a gate . // gate/failover.ts export class GateFailover { async evaluateWithFailover request: GateRequest, gateAction: GateAction : Promise<GateDecision { try { return await this.primary.evaluate request ; } catch error { console.error ' GateFailover Primary gate failed:', error ; if gateAction === 'BLOCK' { // CRITICAL: Never bypass BLOCK gates throw new Error Gate service unavailable for BLOCK action. + Action ${request.actionId} will NOT be executed. + Retry when service is restored. ; } // For PASS/REVIEW actions, fail-open with audit trail return { actionId: request.actionId, riskScore: 0, gateAction: 'PASS', policyViolations: , reviewerRequired: false, timeoutMs: 0, createdAt: new Date , metadata: { failover: true, originalError: error.message, bypassedGate: true, }, }; } } } BLOCK gates add latency by definition. The mitigation is action batching : group multiple low-risk actions and submit them as a single BLOCK request with a consolidated diff. For truly critical actions IAM, Terraform destroy , the latency is acceptable — these should never be automated without review anyway. Consider also implementing a pre-flight mode where the agent submits a plan, gets approval for the plan, and then executes individual steps without per-step gating. Use a shared gate service with per-agent policy profiles. A shared service gives you centralized auditing, consistent metrics, and a single deployment to maintain. Per-agent profiles loaded via policy-as-code handle the fact that different agents have different trust levels and risk profiles. The agentId field in every request enables per-agent metrics and policy routing. Three strategies: 1 Gate Decay — progressively reduce gates as trust builds, so reviewers only see genuinely risky actions. 2 Randomized audit sampling — randomly select 5% of PASS decisions for post-hoc human review, catching drift without blocking the pipeline. 3 Reviewer rotation — prevent the same person from always approving, which reduces both fatigue and single-point-of-failure risk. Track approval rates per reviewer; a 99% approval rate is a signal to investigate, not celebrate. Building safe autonomous pipelines isn't about preventing automation — it's about making automation accountable. Every gate you build is a contract between your organization and your AI agents: you'll give you freedom, you'll keep us safe. The architecture described here gives you that contract, enforced in code, audited in logs, and reviewed by humans who know the difference between a typo fix and a production outage. For more on production AI safety patterns, see Tamiz's Insights https://tamiz.pro/insights . The conceptual model is sound, but DevOps lives or dies on implementation. Let's build a production-grade gate engine that sits between your AI agent's proposed changes and your deployment pipeline. The architecture we're targeting is straightforward: every agent-generated artifact passes through a deterministic gate that evaluates it against policy, risk scoring, and human review thresholds before anything reaches a merge queue or production environment. python from dataclasses import dataclass, field from enum import Enum from typing import Optional import hashlib import json import time import uuid class GateDecision Enum : AUTO APPROVE = "auto approve" HUMAN REVIEW = "human review" BLOCK = "block" ESCALATE = "escalate" @dataclass class RiskProfile: """Quantified risk assessment for a proposed change.""" scope score: float = 0.0 0-1: how broad is the change? blast radius: float = 0.0 0-1: how many systems affected? reversibility: float = 0.0 0-1: how easily can we undo? novelty: float = 0.0 0-1: how different from prior changes? confidence: float = 0.0 0-1: how sure is the agent? test coverage delta: float = 0.0 change in test coverage @property def composite risk self - float: """Weighted composite risk score 0-1, higher = riskier .""" return 0.25 self.scope score + 0.30 self.blast radius + 0.20 1.0 - self.reversibility + 0.15 self.novelty + 0.10 1.0 - self.confidence @dataclass class GateResult: decision: GateDecision risk profile: RiskProfile gate id: str timestamp: float metadata: dict = field default factory=dict reviewer id: Optional str = None reviewer decision: Optional str = None reviewer notes: Optional str = None class GateEngine: """ Deterministic gate engine that evaluates AI agent proposals against configurable policy thresholds. """ def init self, config: dict : self.config = config self.audit log: list dict = self. review queue: list GateResult = def evaluate self, proposal: dict - GateResult: """ Evaluate a single agent proposal against gate policy. Args: proposal: Dict containing: - change id: Unique identifier - diff stats: {files changed, lines added, lines removed} - affected services: List of service names - test results: {passed, failed, coverage before, coverage after} - agent confidence: 0-1 confidence score - change type: 'refactor' | 'feature' | 'bugfix' | 'hotfix' - dependencies: List of dependency changes - is security related: bool """ gate id = str uuid.uuid4 risk = self. compute risk proposal decision = self. apply policy risk, proposal result = GateResult decision=decision, risk profile=risk, gate id=gate id, timestamp=time.time , metadata={ "change id": proposal.get "change id" , "change type": proposal.get "change type" , "diff stats": proposal.get "diff stats" , } self. audit result if decision == GateDecision.HUMAN REVIEW: self. review queue.append result return result def compute risk self, proposal: dict - RiskProfile: """Compute risk profile from proposal metadata.""" diff stats = proposal.get "diff stats", {} test results = proposal.get "test results", {} files changed = diff stats.get "files changed", 0 lines changed = diff stats.get "lines added", 0 + diff stats.get "lines removed", 0 affected = len proposal.get "affected services", deps changed = len proposal.get "dependencies", Scope: based on breadth of change scope score = min 1.0, files changed / 20 + lines changed / 500 Blast radius: number of services + dependency changes blast radius = min 1.0, affected / 5 + deps changed / 3 Reversibility: migrations and schema changes are harder to reverse reversibility = 1.0 if proposal.get "is security related" : reversibility -= 0.3 if deps changed 0: reversibility -= 0.2 if test results.get "failed", 0 0: reversibility -= 0.2 reversibility = max 0.0, reversibility Novelty: new files or new dependency patterns novelty = min 1.0, files changed / 15 + deps changed / 5 Coverage delta cov before = test results.get "coverage before", 0.0 cov after = test results.get "coverage after", 0.0 cov delta = cov after - cov before return RiskProfile scope score=scope score, blast radius=blast radius, reversibility=reversibility, novelty=novelty, confidence=proposal.get "agent confidence", 0.5 , test coverage delta=cov delta, def apply policy self, risk: RiskProfile, proposal: dict - GateDecision: """ Apply deterministic policy rules. These are NOT learned — they are explicit, auditable, and versioned. """ Hard blocks: never auto-approve these if proposal.get "is security related" and risk.blast radius 0.5: return GateDecision.BLOCK if risk.composite risk self.config.get "hard block threshold", 0.8 : return GateDecision.BLOCK if proposal.get "test results", {} .get "failed", 0 0: return GateDecision.BLOCK Escalation: unusual patterns require senior review if risk.composite risk self.config.get "escalation threshold", 0.6 : return GateDecision.ESCALATE Human review: moderate risk requires a human if risk.composite risk self.config.get "review threshold", 0.4 : return GateDecision.HUMAN REVIEW Coverage regression check if risk.test coverage delta < self.config.get "min coverage delta", -0.02 : return GateDecision.HUMAN REVIEW Low-risk changes: auto-approve with logging if risk.composite risk <= self.config.get "auto approve threshold", 0.3 : return GateDecision.AUTO APPROVE return GateDecision.HUMAN REVIEW def audit self, result: GateResult : """Append to immutable audit log.""" entry = { "gate id": result.gate id, "decision": result.decision.value, "composite risk": result.risk profile.composite risk, "timestamp": result.timestamp, "metadata": result.metadata, } self.audit log.append entry In production: append to append-only log store S3, Kafka, etc. The gate engine's power comes from its policy being explicit and versioned. Store it alongside your infrastructure code: gates/policy.yaml version: "2.3.1" last updated: "2025-01-15" approved by: "platform-security-team" thresholds: auto approve: 0.30 Below this: machine approves review: 0.40 Between review and escalation: human reviews escalation: 0.60 Between escalation and hard block: senior review hard block: 0.80 Above this: never auto-approve coverage: min coverage delta: -0.02 Allow 2% regression before requiring review require green tests: true overrides: Emergency hotfix path: faster but still gated hotfix: review threshold: 0.55 max files changed: 10 require: "oncall-approval" Security changes: always require human review regardless of score security: always review: true require: "security-team-approval" Schema migrations: elevated scrutiny migrations: review threshold: 0.25 require: "db-team-approval" max tables affected: 3 review routing: default reviewers: "platform-team" max review time hours: 4 escalation after hours: 2 backup reviewers: "senior-engineers-pool" audit: retention days: 365 export format: "jsonl" destination: "s3://org-audit-logs/gate-decisions/" The gate engine doesn't live in isolation — it hooks into your existing pipeline. Here's how to wire it into a GitHub Actions workflow: .github/workflows/agent-gate.yml name: Agent Gate Evaluation on: pull request: types: opened, synchronize, reopened jobs: gate-evaluation: runs-on: ubuntu-latest permissions: contents: read pull-requests: write steps: - uses: actions/checkout@v4 - name: Run Gate Evaluation id: gate run: | python scripts/run gate.py \ --proposal "${{ github.event.pull request.number }}" \ --policy gates/policy.yaml \ --output gate-result.json - name: Process Gate Decision run: | DECISION=$ jq -r '.decision' gate-result.json RISK=$ jq -r '.composite risk' gate-result.json GATE ID=$ jq -r '.gate id' gate-result.json case "$DECISION" in "auto approve" echo "✅ Auto-approved risk: $RISK " echo "GATE STATUS=approved" $GITHUB ENV ;; "human review" echo "🔍 Requires human review risk: $RISK " echo "GATE STATUS=needs review" $GITHUB ENV Create review request python scripts/request review.py \ --gate-id "$GATE ID" \ --pr "${{ github.event.pull request.number }}" \ --risk "$RISK" exit 1 ;; "escalate" echo "⚠️ Escalated for senior review risk: $RISK " python scripts/escalate.py \ --gate-id "$GATE ID" \ --pr "${{ github.event.pull request.number }}" exit 1 ;; "block" echo "🚫 Blocked by policy risk: $RISK " python scripts/block pr.py \ --gate-id "$GATE ID" \ --pr "${{ github.event.pull request.number }}" \ --reason "Exceeds hard block threshold" exit 1 ;; esac - name: Record Audit Entry if: always run: | python scripts/audit.py \ --gate-result gate-result.json \ --destination "s3://org-audit-logs/gate-decisions/" A gate is only as effective as the quality of human review it enables. The review interface must be purpose-built: showing reviewers exactly what they need to decide, in the minimum time required. @dataclass class ReviewPayload: """ What a human reviewer sees. Optimized for 2-minute decisions on low-risk items and 15-minute decisions on complex items. """ Summary visible immediately change title: str change type: str risk score: float risk factors: list str "3 services affected", "new dependency" recommended action: str "Approve", "Request changes", "Reject" Context one click away diff summary: str "12 files, +87 lines, -23 lines" test results: str "All 342 tests passing, coverage: 87% → 88%" affected services: list str agent rationale: str Why the agent made this change Deep dive expandable full diff: str agent confidence breakdown: dict similar past changes: list dict "3 similar changes approved in last 30d" Actions approve: callable request changes: callable reject: callable escalate: callable Not all human reviews are equal. The interface should adapt to the risk level: class ReviewWorkflow: """ Adapts the review experience based on risk tier. """ def get review mode self, risk: RiskProfile - str: if risk.composite risk < 0.4: return "quick approve" 1-click with diff preview elif risk.composite risk < 0.6: return "standard review" Full diff + test results elif risk.composite risk < 0.8: return "deep review" Context, history, architecture impact else: return "adversarial review" Second reviewer + architecture review def build quick approve payload self, result: GateResult - dict: """ For low-risk changes: show the reviewer everything they need to approve in under 60 seconds. """ return { "mode": "quick approve", "summary": self. summarize change result , "risk factors": self. extract risk factors result.risk profile , "diff preview": self. get diff preview result, max lines=50 , "test status": "✅ All passing", "quick actions": "Approve", "Flag for deeper review" , "confidence note": f"Agent confidence: {result.risk profile.confidence:.0%} | " f"Risk score: {result.risk profile.composite risk:.2f}" , } def build adversarial review payload self, result: GateResult - dict: """ For high-risk changes: structured adversarial review. Reviewer is prompted to actively try to find problems. """ return { "mode": "adversarial review", "summary": self. summarize change result , "risk factors": self. extract risk factors result.risk profile , "questions": "Could this change cause a production outage?", "Is there a safer way to achieve the same goal?", "Are the tests actually testing the right behavior?", "What happens if this runs during peak traffic?", "Can we roll this back in under 5 minutes?", , "required findings": "Rollback procedure verified", "Monitoring alerts identified", "Blast radius confirmed", , "second reviewer required": True, "architecture review required": True, "full diff": self. get full diff result , } A gate system without observability is a black box. You need to answer: Is the gate blocking too much? Too little? Are reviewers rubber-stamping? Is the risk model calibrated? class GateMetrics: """ Metrics that answer the critical questions about gate effectiveness. """ def init self, audit log: list dict : self.audit log = audit log def get health metrics self - dict: return { "decision distribution": self. decision distribution , "review latency": self. review latency stats , "override rate": self. override rate , "false positive rate": self. estimate false positives , "auto approve pass rate": self. auto approve quality , "risk calibration": self. calibration check , } def decision distribution self - dict: """What percentage of changes get each decision?""" total = len self.audit log if total == 0: return {} counts = {} for entry in self.audit log: decision = entry "decision" counts decision = counts.get decision, 0 + 1 return {k: v / total for k, v in counts.items } def review latency stats self - dict: """How long do human reviews take? Are we blocking velocity?""" review times = entry "review time seconds" for entry in self.audit log if entry "decision" in "human review", "escalate" and "review time seconds" in entry if not review times: return {"count": 0} review times.sort return { "count": len review times , "p50 seconds": review times len review times // 2 , "p95 seconds": review times int len review times 0.95 , "p99 seconds": review times int len review times 0.99 , "max seconds": review times -1 , } def override rate self - dict: """ How often do humans override the gate's recommendation? High override rate = model needs recalibration. """ overrides = entry for entry in self.audit log if entry.get "gate decision" = entry.get "final decision" total = len self.audit log return { "total overrides": len overrides , "override rate": len overrides / total if total 0 else 0, "by gate decision": self. overrides by decision overrides, total , } def auto approve quality self - dict: """ Of changes that were auto-approved, how many were later reverted or caused incidents? """ auto approved = entry for entry in self.audit log if entry "decision" == "auto approve" incidents = entry for entry in auto approved if entry.get "caused incident", False reverts = entry for entry in auto approved if entry.get "was reverted", False total = len auto approved return { "total auto approved": total, "incident rate": len incidents / total if total 0 else 0, "revert rate": len reverts / total if total 0 else 0, "target incident rate": 0.001, < 0.1% } def calibration check self - dict: """ Is the risk score actually predictive of outcomes? If high-risk changes rarely cause problems, thresholds are too conservative. If low-risk changes cause incidents, thresholds are too permissive. """ risk buckets = {"low": , "medium": , "high": } for entry in self.audit log: risk = entry.get "composite risk", 0 if risk < 0.4: risk buckets "low" .append entry elif risk < 0.7: risk buckets "medium" .append entry else: risk buckets "high" .append entry return { bucket: { "count": len entries , "incident rate": sum 1 for e in entries if e.get "caused incident", False / len entries if entries else 0, "revert rate": sum 1 for e in entries if e.get "was reverted", False / len entries if entries else 0, } for bucket, entries in risk buckets.items } The gate's thresholds shouldn't be static. They should adapt based on what actually happens: class GateCalibrator: """ Periodically recalibrates gate thresholds based on outcomes. Runs as a scheduled job, not in the hot path. """ def init self, config: dict : self.config = config self.min samples per bucket = 100 self.lookback days = 30 def propose threshold adjustments self, metrics: dict - list dict : """ Generate proposed threshold changes with justification. These proposals go through their own gate human approval . """ proposals = calibration = metrics.get "risk calibration", {} override data = metrics.get "override rate", {} auto quality = metrics.get "auto approve quality", {} Check if auto-approve is too permissive if auto quality.get "incident rate", 0 auto quality.get "target incident rate", 0.001 : proposals.append { "parameter": "auto approve threshold", "current": self.config.get "thresholds", {} .get "auto approve", 0.3 , "proposed": self.config.get "thresholds", {} .get "auto approve", 0.3 - 0.05, "reason": f"Auto-approved changes caused incidents at rate " f"{auto quality 'incident rate' :.4f} target: " f"{auto quality 'target incident rate' :.4f} . " f"Lowering threshold to reduce auto-approve volume." , "confidence": "high" if auto quality.get "total auto approved", 0 500 else "medium", } Check if review threshold is too conservative if override data.get "override rate", 0 0.3: proposals.append { "parameter": "review threshold", "current": self.config.get "thresholds", {} .get "review", 0.4 , "proposed": self.config.get "thresholds", {} .get "review", 0.4 + 0.05, "reason": f"Reviewers override gate decisions {override data 'override rate' :.1%} " f"of the time. Gate may be too conservative — consider raising threshold." , "confidence": "medium", } Check risk bucket calibration for bucket, stats in calibration.items : if stats.get "count", 0 < self.min samples per bucket: continue if bucket == "high" and stats.get "incident rate", 0 < 0.01: proposals.append { "parameter": "hard block threshold", "current": self.config.get "thresholds", {} .get "hard block", 0.8 , "proposed": self.config.get "thresholds", {} .get "hard block", 0.8 + 0.05, "reason": f"High-risk bucket risk 0.7 has incident rate " f"{stats 'incident rate' :.4f}, well below expected. " f"Hard block threshold may be too aggressive." , "confidence": "medium", } return proposals No system is perfect. The gate engine itself can fail, and the failure modes matter: class ResilientGateEngine: """ Wraps the gate engine with failure handling. Key principle: the gate's failure mode must be safe. """ def init self, engine: GateEngine, config: dict : self.engine = engine self.config = config self.failure mode = config.get "failure mode", "fail closed" Options: "fail closed" block all , "fail open" approve all , "fail review" require human review def evaluate safe self, proposal: dict - GateResult: """ Evaluate with guaranteed safe failure behavior. """ try: result = self.engine.evaluate proposal self. record success result return result except Exception as e: self. record failure e return self. handle failure proposal, e def handle failure self, proposal: dict, error: Exception - GateResult: """ When the gate can't evaluate, what do we do? """ gate id = str uuid.uuid4 timestamp = time.time if self.failure mode == "fail closed": Safest: block everything when uncertain decision = GateDecision.BLOCK metadata = {"failure reason": str error , "mode": "fail closed"} elif self.failure mode == "fail open": Riskiest: approve everything only for non-critical paths decision = GateDecision.AUTO APPROVE metadata = {"failure reason": str error , "mode": "fail open"} else: fail review Default: require human judgment decision = GateDecision.HUMAN REVIEW metadata = {"failure reason": str error , "mode": "fail review"} result = GateResult decision=decision, risk profile=RiskProfile , Empty — couldn't compute gate id=gate id, timestamp=timestamp, metadata=metadata, self. audit failure result, error self. alert on failure error return result def alert on failure self, error: Exception : """ Gate failures are operational incidents. Alert the team immediately. """ alert = { "severity": "warning", "source": "gate engine", "error": str error , "failure mode": self.failure mode, "timestamp": time.time , "action": "Gate engine unavailable — operating in degraded mode", } In production: send to PagerDuty, Slack, etc. print f"🚨 GATE ENGINE FAILURE: {alert}" class GateCircuitBreaker: """ Prevents cascading failures when the gate engine is repeatedly failing. After N consecutive failures, opens the circuit and applies a pre-defined emergency policy. """ def init self, max failures: int = 5, recovery seconds: int = 300 : self.max failures = max failures self.recovery seconds = recovery seconds self.consecutive failures = 0 self.last failure time = 0 self.state = "closed" closed, open, half open def should use fallback self - bool: if self.state == "open": if time.time - self.last failure time self.recovery seconds: self.state = "half open" return True Try once to see if it recovered return True Still open, use fallback if self.state == "half open": return True Let one through to test return False Closed — use normal path def record failure self : self.consecutive failures += 1 self.last failure time = time.time if self.consecutive failures = self.max failures: self.state = "open" def record success self : self.consecutive failures = 0 self.state = "closed" def get fallback policy self - dict: """ Emergency policy when circuit is open. Conservative by default — requires human review for everything. """ return { "all changes require review": True, "auto approve threshold": 0.0, Effectively disabled "escalation threshold": 0.0, Everything escalates "alert team": True, "message": "Gate engine circuit breaker OPEN. All changes require " "manual review until engine recovers." , } Different environments warrant different gate strictness: gates/environments.yaml environments: development: thresholds: auto approve: 0.70 review: 0.70 escalation: 1.0 hard block: 1.0 review required: false failure mode: "fail open" rationale: "Developer velocity matters more than safety in dev" staging: thresholds: auto approve: 0.40 review: 0.55 escalation: 0.75 hard block: 0.90 review required: true failure mode: "fail review" rationale: "Balance safety and velocity; staging is for catching issues" production: thresholds: auto approve: 0.20 review: 0.35 escalation: 0.55 hard block: 0.70 review required: true failure mode: "fail closed" rationale: "Safety first — production changes affect real users" critical production: For payment systems, authentication, etc. thresholds: auto approve: 0.0 review: 0.25 escalation: 0.40 hard block: 0.55 review required: true double review: true failure mode: "fail closed" rationale: "Zero tolerance for automated changes in critical systems" class AgentRateLimiter: """ Prevents agent-driven change flooding. Even if individual changes pass the gate, too many changes in rapid succession create systemic risk. """ def init self, config: dict : self.window seconds = config.get "window seconds", 3600 self.max changes per window = config.get "max changes per window", 10 self.max concurrent reviews = config.get "max concurrent reviews", 3 self. change timestamps: list float = self. active reviews: int = 0 def can proceed self, proposal: dict - tuple bool, str : """Check if this change can proceed given rate limits.""" now = time.time Clean expired timestamps self. change timestamps = t for t in self. change timestamps if now - t < self.window seconds Rate limit check if len self. change timestamps = self.max changes per window: return False, f"Rate limit exceeded: {self.max changes per window} changes " f"per {self.window seconds}s window. " f"Try again in {self.window seconds - now - self. change timestamps 0 :.0f}s" Concurrent review check if self. active reviews = self.max concurrent reviews: return False, f"Too many concurrent reviews {self.max concurrent reviews} max . " f"Please wait for existing reviews to complete." self. change timestamps.append now return True, "OK" def record review start self : self. active reviews += 1 def record review complete self : self. active reviews = max 0, self. active reviews - 1 Here's the complete integration showing how all pieces connect: class AgentSafePipeline: """ Complete pipeline orchestrating agent proposals through rate limiting, gate evaluation, human review, and deployment. """ def init self, config: dict : self.config = config self.gate engine = GateEngine config self.resilient gate = ResilientGateEngine self.gate engine, config.get "failure handling", {} self.circuit breaker = GateCircuitBreaker max failures=config.get "circuit breaker", {} .get "max failures", 5 , recovery seconds=config.get "circuit breaker", {} .get "recovery seconds", 300 , self.rate limiter = AgentRateLimiter config.get "rate limits", {} self.metrics = GateMetrics self.gate engine.audit log def process proposal self, proposal: dict - GateResult: """ Full pipeline for processing an agent proposal. """ Step 1: Rate limiting can proceed, reason = self.rate limiter.can proceed proposal if not can proceed: return GateResult decision=GateDecision.BLOCK, risk profile=RiskProfile , gate id=str uuid.uuid4 , timestamp=time.time , metadata={"reason": reason, "stage": "rate limit"} Step 2: Circuit breaker check if self.circuit breaker.should use fallback : fallback = self.circuit breaker.get fallback policy result = GateResult decision=GateDecision.HUMAN REVIEW, risk profile=RiskProfile , gate id=str uuid.uuid4 , timestamp=time.time , metadata={"reason": "circuit breaker open", "fallback": fallback} return result Step 3: Gate evaluation try: result = self.resilient gate.evaluate safe proposal self.circuit breaker.record success Step 4: If human review needed, initiate review flow if result.decision in GateDecision.HUMAN REVIEW, GateDecision.ESCALATE : self.rate limiter.record review start result = self. initiate review result, proposal return result except Exception as e: self.circuit breaker.record failure return