feat(AGX1-272): dual-write agent_api_keys to spark-authz behind FGAC flag

agentex/src/adapters/authorization/adapter_agentex_authz_proxy.py

@@ -85,6 +85,34 @@ async def list_resources(
         )
         return response["items"]
 
+    async def register_resource(
+        self,
+        principal: AgentexAuthPrincipalContext,
+        resource: AgentexResource,
+        parent: AgentexResource | None = None,
+    ) -> None:
+        payload = {
+            "principal": principal,
+            "resource": resource.model_dump(),
+            "parent": parent.model_dump() if parent is not None else None,
+        }
+        await HttpRequestHandler.post_with_error_handling(
+            self.agentex_auth_url, "/v1/authz/register", json=payload
+        )
+
+    async def deregister_resource(
+        self,
+        principal: AgentexAuthPrincipalContext,
+        resource: AgentexResource,
+    ) -> None:
+        payload = {
+            "principal": principal,
+            "resource": resource.model_dump(),
+        }
+        await HttpRequestHandler.post_with_error_handling(
+            self.agentex_auth_url, "/v1/authz/deregister", json=payload
+        )
+
 
 DAgentexAuthorization = Annotated[
     AgentexAuthorizationProxy, Depends(AgentexAuthorizationProxy)

agentex/src/adapters/authorization/port.py

@@ -49,3 +49,29 @@ async def list_resources(
         filter_operation: AuthorizedOperationType = AuthorizedOperationType.read,
     ) -> Iterable[str]:
         """List resource_ids for a given principal"""
+
+    @abstractmethod
+    async def register_resource(
+        self,
+        principal: PrincipalT,
+        resource: AgentexResource,
+        parent: AgentexResource | None = None,
+    ) -> None:
+        """Register a newly created resource in SpiceDB with the principal as
+        owner. Optionally writes a lifecycle parent edge.
+
+        Use this on resource create instead of ``grant`` when the resource
+        type's SpiceDB definition has a parent relation that permission
+        checks cascade through (e.g. ``agent_api_key`` declares
+        ``parent_agent``). Without writing that edge here the cascade fails
+        closed.
+        """
+
+    @abstractmethod
+    async def deregister_resource(
+        self,
+        principal: PrincipalT,
+        resource: AgentexResource,
+    ) -> None:
+        """Deregister a deleted resource and all of its relationships
+        (owner, parent, grantees) in a single atomic call."""

agentex/src/api/routes/agent_api_keys.py

@@ -192,7 +192,9 @@ async def delete_agent_api_key_by_name(
         )
     agent = await agent_use_case.get(id=agent_id, name=agent_name)
     await agent_api_key_use_case.delete_by_agent_id_and_key_name(
-        agent_id=agent.id, key_name=api_key_name, api_key_type=api_key_type
+        agent_id=agent.id,
+        key_name=api_key_name,
+        api_key_type=api_key_type,
     )
 
     return f"Agent api_key '{api_key_name}' deleted"

agentex/src/api/schemas/authorization_types.py

@@ -14,6 +14,7 @@ class AuthorizedOperationType(StrEnum):
 class AgentexResourceType(StrEnum):
     agent = "agent"
     task = "task"
+    api_key = "api_key"
 
 
 # Resources that inherit permissions from their parent task
@@ -37,6 +38,10 @@ def agent(cls, selector: str) -> "AgentexResource":
     def task(cls, selector: str) -> "AgentexResource":
         return cls(type=AgentexResourceType.task, selector=selector)
 
+    @classmethod
+    def api_key(cls, selector: str) -> "AgentexResource":
+        return cls(type=AgentexResourceType.api_key, selector=selector)
+
 
 class AgentexResourceOptionalSelector(BaseModel):
     type: AgentexResourceType
@@ -49,3 +54,7 @@ def agent(cls, selector: str | None = None) -> "AgentexResourceOptionalSelector"
     @classmethod
     def task(cls, selector: str | None = None) -> "AgentexResourceOptionalSelector":
         return cls(type=AgentexResourceType.task, selector=selector)
+
+    @classmethod
+    def api_key(cls, selector: str | None = None) -> "AgentexResourceOptionalSelector":
+        return cls(type=AgentexResourceType.api_key, selector=selector)

agentex/src/domain/services/authorization_service.py

@@ -188,5 +188,62 @@ async def list_resources(
         )
         return result
 
+    async def register_resource(
+        self,
+        resource: AgentexResource,
+        parent: AgentexResource | None = None,
+        *,
+        principal_context=...,
+    ) -> None:
+        """Register a newly created resource with the principal as owner.
+
+        Prefer this over ``grant`` when the resource's SpiceDB schema has
+        a parent relation that permissions cascade through (e.g.
+        ``agent_api_key`` declares ``parent_agent``). Pass ``parent`` to
+        link the child to its parent atomically; without it the cascade
+        fails closed.
+        """
+        if self._bypass():
+            logger.info(f"Authorization bypassed for register_resource on {resource}")
+            return None
+
+        effective_principal = (
+            principal_context
+            if principal_context is not ...
+            else self.principal_context
+        )
+        logger.info(
+            "[authorization_service] Registering %s:%s for principal %s (parent=%s)",
+            resource.type,
+            resource.selector,
+            effective_principal,
+            f"{parent.type}:{parent.selector}" if parent is not None else None,
+        )
+        await self.gateway.register_resource(effective_principal, resource, parent)
+
+    async def deregister_resource(
+        self,
+        resource: AgentexResource,
+        *,
+        principal_context=...,
+    ) -> None:
+        """Deregister a deleted resource and all of its relationships."""
+        if self._bypass():
+            logger.info(f"Authorization bypassed for deregister_resource on {resource}")
+            return None
+
+        effective_principal = (
+            principal_context
+            if principal_context is not ...
+            else self.principal_context
+        )
+        logger.info(
+            "[authorization_service] Deregistering %s:%s for principal %s",
+            resource.type,
+            resource.selector,
+            effective_principal,
+        )
+        await self.gateway.deregister_resource(effective_principal, resource)
+
 
 DAuthorizationService = Annotated[AuthorizationService, Depends(AuthorizationService)]

agentex/src/domain/use_cases/agent_api_keys_use_case.py

@@ -13,6 +13,7 @@
 )
 from src.adapters.crud_store.exceptions import ItemDoesNotExist
 from src.api.middleware_utils import get_request_headers_to_forward, verify_auth_gateway
+from src.api.schemas.authorization_types import AgentexResource
 from src.config.dependencies import (
     DHttpxClient,
     resolve_environment_variable_dependency,
@@ -27,6 +28,7 @@
     DAgentAPIKeyRepository,
 )
 from src.domain.repositories.agent_repository import DAgentRepository
+from src.domain.services.authorization_service import DAuthorizationService
 from src.utils.ids import orm_id
 from src.utils.logging import make_logger
 
@@ -39,10 +41,12 @@ def __init__(
         agent_api_key_repository: DAgentAPIKeyRepository,
         agent_repository: DAgentRepository,
         client: DHttpxClient,
+        authorization_service: DAuthorizationService,
     ):
         self.agent_api_key_repo = agent_api_key_repository
         self.agent_repo = agent_repository
         self.client = client
+        self.authorization_service = authorization_service
         self.auth_gateway_enabled = bool(
             resolve_environment_variable_dependency(EnvVarKeys.AGENTEX_AUTH_URL)
         )
@@ -83,17 +87,95 @@ async def create(
                 status_code=404,
                 detail=f"Agent ID {agent_id} not found.",
             )
+
+        api_key_id = orm_id()
+
+        # Unconditional — agentex-auth decides per-account whether the call
+        # routes to Spark or the legacy backend.
+        await self._register_api_key_in_auth(
+            api_key_id=api_key_id,
+            agent_id=agent.id,
+        )
+
         # TODO: encrypt API key before storing it
         # Initialize a new agent api_key
         agent_api_key = AgentAPIKeyEntity(
-            id=orm_id(),
+            id=api_key_id,
             name=name,
             agent_id=agent.id,
             api_key_type=api_key_type,
             api_key=api_key,
         )
         return await self.agent_api_key_repo.create(item=agent_api_key)
 
+    async def _register_api_key_in_auth(
+        self,
+        *,
+        api_key_id: str,
+        agent_id: str,
+    ) -> None:
+        """Register a new agent_api_key with the auth service, including the
+        parent_agent edge so permissions cascade from the owning agent.
+
+        Called BEFORE the Postgres write — a failure raises and prevents the
+        row from being persisted, so there is no compensating action. Skipped
+        with a warning when no usable creator identity is available on the
+        principal context (e.g. internal-key creation paths without an
+        authenticated user).
+        """
+        principal_context = self.authorization_service.principal_context
+        user_id = getattr(principal_context, "user_id", None)
+        service_account_id = getattr(principal_context, "service_account_id", None)
+        if user_id is None and service_account_id is None:
+            logger.warning(
+                "Skipping auth registration for api_key: no creator resolvable",
+                extra={
+                    "api_key_id": api_key_id,
+                    "agent_id": agent_id,
+                },
+            )
+            return
+        try:
+            await self.authorization_service.register_resource(
+                resource=AgentexResource.api_key(api_key_id),
+                parent=AgentexResource.agent(agent_id),
+            )
+        except Exception as exc:
+            # Fail closed: log + re-raise so the Postgres row is never written.
+            logger.exception(
+                "Auth register_resource failed for agent_api_key; aborting create",
+                extra={
+                    "api_key_id": api_key_id,
+                    "agent_id": agent_id,
+                    "error_type": type(exc).__name__,
+                },
+            )
+            raise
+
+    async def _deregister_api_key_from_auth(self, *, api_key_id: str) -> None:
+        """Best-effort deregistration of an api_key's auth tuples on delete.
+
+        ``deregister_resource`` removes the resource and all of its
+        relationships (owner, parent, grantees) atomically. Always invoked;
+        agentex-auth's per-account routing (#353) decides whether the call
+        targets Spark or the legacy backend. Failures are logged but do not
+        block the delete.
+        """
+        try:
+            await self.authorization_service.deregister_resource(
+                resource=AgentexResource.api_key(api_key_id),
+            )
+        except Exception as exc:
+            # Best-effort: log and continue. Postgres row already deleted.
+            logger.warning(
+                "Auth deregister failed for agent_api_key; tuple may be orphaned",
+                extra={
+                    "api_key_id": api_key_id,
+                    "error_type": type(exc).__name__,
+                },
+                exc_info=True,
+            )
+
     async def get(self, id: str) -> AgentAPIKeyEntity:
         return await self.agent_api_key_repo.get(id=id)
 
@@ -124,21 +206,44 @@ async def get_external_by_agent_id_and_key(
         )
 
     async def delete(self, id: str) -> None:
-        return await self.agent_api_key_repo.delete(id=id)
+        # Pre-fetch so we skip both the delete and the deregister when the row
+        # never existed — no DB round-trip, no auth round-trip for a no-op.
+        try:
+            await self.agent_api_key_repo.get(id=id)
+        except ItemDoesNotExist:
+            return
+        await self.agent_api_key_repo.delete(id=id)
+        await self._deregister_api_key_from_auth(api_key_id=id)
 
     async def delete_by_agent_id_and_key_name(
-        self, agent_id: str, key_name: str, api_key_type: AgentAPIKeyType
+        self,
+        agent_id: str,
+        key_name: str,
+        api_key_type: AgentAPIKeyType,
     ) -> None:
-        return await self.agent_api_key_repo.delete_by_agent_id_and_key_name(
+        existing = await self.agent_api_key_repo.get_by_agent_id_and_name(
+            agent_id=agent_id, name=key_name, api_key_type=api_key_type
+        )
+        await self.agent_api_key_repo.delete_by_agent_id_and_key_name(
             agent_id=agent_id, key_name=key_name, api_key_type=api_key_type
         )
+        if existing is not None:
+            await self._deregister_api_key_from_auth(api_key_id=existing.id)
 
     async def delete_by_agent_name_and_key_name(
-        self, agent_name: str, key_name: str, api_key_type: AgentAPIKeyType
+        self,
+        agent_name: str,
+        key_name: str,
+        api_key_type: AgentAPIKeyType,
     ) -> None:
-        return await self.agent_api_key_repo.delete_by_agent_name_and_key_name(
+        existing = await self.agent_api_key_repo.get_by_agent_name_and_key_name(
+            agent_name, key_name, api_key_type
+        )
+        await self.agent_api_key_repo.delete_by_agent_name_and_key_name(
             agent_name=agent_name, key_name=key_name, api_key_type=api_key_type
         )
+        if existing is not None:
+            await self._deregister_api_key_from_auth(api_key_id=existing.id)
 
     async def list(
         self, agent_id: str, limit: int, page_number: int

agentex/tests/integration/fixtures/integration_client.py

@@ -6,7 +6,7 @@
 import asyncio
 import os
 import uuid
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import pymongo
 import pytest
@@ -398,10 +398,17 @@ def create_agents_use_case():
         )
 
     def create_agent_api_keys_use_case():
+        noop_authorization_service = Mock()
+        noop_authorization_service.principal_context = None
+        noop_authorization_service.grant = AsyncMock(return_value={})
+        noop_authorization_service.revoke = AsyncMock(return_value=None)
+        noop_authorization_service.register_resource = AsyncMock(return_value=None)
+        noop_authorization_service.deregister_resource = AsyncMock(return_value=None)
         return AgentAPIKeysUseCase(
             agent_api_key_repository=isolated_repositories["agent_api_key_repository"],
             agent_repository=isolated_repositories["agent_repository"],
             client=isolated_api_key_http_client,  # Use mock client for forwarding requests
+            authorization_service=noop_authorization_service,
         )
 
     def create_deployment_history_use_case():