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

agentex/database/migrations/alembic/versions/2026_05_26_1200_add_agent_api_key_creator_and_zedtoken_b2c84edb77d6.py

@@ -0,0 +1,48 @@
+"""add_agent_api_key_creator_and_zedtoken
+
+Revision ID: b2c84edb77d6
+Revises: a9959ebcbe98
+Create Date: 2026-05-26 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'b2c84edb77d6'
+down_revision: Union[str, None] = 'a9959ebcbe98'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('agent_api_keys', sa.Column('creator_user_id', sa.String(), nullable=True))
+    op.add_column('agent_api_keys', sa.Column('creator_service_account_id', sa.String(), nullable=True))
+    op.add_column('agent_api_keys', sa.Column('spark_authz_zedtoken', sa.Text(), nullable=True))
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_agent_api_keys_creator_user_id "
+            "ON agent_api_keys (creator_user_id)"
+        )
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_agent_api_keys_creator_service_account_id "
+            "ON agent_api_keys (creator_service_account_id)"
+        )
+    op.create_check_constraint(
+        'ck_agent_api_keys_one_creator',
+        'agent_api_keys',
+        '(creator_user_id IS NULL) OR (creator_service_account_id IS NULL)',
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint('ck_agent_api_keys_one_creator', 'agent_api_keys', type_='check')
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_agent_api_keys_creator_service_account_id")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_agent_api_keys_creator_user_id")
+    op.drop_column('agent_api_keys', 'spark_authz_zedtoken')
+    op.drop_column('agent_api_keys', 'creator_service_account_id')
+    op.drop_column('agent_api_keys', 'creator_user_id')

agentex/database/migrations/migration_history.txt

@@ -1,4 +1,6 @@
-9ff3ee32c81b -> e9c4ff9e6542 (head), add_tasks_metadata_gin_index
+a9959ebcbe98 -> b2c84edb77d6 (head), add_agent_api_key_creator_and_zedtoken
+e9c4ff9e6542 -> a9959ebcbe98, finalize_spans_task_id
+9ff3ee32c81b -> e9c4ff9e6542, add_tasks_metadata_gin_index
 57c5ed4f59ae -> 9ff3ee32c81b, uppercase deployment status enum labels
 4a9b7787ccd7 -> 57c5ed4f59ae, add_task_id_to_spans
 d1a6cde41b3f -> 4a9b7787ccd7, deployments

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/adapters/orm.py

@@ -180,6 +180,9 @@ class AgentAPIKeyORM(BaseORM):
     name = Column(String(256), nullable=False, index=True)
     api_key_type = Column(SQLAlchemyEnum(AgentAPIKeyType), nullable=False)
     api_key = Column(String, nullable=False)
+    creator_user_id = Column(String, nullable=True, index=True)
+    creator_service_account_id = Column(String, nullable=True, index=True)
+    spark_authz_zedtoken = Column(Text, nullable=True)
 
     # Indexes for efficient querying
     __table_args__ = (

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/entities/agent_api_keys.py

@@ -25,3 +25,15 @@ class AgentAPIKeyEntity(BaseModel):
         description="The type of the API key (either internal or external)",
     )
     api_key: str = Field(..., description="The API key")
+    creator_user_id: str | None = Field(
+        None,
+        description="Identity ID of the user who created this API key (granted as FGAC owner)",
+    )
+    creator_service_account_id: str | None = Field(
+        None,
+        description="Service identity ID of the service account that created this API key",
+    )
+    spark_authz_zedtoken: str | None = Field(
+        None,
+        description="ZedToken from the Spark AuthZ grant for new-write isolation",
+    )

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)]