Add bounded policy conditions

api/nexus/authorization/http_models.py

@@ -57,3 +57,19 @@ class ToolPermissionResponse(BaseModel):
     agent_key: str
     tool_name: str
     enabled: bool
+
+
+class PolicyConditionRequest(BaseModel):
+    policy_id: UUID
+    agent_key: str = Field(min_length=1, max_length=128)
+    condition_type: str = Field(min_length=1, max_length=64)
+    condition_value: str = Field(min_length=1, max_length=256)
+    enabled: bool = True
+
+
+class PolicyConditionResponse(BaseModel):
+    policy_id: UUID
+    agent_key: str
+    condition_type: str
+    condition_value: str
+    enabled: bool

api/nexus/authorization/models.py

@@ -18,6 +18,7 @@ class DecisionReason(StrEnum):
     KILL_SWITCH_ACTIVE = "denied: kill switch active"
     BUDGET_EXCEEDED = "denied: budget exceeded"
     TOOL_NOT_ALLOWED = "denied: tool not allowed"
+    POLICY_CONDITION_MATCHED = "denied: policy condition matched"
 
 
 @dataclass(frozen=True)
@@ -45,3 +46,4 @@ class AuthorizationDecision:
     fail_mode_applied: bool
     reason: DecisionReason
     decided_at: datetime
+    policy_id: UUID | None = None

api/nexus/authorization/routes.py

@@ -13,6 +13,8 @@
     BudgetResponse,
     ToolPermissionRequest,
     ToolPermissionResponse,
+    PolicyConditionRequest,
+    PolicyConditionResponse,
 )
 from nexus.authorization.models import AuthorizationRequest
 from nexus.authorization.service import AuthorizationService
@@ -35,6 +37,7 @@ def _authorization_service(request: Request) -> AuthorizationService:
         request.app.state.kill_switch_repo,
         request.app.state.budget_repo,
         request.app.state.tool_permission_repo,
+        request.app.state.policy_condition_repo,
     )
 
 
@@ -194,3 +197,41 @@ async def set_tool_permission(request: Request):
         tool_name=payload.tool_name,
         enabled=payload.enabled,
     )
+
+
+@router.post("/policy-conditions", response_model=PolicyConditionResponse)
+async def set_policy_condition(request: Request):
+    try:
+        body = await request.json()
+    except Exception:
+        return _error(400, "malformed_request", "request body is not valid JSON")
+
+    try:
+        payload = PolicyConditionRequest.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.policy_condition_repo.set_policy(
+        policy_id=payload.policy_id,
+        agent_key=payload.agent_key,
+        condition_type=payload.condition_type,
+        condition_value=payload.condition_value,
+        enabled=payload.enabled,
+    )
+
+    return PolicyConditionResponse(
+        policy_id=payload.policy_id,
+        agent_key=payload.agent_key,
+        condition_type=payload.condition_type,
+        condition_value=payload.condition_value,
+        enabled=payload.enabled,
+    )

api/nexus/authorization/service.py

@@ -48,18 +48,30 @@ async def is_allowed(self, agent_key: str, tool_name: str) -> bool:
         """Return whether this agent may invoke this tool."""
 
 
+class PolicyConditionReader(Protocol):
+    async def matching_policy(
+        self,
+        *,
+        agent_key: str,
+        tool_name: str,
+    ):
+        """Return matched bounded policy condition, if any."""
+
+
 class AuthorizationService:
     def __init__(
         self,
         writer: DecisionLogWriter,
         kill_switch: KillSwitchReader | None = None,
         budget: BudgetReader | None = None,
         tool_permissions: ToolPermissionReader | None = None,
+        policy_conditions: PolicyConditionReader | None = None,
     ):
         self._writer = writer
         self._kill_switch = kill_switch
         self._budget = budget
         self._tool_permissions = tool_permissions
+        self._policy_conditions = policy_conditions
 
     async def _budget_exceeded(self, request: AuthorizationRequest) -> bool:
         if self._budget is None:
@@ -89,6 +101,15 @@ async def _tool_not_allowed(self, request: AuthorizationRequest) -> bool:
             request.tool_name,
         )
 
+    async def _matched_policy(self, request: AuthorizationRequest):
+        if self._policy_conditions is None:
+            return None
+
+        return await self._policy_conditions.matching_policy(
+            agent_key=request.agent_key,
+            tool_name=request.tool_name,
+        )
+
     async def authorize(
         self,
         request: AuthorizationRequest,
@@ -97,6 +118,8 @@ async def authorize(
         if self._kill_switch is not None:
             kill_switch_active = await self._kill_switch.is_halted(request.agent_key)
 
+        matched_policy = await self._matched_policy(request)
+
         if kill_switch_active:
             decision = AuthorizationDecision(
                 decision_id=uuid4(),
@@ -121,6 +144,15 @@ async def authorize(
                 reason=DecisionReason.TOOL_NOT_ALLOWED,
                 decided_at=datetime.now(UTC),
             )
+        elif matched_policy is not None:
+            decision = AuthorizationDecision(
+                decision_id=uuid4(),
+                effect=DecisionEffect.DENY,
+                fail_mode_applied=False,
+                reason=DecisionReason.POLICY_CONDITION_MATCHED,
+                decided_at=datetime.now(UTC),
+                policy_id=matched_policy.policy_id,
+            )
         elif request.tool_name == "__nexus_deny_test__":
             decision = AuthorizationDecision(
                 decision_id=uuid4(),
@@ -155,4 +187,5 @@ async def authorize(
             fail_mode_applied=recorded.fail_mode_applied,
             reason=DecisionReason(recorded.reason),
             decided_at=recorded.decided_at,
+            policy_id=recorded.policy_id,
         )

api/nexus/main.py

@@ -37,6 +37,8 @@
 from nexus.repositories.kill_switch import KillSwitchRepository
 from nexus.repositories.budget import BudgetRepository
 from nexus.repositories.tool_permissions import ToolPermissionRepository
+from nexus.repositories.policy_conditions import PolicyConditionRepository
+
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -58,6 +60,7 @@ async def lifespan(app: FastAPI):
         app.state.kill_switch_repo = KillSwitchRepository(pool)
         app.state.budget_repo = BudgetRepository(pool)
         app.state.tool_permission_repo = ToolPermissionRepository(pool)
+        app.state.policy_condition_repo = PolicyConditionRepository(pool)
         try:
             yield
         finally:

api/nexus/repositories/decisions.py

@@ -17,6 +17,7 @@ class RecordedDecision:
     effect: str
     fail_mode_applied: bool
     reason: str
+    policy_id: UUID | None
     decided_at: datetime
 
 
@@ -31,12 +32,12 @@ async def record(
                 """
                 INSERT INTO decisions
                   (tenant_id, decision_id, tool_call_id, agent_key, tool_name,
-                   effect, fail_mode_applied, reason, decided_at)
+                   effect, fail_mode_applied, reason, policy_id, decided_at)
                 VALUES (current_setting('app.tenant_id')::uuid,
-                        $1, $2, $3, $4, $5, $6, $7, $8)
+                        $1, $2, $3, $4, $5, $6, $7, $8, $9)
                 ON CONFLICT (tenant_id, tool_call_id)
                 DO NOTHING
-                RETURNING decision_id, effect, fail_mode_applied, reason, decided_at
+                RETURNING decision_id, effect, fail_mode_applied, reason, policy_id, decided_at
                 """,
                 decision.decision_id,
                 request.tool_call_id,
@@ -45,13 +46,14 @@ async def record(
                 decision.effect.value,
                 decision.fail_mode_applied,
                 decision.reason.value,
+                decision.policy_id,
                 decision.decided_at,
             )
 
             if row is None:
                 row = await conn.fetchrow(
                     """
-                    SELECT decision_id, effect, fail_mode_applied, reason, decided_at
+                    SELECT decision_id, effect, fail_mode_applied, reason, policy_id, decided_at
                     FROM decisions
                     WHERE tool_call_id = $1
                     """,
@@ -63,5 +65,6 @@ async def record(
                 effect=row["effect"],
                 fail_mode_applied=row["fail_mode_applied"],
                 reason=row["reason"],
+                policy_id=row["policy_id"],
                 decided_at=row["decided_at"],
             )

api/nexus/repositories/policy_conditions.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import UUID
+
+from nexus.repositories.base import BaseRepository
+
+
+@dataclass(frozen=True)
+class PolicyCondition:
+    policy_id: UUID
+    agent_key: str
+    condition_type: str
+    condition_value: str
+
+
+class PolicyConditionRepository(BaseRepository):
+    async def matching_policy(
+        self,
+        *,
+        agent_key: str,
+        tool_name: str,
+    ) -> PolicyCondition | None:
+        async with self._tx() as conn:
+            row = await conn.fetchrow(
+                """
+                SELECT
+                    policy_id,
+                    agent_key,
+                    condition_type,
+                    condition_value
+                FROM policy_conditions
+                WHERE enabled = true
+                  AND agent_key = $1
+                  AND condition_type = 'agent_tool'
+                  AND condition_value = $2
+                LIMIT 1
+                """,
+                agent_key,
+                tool_name,
+            )
+
+            if row is None:
+                return None
+
+            return PolicyCondition(
+                policy_id=row["policy_id"],
+                agent_key=row["agent_key"],
+                condition_type=row["condition_type"],
+                condition_value=row["condition_value"],
+            )
+
+    async def set_policy(
+        self,
+        *,
+        policy_id: UUID,
+        agent_key: str,
+        condition_type: str,
+        condition_value: str,
+        enabled: bool = True,
+    ) -> None:
+        async with self._tx() as conn:
+            await conn.execute(
+                """
+                INSERT INTO policy_conditions
+                  (
+                    tenant_id,
+                    policy_id,
+                    agent_key,
+                    condition_type,
+                    condition_value,
+                    enabled
+                  )
+                VALUES (
+                    current_setting('app.tenant_id')::uuid,
+                    $1, $2, $3, $4, $5
+                )
+                ON CONFLICT (tenant_id, policy_id)
+                DO UPDATE SET
+                    agent_key = EXCLUDED.agent_key,
+                    condition_type = EXCLUDED.condition_type,
+                    condition_value = EXCLUDED.condition_value,
+                    enabled = EXCLUDED.enabled
+                """,
+                policy_id,
+                agent_key,
+                condition_type,
+                condition_value,
+                enabled,
+            )

api/tests/conftest.py

@@ -119,6 +119,7 @@ async def _seeded_key():
                 tenant_id,
             )
             for tbl in (
+                "policy_conditions",
                 "tool_permissions",
                 "budget_controls",
                 "decisions",