Agent Kill-Switch

Field	Value
Document ID	ASCEND-AGENT-002
Version	1.0.0
Last Updated	December 19, 2025
Author	Ascend Engineering Team
Publisher	OW-KAI Technologies Inc.
Classification	Enterprise Client Documentation
Compliance	SOC 2 CC6.1/CC6.2, PCI-DSS 7.1/8.3, HIPAA 164.312, NIST 800-53 AC-2/SI-4

Reading Time: 10 minutes | Skill Level: Advanced

Overview

The ASCEND Kill-Switch provides sub-500ms emergency shutdown capability for AI agents. When activated, agents immediately stop all operations, providing critical incident response capability.

Architecture

┌─────────────────────────────────────────────────────────────────────────────────────┐
│                        KILL-SWITCH ARCHITECTURE                                      │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                      │
│   Admin Dashboard          Backend                SNS Topic              SQS Queue   │
│                                                                                      │
│   ┌─────────────┐    ┌──────────────┐    ┌─────────────────┐    ┌───────────────┐  │
│   │   BLOCK     │───▶│ Agent        │───▶│ ascend-agent-   │───▶│ org-34        │  │
│   │   Button    │    │ Control      │    │ control         │    │ queue         │  │
│   │             │    │ Service      │    │                 │    │               │  │
│   │ Click!      │    │              │    │ Fan-out to      │    │ filter:       │  │
│   └─────────────┘    │ • Validate   │    │ per-org queues  │    │ org_id=34     │  │
│                      │ • Audit log  │    │                 │    │               │  │
│                      │ • Publish    │    │ < 50ms          │    │               │  │
│                      └──────────────┘    └─────────────────┘    └───────┬───────┘  │
│                                                                          │          │
│                                                                          │ < 100ms  │
│                                                                          ▼          │
│                                                                   ┌───────────────┐ │
│                      Target Latency: < 500ms                      │ SDK Agent     │ │
│                      Button Click → Agent Stop                    │               │ │
│                                                                   │ • Poll SQS    │ │
│                                                                   │ • Receive CMD │ │
│                                                                   │ • STOP OPS    │ │
│                                                                   └───────────────┘ │
│                                                                                      │
└─────────────────────────────────────────────────────────────────────────────────────┘

Command Types

# Source: models_agent_registry.py:402
class ControlCommandType(str, enum.Enum):
    """Types of agent control commands."""
    BLOCK = "BLOCK"              # Immediately stop all operations
    UNBLOCK = "UNBLOCK"          # Resume agent operations
    SUSPEND = "SUSPEND"          # Graceful suspension
    RESUME = "RESUME"            # Resume from suspension
    RATE_LIMIT = "RATE_LIMIT"    # Apply rate limiting
    QUARANTINE = "QUARANTINE"    # Isolate for investigation

Command	Effect	Speed	Use Case
BLOCK	Immediate stop	< 500ms	Security incident
UNBLOCK	Resume ops	< 500ms	Incident resolved
SUSPEND	Graceful stop	< 5s	Maintenance
RESUME	Resume ops	< 5s	Maintenance complete
RATE_LIMIT	Throttle agent	< 1s	Resource protection
QUARANTINE	Isolate agent	< 500ms	Investigation

Using the Kill-Switch

Block Agent (Emergency Stop)

curl -X POST "https://pilot.owkai.app/api/agents/control/block" \
  -H "Authorization: Bearer owkai_..." \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "compromised-agent-001",
    "reason": "Unauthorized data access detected"
  }'

Response:

{
  "success": true,
  "command_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "status": "delivered",
  "sns_message_id": "abc123-def456",
  "organization_id": 34,
  "agent_id": "compromised-agent-001",
  "command_type": "BLOCK",
  "message": "Control command BLOCK sent successfully"
}

Block All Organization Agents

curl -X POST "https://pilot.owkai.app/api/agents/control/block" \
  -H "Authorization: Bearer owkai_..." \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": null,
    "reason": "Organization-wide security incident"
  }'

Unblock Agent

curl -X POST "https://pilot.owkai.app/api/agents/control/unblock" \
  -H "Authorization: Bearer owkai_..." \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "compromised-agent-001",
    "reason": "Security review completed - cleared for operation"
  }'

Implementation

Backend Service

# Source: services/agent_control_service.py:46
class AgentControlService:
    """Enterprise agent control service for real-time kill-switch."""

    def __init__(self):
        """Initialize SNS client."""
        self.sns_client = boto3.client('sns', region_name="us-east-2")
        self.topic_arn = "arn:aws:sns:us-east-2:xxx:ascend-agent-control"

    def send_block_command(
        self,
        db: Session,
        organization_id: int,
        reason: str,
        issued_by: str,
        agent_id: Optional[str] = None
    ) -> dict:
        """Send BLOCK (kill-switch) command."""

        # Generate unique command ID
        command_id = str(uuid.uuid4())

        # Create audit record FIRST
        control_command = AgentControlCommand(
            command_id=command_id,
            organization_id=organization_id,
            agent_id=agent_id,
            command_type="BLOCK",
            reason=reason,
            status="pending",
            issued_by=issued_by
        )
        db.add(control_command)
        db.flush()

        # Build SNS message
        message_payload = {
            "command_id": command_id,
            "organization_id": organization_id,
            "agent_id": agent_id,
            "command_type": "BLOCK",
            "reason": reason,
            "issued_by": issued_by,
            "issued_at": datetime.now(UTC).isoformat()
        }

        # Publish to SNS
        response = self.sns_client.publish(
            TopicArn=self.topic_arn,
            Message=json.dumps(message_payload),
            MessageAttributes={
                "organization_id": {
                    "DataType": "String",
                    "StringValue": str(organization_id)
                },
                "command_type": {
                    "DataType": "String",
                    "StringValue": "BLOCK"
                }
            }
        )

        # Update audit record
        control_command.sns_message_id = response.get('MessageId')
        control_command.status = "delivered"
        control_command.delivered_at = datetime.now(UTC)
        db.commit()

        return {
            "success": True,
            "command_id": command_id,
            "status": "delivered"
        }

SDK Agent Handler

# SDK agent polling for control commands
import boto3
import json

class AgentControlHandler:
    """Handle kill-switch commands in SDK."""

    def __init__(self, agent_id: str, organization_id: int):
        self.agent_id = agent_id
        self.organization_id = organization_id
        self.sqs = boto3.client('sqs')
        self.queue_url = f"https://sqs.us-east-2.amazonaws.com/xxx/org-{organization_id}"
        self.blocked = False

    def poll_commands(self):
        """Poll SQS for control commands (long-polling)."""
        while True:
            response = self.sqs.receive_message(
                QueueUrl=self.queue_url,
                MaxNumberOfMessages=10,
                WaitTimeSeconds=5,  # Long polling
                MessageAttributeNames=['All']
            )

            for message in response.get('Messages', []):
                self.handle_command(message)
                self.sqs.delete_message(
                    QueueUrl=self.queue_url,
                    ReceiptHandle=message['ReceiptHandle']
                )

    def handle_command(self, message: dict):
        """Process a control command."""
        payload = json.loads(message['Body'])
        command_type = payload.get('command_type')
        target_agent = payload.get('agent_id')

        # Check if command targets this agent (or all agents)
        if target_agent and target_agent != self.agent_id:
            return

        if command_type == 'BLOCK':
            self.blocked = True
            self.stop_all_operations()
            self.acknowledge_command(payload['command_id'])
            print(f"BLOCKED: {payload['reason']}")

        elif command_type == 'UNBLOCK':
            self.blocked = False
            self.resume_operations()
            self.acknowledge_command(payload['command_id'])
            print(f"UNBLOCKED: {payload['reason']}")

    def is_blocked(self) -> bool:
        """Check if agent is currently blocked."""
        return self.blocked

    def stop_all_operations(self):
        """Immediately stop all agent operations."""
        # Implementation specific to your agent
        pass

    def resume_operations(self):
        """Resume agent operations."""
        # Implementation specific to your agent
        pass

SDK Integration

from ascend import AscendClient

client = AscendClient(
    api_key="owkai_...",
    agent_id="my-agent-001",
    enable_kill_switch=True  # Enable kill-switch polling
)

# Before any action, check if blocked
if client.is_blocked():
    print("Agent is blocked - skipping action")
    return

# Proceed with action
result = client.evaluate_action(...)

Command Acknowledgment

Acknowledge Receipt

# Source: services/agent_control_service.py:231
def acknowledge_command(db, command_id: str, agent_id: str) -> bool:
    """Record that an agent has acknowledged a control command."""
    command = db.query(AgentControlCommand).filter(
        AgentControlCommand.command_id == command_id
    ).first()

    if not command:
        return False

    # Add agent to acknowledged list
    acknowledged = command.acknowledged_by_agents or []
    if agent_id not in acknowledged:
        acknowledged.append(agent_id)
        command.acknowledged_by_agents = acknowledged
        command.status = "acknowledged"
        command.acknowledged_at = datetime.now(UTC)
        db.commit()

    return True

Check Command Status

curl "https://pilot.owkai.app/api/agents/control/commands/f47ac10b-58cc-4372-a567-0e02b2c3d479" \
  -H "Authorization: Bearer owkai_..."

Response:

{
  "command_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "command_type": "BLOCK",
  "status": "acknowledged",
  "agent_id": "compromised-agent-001",
  "reason": "Unauthorized data access detected",
  "issued_by": "security-admin@company.com",
  "issued_at": "2025-12-15T10:30:00Z",
  "delivered_at": "2025-12-15T10:30:00.150Z",
  "acknowledged_at": "2025-12-15T10:30:00.450Z",
  "acknowledged_by_agents": ["compromised-agent-001"],
  "latency_ms": 450
}

Fallback Polling

If SQS delivery fails, agents can poll directly:

# Source: services/agent_control_service.py:263
def get_pending_commands(db, organization_id: int, agent_id: str = None):
    """Get pending control commands (fallback if SQS fails)."""
    query = db.query(AgentControlCommand).filter(
        AgentControlCommand.organization_id == organization_id,
        AgentControlCommand.status.in_(["pending", "delivered"])
    )

    if agent_id:
        # Get commands for specific agent OR broadcast commands
        query = query.filter(
            (AgentControlCommand.agent_id == agent_id) |
            (AgentControlCommand.agent_id.is_(None))
        )

    return query.order_by(
        AgentControlCommand.created_at.desc()
    ).limit(10).all()

SDK Fallback

# Poll API as fallback
def check_for_commands_fallback():
    response = requests.get(
        f"https://pilot.owkai.app/api/agents/control/pending/{agent_id}",
        headers={"Authorization": f"Bearer {api_key}"}
    )

    for command in response.json().get('commands', []):
        handle_command(command)

Audit Trail

Command Audit Record

# Source: models_agent_registry.py:412
class AgentControlCommand(Base):
    """Agent control command audit log."""
    __tablename__ = "agent_control_commands"

    id = Column(Integer, primary_key=True)
    command_id = Column(String(36), unique=True)      # UUID
    organization_id = Column(Integer, nullable=False)
    agent_id = Column(String(255), nullable=True)     # Null = broadcast

    command_type = Column(String(50), nullable=False)
    reason = Column(Text, nullable=False)
    parameters = Column(JSONB, default=dict)

    status = Column(String(50), default="pending")    # pending, delivered, acknowledged
    sns_message_id = Column(String(100), nullable=True)
    delivered_at = Column(DateTime, nullable=True)
    acknowledged_at = Column(DateTime, nullable=True)
    acknowledged_by_agents = Column(JSONB, default=list)

    issued_by = Column(String(255), nullable=False)
    issued_via = Column(String(50), default="api")
    created_at = Column(DateTime, nullable=False)
    expires_at = Column(DateTime, nullable=True)

Query Command History

curl "https://pilot.owkai.app/api/agents/control/history?limit=50" \
  -H "Authorization: Bearer owkai_..."

Latency Optimization

Target: < 500ms End-to-End

Stage	Target Latency	Actual
Dashboard → Backend	< 50ms	~30ms
Backend → SNS	< 50ms	~20ms
SNS → SQS	< 100ms	~50ms
SQS → Agent (poll)	< 300ms	~200ms
Total	< 500ms	~300ms

Optimization Tips

1. Use Long Polling — 5-second wait reduces polling frequency
2. Regional Deployment — Deploy in same region as SNS/SQS
3. Connection Pooling — Reuse HTTP connections
4. Fast Serialization — Use efficient JSON libraries

Security Considerations

Authorization

Only organization admins can issue kill-switch commands:

# Requires admin role
@router.post("/control/block")
async def block_agent(
    request: BlockRequest,
    current_user: dict = Depends(require_admin_role)
):
    ...

Audit Requirements

All commands are logged with:

• Who — issued_by email
• What — Command type and target
• When — Timestamps for all stages
• Why — Required reason field
• How — issued_via (api, dashboard, automation)

Compliance

Standard	Requirement	Implementation
SOC 2 CC6.2	Logical access	Admin-only access
NIST IR-4	Incident handling	Kill-switch capability
HIPAA 164.308	Security incidents	Immediate response
PCI-DSS 10.6	Log review	Command audit trail

Best Practices

1. Test Kill-Switch Regularly

# Schedule monthly kill-switch drills
def monthly_kill_switch_test():
    # Use test agent in staging
    result = client.send_block_command(
        agent_id="test-agent-staging",
        reason="Monthly kill-switch drill"
    )

    # Verify latency
    assert result['latency_ms'] < 500

    # Unblock
    client.send_unblock_command(
        agent_id="test-agent-staging",
        reason="Drill complete"
    )

2. Document Incident Procedures

## Kill-Switch Runbook

1. **Identify threat** - Confirm security incident
2. **Block agent** - Use kill-switch immediately
3. **Investigate** - Review audit logs
4. **Remediate** - Fix root cause
5. **Unblock** - Resume operations with approval
6. **Post-mortem** - Document lessons learned

3. Configure Alerting

# Alert on kill-switch activation
{
    "alert_type": "kill_switch_activated",
    "channels": ["pagerduty", "slack"],
    "recipients": ["security-team@company.com"]
}

4. Use Meaningful Reasons

# Good - actionable reason
reason = "SEC-2025-001: Unauthorized PII access detected in audit logs"

# Bad - vague reason
reason = "Security issue"

Next Steps

• Smart Rules — Automated kill-switch triggers
• Audit Logs — Review command history
• Notifications — Alert configuration

---

Document Version: 1.0.0 | Last Updated: December 2025