fix(extraction): resolve gcloud ADC HOME for non-nested mounts (#786)

* fix(extraction): resolve gcloud ADC HOME for non-nested mounts

_home_for_adc faked HOME by walking two parents up from
gcloud_config_mount, assuming the mount is always shaped
<home>/.config/gcloud. That holds for dev compose (mounts the
developer's real $HOME/.config/gcloud) but not for stage/prod, where
the Vault-sourced kartograph-extraction-runtime secret is mounted
flatly at /var/secrets/gcloud. There, the derived HOME (/var) has no
.config/gcloud, so openshell's --from-gcloud-adc credential
resolution fails with "no credentials resolved for provider type
'google-vertex-ai'" even though the ADC file exists at the configured
mount.

Stage a scratch HOME per invocation with a symlinked
.config/gcloud/application_default_credentials.json instead, so
credential resolution works regardless of the mount's actual shape.

Co-authored-by: Cursor <cursoragent@cursor.com>

* style: satisfy ruff format for vertex provider tests

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: aredenba-rh

src/api/extraction/infrastructure/openshell/vertex_provider.py

@@ -4,6 +4,7 @@
 
 import json
 import os
+import tempfile
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Iterator, Literal
@@ -24,20 +25,32 @@ def _adc_path(*, gcloud_config_mount: str | None) -> Path:
 
 @contextmanager
 def _home_for_adc(*, gcloud_config_mount: str | None) -> Iterator[None]:
-    """Point HOME at the mounted gcloud directory so ``--from-gcloud-adc`` resolves ADC."""
+    """Point HOME at a scratch dir whose ``.config/gcloud`` resolves the mounted ADC.
+
+    ``--from-gcloud-adc`` resolves credentials via the gcloud SDK convention of
+    ``$HOME/.config/gcloud/application_default_credentials.json``. The mount
+    itself is not guaranteed to already sit two directories below a usable
+    HOME (dev compose mounts ``$HOME/.config/gcloud`` directly, but prod
+    mounts a flat Vault-sourced secret at e.g. ``/var/secrets/gcloud``), so
+    build the nested layout explicitly instead of assuming one.
+    """
     if not gcloud_config_mount:
         yield
         return
-    home = str(Path(gcloud_config_mount).expanduser().resolve().parent.parent)
-    previous = os.environ.get("HOME")
-    os.environ["HOME"] = home
-    try:
-        yield
-    finally:
-        if previous is None:
-            os.environ.pop("HOME", None)
-        else:
-            os.environ["HOME"] = previous
+    adc_source = _adc_path(gcloud_config_mount=gcloud_config_mount)
+    with tempfile.TemporaryDirectory(prefix="kartograph-adc-home-") as scratch_home:
+        gcloud_dir = Path(scratch_home) / ".config" / "gcloud"
+        gcloud_dir.mkdir(parents=True, exist_ok=True)
+        (gcloud_dir / "application_default_credentials.json").symlink_to(adc_source)
+        previous = os.environ.get("HOME")
+        os.environ["HOME"] = scratch_home
+        try:
+            yield
+        finally:
+            if previous is None:
+                os.environ.pop("HOME", None)
+            else:
+                os.environ["HOME"] = previous
 
 
 def provider_exists(*, provider_name: str) -> bool:

src/api/tests/unit/extraction/infrastructure/test_openshell_vertex_provider.py

@@ -2,12 +2,17 @@
 
 from __future__ import annotations
 
+import os
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 
 from extraction.infrastructure.openshell.cli import OpenShellCliError
-from extraction.infrastructure.openshell.vertex_provider import ensure_vertex_provider
+from extraction.infrastructure.openshell.vertex_provider import (
+    _home_for_adc,
+    ensure_vertex_provider,
+)
 
 
 def test_ensure_vertex_provider_skips_when_provider_exists() -> None:
@@ -77,6 +82,64 @@ def test_ensure_vertex_provider_creates_google_vertex_ai_from_adc(tmp_path) -> N
     )
 
 
+def _adc_lookup_path_from_home() -> Path:
+    """Where gcloud/OpenShell's ``--from-gcloud-adc`` looks, given the active HOME."""
+    return (
+        Path(os.environ["HOME"])
+        / ".config"
+        / "gcloud"
+        / "application_default_credentials.json"
+    )
+
+
+def test_home_for_adc_resolves_nested_dev_style_mount(tmp_path) -> None:
+    """Dev compose mounts the host's real ``$HOME/.config/gcloud`` directly."""
+    fake_home = tmp_path / "home" / "dev"
+    gcloud_dir = fake_home / ".config" / "gcloud"
+    gcloud_dir.mkdir(parents=True)
+    adc_file = gcloud_dir / "application_default_credentials.json"
+    adc_file.write_text('{"type":"authorized_user"}', encoding="utf-8")
+
+    with _home_for_adc(gcloud_config_mount=str(gcloud_dir)):
+        assert _adc_lookup_path_from_home().read_text(encoding="utf-8") == (
+            adc_file.read_text(encoding="utf-8")
+        )
+
+
+def test_home_for_adc_resolves_flat_prod_style_mount(tmp_path) -> None:
+    """Prod mounts a Vault-sourced secret flatly (not nested under a home dir).
+
+    Regression test: OpenShift mounts the ``kartograph-extraction-runtime``
+    ExternalSecret at ``/var/secrets/gcloud`` (see hp-fleet-gitops
+    apps/kartograph/base/api-deployment.yaml), so the mount is *not* two
+    directories below a usable HOME. ``_home_for_adc`` must still produce a
+    HOME whose ``.config/gcloud/application_default_credentials.json``
+    resolves to the actual mounted ADC file.
+    """
+    flat_mount = tmp_path / "var" / "secrets" / "gcloud"
+    flat_mount.mkdir(parents=True)
+    adc_file = flat_mount / "application_default_credentials.json"
+    adc_file.write_text('{"type":"service_account"}', encoding="utf-8")
+
+    with _home_for_adc(gcloud_config_mount=str(flat_mount)):
+        assert _adc_lookup_path_from_home().read_text(encoding="utf-8") == (
+            adc_file.read_text(encoding="utf-8")
+        )
+
+
+def test_home_for_adc_restores_previous_home(tmp_path) -> None:
+    flat_mount = tmp_path / "secrets" / "gcloud"
+    flat_mount.mkdir(parents=True)
+    (flat_mount / "application_default_credentials.json").write_text(
+        "{}", encoding="utf-8"
+    )
+
+    previous = os.environ.get("HOME")
+    with _home_for_adc(gcloud_config_mount=str(flat_mount)):
+        assert os.environ.get("HOME") != previous
+    assert os.environ.get("HOME") == previous
+
+
 def test_ensure_vertex_provider_raises_when_adc_missing(tmp_path) -> None:
     with patch(
         "extraction.infrastructure.openshell.vertex_provider.provider_exists",