# Python 3.10+
pip install "langgraph>=0.2.55" pytest redis   
# Optional for the production PostgresSaver:
pip install langgraph-checkpoint-postgres

import time, random
def tool_call(query: str) -> str:
    # Pretend that each call costs money and takes time to make
    time.sleep(0.2)
    return f"result for {query}"
def naive_agent(goal: str):
    """The dangerous version: loop until it 'decides' it's done.
    A drifting model may never decide it is done!"""
    step = 0
    while True:                         # <-- No termination guarantee
        step += 1
        out = tool_call(goal)
        # A real agent asks the LLM 'are we done?'. A looping one keeps saying 'No'.
        done = random.random() < 0.0    # Force it to never finish, for the demo
        print(f"step {step}: {out}")
        if done:
            break
naive_agent("summarise market data")  # <-- would run forever

import os
def kill_switch_enabled(agent_id: str) -> bool:
    # External source of truth. Here: an env var. In prod: Redis / feature flag.
    # Missing value defaults to 'enabled'.
    return os.environ.get(f"AGENT_{agent_id}_KILL", "off") != "on"
def guarded_agent(agent_id: str, goal: str, max_steps: int = 50):
    for step in range(1, max_steps + 1):     # Belt: a hard iteration cap
        if not kill_switch_enabled(agent_id): # Braces: the external kill switch
            print(f"step {step}: kill switch thrown — halting")
            return "halted"
        print(f"step {step}: {tool_call(goal)}")
    print(f"reached max_steps ({max_steps}) — stopping")
    return "max_steps"

# macOS / Linux: Set the flag and launch in one command
AGENT_demo_KILL=on python guarded_agent.py
# Windows PowerShell
$env:AGENT_demo_KILL = "on"; python guarded_agent.py

import os, redis
r = redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))
def kill_switch_enabled_live(agent_id: str) -> bool:
    """Read the flag from Redis on EVERY check so changes apply immediately."""
    return r.get(f"agent:{agent_id}:kill_switch") != b"on"
def guarded_agent_live(agent_id: str, goal: str, max_steps: int = 50):
    for step in range(1, max_steps + 1):           # Belt: hard iteration cap
        if not kill_switch_enabled_live(agent_id):  # Braces: live external flag
            print(f"step {step}: kill switch thrown — halting")
            return "halted"
        print(f"step {step}: {tool_call(goal)}")
    print(f"reached max_steps ({max_steps}) — stopping")
    return "max_steps"
if __name__ == "__main__":
    r.delete("agent:demo:kill_switch")             # Start with the switch clear
    print("return:", guarded_agent_live("demo", "summarise market data"))

redis-cli SET agent:demo:kill_switch on      # Halt
redis-cli DEL agent:demo:kill_switch         # Release

step 1: result for summarise market data
step 2: result for summarise market data
step 3: result for summarise market data
step 4: result for summarise market data
step 5: kill switch thrown - halting
return: halted

step 1: result for summarise market data
step 2: result for summarise market data
step 3: result for summarise market data
step 4: result for summarise market data
step 5: result for summarise market data
reached max_steps (5) - stopping
return: max_steps

import redis
def kill_switch_enabled_redis(agent_id: str, redis_client) -> bool:
    # Check Redis instead of env vars - works across multiple processes
    return redis_client.get(f"agent:{agent_id}:kill_switch") != b"on"
# Usage: export REDIS_URL=redis://localhost:6379
redis_client = redis.from_url(os.environ.get("REDIS_URL", "redis://localhost"))

import os, time, json, hmac, hashlib, random
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta, timezone
from threading import Lock
from collections import deque
import redis
# One shared client, created once. decode_responses=False (the default)
# Means Redis returns bytes, which is why the checks compare with b"on".
REDIS = redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))
def kill_switch_check(agent_id: str, client=REDIS) -> tuple[bool, str]:
    """Return (allowed, reason). Fails CLOSED: any error means stop."""
    try:
        val = client.get(f"agent:{agent_id}:kill_switch")
        if val == b"on":
            return False, "kill switch flag set to 'on'"
        return True, ""
    except Exception as e:
        print(f"⚠️  kill switch check failed ({e}) — defaulting to HALT")
        return False, f"kill switch check failed: {e}"
def kill_switch_enabled_failclosed(agent_id: str, client=REDIS) -> bool:
    """Boolean wrapper, kept for the tests and earlier examples."""
    return kill_switch_check(agent_id, client)[0]

def fleet_kill_switch(client, agent_id: str, team: str | None = None) -> bool:
    """True = allowed to run. Checks global, team and agent in one call.
    Key scheme (used everywhere in the post):
      kill_switch:global
      kill_switch:team:<name>
      agent:<id>:kill_switch
    """
    keys = ["kill_switch:global"]
    if team:
        keys.append(f"kill_switch:team:{team}")
    keys.append(f"agent:{agent_id}:kill_switch")
    try:
        return all(v != b"on" for v in client.mget(keys))
    except Exception as e:
        print(f"⚠️  fleet kill switch check failed ({e}) — defaulting to HALT")
        return False
# To halt every agent in the organisation from a single command:
#   redis-cli SET kill_switch:global on
#
# To halt all agents in the finance team:
#   redis-cli SET kill_switch:team:finance on
#
# To release:
#   redis-cli DEL kill_switch:global

# In production, load from a secrets manager. The agent must not have it.
AUDIT_KEY = os.environ.get("AUDIT_HMAC_KEY", "dev-only-key").encode()
@dataclass(frozen=True)
class KillSwitchEvent:
    """Audit record. frozen=True blocks edits in code; the HMAC signature
    detects edits to the stored copy (a plain hash could just be recomputed)."""
    timestamp:      str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    agent_id:       str = ""
    trigger_type:   str = ""   # external_flag | spend_ceiling | pattern_breaker | human_reject
                               # | iteration_cap | approval_timeout | quarantine
    trigger_reason: str = ""
    agent_action:   str = ""
    agent_step:     int = 0
    agent_state:    str = ""   # halted | deferred | paused_awaiting_human | tripped | quarantined
    triggered_by:   str = ""   # system | operator:<name> | human_reviewer
    signature:      str = ""
    def _payload(self) -> bytes:
        body = {k: v for k, v in asdict(self).items() if k != "signature"}
        return json.dumps(body, sort_keys=True).encode()
    def __post_init__(self):
        sig = hmac.new(AUDIT_KEY, self._payload(), hashlib.sha256).hexdigest()
        object.__setattr__(self, "signature", sig)   # allowed once, at creation
    def verify(self) -> bool:
        expected = hmac.new(AUDIT_KEY, self._payload(), hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected, self.signature)
def log_kill_switch_event(event: KillSwitchEvent, sink: str = "stdout") -> str:
    record = json.dumps(asdict(event), indent=2)
    if sink == "stdout":
        print(f"📋 KILL SWITCH AUDIT LOG:\n{record}")
    # Production: write to an append-only store (CloudWatch, Splunk,
    # S3 Object Lock, or a table with INSERT-only permissions).
    return record
def guarded_agent_audited(agent_id: str, goal: str, max_steps: int = 50):
    for step in range(1, max_steps + 1):
        allowed, reason = kill_switch_check(agent_id)
        if not allowed:
            failed = reason.startswith("kill switch check failed")
            log_kill_switch_event(KillSwitchEvent(
                agent_id=agent_id,
                trigger_type="control_plane_failure" if failed else "external_flag",
                trigger_reason=reason,
                agent_action=f"about to execute tool_call('{goal}')",
                agent_step=step,
                agent_state="halted",
                triggered_by="system" if failed else "operator",
            ))
            print(f"step {step}: halting — {reason}")
            return "halted"
        print(f"step {step}: {tool_call(goal)}")
    log_kill_switch_event(KillSwitchEvent(
        agent_id=agent_id, trigger_type="iteration_cap",
        trigger_reason=f"reached max_steps ({max_steps})",
        agent_action=f"completed step {max_steps}", agent_step=max_steps,
        agent_state="halted", triggered_by="system",
    ))
    print(f"reached max_steps ({max_steps}) — stopping")
    return "max_steps"

# KILL SWITCH AUDIT LOG:
{
  "timestamp": "2026-10-24T12:00:00.000000+00:00",
  "agent_id": "demo",
  "trigger_type": "external_flag",
  "trigger_reason": "kill switch flag set to 'on'",
  "agent_action": "about to execute tool_call('summarise market data')",
  "agent_step": 5,
  "agent_state": "halted",
  "triggered_by": "operator",
  "signature": "e96a...e1db"
}

class SpendGate:
    """Control plane between the agent and its tools. Tracks spend in
    whole cents to avoid floating-point drift."""
    def __init__(self, daily_max_usd: float, cost_per_call: float, agent_id: str = "default"):
        self.daily_max_c = round(daily_max_usd * 100)
        self.cost_c = round(cost_per_call * 100)
        self.spent_c = 0
        self.agent_id = agent_id
        self.current_day = datetime.now(timezone.utc).date()
        self.lock = Lock()
        self.call_history = deque(maxlen=10_000)   # bounded
    @property
    def daily_max(self) -> float: return self.daily_max_c / 100
    @property
    def spent(self) -> float: return self.spent_c / 100
    def _reset_if_new_day(self):
        today = datetime.now(timezone.utc).date()
        if today != self.current_day:
            self.current_day, self.spent_c = today, 0
    def call(self, fn, *args, **kwargs):
        with self.lock:
            self._reset_if_new_day()
            if self.spent_c + self.cost_c > self.daily_max_c:
                raise RuntimeError(
                    f"budget ceiling ${self.daily_max:.2f} reached "
                    f"(spent ${self.spent:.2f}) — deferring to human")
            self.spent_c += self.cost_c
            self.call_history.append({"timestamp": datetime.now(timezone.utc),
                                      "cost": self.cost_c / 100,
                                      "total_spent": self.spent})
        return fn(*args, **kwargs)   # run the tool outside the lock
    def get_spend_rate(self, window_minutes: int = 5) -> float:
        cutoff = datetime.now(timezone.utc) - timedelta(minutes=window_minutes)
        recent = [x for x in self.call_history if x["timestamp"] > cutoff]
        return len(recent) / window_minutes if window_minutes else 0.0
gate = SpendGate(daily_max_usd=50.0, cost_per_call=0.10, agent_id="demo")
def budgeted_agent(goal: str):
    step = 0
    while True:
        step += 1
        try:
            out = gate.call(tool_call, goal)      # every call passes the gate first
        except RuntimeError as e:
            print(f"step {step}: {e}")
            return "deferred"
        print(f"step {step}: {out}  (spent ${gate.spent:.2f})")

step 1: result for summarise market data  (spent $0.10)
step 2: result for summarise market data  (spent $0.20)
step 3: result for summarise market data  (spent $0.30)
step 100: result for summarise market data  (spent $10.00)
step 200: result for summarise market data  (spent $20.00)
step 300: result for summarise market data  (spent $30.00)
step 400: result for summarise market data  (spent $40.00)
step 500: result for summarise market data  (spent $50.00)
step 501: budget ceiling $50.00 reached (spent $50.00) — deferring to human
return: deferred at step 501
total spent before stop: $50.00 over 500 successful calls

# Spend is stored in integer cents, so there is no float drift in Redis.
SPEND_SCRIPT = """
local spent = tonumber(redis.call('GET', KEYS[1]) or '0')
local cost  = tonumber(ARGV[1])
local cap   = tonumber(ARGV[2])
if spent + cost > cap then
  return {0, spent}
end
local new = redis.call('INCRBY', KEYS[1], cost)
redis.call('EXPIRE', KEYS[1], 172800) -- clean up after 2 days
return {1, new}
"""
class RedisSpendGate:
    """Shared, persistent spend gate. One budget per agent per UTC day."""
    def __init__(self, agent_id: str, redis_client, daily_max_usd: float = 50.0,
                 cost_per_call: float = 0.10):
        self.agent_id = agent_id
        self.redis = redis_client
        self.daily_max_c = round(daily_max_usd * 100)
        self.cost_c = round(cost_per_call * 100)
        self._spend = redis_client.register_script(SPEND_SCRIPT)
    @property
    def daily_max(self) -> float: return self.daily_max_c / 100
    def _key(self) -> str:
        day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        return f"agent:{self.agent_id}:spend:{day}"
    @property
    def spent(self) -> float:
        return int(self.redis.get(self._key()) or 0) / 100
    def call(self, fn, *args, **kwargs):
        try:
            allowed, total = self._spend(keys=[self._key()],
                                         args=[self.cost_c, self.daily_max_c])
        except redis.RedisError as e:
            raise RuntimeError(f"spend check failed ({e}) — deferring to human")
        if not int(allowed):
            raise RuntimeError(
                f"budget ceiling ${self.daily_max:.2f} reached "
                f"(spent ${int(total) / 100:.2f}) — deferring to human")
        return fn(*args, **kwargs)
# Usage: a drop-in replacement for SpendGate in budgeted_agent()
r = redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))
gate = RedisSpendGate("demo", r, daily_max_usd=50.0, cost_per_call=0.10)

from typing import TypedDict
from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
# Production: pip install langgraph-checkpoint-postgres, then
# from langgraph.checkpoint.postgres import PostgresSaver
class ToolState(TypedDict):
    tool_name: str
    approved: bool
    result: str | None
DESTRUCTIVE = {"delete_records", "send_email", "execute_trade"}
def approval_node(state: ToolState) -> ToolState:
    if state["tool_name"] not in DESTRUCTIVE:
        return {"approved": True}                 # low-risk tools pass straight through
    answer = interrupt({                           # <-- graph suspends here
        "message": f"Approve '{state['tool_name']}'?",
        "options": ["approve", "reject"],
    })
    return {"approved": answer == "approve"}
def execute_node(state: ToolState) -> ToolState:
    if not state["approved"]:
        return {"result": "Blocked by human."}
    return {"result": f"ran {state['tool_name']}"}
builder = StateGraph(ToolState)
builder.add_node("approval", approval_node)
builder.add_node("execute", execute_node)
builder.set_entry_point("approval")
builder.add_edge("approval", "execute")
builder.add_edge("execute", END)
# A checkpointer is REQUIRED — without it there's no state to resume from.
# MemorySaver for dev/testing only. Use PostgresSaver in production.
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "demo-1"}}  # ties the two calls together
first = graph.invoke(
    {"tool_name": "send_email", "approved": False, "result": None},
    config=config,
)
print(first)   # -> contains '__interrupt__': the graph is paused, waiting
# ...surface that question to a human, then resume:
final = graph.invoke(Command(resume="reject"), config=config)
print(final["result"])   # -> 'Blocked by human.'

import time
from langgraph.types import Command
APPROVAL_TIMEOUT_S = 300        # 5 minutes
pending_approvals = {}          # thread_id -> request time. In production: Redis/DB
def start_run(graph, thread_id: str, tool_name: str):
    """Start a run. If it pauses for approval, record when."""
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(
        {"tool_name": tool_name, "approved": False, "result": None},
        config=config,
    )
    if "__interrupt__" in result:
        pending_approvals[thread_id] = time.time()
        question = result["__interrupt__"][0].value["message"]
        print(f"[{thread_id}] waiting for human: {question}")
    return result
def human_responds(graph, thread_id: str, answer: str):
    """Called by your UI / Slack bot when a human answers."""
    if pending_approvals.pop(thread_id, None) is None:
        print(f"[{thread_id}] too late — already auto-rejected")
        return None
    config = {"configurable": {"thread_id": thread_id}}
    return graph.invoke(Command(resume=answer), config=config)
def expire_stale_approvals(graph):
    """Run every few seconds from a scheduler or background worker.
    Any approval older than the timeout is resumed with 'reject'."""
    now = time.time()
    for thread_id, asked_at in list(pending_approvals.items()):
        if now - asked_at > APPROVAL_TIMEOUT_S:
            pending_approvals.pop(thread_id)
            config = {"configurable": {"thread_id": thread_id}}
            final = graph.invoke(Command(resume="reject"), config=config)
            print(f"[{thread_id}] no response in {APPROVAL_TIMEOUT_S}s — "
                  f"auto-rejected: {final['result']}")
            # Log a KillSwitchEvent(trigger_type="approval_timeout", ...) here

[demo-timeout] waiting for human: Approve 'send_email'?
[demo-timeout] no response in 2s — auto-rejected: Blocked by human.

from collections import deque
import time
from threading import Lock
class PatternBreaker:
    """Detects when an agent repeats the same action too many times.
    Catches slow loops that slip past rate limiters."""
    
    def __init__(self, agent_id: str, window_s: float = 2.0, max_repeats: int = 5):
        self.agent_id = agent_id
        self.window_s = window_s
        self.max_repeats = max_repeats
        self.history = deque(maxlen=1000)
        self.lock = Lock()
        self.allowlist = set()  # actions that legitimately repeat (polling, pagination)
    def add_to_allowlist(self, action: str):
        """Legitimate repeating actions (e.g., 'retry_request', 'fetch_page') won't trip."""
        self.allowlist.add(action)
    def should_trip(self, action: str) -> bool:
        """Trip if the same action occurs > max_repeats times within window_s."""
        if action in self.allowlist:
            return False
        
        now = time.time()
        with self.lock:
            self.history.append((action, now))
            recent = [x for x in self.history if now - x[1] < self.window_s]
            repeated = [x for x in recent if x[0] == action]
            
            if len(repeated) > self.max_repeats:
                return True
        return False
# Per-agent instance to isolate agents from each other
breakers = {}  # in production, keyed by agent_id
def get_breaker(agent_id: str) -> PatternBreaker:
    if agent_id not in breakers:
        breakers[agent_id] = PatternBreaker(agent_id, window_s=2.0, max_repeats=5)
    return breakers[agent_id]
def pattern_guarded_agent(agent_id: str, goal: str, max_steps: int = 50):
    breaker = get_breaker(agent_id)
    
    for step in range(1, max_steps + 1):
        action = "query_market_data"            # agent stuck repeating one action
        if breaker.should_trip(action):
            print(f"step {step}: circuit breaker tripped — "
                  f"'{action}' repeated too often — halting")
            return "tripped"
        print(f"step {step}: ran {action}")
        time.sleep(0.05)                         # slow loop: well under any sane rate limit
    print("reached max_steps — stopping")
    return "max_steps"

step 1: ran query_market_data
step 2: ran query_market_data
step 3: ran query_market_data
step 4: ran query_market_data
step 5: ran query_market_data
step 6: circuit breaker tripped — 'query_market_data' repeated too often — halting
return: tripped at step 6

import uuid
class RedisPatternBreaker:
    """Shared repetition breaker. One sorted set per agent and action, so
    'fetch' never counts 'fetch_next_page'."""
    def __init__(self, agent_id: str, redis_client, window_s: float = 2.0, max_repeats: int = 5):
        self.agent_id = agent_id
        self.redis = redis_client
        self.window_s = window_s
        self.max_repeats = max_repeats
        self.allowlist: set[str] = set()
    def add_to_allowlist(self, action: str):
        self.allowlist.add(action)
    def should_trip(self, action: str) -> bool:
        if action in self.allowlist:
            return False
        now = time.time()
        key = f"agent:{self.agent_id}:actions:{action}"
        pipe = self.redis.pipeline(transaction=True)      # MULTI/EXEC: atomic
        pipe.zadd(key, {f"{now}:{uuid.uuid4().hex}": now}) # unique member
        pipe.zremrangebyscore(key, 0, now - self.window_s)
        pipe.zcard(key)
        pipe.expire(key, max(60, int(self.window_s * 2)))
        _, _, count, _ = pipe.execute()
        return count > self.max_repeats

from enum import Enum
from typing import Any
class AgentMode(Enum):
    RUNNING     = "running"
    QUARANTINED = "quarantined"   # isolated: can't call tools, but state preserved
    HALTED      = "halted"        # fully stopped
class QuarantineManager:
    """Isolates a misbehaving agent without destroying its state.
    A quarantined agent can be inspected, debugged, and either
    resumed or terminated — unlike a hard kill, which loses context."""
    def __init__(self):
        self.agent_modes: dict[str, AgentMode] = {}
    def quarantine(self, agent_id: str, reason: str):
        """Move agent to quarantine: it can no longer call tools,
        but its state is preserved for inspection."""
        self.agent_modes[agent_id] = AgentMode.QUARANTINED
        event = KillSwitchEvent(
            agent_id=agent_id,
            trigger_type="quarantine",
            trigger_reason=reason,
            agent_state="quarantined",
            triggered_by="system",
        )
        log_kill_switch_event(event)
        print(f"🔒 Agent {agent_id} quarantined: {reason}")
    def release(self, agent_id: str):
        """Release from quarantine after human review."""
        self.agent_modes[agent_id] = AgentMode.RUNNING
        print(f"🔓 Agent {agent_id} released from quarantine")
    def is_allowed(self, agent_id: str) -> bool:
        mode = self.agent_modes.get(agent_id, AgentMode.RUNNING)
        return mode == AgentMode.RUNNING
class TransactionLog:
    """Append-only log of agent actions that supports rollback.
    Each action records enough context to undo it."""
    def __init__(self):
        self.log: list[dict] = []
    def record(self, agent_id: str, action: str, target: str,
               before_state: Any, after_state: Any):
        """Record an action with its before/after state for rollback."""
        self.log.append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "agent_id": agent_id,
            "action": action,
            "target": target,
            "before": before_state,
            "after": after_state,
            "status": "pending",        # becomes "committed" once executed
            "rolled_back": False,
        })
        return len(self.log) - 1    # entry index, used to commit
    def commit(self, index: int):
        """Mark an entry as actually executed. Only committed entries roll back."""
        self.log[index]["status"] = "committed"
    def rollback(self, agent_id: str, steps: int = 1) -> list[dict]:
        """Undo the last N actions by this agent.
        Returns the list of rolled-back entries."""
        agent_entries = [e for e in reversed(self.log)
                         if e["agent_id"] == agent_id and e["status"] == "committed" and not e["rolled_back"]]
        rolled_back = []
        for entry in agent_entries[:steps]:
            # In production: actually restore entry["before"] to entry["target"]
            # e.g., db.update(entry["target"], entry["before"])
            entry["rolled_back"] = True
            rolled_back.append(entry)
            print(f"↩️  Rolled back: {entry['action']} on {entry['target']} "
                  f"(restored to {entry['before']})")
        return rolled_back
# --- Example: quarantine + rollback in action ---
qm = QuarantineManager()
txlog = TransactionLog()
def quarantine_aware_agent(agent_id: str, goal: str, max_steps: int = 10):
    """An agent that records actions and can be quarantined mid-run."""
    breaker = get_breaker(agent_id)
    for step in range(1, max_steps + 1):
        # Check quarantine status before every action
        if not qm.is_allowed(agent_id):
            print(f"step {step}: agent is quarantined — pausing")
            return "quarantined"
        action = "update_customer_record"
        before = {"status": "active", "balance": 100}
        after  = {"status": "active", "balance": 90}
        # Check pattern breaker FIRST, so a blocked action is never logged
        if breaker.should_trip(action):
            qm.quarantine(agent_id, f"'{action}' repeated too often")
            txlog.rollback(agent_id, steps=3)   # undo the last 3 committed actions
            return "quarantined_and_rolled_back"
        # Record BEFORE executing (pending), then commit once it succeeds
        idx = txlog.record(agent_id, action, target="customer:12345",
                           before_state=before, after_state=after)
        print(f"step {step}: executed {action}")   # in production: the real write
        txlog.commit(idx)
    return "completed"

step 1: executed update_customer_record
step 2: executed update_customer_record
step 3: executed update_customer_record
step 4: executed update_customer_record
step 5: executed update_customer_record
🔒 Agent demo quarantined: 'update_customer_record' repeated too often
↩️  Rolled back: update_customer_record on customer:12345 (restored to {'status': 'active', 'balance': 100})
↩️  Rolled back: update_customer_record on customer:12345 (restored to {'status': 'active', 'balance': 100})
↩️  Rolled back: update_customer_record on customer:12345 (restored to {'status': 'active', 'balance': 100})
return: quarantined_and_rolled_back

from typing import Dict
@dataclass
class AgentMetrics:
    agent_id: str
    total_calls: int = 0
    total_cost: float = 0.0
    start_time: float = field(default_factory=time.time)
    action_frequency: Dict[str, int] = field(default_factory=dict)
    last_action: str = ""
    last_action_time: float = 0.0
    def record_action(self, action: str, cost: float):
        self.total_calls += 1
        self.total_cost += cost
        self.action_frequency[action] = self.action_frequency.get(action, 0) + 1
        self.last_action, self.last_action_time = action, time.time()
    def cost_rate_per_minute(self, min_elapsed_s: float = 60.0) -> float:
        elapsed = time.time() - self.start_time
        if elapsed < min_elapsed_s:          # too early to judge a rate
            return 0.0
        return self.total_cost / (elapsed / 60)
    def get_most_common_action(self) -> str:
        return max(self.action_frequency, key=self.action_frequency.get) if self.action_frequency else ""
COST_ALERT_MULTIPLIER = 3.0       # keep in sync with the checklist
REPETITION_ALERT_RATIO = 0.8      # keep in sync with the checklist
def check_cost_anomaly(m: AgentMetrics, baseline_rate: float) -> bool:
    rate = m.cost_rate_per_minute()
    if rate > baseline_rate * COST_ALERT_MULTIPLIER:
        print(f"⚠️  ALERT: cost rate ${rate:.2f}/min vs baseline ${baseline_rate:.2f}/min")
        return True
    return False
def check_repetition_anomaly(m: AgentMetrics) -> bool:
    if m.total_calls < 5:
        return False
    top = m.get_most_common_action()
    ratio = m.action_frequency[top] / m.total_calls
    if ratio > REPETITION_ALERT_RATIO:
        print(f"⚠️  ALERT: '{top}' is {ratio:.0%} of calls — likely loop")
        return True
    return False

def send_alert(channel: str, message: str):
    print(f"[ALERT to {channel}] {message}")   # replace with Slack / PagerDuty
# In your agent loop:
max_steps, action, cost = 200, "query_market_data", 0.10
metrics = AgentMetrics(agent_id="demo")
for step in range(1, max_steps + 1):
    # ... agent logic ...
    metrics.record_action(action, cost)
    if step % 10 == 0:                         # steps 10, 20, 30...
        if check_cost_anomaly(metrics, baseline_rate=1.0):
            send_alert("ops", f"Agent {metrics.agent_id} cost spike — ${metrics.total_cost:.2f} spent")
        if check_repetition_anomaly(metrics):
            send_alert("ops", f"Agent {metrics.agent_id} looping on '{metrics.get_most_common_action()}'")

# DON'T: trip on all "retry_request" repeats
# DO: allowlist it
breaker.add_to_allowlist("retry_request")
breaker.add_to_allowlist("fetch_next_page")
# DON'T: set window_s=1 for all agents
# DO: tune by use case
if agent_type == "polling":
    window_s = 60  # looser
elif agent_type == "research":
    window_s = 2   # tighter

# DON'T: hard-code a ceiling for all agents
# DO: base it on historical usage
historical_daily_spend = 10.0
spend_gate = SpendGate(daily_max_usd=historical_daily_spend * 1.5,  # 1.5x buffer
                       cost_per_call=0.10)

# DON'T: pause and assume someone will eventually answer
answer = interrupt({"message": "Approve?"})
# DO: track the request and auto-reject it from outside the graph
start_run(graph, thread_id, "delete_records")
# A scheduler runs expire_stale_approvals(graph) every 10s; no answer in 5 min = reject

# DON'T: each process has its own deque/spend counter
# DO: store in Redis so all instances share it
pattern_breaker = RedisPatternBreaker(agent_id, redis_client)
spend_gate = RedisSpendGate(agent_id, redis_client)  # see Experiment 2 production note

import pytest
from unittest.mock import Mock, patch
class TestKillSwitches:
    
    def test_iteration_cap_stops_loop(self):
        """Hard cap halts after N steps."""
        result = guarded_agent("test_agent", "goal", max_steps=5)
        assert result == "max_steps"
    
    def test_external_flag_halts_immediately(self):
        """Kill flag halts on next check."""
        with patch.dict(os.environ, {"AGENT_test_KILL": "on"}):
            result = guarded_agent("test", "goal", max_steps=100)
            # Should stop before max_steps
            assert result == "halted"
    
    def test_spend_gate_blocks_at_ceiling(self):
        """Spend gate allows exactly $1.00 of calls, blocks the next."""
        gate = SpendGate(daily_max_usd=1.0, cost_per_call=0.10)
        tool = MagicMock()
        with pytest.raises(RuntimeError):
            for _ in range(20):
                gate.call(tool)
        assert tool.call_count == 10
        assert gate.spent == 1.0
    def test_pattern_breaker_trips_on_repetition(self):
        """Pattern breaker catches repeated actions."""
        breaker = PatternBreaker("test", window_s=2.0, max_repeats=3)
        
        # First 3 calls OK
        assert not breaker.should_trip("fetch")
        assert not breaker.should_trip("fetch")
        assert not breaker.should_trip("fetch")
        
        # 4th call trips
        assert breaker.should_trip("fetch")
    def test_pattern_breaker_allowlist_works(self):
        """Allowlisted actions don't trip."""
        breaker = PatternBreaker("test", window_s=2.0, max_repeats=3)
        breaker.add_to_allowlist("retry")
        
        # Repeat allowlisted action 100x — no trip
        for _ in range(100):
            assert not breaker.should_trip("retry")
    def test_langgraph_interrupt_pauses_and_resumes(self):
        """Interrupt suspends; resume continues with new state."""
        # (Requires LangGraph setup; see experiment 3)
        config = {"configurable": {"thread_id": "test-1"}}
        
        first = graph.invoke(
            {"tool_name": "delete_records", "approved": False, "result": None},
            config=config,
        )
        assert "__interrupt__" in str(first)  # paused
        
        final = graph.invoke(Command(resume="approve"), config=config)
        assert "delete_records" in final["result"]
    def test_approval_auto_rejects_on_timeout(self):
        """Unanswered approvals are auto-rejected from outside the graph."""
        global APPROVAL_TIMEOUT_S
        APPROVAL_TIMEOUT_S = 0
        start_run(graph, "test-timeout", "send_email")
        assert "test-timeout" in pending_approvals
        expire_stale_approvals(graph)
        assert "test-timeout" not in pending_approvals
# Run with: pytest test_kill_switches.py -v
import pytest
from unittest.mock import patch, MagicMock
class TestKillSwitchesAdditions:
    """Additional tests for the new code blocks."""
    def test_fail_closed_on_redis_failure(self):
        """If Redis is unreachable, kill switch defaults to HALT."""
        bad = MagicMock()
        bad.get.side_effect = ConnectionError("Redis down")
        assert kill_switch_enabled_failclosed("test_agent", client=bad) is False
    def test_audit_event_has_all_fields(self):
        """Every audit event captures who, when, why, what, and state."""
        event = KillSwitchEvent(
            agent_id="test",
            trigger_type="external_flag",
            trigger_reason="manual halt",
            agent_action="tool_call('query')",
            agent_step=5,
            agent_state="halted",
            triggered_by="operator:jane",
        )
        assert event.agent_id == "test"
        assert event.trigger_type == "external_flag"
        assert event.triggered_by == "operator:jane"
        assert event.agent_state == "halted"
        assert event.verify() and len(event.signature) == 64  # HMAC-SHA256
    def test_audit_signature_detects_tampering(self):
        """Edits are blocked in code, and a tampered stored copy fails verify()."""
        event = KillSwitchEvent(agent_id="test", trigger_type="spend_ceiling",
                                trigger_reason="budget exceeded", agent_state="deferred",
                                triggered_by="system")
        assert event.verify()
        with pytest.raises(Exception):          # frozen dataclass
            event.trigger_reason = "everything is fine"
        stored = asdict(event)
        stored["trigger_reason"] = "everything is fine"   # tamper with the stored copy
        forged = KillSwitchEvent.__new__(KillSwitchEvent)
        for k, v in stored.items():
            object.__setattr__(forged, k, v)
        assert not forged.verify()
    def test_quarantine_blocks_agent(self):
        """Quarantined agent cannot execute actions."""
        qm_test = QuarantineManager()
        assert qm_test.is_allowed("agent_1") is True
        qm_test.quarantine("agent_1", "loop detected")
        assert qm_test.is_allowed("agent_1") is False
        qm_test.release("agent_1")
        assert qm_test.is_allowed("agent_1") is True
    def test_rollback_undoes_actions(self):
        """Transaction log rollback restores previous state."""
        txlog_test = TransactionLog()
        i1 = txlog_test.record("agent_1", "update", "customer:1",
                          before_state={"balance": 100}, after_state={"balance": 90})
        txlog_test.commit(i1)
        i2 = txlog_test.record("agent_1", "update", "customer:1",
                          before_state={"balance": 90}, after_state={"balance": 80})
        txlog_test.commit(i2)
        rolled = txlog_test.rollback("agent_1", steps=2)
        assert len(rolled) == 2
        assert rolled[0]["before"] == {"balance": 90}   # most recent first
        assert rolled[1]["before"] == {"balance": 100}
    def test_fleet_kill_switch_global(self):
        """Global kill switch halts all agents."""
        r = MagicMock()
        r.mget.side_effect = lambda keys: [b"on" if k == "kill_switch:global" else None for k in keys]
        assert fleet_kill_switch(r, "a1") is False
    def test_fleet_kill_switch_team_scope(self):
        """Team-level kill switch halts only that team's agents."""
        r = MagicMock()
        r.mget.side_effect = lambda keys: [b"on" if k == "kill_switch:team:finance" else None for k in keys]
        assert fleet_kill_switch(r, "a1", team="finance") is False
        assert fleet_kill_switch(r, "a1", team="engineering") is True
    def test_fleet_kill_switch_fails_closed(self):
        """Fleet kill switch fails closed on Redis error."""
        r = MagicMock()
        r.mget.side_effect = ConnectionError("Redis down")
        assert fleet_kill_switch(r, "a1") is False
    # --- RedisSpendGate (Experiment 2 production note) ---
    # The Lua script runs inside Redis, so these tests mock register_script()
    # and simulate the script's behaviour with a shared in-memory total.
    def _fake_redis(self):
        store = {}
        def script(keys, args):                 # amounts in integer cents
            key, cost, cap = keys[0], int(args[0]), int(args[1])
            spent = store.get(key, 0)
            if spent + cost > cap:
                return [0, spent]
            store[key] = spent + cost
            return [1, store[key]]
        mock_redis = MagicMock()
        mock_redis.register_script.return_value = script
        mock_redis.get.side_effect = lambda k: store.get(k)
        return mock_redis, store
    def test_redis_spend_gate_blocks_at_ceiling(self):
        """Allows calls up to the ceiling, blocks the next one before it runs."""
        mock_redis, _ = self._fake_redis()
        gate = RedisSpendGate("test", mock_redis, daily_max_usd=1.0, cost_per_call=0.10)
        tool = MagicMock(return_value="ok")
        with pytest.raises(RuntimeError, match="budget ceiling"):
            for _ in range(20):
                gate.call(tool)
        assert tool.call_count == 10          # exactly $1.00 of calls ran
        assert gate.spent <= 1.0
    def test_redis_spend_gate_shared_across_instances(self):
        """Two gates for the same agent share one budget."""
        mock_redis, _ = self._fake_redis()
        a = RedisSpendGate("shared", mock_redis, daily_max_usd=0.50, cost_per_call=0.10)
        b = RedisSpendGate("shared", mock_redis, daily_max_usd=0.50, cost_per_call=0.10)
        tool = MagicMock()
        for gate in (a, b, a, b, a):          # 5 calls = $0.50 across both
            gate.call(tool)
        with pytest.raises(RuntimeError):
            b.call(tool)                      # 6th call blocked, whichever instance
        assert tool.call_count == 5
    def test_redis_spend_gate_key_resets_daily(self):
        """The key includes the UTC date, so a new day starts a new budget."""
        mock_redis, _ = self._fake_redis()
        gate = RedisSpendGate("test", mock_redis)
        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        assert gate._key() == f"agent:test:spend:{today}"
    def test_redis_spend_gate_fails_closed(self):
        """If Redis errors, the call is refused and the tool never runs."""
        import redis
        mock_redis = MagicMock()
        mock_redis.register_script.return_value = MagicMock(
            side_effect=redis.ConnectionError("Redis down"))
        gate = RedisSpendGate("test", mock_redis)
        tool = MagicMock()
        with pytest.raises(RuntimeError, match="spend check failed"):
            gate.call(tool)
        tool.assert_not_called()
# Run with: pytest test_kill_switches.py -v