refactor(AGX1-272): query egp-api-backend for FGAC_AGENT_API_KEYS_DUAL_WRITE

Per team discussion: rather than maintain a parallel env-var flag system
in scale-agentex, route api_key dual-write flag checks through
egp-api-backend's existing flag service. One source of truth across services,
single flip surface for ops, fewer per-env env-var allowlists to keep in sync.

Changes:

- EnvVarKeys.EGP_API_BACKEND_URL — new env var for the egp-api-backend
  base URL. Used by the new HTTP-backed flag provider.

- FeatureFlagProvider rewritten as an HTTP client of egp-api-backend's
  GET /feature-flag/{id} endpoint:
    * Forwards x-api-key / x-user-id / x-service-account-id /
      x-selected-account-id from the caller's principal_context so the
      endpoint's REQUIRE_IDENTITY_AND_OPTIONAL_ACCOUNT policy admits the
      request.
    * Coerces the response's `value` field to bool.
    * Fails closed to False on any error (config missing, no identity,
      non-2xx, transport failure, JSON parse failure) — the legacy
      no-Spark code path is the safe default.
    * `is_enabled` is now async (HTTP call). Signature is
      `is_enabled(name, *, principal_context, account_id)`.

- AgentAPIKeysUseCase: both call sites now await is_enabled and pass
  principal_context. _deregister grabs principal_context from
  self.authorization_service.

- Test fixtures: mock FeatureFlagProvider directly (Mock with
  is_enabled = AsyncMock(return_value=flag_on)) so dual-write tests stay
  hermetic. The pre-existing FeatureFlagProvider() no-arg constructions
  in test_agents_api_keys_use_case.py and integration_client.py now pass
  egp_api_backend_url=None (provider returns False without it, matching
  the prior "flag never enabled in unit tests" behavior).

Out of scope:

- Migrating Asher's FGAC_TASKS_DUAL_WRITE flag check off env vars.
  That's task-team-owned and we leave their existing pattern alone per
  the team discussion (new-work-only).
- Caching the flag response. Each is_enabled is a fresh HTTP call.
  Egp-api-backend's flag endpoint is fast and the caller paths are
  already crossing the network for the actual register/deregister, so
  one extra round-trip is acceptable for now. Add caching later if
  load profiling shows it matters.

Test plan:

- uv run pytest agentex/tests/integration/services/test_agent_api_key_service_dual_write.py — 8/8 pass.
- Existing 4 unrelated test_agents_api_keys_use_case.py docker-fixture
  errors predate this commit (verified via `git stash`).

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

Author: dm36

agentex/src/config/environment_variables.py

@@ -44,6 +44,7 @@ class EnvVarKeys(str, Enum):
     REDIS_STREAM_TTL_SECONDS = "REDIS_STREAM_TTL_SECONDS"
     IMAGE_PULL_SECRET_NAME = "IMAGE_PULL_SECRET_NAME"
     AGENTEX_AUTH_URL = "AGENTEX_AUTH_URL"
+    EGP_API_BACKEND_URL = "EGP_API_BACKEND_URL"
     ALLOWED_ORIGINS = "ALLOWED_ORIGINS"
     DD_AGENT_HOST = "DD_AGENT_HOST"
     DD_STATSD_PORT = "DD_STATSD_PORT"
@@ -100,6 +101,7 @@ class EnvironmentVariables(BaseModel):
     )
     IMAGE_PULL_SECRET_NAME: str | None = None
     AGENTEX_AUTH_URL: str | None = None
+    EGP_API_BACKEND_URL: str | None = None
     ALLOWED_ORIGINS: str | None = None
     HTTPX_MAX_CONNECTIONS: int = 200  # Max total connections allowed
     HTTPX_MAX_KEEPALIVE_CONNECTIONS: int = 100  # Max connections to keep alive
@@ -166,6 +168,7 @@ def refresh(cls, force_refresh: bool = False) -> EnvironmentVariables | None:
             ),
             IMAGE_PULL_SECRET_NAME=os.environ.get(EnvVarKeys.IMAGE_PULL_SECRET_NAME),
             AGENTEX_AUTH_URL=os.environ.get(EnvVarKeys.AGENTEX_AUTH_URL),
+            EGP_API_BACKEND_URL=os.environ.get(EnvVarKeys.EGP_API_BACKEND_URL),
             ALLOWED_ORIGINS=os.environ.get(EnvVarKeys.ALLOWED_ORIGINS, "*"),
             DD_AGENT_HOST=os.environ.get(EnvVarKeys.DD_AGENT_HOST),
             DD_STATSD_PORT=os.environ.get(EnvVarKeys.DD_STATSD_PORT),

agentex/src/domain/use_cases/agent_api_keys_use_case.py

@@ -101,8 +101,10 @@ async def create(
         api_key_id = orm_id()
         zedtoken: str | None = None
 
-        if self.feature_flags.is_enabled(
-            FeatureFlagName.FGAC_AGENT_API_KEYS_DUAL_WRITE, account_id
+        if await self.feature_flags.is_enabled(
+            FeatureFlagName.FGAC_AGENT_API_KEYS_DUAL_WRITE,
+            principal_context=principal_context,
+            account_id=account_id,
         ):
             zedtoken = await self._register_api_key_in_spark_authz(
                 api_key_id=api_key_id,
@@ -198,8 +200,10 @@ async def _deregister_api_key_from_spark_authz(
         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
+        if not await self.feature_flags.is_enabled(
+            FeatureFlagName.FGAC_AGENT_API_KEYS_DUAL_WRITE,
+            principal_context=self.authorization_service.principal_context,
+            account_id=account_id,
         ):
             return
         try:

agentex/src/utils/feature_flags.py

@@ -1,9 +1,30 @@
-import os
+"""Per-account feature flag provider.
+
+Queries egp-api-backend's ``GET /feature-flag/{id}`` endpoint with the
+caller's identity headers. The endpoint evaluates the flag against the
+account_id resolved from those headers and returns a ``FeatureFlag``
+payload whose ``value`` field is a typed flag value (bool for ``boolean``
+flags, etc.).
+
+Falls back to disabled (``False``) on any error so a transient
+egp-api-backend outage doesn't break the dual-write path — the flag-off
+behavior is the safe legacy path.
+"""
+
+from __future__ import annotations
+
 from enum import StrEnum
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastapi import Depends
 
+from src.config.dependencies import DEnvironmentVariable
+from src.config.environment_variables import EnvVarKeys
+from src.utils.cached_httpx_client import get_async_client
+from src.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
 
 class FeatureFlagName(StrEnum):
     FGAC_TASKS = "fgac-tasks"
@@ -12,19 +33,112 @@ class FeatureFlagName(StrEnum):
 
 
 class FeatureFlagProvider:
-    """Per-account feature flag provider.
+    """Per-account feature flag provider backed by egp-api-backend.
 
-    v1: env-var allowlist (per-account, comma-separated). The env var name is
-    derived from the flag name, e.g. ``FGAC_AGENT_API_KEYS_DUAL_WRITE_ACCOUNTS``.
-    A follow-up will swap this for LaunchDarkly with an account_id context.
+    Calls ``GET {EGP_API_BACKEND_URL}/feature-flag/{name}`` with the
+    caller's identity headers. The endpoint evaluates the flag against the
+    caller's account and returns a ``FeatureFlag`` with a ``value`` field.
+    For boolean flags this method coerces the value to ``bool``.
+
+    Returns ``False`` (flag off) when:
+    - ``EGP_API_BACKEND_URL`` is not configured;
+    - the caller's principal has no usable identity headers;
+    - egp-api-backend returns a non-2xx response;
+    - any network or parsing error occurs.
+
+    Fail-closed-to-disabled is intentional: the legacy code path is the
+    safe default if FGAC dual-write is unreachable.
     """
 
-    def is_enabled(self, name: FeatureFlagName, account_id: str | None) -> bool:
-        if not account_id:
+    def __init__(
+        self,
+        egp_api_backend_url: DEnvironmentVariable(EnvVarKeys.EGP_API_BACKEND_URL),
+    ):
+        self.egp_api_backend_url = egp_api_backend_url
+
+    async def is_enabled(
+        self,
+        name: FeatureFlagName,
+        *,
+        principal_context: Any,
+        account_id: str | None,
+    ) -> bool:
+        if not self.egp_api_backend_url:
             return False
-        env_key = f"{name.value.upper().replace('-', '_')}_ACCOUNTS"
-        allowed = os.environ.get(env_key, "")
-        return account_id in {a.strip() for a in allowed.split(",") if a.strip()}
+
+        headers = self._principal_headers(principal_context, account_id)
+        if not headers:
+            return False
+
+        url = f"{self.egp_api_backend_url.rstrip('/')}/feature-flag/{name.value}"
+        try:
+            client = get_async_client()
+            response = await client.get(url, headers=headers)
+        except Exception as exc:
+            logger.warning(
+                "Feature flag fetch failed; treating as disabled",
+                extra={
+                    "flag": name.value,
+                    "account_id": account_id,
+                    "error_type": type(exc).__name__,
+                },
+            )
+            return False
+
+        if response.status_code != 200:
+            logger.warning(
+                "Feature flag non-2xx response; treating as disabled",
+                extra={
+                    "flag": name.value,
+                    "account_id": account_id,
+                    "status_code": response.status_code,
+                },
+            )
+            return False
+
+        try:
+            payload = response.json()
+            value = payload.get("value")
+        except Exception:
+            logger.warning(
+                "Feature flag response not JSON-parseable; treating as disabled",
+                extra={"flag": name.value, "account_id": account_id},
+            )
+            return False
+
+        return bool(value)
+
+    @staticmethod
+    def _principal_headers(
+        principal_context: Any, account_id: str | None
+    ) -> dict[str, str]:
+        """Build identity headers from the caller's principal_context so
+        egp-api-backend's ``REQUIRE_IDENTITY_AND_OPTIONAL_ACCOUNT`` policy
+        admits the request.
+
+        Returns ``{}`` when no usable identity is present — the caller
+        should treat that as flag-off (the legacy path is safe).
+        """
+        if principal_context is None:
+            return {}
+
+        api_key = getattr(principal_context, "api_key", None)
+        user_id = getattr(principal_context, "user_id", None)
+        service_account_id = getattr(principal_context, "service_account_id", None)
+
+        if not api_key and not user_id and not service_account_id:
+            return {}
+
+        headers: dict[str, str] = {}
+        if api_key:
+            headers["x-api-key"] = api_key
+        if user_id:
+            headers["x-user-id"] = user_id
+        if service_account_id:
+            headers["x-service-account-id"] = service_account_id
+        if account_id:
+            headers["x-selected-account-id"] = account_id
+        return headers
 
 
 DFeatureFlagProvider = Annotated[FeatureFlagProvider, Depends(FeatureFlagProvider)]

agentex/tests/integration/fixtures/integration_client.py

@@ -409,7 +409,7 @@ def create_agent_api_keys_use_case():
             agent_repository=isolated_repositories["agent_repository"],
             client=isolated_api_key_http_client,  # Use mock client for forwarding requests
             authorization_service=noop_authorization_service,
-            feature_flags=FeatureFlagProvider(),
+            feature_flags=FeatureFlagProvider(egp_api_backend_url=None),
         )
 
     def create_deployment_history_use_case():

agentex/tests/integration/services/test_agent_api_key_service_dual_write.py

@@ -37,7 +37,6 @@
 from src.domain.entities.agent_api_keys import AgentAPIKeyEntity, AgentAPIKeyType
 from src.domain.entities.agents import ACPType, AgentEntity, AgentStatus
 from src.domain.use_cases.agent_api_keys_use_case import AgentAPIKeysUseCase
-from src.utils.feature_flags import FeatureFlagProvider
 from src.utils.ids import orm_id
 
 
@@ -62,16 +61,14 @@ def _agent() -> AgentEntity:
 
 def _build_use_case(
     *,
-    flag_accounts: str,
+    flag_on: bool,
     principal: SimpleNamespace | None,
     register_resource: AsyncMock | None = None,
     deregister_resource: AsyncMock | None = None,
     agent: AgentEntity | None = None,
     create_raises: Exception | None = None,
     monkeypatch: pytest.MonkeyPatch,
 ) -> tuple[AgentAPIKeysUseCase, Mock, AsyncMock, AsyncMock]:
-    monkeypatch.setenv("FGAC_AGENT_API_KEYS_DUAL_WRITE_ACCOUNTS", flag_accounts)
-
     sample_agent = agent or _agent()
 
     agent_repository = Mock()
@@ -103,7 +100,11 @@ def _build_use_case(
         return_value=None
     )
 
-    feature_flags = FeatureFlagProvider()
+    # FeatureFlagProvider normally calls egp-api-backend over HTTP. Mock it
+    # so tests are hermetic; behaviour under test is the use case's response
+    # to the flag value, not the provider's transport.
+    feature_flags = Mock()
+    feature_flags.is_enabled = AsyncMock(return_value=flag_on)
 
     # Patch env var lookup inside UseCase __init__ so we don't depend on real
     # env configuration to instantiate.
@@ -133,7 +134,7 @@ async def test_create_api_key_skips_grant_when_flag_off(
 ) -> None:
     agent = _agent()
     use_case, repo, register, _ = _build_use_case(
-        flag_accounts="",
+        flag_on=False,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         agent=agent,
         monkeypatch=monkeypatch,
@@ -161,7 +162,7 @@ async def test_create_api_key_calls_grant_when_flag_on(
 ) -> None:
     agent = _agent()
     use_case, repo, register, _ = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         agent=agent,
         monkeypatch=monkeypatch,
@@ -197,7 +198,7 @@ async def test_delete_api_key_calls_revoke_when_flag_on(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     use_case, repo, _, deregister = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         monkeypatch=monkeypatch,
     )
@@ -218,7 +219,7 @@ async def test_delete_api_key_skips_revoke_when_flag_off(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     use_case, repo, _, deregister = _build_use_case(
-        flag_accounts="",
+        flag_on=False,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         monkeypatch=monkeypatch,
     )
@@ -237,7 +238,7 @@ async def test_create_api_key_grant_failure_prevents_db_row(
     register_resource = AsyncMock(side_effect=RuntimeError("spark unavailable"))
     agent = _agent()
     use_case, repo, _, _ = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         register_resource=register_resource,
         agent=agent,
@@ -263,7 +264,7 @@ async def test_delete_api_key_revoke_failure_does_not_block_delete(
 ) -> None:
     deregister = AsyncMock(side_effect=RuntimeError("spark unavailable"))
     use_case, repo, _, deregister_ref = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         deregister_resource=deregister,
         monkeypatch=monkeypatch,
@@ -285,7 +286,7 @@ async def test_create_api_key_skips_grant_when_no_creator_resolvable(
     the dual-write is a no-op (logged) and the row still lands without a tuple."""
     agent = _agent()
     use_case, repo, register, _ = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id=None, account_id="acct-1"),
         agent=agent,
         monkeypatch=monkeypatch,
@@ -313,7 +314,7 @@ async def test_delete_by_agent_id_and_key_name_revokes_existing(
     agent = _agent()
     existing_id = orm_id()
     use_case, repo, _, deregister = _build_use_case(
-        flag_accounts="acct-1",
+        flag_on=True,
         principal=_principal(user_id="user-A", account_id="acct-1"),
         agent=agent,
         monkeypatch=monkeypatch,

agentex/tests/unit/use_cases/test_agents_api_keys_use_case.py

@@ -45,7 +45,7 @@ def agent_api_keys_use_case(
         agent_repository=agent_repository,
         client=mock_http_client,
         authorization_service=authorization_service,
-        feature_flags=FeatureFlagProvider(),
+        feature_flags=FeatureFlagProvider(egp_api_backend_url=None),
     )