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

@@ -8,6 +8,7 @@
     CreateAPIKeyResponse,
 )
 from src.domain.entities.agent_api_keys import AgentAPIKeyType
+from src.domain.services.authorization_service import DAuthorizationService
 from src.domain.use_cases.agent_api_keys_use_case import DAgentAPIKeysUseCase
 from src.domain.use_cases.agents_use_case import DAgentsUseCase
 from src.utils.logging import make_logger
@@ -28,6 +29,7 @@ async def create_api_key(
     request: CreateAPIKeyRequest,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorization_service: DAuthorizationService,
 ) -> CreateAPIKeyResponse:
     if not request.agent_id and not request.agent_name:
         raise HTTPException(
@@ -52,11 +54,13 @@ async def create_api_key(
         raise HTTPException(status_code=409, detail=error_msg)
 
     new_api_key = request.api_key or secrets.token_hex(32)
+    account_id = getattr(authorization_service.principal_context, "account_id", None)
     agent_api_key_entity = await agent_api_key_use_case.create(
         agent_id=agent.id,
         api_key=str(new_api_key),
         name=request.name,
         api_key_type=request.api_key_type,
+        account_id=account_id,
     )
     return CreateAPIKeyResponse(
         id=agent_api_key_entity.id,
@@ -161,8 +165,10 @@ async def get_agent_api_key(
 async def delete_agent_api_key(
     id: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
+    authorization_service: DAuthorizationService,
 ) -> str:
-    await agent_api_key_use_case.delete(id=id)
+    account_id = getattr(authorization_service.principal_context, "account_id", None)
+    await agent_api_key_use_case.delete(id=id, account_id=account_id)
     return f"Agent API key with ID {id} deleted"
 
 
@@ -176,6 +182,7 @@ async def delete_agent_api_key_by_name(
     api_key_name: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorization_service: DAuthorizationService,
     agent_id: str | None = None,
     agent_name: str | None = None,
     api_key_type: AgentAPIKeyType = AgentAPIKeyType.EXTERNAL,
@@ -191,8 +198,12 @@ async def delete_agent_api_key_by_name(
             detail="Only one of 'agent_id' or 'agent_name' should be provided to delete an agent api_key.",
         )
     agent = await agent_use_case.get(id=agent_id, name=agent_name)
+    account_id = getattr(authorization_service.principal_context, "account_id", None)
     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,
+        account_id=account_id,
     )
 
     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)]