Add tenant-scoped kill switch

api/nexus/authorization/http_models.py

@@ -18,3 +18,13 @@ class AuthorizeResponse(BaseModel):
     effect: DecisionEffect
     fail_mode_applied: bool
     reason: str
+
+
+class KillSwitchRequest(BaseModel):
+    halted: bool
+    agent_key: str = Field(default="*", min_length=1, max_length=128)
+
+
+class KillSwitchResponse(BaseModel):
+    agent_key: str
+    halted: bool

api/nexus/authorization/models.py

@@ -10,10 +10,12 @@ class DecisionEffect(StrEnum):
     ALLOW = "allow"
     DENY = "deny"
 
+
 class DecisionReason(StrEnum):
     DEFAULT_ALLOW = "default_allow"
     FAIL_CLOSED = "fail_closed"
     STATIC_DENY_TEST_TOOL = "denied: static deny test tool"
+    KILL_SWITCH_ACTIVE = "denied: kill switch active"
 
 
 @dataclass(frozen=True)

api/nexus/authorization/routes.py

@@ -7,6 +7,8 @@
 from nexus.authorization.http_models import (
     AuthorizeRequest,
     AuthorizeResponse,
+    KillSwitchRequest,
+    KillSwitchResponse,
 )
 from nexus.authorization.models import AuthorizationRequest
 from nexus.authorization.service import AuthorizationService
@@ -15,6 +17,7 @@
 router = APIRouter(prefix="/v1")
 CONTRACT_HEADER = "x-nexus-contract-version"
 
+
 def _error(status: int, code: str, message: str, details: list | None = None):
     body = {"error": {"code": code, "message": message}}
     if details is not None:
@@ -23,7 +26,10 @@ def _error(status: int, code: str, message: str, details: list | None = None):
 
 
 def _authorization_service(request: Request) -> AuthorizationService:
-    return AuthorizationService(request.app.state.decision_repo)
+    return AuthorizationService(
+        request.app.state.decision_repo,
+        request.app.state.kill_switch_repo,
+    )
 
 
 @router.post("/authorize", response_model=AuthorizeResponse)
@@ -78,3 +84,35 @@ async def authorize(request: Request):
         fail_mode_applied=decision.fail_mode_applied,
         reason=decision.reason.value,
     )
+
+
+@router.post("/kill-switch", response_model=KillSwitchResponse)
+async def set_kill_switch(request: Request):
+    try:
+        body = await request.json()
+    except Exception:
+        return _error(400, "malformed_request", "request body is not valid JSON")
+
+    try:
+        payload = KillSwitchRequest.model_validate(body)
+    except PydanticValidationError as e:
+        details = [
+            {"field": ".".join(str(p) for p in err["loc"]), "issue": err["msg"]}
+            for err in e.errors()
+        ]
+        return _error(
+            422,
+            "validation_failed",
+            "request failed schema validation",
+            details,
+        )
+
+    await request.app.state.kill_switch_repo.set_halted(
+        payload.halted,
+        payload.agent_key,
+    )
+
+    return KillSwitchResponse(
+        agent_key=payload.agent_key,
+        halted=payload.halted,
+    )

api/nexus/authorization/service.py

@@ -21,15 +21,37 @@ async def record(
         """Persist the authorization decision before it is returned."""
 
 
+class KillSwitchReader(Protocol):
+    async def is_halted(self, agent_key: str) -> bool:
+        """Return whether authorization must be denied by kill switch."""
+
+
 class AuthorizationService:
-    def __init__(self, writer: DecisionLogWriter):
+    def __init__(
+        self,
+        writer: DecisionLogWriter,
+        kill_switch: KillSwitchReader | None = None,
+    ):
         self._writer = writer
+        self._kill_switch = kill_switch
 
     async def authorize(
         self,
         request: AuthorizationRequest,
     ) -> AuthorizationDecision:
-        if request.tool_name == "__nexus_deny_test__":
+        kill_switch_active = False
+        if self._kill_switch is not None:
+            kill_switch_active = await self._kill_switch.is_halted(request.agent_key)
+
+        if kill_switch_active:
+            decision = AuthorizationDecision(
+                decision_id=uuid4(),
+                effect=DecisionEffect.DENY,
+                fail_mode_applied=False,
+                reason=DecisionReason.KILL_SWITCH_ACTIVE,
+                decided_at=datetime.now(UTC),
+            )
+        elif request.tool_name == "__nexus_deny_test__":
             decision = AuthorizationDecision(
                 decision_id=uuid4(),
                 effect=DecisionEffect.DENY,

api/nexus/main.py

@@ -34,6 +34,8 @@
 from nexus.repositories.agent_run import AgentRunRepository
 from nexus.tenancy.boot_assertion import assert_safe_role
 from nexus.read.routes import router as read_router
+from nexus.repositories.kill_switch import KillSwitchRepository
+
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     settings = settings or Settings.from_env()
@@ -51,6 +53,7 @@ async def lifespan(app: FastAPI):
         limits = Limits.from_env()
         app.state.agent_run_repo = AgentRunRepository(pool, crypto, limits)
         app.state.decision_repo = DecisionRepository(pool)
+        app.state.kill_switch_repo = KillSwitchRepository(pool)
         try:
             yield
         finally:

api/nexus/repositories/kill_switch.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from nexus.repositories.base import BaseRepository
+
+
+class KillSwitchRepository(BaseRepository):
+    async def is_halted(self, agent_key: str) -> bool:
+        async with self._tx() as conn:
+            halted = await conn.fetchval(
+                """
+                SELECT halted
+                FROM kill_switch_state
+                WHERE agent_key IN ('*', $1)
+                  AND halted = true
+                LIMIT 1
+                """,
+                agent_key,
+            )
+            return bool(halted)
+
+    async def set_halted(self, halted: bool, agent_key: str = "*") -> None:
+        async with self._tx() as conn:
+            await conn.execute(
+                """
+                INSERT INTO kill_switch_state
+                  (tenant_id, agent_key, halted, updated_at)
+                VALUES (current_setting('app.tenant_id')::uuid, $1, $2, now())
+                ON CONFLICT (tenant_id, agent_key)
+                DO UPDATE SET halted = EXCLUDED.halted, updated_at = now()
+                """,
+                agent_key,
+                halted,
+            )
+
+            await conn.execute(
+                """
+                INSERT INTO kill_switch_events
+                  (tenant_id, event_id, agent_key, halted, changed_at)
+                VALUES (current_setting('app.tenant_id')::uuid, $1, $2, $3, now())
+                """,
+                uuid4(),
+                agent_key,
+                halted,
+            )

api/tests/test_authorization_service.py

@@ -29,6 +29,12 @@ class FailingWriter:
     async def record(self, request, decision):
         raise RuntimeError("write failed")
 
+class FakeKillSwitch:
+    def __init__(self, halted: bool):
+        self.halted = halted
+
+    async def is_halted(self, agent_key: str) -> bool:
+        return self.halted
 
 def make_request():
     return AuthorizationRequest(
@@ -92,3 +98,37 @@ async def test_authorization_preserves_request_fields_for_logging():
     assert logged_decision.effect == decision.effect
     assert logged_decision.fail_mode_applied == decision.fail_mode_applied
     assert logged_decision.decided_at == decision.decided_at
+
+
+@pytest.mark.asyncio
+async def test_kill_switch_denies_before_default_allow():
+    writer = FakeWriter()
+    service = AuthorizationService(
+        writer,
+        FakeKillSwitch(True),
+    )
+
+    decision = await service.authorize(make_request())
+
+    assert decision.effect == DecisionEffect.DENY
+    assert decision.fail_mode_applied is False
+    assert decision.reason == DecisionReason.KILL_SWITCH_ACTIVE
+
+
+@pytest.mark.asyncio
+async def test_kill_switch_decision_is_logged():
+    writer = FakeWriter()
+    service = AuthorizationService(
+        writer,
+        FakeKillSwitch(True),
+    )
+    request = make_request()
+
+    decision = await service.authorize(request)
+
+    logged_request, logged_decision = writer.calls[0]
+
+    assert logged_request == request
+    assert logged_decision.effect == DecisionEffect.DENY
+    assert logged_decision.reason == DecisionReason.KILL_SWITCH_ACTIVE
+    assert logged_decision.decision_id == decision.decision_id

api/tests/test_authorize_endpoint.py

@@ -158,4 +158,80 @@ async def test_authorize_static_deny_is_tenant_isolated(client, two_auth_headers
     assert r2.status_code == 200, r2.text
     assert r1.json()["effect"] == "deny"
     assert r2.json()["effect"] == "deny"
-    assert r1.json()["decision_id"] != r2.json()["decision_id"]
+    assert r1.json()["decision_id"] != r2.json()["decision_id"]
+
+
+async def test_kill_switch_halt_forces_deny(client, auth_headers):
+    r_set = await client.post(
+        "/v1/kill-switch",
+        json={"halted": True},
+        headers=auth_headers,
+    )
+
+    assert r_set.status_code == 200, r_set.text
+    assert r_set.json()["halted"] is True
+
+    r = await client.post(
+        "/v1/authorize",
+        json=_authorize_body(),
+        headers=auth_headers,
+    )
+
+    assert r.status_code == 200, r.text
+    assert r.json()["effect"] == "deny"
+    assert r.json()["reason"] == "denied: kill switch active"
+
+
+async def test_kill_switch_clear_restores_default_allow(client, auth_headers):
+    await client.post(
+        "/v1/kill-switch",
+        json={"halted": True},
+        headers=auth_headers,
+    )
+    r_clear = await client.post(
+        "/v1/kill-switch",
+        json={"halted": False},
+        headers=auth_headers,
+    )
+
+    assert r_clear.status_code == 200, r_clear.text
+    assert r_clear.json()["halted"] is False
+
+    r = await client.post(
+        "/v1/authorize",
+        json=_authorize_body(),
+        headers=auth_headers,
+    )
+
+    assert r.status_code == 200, r.text
+    assert r.json()["effect"] == "allow"
+    assert r.json()["reason"] == "default_allow"
+
+
+async def test_kill_switch_deny_is_idempotent(client, auth_headers):
+    tool_call_id = str(uuid.uuid4())
+
+    await client.post(
+        "/v1/kill-switch",
+        json={"halted": True},
+        headers=auth_headers,
+    )
+
+    r1 = await client.post(
+        "/v1/authorize",
+        json=_authorize_body(tool_call_id),
+        headers=auth_headers,
+    )
+    r2 = await client.post(
+        "/v1/authorize",
+        json=_authorize_body(tool_call_id),
+        headers=auth_headers,
+    )
+
+    assert r1.status_code == 200, r1.text
+    assert r2.status_code == 200, r2.text
+    assert r1.json()["decision_id"] == r2.json()["decision_id"]
+    assert r1.json()["effect"] == "deny"
+    assert r2.json()["effect"] == "deny"
+    assert r1.json()["reason"] == r2.json()["reason"]
+    assert r1.json()["reason"] == "denied: kill switch active"

migrations/0015_kill_switch.sql

@@ -0,0 +1,52 @@
+-- 0015_kill_switch.sql
+-- Run as nexus_owner.
+-- PR #32: tenant-scoped kill switch v1.
+--
+-- Current halt state is mutable.
+-- Halt changes are append-only governance records.
+
+SET search_path = nexus, public;
+
+CREATE TABLE kill_switch_state (
+    tenant_id uuid NOT NULL,
+    agent_key text NOT NULL DEFAULT '*',
+    halted boolean NOT NULL,
+    updated_at timestamptz NOT NULL DEFAULT now(),
+
+    PRIMARY KEY (tenant_id, agent_key)
+);
+
+CREATE TABLE kill_switch_events (
+    tenant_id uuid NOT NULL,
+    event_id uuid NOT NULL,
+    agent_key text NOT NULL DEFAULT '*',
+    halted boolean NOT NULL,
+    changed_at timestamptz NOT NULL DEFAULT now(),
+
+    PRIMARY KEY (tenant_id, event_id)
+);
+
+ALTER TABLE kill_switch_state ENABLE ROW LEVEL SECURITY;
+ALTER TABLE kill_switch_state FORCE ROW LEVEL SECURITY;
+
+CREATE POLICY kill_switch_state_tenant_isolation
+    ON kill_switch_state
+    USING (tenant_id = current_setting('app.tenant_id')::uuid)
+    WITH CHECK (tenant_id = current_setting('app.tenant_id')::uuid);
+
+ALTER TABLE kill_switch_events ENABLE ROW LEVEL SECURITY;
+ALTER TABLE kill_switch_events FORCE ROW LEVEL SECURITY;
+
+CREATE POLICY kill_switch_events_tenant_isolation
+    ON kill_switch_events
+    USING (tenant_id = current_setting('app.tenant_id')::uuid)
+    WITH CHECK (tenant_id = current_setting('app.tenant_id')::uuid);
+
+CREATE INDEX kill_switch_events_tenant_time_idx
+    ON kill_switch_events (tenant_id, changed_at DESC);
+
+GRANT SELECT, INSERT, UPDATE ON kill_switch_state TO nexus_app;
+REVOKE DELETE ON kill_switch_state FROM nexus_app;
+
+GRANT SELECT, INSERT ON kill_switch_events TO nexus_app;
+REVOKE UPDATE, DELETE ON kill_switch_events FROM nexus_app;