Configuring the KillSwitch for Emergency Agent Termination in SRE Workflows

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. 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.

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.

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.

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.

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, 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

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

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

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 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, 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}