feat(AGX1-274): task FGAC dual-write call sites + FGAC_TASKS_DUAL_WRITE flag

Wires `register_resource` / `deregister_resource` into
`AgentTaskService.create_task` / `delete_task`, gated by a new
`FGAC_TASKS_DUAL_WRITE` env-var (off by default; resolved at
DI-resolve time so mid-process flips are intentionally invisible —
rollout assumes a redeploy cycles pods).

- `create_task`: after the Postgres row persists, register the
  task in the authorization graph with `parent=agent` so the
  tenant + owner + parent_agent tuples are written atomically.
- `delete_task`: pre-resolves the task id by name before the
  Postgres delete (lookup-after-delete would race), then
  deregisters once the row is gone. The name-lookup
  `ItemDoesNotExist` is swallowed so the subsequent `delete()`
  surfaces its own native error — flipping the flag must not
  change the error contract for missing tasks.
- Both call sites share `_dual_write_with_retry(op_name, do_call,
  task_id)`, which retries transient
  `AuthenticationServiceUnavailableError` /
  `AuthenticationGatewayError` with exponential backoff + jitter
  (3 retries → 4 attempts max). Mirrors
  `agents_acp_use_case.grant_with_retry`, but with no `fail_task`
  fallback: the Postgres row is the durable record and orphan auth
  tuples are preferable to losing the task. The AGX1-291 operator
  runbook covers backfill using the creator-audit columns added in
  the parent commit.
- Emits `task_fgac_dual_write.{attempt,success,retry,failure}`
  statsd counters (tagged `op:register|deregister` and
  `exception_class` on failure) — the rollout signal for the
  FGAC_TASKS_DUAL_WRITE flip dashboard.

The `Port` interface gains `register_resource` /
`deregister_resource`, and the agentex-auth proxy adapter calls
`POST /v1/authz/register` and `POST /v1/authz/deregister`. The
endpoints themselves already live on agentex-auth `main` via #354;
per-account routing across them is set by scaleapi/agentex#353.

Part of the AGX1-264 stack: scaleapi/scaleapi NEW2
(per-account FF endpoint) → scaleapi/agentex#353 (agentex-auth
routing + cancel) → this PR → #249 (per-RPC
route migration).

Author: asherfink

agentex/src/adapters/authorization/adapter_agentex_authz_proxy.py

@@ -85,6 +85,35 @@ async def list_resources(
         )
         return response["items"]
 
+    async def register_resource(
+        self,
+        principal: AgentexAuthPrincipalContext,
+        resource: AgentexResource,
+        parent: AgentexResource | None = None,
+    ) -> None:
+        payload: dict = {
+            "principal": principal,
+            "resource": resource.model_dump(),
+        }
+        if parent is not None:
+            payload["parent"] = parent.model_dump()
+        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,31 @@ 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 the authorization graph.
+
+        Atomically writes the relation tuples the schema requires (tenant +
+        owner, plus an optional typed parent like ``task.parent_agent``).
+        Distinct from ``grant`` because ``grant`` writes a single role
+        relation, which is insufficient for schemas that gate access on
+        ``tenant->membership``.
+        """
+
+    @abstractmethod
+    async def deregister_resource(
+        self,
+        principal: PrincipalT,
+        resource: AgentexResource,
+    ) -> None:
+        """Deregister a resource being deleted from the authorization graph.
+
+        Removes every relation tuple written for the resource — keeps the
+        graph in sync with the application database on row delete.
+        """

agentex/src/config/environment_variables.py

@@ -58,6 +58,7 @@ class EnvVarKeys(str, Enum):
     AGENTEX_SERVER_TASK_QUEUE = "AGENTEX_SERVER_TASK_QUEUE"
     ENABLE_HEALTH_CHECK_WORKFLOW = "ENABLE_HEALTH_CHECK_WORKFLOW"
     WEBHOOK_REQUEST_TIMEOUT = "WEBHOOK_REQUEST_TIMEOUT"
+    FGAC_TASKS_DUAL_WRITE = "FGAC_TASKS_DUAL_WRITE"
 
 
 class Environment(str, Enum):
@@ -114,6 +115,10 @@ class EnvironmentVariables(BaseModel):
     AGENTEX_SERVER_TASK_QUEUE: str | None = None
     ENABLE_HEALTH_CHECK_WORKFLOW: bool = False
     WEBHOOK_REQUEST_TIMEOUT: float = 15.0  # Webhook request timeout in seconds
+    # AGX1-274: gate the task FGAC dual-write call sites. Off by default so
+    # rollout is operator-controlled per environment. Mirrors KB's
+    # ``FGAC_KNOWLEDGE_BASES_DUAL_WRITE`` shape.
+    FGAC_TASKS_DUAL_WRITE: bool = False
 
     @classmethod
     def refresh(cls, force_refresh: bool = False) -> EnvironmentVariables | None:
@@ -203,6 +208,10 @@ def refresh(cls, force_refresh: bool = False) -> EnvironmentVariables | None:
             WEBHOOK_REQUEST_TIMEOUT=float(
                 os.environ.get(EnvVarKeys.WEBHOOK_REQUEST_TIMEOUT, "15.0")
             ),
+            FGAC_TASKS_DUAL_WRITE=(
+                os.environ.get(EnvVarKeys.FGAC_TASKS_DUAL_WRITE, "false").lower()
+                == "true"
+            ),
         )
         refreshed_environment_variables = environment_variables
         return refreshed_environment_variables

agentex/src/domain/services/authorization_service.py

@@ -1,5 +1,5 @@
 from collections.abc import Iterable
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastapi import Depends, Request
 
@@ -17,6 +17,11 @@
 
 logger = make_logger(__name__)
 
+# Sentinel for "caller did not pass an explicit principal_context" — None is a
+# valid value (the bypass path logs ``for principal None``) so it can't be the
+# default. Using a named object reads better at the call site than ``...``.
+_UNSET: Any = object()
+
 
 class AuthorizationService:
     def __init__(
@@ -40,7 +45,11 @@ def is_enabled(self) -> bool:
         return self.enabled
 
     async def grant(
-        self, resource: AgentexResource, *, commit: bool = True, principal_context=...
+        self,
+        resource: AgentexResource,
+        *,
+        commit: bool = True,
+        principal_context: Any = _UNSET,
     ) -> None:
         if self._bypass():
             logger.info(
@@ -57,15 +66,19 @@ async def grant(
         )
         result = await self.gateway.grant(
             principal_context
-            if principal_context is not ...
+            if principal_context is not _UNSET
             else self.principal_context,
             resource,
             AuthorizedOperationType.create,
         )
         return result
 
     async def revoke(
-        self, resource: AgentexResource, *, commit: bool = True, principal_context=...
+        self,
+        resource: AgentexResource,
+        *,
+        commit: bool = True,
+        principal_context: Any = _UNSET,
     ) -> None:
         if self._bypass():
             logger.info("Authorization bypassed for revoke operation")
@@ -81,7 +94,7 @@ async def revoke(
 
         result = await self.gateway.revoke(
             principal_context
-            if principal_context is not ...
+            if principal_context is not _UNSET
             else self.principal_context,
             resource,
             AuthorizedOperationType.delete,
@@ -96,7 +109,7 @@ async def check(
         resource: AgentexResource,
         operation: AuthorizedOperationType,
         *,
-        principal_context=...,
+        principal_context: Any = _UNSET,
     ) -> bool:
         if self._bypass():
             logger.info("Authorization bypassed for check operation")
@@ -105,7 +118,7 @@ async def check(
         # Determine which principal context to use
         effective_principal = (
             principal_context
-            if principal_context is not ...
+            if principal_context is not _UNSET
             else self.principal_context
         )
 
@@ -157,12 +170,75 @@ async def check(
         )
         return result
 
+    async def register_resource(
+        self,
+        resource: AgentexResource,
+        *,
+        parent: AgentexResource | None = None,
+        principal_context: Any = _UNSET,
+    ) -> None:
+        """Register a freshly-created resource with the authorization graph.
+
+        Used immediately after persisting a new row to write the tenant +
+        owner (and optionally typed parent) relation tuples atomically.
+        Distinct from ``grant`` because ``grant`` only writes a single
+        role relation, which is insufficient for schemas (e.g. ``task``)
+        that require a ``tenant->membership`` gate.
+        """
+        if self._bypass():
+            logger.info(
+                f"Authorization bypassed for register_resource on {resource.type}:{resource.selector}"
+            )
+            return None
+
+        logger.info(
+            "[authorization_service] Registering resource %s:%s for principal %s (parent=%s)",
+            resource.type,
+            resource.selector,
+            self.principal_context,
+            parent,
+        )
+        await self.gateway.register_resource(
+            principal_context
+            if principal_context is not _UNSET
+            else self.principal_context,
+            resource,
+            parent,
+        )
+
+    async def deregister_resource(
+        self,
+        resource: AgentexResource,
+        *,
+        principal_context: Any = _UNSET,
+    ) -> None:
+        """Remove every relation tuple written for the resource — used when
+        deleting the underlying database row."""
+        if self._bypass():
+            logger.info(
+                f"Authorization bypassed for deregister_resource on {resource.type}:{resource.selector}"
+            )
+            return None
+
+        logger.info(
+            "[authorization_service] Deregistering resource %s:%s for principal %s",
+            resource.type,
+            resource.selector,
+            self.principal_context,
+        )
+        await self.gateway.deregister_resource(
+            principal_context
+            if principal_context is not _UNSET
+            else self.principal_context,
+            resource,
+        )
+
     async def list_resources(
         self,
         filter_resource: AgentexResourceType,
         filter_operation: AuthorizedOperationType = AuthorizedOperationType.read,
         *,
-        principal_context=...,
+        principal_context: Any = _UNSET,
     ) -> Iterable[str] | None:
         """List resource identifiers for which the current principal has *filter_operation* permission."""
 
@@ -178,7 +254,7 @@ async def list_resources(
         )
         result = await self.gateway.list_resources(
             principal_context
-            if principal_context is not ...
+            if principal_context is not _UNSET
             else self.principal_context,
             filter_resource,
             filter_operation,