# Configuring the KillSwitch for Emergency Agent Termination in SRE Workflows > Implement emergency agent termination with the KillSwitch in your SRE workflows. Securely halt autonomous agents and ensure system stability with this critical safety net. - Repository: [Microsoft/agent-governance-toolkit](https://github.com/microsoft/agent-governance-toolkit) - Tags: how-to-guide - Published: 2026-05-29 --- **The KillSwitch provides immediate, hard termination of autonomous agents with configurable callback timeouts, immutable audit trails, and automatic compensation of in-flight saga steps, serving as the final safety net in multi-layer SRE defensive architectures.** Configuring the KillSwitch for emergency agent termination is essential for Site Reliability Engineering teams operating autonomous AI systems at scale. This implementation guide covers the microsoft/agent-governance-toolkit codebase, demonstrating how to deploy the "big red button" that instantly stops rogue agents while preserving forensic data for post-mortem analysis. The architecture is intentionally stateless beyond in-memory history, making it safe to embed directly into Python services running the Agent Hypervisor runtime. ## KillSwitch Architecture and Core Components The KillSwitch implementation resides in [`agent-governance-python/agent-hypervisor/src/hypervisor/security/kill_switch.py`](https://github.com/microsoft/agent-governance-toolkit/blob/main/agent-governance-python/agent-hypervisor/src/hypervisor/security/kill_switch.py). It operates as a central registry and execution manager with the following key structures: - **KillSwitch class**: Central manager that registers agent termination callbacks, handles substitute routing, and executes kills with a `DEFAULT_CALLBACK_TIMEOUT_SECONDS = 5.0` wall-clock timeout to prevent hung callbacks from blocking termination. - **KillReason enum**: Structured taxonomy including `BEHAVIORAL_DRIFT`, `RATE_LIMIT`, `RING_BREACH`, `MANUAL`, `QUARANTINE_TIMEOUT`, and `SESSION_TIMEOUT` for searchable audit logs. - **StepHandoff dataclass**: Tracks the disposition of each in-flight saga step, marking them either `HANDED_OFF` (Enterprise edition) or `COMPENSATED` (public preview). - **KillResult dataclass**: Immutable record containing `kill_id`, timestamps, `agent_did`, `reason`, `handoffs` list, and `terminated` boolean indicating callback success. - **_agents registry**: Private dictionary mapping agent DIDs to callable termination functions. - **_substitutes registry**: Maps session IDs to backup agent DIDs for automatic failover (Enterprise feature). - **_kill_history**: In-memory list of `KillResult` objects exposed via the `kill_history` property and `total_kills` counter. ## Step-by-Step KillSwitch Configuration ### Registering an Agent Before issuing kills, register each agent with its termination callback. The callable is stored in `_agents` and invoked when `kill()` is requested. ```python from hypervisor.security.kill_switch import KillSwitch def my_agent_termination(): # Clean-up resources, shut down subprocesses, close sockets import os os._exit(1) # Hard exit example kill_switch = KillSwitch() kill_switch.register_agent("did:example:my-agent", my_agent_termination) ``` ### (Optional) Registering a Substitute Enterprise editions can register backup agents to receive handoffs from terminated primaries. ```python kill_switch.register_substitute( session_id="session-123", agent_did="did:example:backup" ) ``` When `kill()` executes, `KillSwitch._find_substitute()` checks for substitutes matching the session ID. If found, the system creates `StepHandoff` entries with `status=HANDED_OFF`; otherwise, steps are marked `COMPENSATED`. ### Issuing an Emergency Kill Invoke `kill()` with the agent DID, session ID, reason, and any in-flight saga steps requiring compensation. ```python from hypervisor.security.kill_switch import KillReason result = kill_switch.kill( agent_did="did:example:rogue-agent", session_id="session-001", reason=KillReason.MANUAL, in_flight_steps=[ {"step_id": "create-pr", "saga_id": "deploy-42"}, {"step_id": "run-tests", "saga_id": "deploy-42"}, ], details="Agent attempted unauthorized API access" ) print(f"Killed {result.agent_did} for {result.reason}") print(f"Audit ID: {result.kill_id}") print(f"Terminated successfully: {result.terminated}") ``` The method builds `StepHandoff` objects, runs the termination callback **outside the lock** with the 5-second timeout via `_invoke_callback_with_timeout`, and unregisters the agent regardless of callback success. If the callback exceeds the timeout or raises an exception, `result.terminated` returns `False` but the kill is still recorded. ### Querying the Audit Trail Access kill history through the `kill_history` property, which returns a shallow copy (`list(self._kill_history)`) for safe iteration. ```python for entry in kill_switch.kill_history: print(f"[{entry.timestamp}] {entry.agent_did} → {entry.reason} (ID {entry.kill_id})") print(f"Total kills: {kill_switch.total_kills}") ``` ## Integrating KillSwitch into SRE Enforcement Pipelines The KillSwitch sits at the bottom layer of defense-in-depth architectures. According to the tutorial in [`docs/tutorials/14-kill-switch-and-rate-limiting.md`](https://github.com/microsoft/agent-governance-toolkit/blob/main/docs/tutorials/14-kill-switch-and-rate-limiting.md), the standard SRE flow wires the KillSwitch with `RingBreachDetector`, `AgentRateLimiter`, and `RingElevationManager` inside an `enforce_action` function: 1. **Circuit-breaker check**: If `breach_detector.is_breaker_tripped()` returns True, invoke `kill_switch.kill(..., reason=KillReason.RING_BREACH)`. 2. **Rate-limit evaluation**: Persistent violations from `AgentRateLimiter` trigger `KillReason.RATE_LIMIT`. 3. **Breach recording**: High-severity breaches detected via `breach_detector.record_call()` result in immediate termination with `KillReason.RING_BREACH`. This pattern ensures the KillSwitch acts as the final arbiter when softer controls fail. ## Production Implementation Examples ### Minimal 10-Line Configuration ```python from hypervisor.security.kill_switch import KillSwitch, KillReason kill_switch = KillSwitch() result = kill_switch.kill( agent_did="did:example:rogue-agent", session_id="session-001", reason=KillReason.MANUAL, details="Agent exceeded budget and accessed unauthorized APIs", ) print(f"Kill ID: {result.kill_id}") print(f"Total kills: {kill_switch.total_kills}") ``` ### Full Defense-in-Depth Pipeline ```python from hypervisor.models import ExecutionRing from hypervisor.security.kill_switch import KillSwitch, KillReason from hypervisor.security.rate_limiter import AgentRateLimiter, RateLimitExceeded from hypervisor.rings.breach_detector import RingBreachDetector, BreachSeverity from hypervisor.rings.elevation import RingElevationManager kill_switch = KillSwitch() rate_limiter = AgentRateLimiter() breach_detector = RingBreachDetector(window_seconds=60, baseline_rate=10.0) elevation_manager = RingElevationManager() def enforce_action(agent_did: str, session_id: str, agent_ring: ExecutionRing, target_ring: ExecutionRing, in_flight_steps: list[dict] | None = None) -> bool: """Return True if allowed; otherwise kill and return False.""" # Circuit-breaker layer if breach_detector.is_breaker_tripped(agent_did, session_id): kill_switch.kill(agent_did, session_id, KillReason.RING_BREACH, in_flight_steps, details="Circuit breaker tripped") return False # Rate-limit layer if not rate_limiter.try_check(agent_did, session_id, agent_ring): stats = rate_limiter.get_stats(agent_did, session_id) if stats and stats.rejected_requests > 10: kill_switch.kill(agent_did, session_id, KillReason.RATE_LIMIT, in_flight_steps, details=f"Rate limit exceeded {stats.rejected_requests} times") return False # Breach detection layer breach = breach_detector.record_call(agent_did, session_id, agent_ring, target_ring) if breach and breach.severity in (BreachSeverity.HIGH, BreachSeverity.CRITICAL): kill_switch.kill(agent_did, session_id, KillReason.RING_BREACH, in_flight_steps, details=f"Breach: {breach.details}") return False # Ring elevation check effective_ring = elevation_manager.get_effective_ring(agent_did, session_id, agent_ring) if effective_ring.value > target_ring.value: return False return True ``` ### HTTP Edge Service Integration ```python from agentmesh.services.rate_limiter import RateLimiter from agentmesh.services.rate_limit_middleware import ( RateLimitMiddleware, SimpleRequest, SimpleResponse, HEADER_AGENT_DID ) from hypervisor.security.kill_switch import KillSwitch, KillReason http_limiter = RateLimiter(global_rate=100, global_capacity=200, per_agent_rate=10, per_agent_capacity=20) http_middleware = RateLimitMiddleware(http_limiter) kill_switch = KillSwitch() def handler(request: SimpleRequest) -> SimpleResponse: agent_did = request.headers.get(HEADER_AGENT_DID, "anonymous") if not http_limiter.allow(agent_did): kill_switch.kill(agent_did, "http-session", KillReason.RATE_LIMIT, details="Too many HTTP requests") return SimpleResponse(status_code=429, body={"error": "Too Many Requests"}) return SimpleResponse(status_code=200, body={"msg": "OK"}) req = SimpleRequest(headers={HEADER_AGENT_DID: "did:mesh:agent-1"}, path="/api/do", method="POST") resp = http_middleware.handle(req, handler) ``` ## Summary - The **KillSwitch** class in [`hypervisor/security/kill_switch.py`](https://github.com/microsoft/agent-governance-toolkit/blob/main/hypervisor/security/kill_switch.py) provides stateless, immediate agent termination with a mandatory 5-second callback timeout. - **KillReason** enums standardize termination causes for searchable audit trails covering behavioral drift, rate limits, ring breaches, and manual interventions. - The **StepHandoff** and **KillResult** dataclasses capture immutable records of compensation actions and termination outcomes. - **Registration** requires mapping agent DIDs to termination callables; substitutes support enterprise failover scenarios. - **Integration** follows a defense-in-depth pattern where circuit breakers and rate limiters invoke the KillSwitch only when softer controls fail. ## Frequently Asked Questions ### What happens if the termination callback hangs or crashes? The KillSwitch enforces a **5.0-second wall-clock timeout** via `_invoke_callback_with_timeout`. If the callback exceeds this limit or raises an exception, the kill is still recorded in the audit trail with `result.terminated` set to `False`, and the agent is unregistered to prevent subsequent calls. This ensures a hung callback cannot block the emergency termination. ### Can I retrieve historical kill data after a service restart? The `_kill_history` is stored in-memory only. According to the source implementation in [`kill_switch.py`](https://github.com/microsoft/agent-governance-toolkit/blob/main/kill_switch.py), the history does not persist to disk or external storage. For persistent audit trails in production SRE workflows, you should export `kill_history` to a logging pipeline or database before process termination. ### How does the substitute agent feature work in Enterprise editions? When `register_substitute(session_id, agent_did)` is called, the KillSwitch stores the mapping in `_substitutes`. During `kill()` execution, `_find_substitute()` queries this registry by session ID. If a substitute exists, the system creates `StepHandoff` records with `status=HANDED_OFF` rather than `COMPENSATED`, allowing the backup agent to assume the terminated agent's in-flight work. In the public preview, this path is disabled and all steps default to `COMPENSATED`. ### What is the difference between RING_BREACH and RATE_LIMIT kill reasons? **RING_BREACH** indicates the agent violated execution ring boundaries (e.g., attempting privileged operations from an untrusted ring), typically detected by `RingBreachDetector`. **RATE_LIMIT** indicates the agent exceeded configured request thresholds managed by `AgentRateLimiter`. Both trigger immediate termination, but the distinction enables SRE teams to categorize incidents by security posture (breach) versus operational capacity (throttling) during post-mortem analysis.