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

Mirrors the AGX1-274 task dual-write pattern (PR #246) for agent_api_keys.

- Adds creator_user_id / creator_service_account_id / spark_authz_zedtoken
  columns to agent_api_keys, with CHECK constraint and concurrent indexes.
- On create, when FGAC_AGENT_API_KEYS_DUAL_WRITE is enabled for the caller's
  account, calls authorization_service.grant(AgentexResource.api_key(id))
  BEFORE the Postgres write. Grant failure aborts the create.
- On delete, best-effort revoke after the Postgres delete. Failures are
  logged but do not block the delete.
- Adds AgentexResourceType.api_key and AgentexResource.api_key(...) factory.
- Creates src/utils/feature_flags.py with both FGAC_TASKS_DUAL_WRITE and
  FGAC_AGENT_API_KEYS_DUAL_WRITE (file does not exist on main yet; if PR #246
  lands first this becomes a rebase concern).

Structural divergence from tasks: agent_api_keys have no service layer, so
the dual-write logic lives in AgentAPIKeysUseCase rather than a separate
service. This keeps the call site simple and avoids inventing a new layer.

Route layer (read-side auth checks) is out of scope; that's PR B (AGX1-273).
agentex-auth spark_mapping.py update is a sibling-repo concern.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dm36

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/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/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,8 @@
     DAgentAPIKeyRepository,
 )
 from src.domain.repositories.agent_repository import DAgentRepository
+from src.domain.services.authorization_service import DAuthorizationService
+from src.utils.feature_flags import DFeatureFlagProvider, FeatureFlagName
 from src.utils.ids import orm_id
 from src.utils.logging import make_logger
 
@@ -39,10 +42,14 @@ def __init__(
         agent_api_key_repository: DAgentAPIKeyRepository,
         agent_repository: DAgentRepository,
         client: DHttpxClient,
+        authorization_service: DAuthorizationService,
+        feature_flags: DFeatureFlagProvider,
     ):
         self.agent_api_key_repo = agent_api_key_repository
         self.agent_repo = agent_repository
         self.client = client
+        self.authorization_service = authorization_service
+        self.feature_flags = feature_flags
         self.auth_gateway_enabled = bool(
             resolve_environment_variable_dependency(EnvVarKeys.AGENTEX_AUTH_URL)
         )
@@ -76,24 +83,115 @@ async def create(
         agent_id: str,
         api_key_type: AgentAPIKeyType,
         api_key: str,
+        account_id: str | None = None,
     ) -> AgentAPIKeyEntity:
         agent = await self.get_agent(agent_id=agent_id)
         if not agent:
             raise HTTPException(
                 status_code=404,
                 detail=f"Agent ID {agent_id} not found.",
             )
+
+        principal_context = self.authorization_service.principal_context
+        creator_user_id = getattr(principal_context, "user_id", None)
+        creator_service_account_id = getattr(
+            principal_context, "service_account_id", None
+        )
+
+        api_key_id = orm_id()
+        zedtoken: str | None = None
+
+        if self.feature_flags.is_enabled(
+            FeatureFlagName.FGAC_AGENT_API_KEYS_DUAL_WRITE, account_id
+        ):
+            zedtoken = await self._register_api_key_in_spark_authz(
+                api_key_id=api_key_id,
+                agent_id=agent.id,
+                account_id=account_id,
+                creator_user_id=creator_user_id,
+                creator_service_account_id=creator_service_account_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,
+            creator_user_id=creator_user_id,
+            creator_service_account_id=creator_service_account_id,
+            spark_authz_zedtoken=zedtoken,
         )
         return await self.agent_api_key_repo.create(item=agent_api_key)
 
+    async def _register_api_key_in_spark_authz(
+        self,
+        *,
+        api_key_id: str,
+        agent_id: str,
+        account_id: str | None,
+        creator_user_id: str | None,
+        creator_service_account_id: str | None,
+    ) -> str | None:
+        """Register a new agent_api_key in Spark AuthZ with creator as owner.
+
+        Called BEFORE the Postgres write — a failure raises and prevents the
+        row from being persisted, so there is no compensating action to take.
+        Mirrors the dual-write pattern used for tasks (AGX1-274).
+
+        The current ``Provider.spark`` adapter returns ``{}`` from ``grant``;
+        no ZedToken is surfaced today, so we always return ``None`` for the
+        new-write-isolation column. A follow-up will plumb the token through
+        once the adapter exposes it.
+
+        Note: the ``agent_api_key`` SpiceDB schema has a ``parent_agent``
+        relation that read/delete permissions cascade through. The current
+        ``AuthorizationGateway.grant`` signature does not accept a parent
+        relation — the agentex-auth adapter is expected to set
+        ``parent_agent`` based on the resource shape. This is the same
+        gap Asher's task PR has and is tracked as a follow-up.
+        """
+        if creator_user_id is None and creator_service_account_id is None:
+            logger.warning(
+                "Skipping Spark AuthZ api_key registration: no creator resolvable",
+                extra={
+                    "api_key_id": api_key_id,
+                    "agent_id": agent_id,
+                    "account_id": account_id,
+                },
+            )
+            return None
+        await self.authorization_service.grant(
+            resource=AgentexResource.api_key(api_key_id),
+        )
+        return None
+
+    async def _deregister_api_key_from_spark_authz(
+        self, *, api_key_id: str, account_id: str | None
+    ) -> None:
+        """Best-effort revocation of an api_key's Spark AuthZ tuples on delete.
+
+        Only invoked when the FGAC_AGENT_API_KEYS_DUAL_WRITE flag is enabled
+        for the caller's account. Failures are logged but do not block the
+        delete.
+        """
+        if not self.feature_flags.is_enabled(
+            FeatureFlagName.FGAC_AGENT_API_KEYS_DUAL_WRITE, account_id
+        ):
+            return
+        try:
+            await self.authorization_service.revoke(
+                resource=AgentexResource.api_key(api_key_id),
+            )
+        except Exception:
+            logger.warning(
+                "Spark AuthZ revoke failed for agent_api_key",
+                extra={"api_key_id": api_key_id, "account_id": account_id},
+                exc_info=True,
+            )
+
     async def get(self, id: str) -> AgentAPIKeyEntity:
         return await self.agent_api_key_repo.get(id=id)
 
@@ -123,22 +221,47 @@ async def get_external_by_agent_id_and_key(
             agent_id=agent_id, api_key=api_key
         )
 
-    async def delete(self, id: str) -> None:
-        return await self.agent_api_key_repo.delete(id=id)
+    async def delete(self, id: str, account_id: str | None = None) -> None:
+        await self.agent_api_key_repo.delete(id=id)
+        await self._deregister_api_key_from_spark_authz(
+            api_key_id=id, account_id=account_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,
+        account_id: str | None = None,
     ) -> 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_spark_authz(
+                api_key_id=existing.id, account_id=account_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,
+        account_id: str | None = None,
     ) -> 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_spark_authz(
+                api_key_id=existing.id, account_id=account_id
+            )
 
     async def list(
         self, agent_id: str, limit: int, page_number: int