feat: register_resource with parent edge for agent_api_keys dual-write

Closes the parent_agent cascade gap surfaced on scaleapi/agentex#354.
The api_key dual-write (AGX1-272, PR #248) currently calls grant() which
writes the owner edge in SpiceDB but NOT the parent_agent edge. The
agent_api_key schema requires `read = ... & parent_agent->read & ...`,
so every downstream read/update fails closed without that edge.

This PR adds register_resource/deregister_resource (Port + adapter + service)
and swaps the api_keys use case from grant→register_resource with
parent=AgentexResource.agent(agent_id). Now the owner edge and parent_agent
edge are written atomically.

Stack:
- scaleapi/scaleapi#144657 — sgp-authz 0.7.0 (parent_resource kwarg).
- scaleapi/agentex#355 — agentex-auth Port + adapter + HTTP routes.
- #248 — original AGX1-272 dual-write (this stacks on it).
- THIS PR — extends #248 to use the parent-aware path.

Changes:
- Port: abstract register_resource(resource, parent=None) and
  deregister_resource(resource).
- Adapter proxy: POST /v1/authz/register and /v1/authz/deregister.
- Service: mirror existing grant/revoke pattern (principal_context override,
  _bypass support, parent in log line for cascade debugging).
- Use case: swap grant→register_resource passing parent=agent;
  swap revoke→deregister_resource. except Exception wrappers preserved
  (fail-closed on register, best-effort on deregister).
- Tests: rename mocks to register_resource/deregister_resource; assert the
  parent edge is passed correctly.

Test plan:
- pytest agentex/tests/integration/services/test_agent_api_key_service_dual_write.py
  → 8 / 8 pass.
- New test ``test_create_api_key_calls_grant_when_flag_on`` asserts
  parent.type == AgentexResourceType.agent and parent.selector == agent.id.

Other resource types' grant→register_resource swap is out of scope.

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

Author: dm36

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/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

@@ -135,23 +135,24 @@ async def _register_api_key_in_spark_authz(
         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.
+        """Register a new agent_api_key in Spark AuthZ with creator as owner
+        AND the parent_agent edge to the owning agent.
 
         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
+        row from being persisted, so there is no compensating action.
+
+        The ``agent_api_key`` SpiceDB schema has a ``parent_agent`` relation
+        that read/update/delete permissions cascade through:
+        ``permission read = internal_effective_viewer & parent_agent->read &
+        internal_tenant_gate``. We MUST write the parent edge here or every
+        downstream permission check fails closed. ``register_resource``
+        (added in agentex-auth and sgp-authz 0.7.0) writes both the owner
+        edge and the parent edge atomically in one round-trip.
+
+        The current ``register_resource`` returns ``None``; no ZedToken is
+        surfaced today, so we always return ``None`` for the
+        spark_authz_zedtoken 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(
@@ -163,16 +164,36 @@ async def _register_api_key_in_spark_authz(
                 },
             )
             return None
-        await self.authorization_service.grant(
-            resource=AgentexResource.api_key(api_key_id),
-        )
+        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.
+            # The dual-write contract requires the SpiceDB tuple (and parent
+            # edge) to exist before the row does.
+            logger.exception(
+                "Spark AuthZ register_resource failed for agent_api_key; aborting create",
+                extra={
+                    "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,
+                    "error_type": type(exc).__name__,
+                },
+            )
+            raise
         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.
+        """Best-effort deregistration of an api_key's Spark AuthZ tuples on delete.
 
+        ``deregister_resource`` removes the resource and all of its
+        relationships (owner, parent_agent, any grantees) atomically.
         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.
@@ -182,13 +203,18 @@ async def _deregister_api_key_from_spark_authz(
         ):
             return
         try:
-            await self.authorization_service.revoke(
+            await self.authorization_service.deregister_resource(
                 resource=AgentexResource.api_key(api_key_id),
             )
-        except Exception:
+        except Exception as exc:
+            # Best-effort: log and continue. Postgres row already deleted.
             logger.warning(
-                "Spark AuthZ revoke failed for agent_api_key",
-                extra={"api_key_id": api_key_id, "account_id": account_id},
+                "Spark AuthZ deregister failed for agent_api_key; tuple may be orphaned",
+                extra={
+                    "api_key_id": api_key_id,
+                    "account_id": account_id,
+                    "error_type": type(exc).__name__,
+                },
                 exc_info=True,
             )