fix(acp): let a generic-ACP agent declare the env vars it authenticates with (#4392)

* fix(acp): let a generic-ACP agent declare the env vars it authenticates with

A generic-ACP agent configured the documented way (an `acp.agents:` row, or
`omnigent setup` -> Custom ACP agent) was spawned with no provider credentials
and no way to be given any, so it started unauthenticated, stalled during the
handshake, and every turn failed.

The spawn env is deny-by-default with an empty prefix family: the executor
drives an arbitrary agent, so it cannot know which vendor family that agent
authenticates with, and guessing would re-widen the leak that filtering closed.
That part is right. The gap was the escape hatch: `env_passthrough` only existed
on a full agent spec's `os_env.sandbox`, which a user configuring an agent
through `acp.agents:` never authors. Measured against a realistic environment,
only HOME/PATH/TERM survived.

Keep deny-by-default and make the hatch reachable per agent:

    acp:
      agents:
        - name: Grok Build
          command: grok agent stdio
          env_passthrough: [XAI_API_KEY]

Names only, never values: the variable is read from the host environment at
spawn, so no secret lands in config.yaml. A `NAME=value` entry is rejected
rather than accepted-and-ignored, since that mistake would write a plaintext
credential and still not reach the agent. Threaded through the existing
plumbing (AcpAgentEntry -> HARNESS_ACP_ENV_PASSTHROUGH -> AcpAgentConfig ->
_build_spawn_env), unioned with any spec-declared names, and also honored for a
spec-embedded one-shot agent.

Also stop the handshake timeout reporting itself as a blank failure.
`asyncio.TimeoutError` carries no message, so a caller reporting it by
`str(exc)` produced `inner executor error: ` with nothing to act on. `_rpc` now
raises a TimeoutError naming the agent, the stalled method and the deadline, at
the one place every handshake RPC routes through.

Before: `inner executor error: `
After:  `inner executor error: ACP agent 'Grok Build' did not answer
         session/new within 30s (command: 'grok agent stdio')`
Signed-off-by: Dhruv Gupta <dhruv.gupta@databricks.com>

* fix(acp): keep the spawn-env canary working with the agent-declared allowlist

The canary drives the real `_build_spawn_env` on an executor built via
`object.__new__` carrying only the attributes the builder reads, so reading
`self._config` unconditionally raised AttributeError there. Read the agent
config defensively, matching the duck-typed style `declared_passthrough`
already uses for the spec chain.

Also extend the canary to the new field: a declared name is an allowlist, not a
bypass, so the declared variable arrives and every planted canary secret still
stays out.

Signed-off-by: Dhruv Gupta <dhruv.gupta@databricks.com>

---------

Signed-off-by: Dhruv Gupta <dhruv.gupta@databricks.com>

omnigent/inner/acp_executor.py

@@ -161,6 +161,12 @@ class AcpAgentConfig:
     :param omnigent_mcp: Expose Omnigent's builtin tools to the agent via
         ``session/new.mcpServers`` (the shared ``serve-mcp`` relay). On by
         default; the global ``OMNIGENT_ACP_MCP=0`` kill switch also disables it.
+    :param env_passthrough: Environment variable *names* this agent may read at
+        spawn, e.g. ``("XAI_API_KEY",)``. The spawn env is deny-by-default and
+        this executor drives an arbitrary agent, so it cannot infer the family
+        the agent authenticates with — an agent that reads a variable must name
+        it here (or in ``os_env.sandbox.env_passthrough``) or it starts
+        unauthenticated. Names only; values come from the host environment.
     """
 
     command: str
@@ -169,6 +175,7 @@ class AcpAgentConfig:
     session_id_mode: str = "server"
     send_model_in_session_new: bool = False
     omnigent_mcp: bool = True
+    env_passthrough: tuple[str, ...] = ()
 
 
 class _AcpRequestError(Exception):
@@ -491,9 +498,14 @@ class AcpExecutor(Executor):
         await self._send({"jsonrpc": "2.0", "id": req_id, "method": method, "params": params})
         try:
             return await asyncio.wait_for(fut, timeout=timeout)
-        except asyncio.TimeoutError:
+        except asyncio.TimeoutError as exc:
             self._pending.pop(req_id, None)
-            raise
+            # asyncio.TimeoutError carries no message, so a caller reporting it
+            # by str() would surface a blank failure. Name the stalled call.
+            raise TimeoutError(
+                f"ACP agent {self._config.name!r} did not answer {method} "
+                f"within {timeout:g}s (command: {self._config.command!r})"
+            ) from exc
 
     # ------------------------------------------------------------------
     # ACP handshake
@@ -502,17 +514,24 @@ class AcpExecutor(Executor):
     def _build_spawn_env(self) -> dict[str, str]:
         """The env handed to the generic ACP subprocess.
 
-        Deny-by-default: base + the spec's ``env_passthrough``. No prefix family
-        is added because the executor cannot know which vendor an arbitrary ACP
-        agent belongs to. Previously ``os.environ.copy()`` handed the CLI every
-        host secret (#3445).
+        Deny-by-default: base + the names declared by the agent's own config and
+        by the spec's ``os_env.sandbox.env_passthrough``. No prefix family is
+        added because the executor cannot know which vendor an arbitrary ACP
+        agent belongs to; an agent that authenticates from a variable names it
+        instead, which keeps every *other* provider's secret out.
 
         Kept as a named builder so the spawn-env canary can drive the real thing
-        rather than a hand-copied prefix list.
+        rather than a hand-copied prefix list. The canary constructs a bare
+        executor carrying only what the builder reads, so the agent config is
+        read defensively rather than assumed present.
         """
+        config = getattr(self, "_config", None)
         return clean_agent_env(
             allow_prefixes=(),
-            extra_allowed=declared_passthrough(self._os_env),
+            extra_allowed=(
+                *getattr(config, "env_passthrough", ()),
+                *declared_passthrough(self._os_env),
+            ),
         )
 
     def _warn_initialize_failed(self, reason: str) -> None:

omnigent/inner/acp_harness.py

@@ -28,6 +28,10 @@ Env vars read at startup:
   ``session/new`` still receives an empty ``mcpServers`` array.
 - ``HARNESS_ACP_OS_ENV``: JSON-encoded :class:`OSEnvSpec`. When unset, falls
   back to ``caller_process`` + ``sandbox=none``.
+- ``HARNESS_ACP_ENV_PASSTHROUGH``: comma-separated environment variable *names*
+  the agent may read at spawn (the spawn env is otherwise deny-by-default, so an
+  agent authenticating from a variable needs it named here). Names only — each
+  value is read from this process's own environment.
 - ``HARNESS_ACP_PROMPT_TIMEOUT_S``: optional idle (time-without-progress) deadline in
   seconds for a prompt turn (default 300); must be positive and finite or the child aborts.
 """
@@ -55,6 +59,7 @@ _ENV_SEND_MODEL = "HARNESS_ACP_SEND_MODEL"
 _ENV_OMNIGENT_MCP = "HARNESS_ACP_OMNIGENT_MCP"
 _ENV_CWD = "HARNESS_ACP_CWD"
 _ENV_OS_ENV = "HARNESS_ACP_OS_ENV"
+_ENV_ENV_PASSTHROUGH = "HARNESS_ACP_ENV_PASSTHROUGH"
 
 
 def _env_enabled(name: str, *, default: bool) -> bool:
@@ -64,6 +69,16 @@ def _env_enabled(name: str, *, default: bool) -> bool:
     return raw.strip().lower() in ("1", "true", "yes", "on")
 
 
+def _env_passthrough_names() -> tuple[str, ...]:
+    """Variable names the configured agent may read, from the spawn env.
+
+    Comma-separated names (never values — the parent forwards only names, and
+    the value is read from this process's own environment at spawn).
+    """
+    raw = os.environ.get(_ENV_ENV_PASSTHROUGH, "")
+    return tuple(part.strip() for part in raw.split(",") if part.strip())
+
+
 def _resolve_os_env() -> OSEnvSpec:
     """Resolve the inner-executor :class:`OSEnvSpec` from env config.
 
@@ -120,6 +135,7 @@ def _build_acp_executor() -> Executor:
         session_id_mode=session_id_mode,
         send_model_in_session_new=send_model,
         omnigent_mcp=omnigent_mcp,
+        env_passthrough=_env_passthrough_names(),
     )
     return AcpExecutor(config=config, cwd=cwd, os_env=_resolve_os_env())
 

omnigent/onboarding/acp_auth.py

@@ -10,13 +10,16 @@ commands in a dedicated top-level ``acp:`` block of ``~/.omnigent/config.yaml``:
         - {name: Gemini CLI,  command: gemini --experimental-acp}
         - {name: Claude Code, command: npx -y @zed-industries/claude-code-acp}
         - {name: Goose,       command: goose acp, model: gpt-5.3}
+        - {name: Grok Build,  command: grok agent stdio, env_passthrough: [XAI_API_KEY]}
 
 Each agent gets a stable ``slug`` derived from its name; a picked
 ``acp:<slug>`` (carried in the spec, resolved at spawn) looks the command back up
 here. Auth is each agent's own — Omnigent stores no credential, so unlike the
-``providers:`` / ``cursor:`` blocks there is no secret reference. A dedicated
-block (not the shared gateway ``auth:``) keeps these commands from being
-mis-consumed by the SDK harnesses.
+``providers:`` / ``cursor:`` blocks there is no secret reference. An agent that
+authenticates from an environment variable names it in ``env_passthrough``
+(names only, never values): the spawn env is deny-by-default, so an undeclared
+variable does not reach the agent. A dedicated block (not the shared gateway
+``auth:``) keeps these commands from being mis-consumed by the SDK harnesses.
 
 This module is pure read + settings-builder (mirroring
 :mod:`omnigent.onboarding.cursor_auth`): the CLI orchestrates writes through
@@ -49,6 +52,12 @@ class AcpAgentEntry:
     :param session_id_mode: ``"server"`` (default) or ``"client"``.
     :param send_model: Send the model in ``session/new`` (Qwen-shaped agents).
     :param omnigent_mcp: Lend Omnigent's builtin MCP relay in ``session/new``.
+    :param env_passthrough: Environment variable *names* the agent may read at
+        spawn, e.g. ``("XAI_API_KEY",)``. The spawn env is deny-by-default and
+        the executor cannot know which variable an arbitrary agent
+        authenticates with, so an agent that reads one must name it here or it
+        starts unauthenticated. Names only — values are read from the host
+        environment at spawn, never stored in the config file.
     """
 
     slug: str
@@ -58,6 +67,7 @@ class AcpAgentEntry:
     session_id_mode: str = "server"
     send_model: bool = False
     omnigent_mcp: bool = True
+    env_passthrough: tuple[str, ...] = ()
 
 
 def slugify(name: str) -> str:
@@ -74,6 +84,38 @@ def slugify(name: str) -> str:
     return slug or "agent"
 
 
+def parse_env_passthrough(raw: object) -> tuple[str, ...]:
+    """Parse an agent's ``env_passthrough`` into a tuple of variable names.
+
+    Accepts a list of names, or a single name as a bare string. ``NAME=value``
+    is rejected rather than accepted-and-ignored: writing a secret here would
+    put it in plaintext in ``config.yaml`` and it would silently not reach the
+    agent, so the mistake has to be loud.
+
+    :param raw: The value read from the config row (any type).
+    :returns: The declared names, de-duplicated in first-seen order.
+    :raises ValueError: When the value isn't names, or a name carries a value.
+    """
+    if raw is None:
+        return ()
+    items = [raw] if isinstance(raw, str) else raw
+    if not isinstance(items, list | tuple):
+        raise ValueError("acp agent env_passthrough must be a list of variable names")
+    names: list[str] = []
+    for item in items:
+        if not isinstance(item, str) or not item.strip():
+            raise ValueError("acp agent env_passthrough entries must be non-empty strings")
+        name = item.strip()
+        if "=" in name:
+            raise ValueError(
+                f"acp agent env_passthrough must list variable NAMES, not values: {name!r}. "
+                "Export the variable in the environment and name it here."
+            )
+        if name not in names:
+            names.append(name)
+    return tuple(names)
+
+
 def acp_agents(config: dict[str, object] | None = None) -> list[AcpAgentEntry]:
     """Return the configured ACP agents, each with a unique derived slug.
 
@@ -124,6 +166,7 @@ def acp_agents(config: dict[str, object] | None = None) -> list[AcpAgentEntry]:
                 session_id_mode=mode if mode in ("server", "client") else "server",
                 send_model=bool(raw.get("send_model", False)),
                 omnigent_mcp=omnigent_mcp,
+                env_passthrough=parse_env_passthrough(raw.get("env_passthrough")),
             )
         )
     return entries
@@ -163,6 +206,8 @@ def acp_agents_settings(entries: list[AcpAgentEntry]) -> dict[str, object]:
             item["send_model"] = True
         if not e.omnigent_mcp:
             item["omnigent_mcp"] = False
+        if e.env_passthrough:
+            item["env_passthrough"] = list(e.env_passthrough)
         agents.append(item)
     return {ACP_CONFIG_KEY: {_AGENTS_FIELD: agents}}
 

omnigent/runtime/workflow.py

@@ -1595,6 +1595,7 @@ def _build_acp_spawn_env(
     from omnigent.onboarding.acp_auth import (
         AcpAgentEntry,
         acp_agents,
+        parse_env_passthrough,
         resolve_acp_agent,
     )
 
@@ -1618,6 +1619,7 @@ def _build_acp_spawn_env(
             name=name.strip(),
             command=command.strip(),
             omnigent_mcp=omnigent_mcp,
+            env_passthrough=parse_env_passthrough(embedded.get("env_passthrough")),
         )
     else:
         agent = resolve_acp_agent(slug) if slug else None
@@ -1632,6 +1634,9 @@ def _build_acp_spawn_env(
         if agent.send_model:
             env["HARNESS_ACP_SEND_MODEL"] = "1"
         env["HARNESS_ACP_OMNIGENT_MCP"] = "1" if agent.omnigent_mcp else "0"
+        if agent.env_passthrough:
+            # Names only; the harness reads each value from its own environment.
+            env["HARNESS_ACP_ENV_PASSTHROUGH"] = ",".join(agent.env_passthrough)
 
         model = _resolve_spec_model(spec)
         if model is not None and not model.startswith(("databricks-", "databricks/")):

tests/inner/test_acp_executor.py

@@ -15,16 +15,20 @@ Two layers:
 from __future__ import annotations
 
 import asyncio
+import os
 import shlex
 import sys
 from pathlib import Path
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch
 
 import pytest
 
+from omnigent.inner import acp_executor as acp_executor_module
 from omnigent.inner._acp_omnigent_mcp import OmnigentAcpMcp, _to_acp_mcp_servers
 from omnigent.inner.acp_executor import AcpAgentConfig, AcpExecutor
+from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
 from omnigent.inner.executor import (
+    ExecutorError,
     ReasoningChunk,
     TextChunk,
     ToolCallComplete,
@@ -333,6 +337,20 @@ def test_harness_wrap_builds_executor(monkeypatch: pytest.MonkeyPatch) -> None:
     assert ex._config.model == "gpt-5.3"
 
 
+def test_harness_wrap_reads_env_passthrough_names(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The wrap decodes the forwarded names, closing parent → child → spawn env."""
+    from omnigent.inner import acp_harness
+
+    monkeypatch.setenv("HARNESS_ACP_COMMAND", "grok agent stdio")
+    monkeypatch.setenv("HARNESS_ACP_ENV_PASSTHROUGH", "XAI_API_KEY, GROK_TOKEN ,")
+    monkeypatch.setenv("XAI_API_KEY", "xai-secret")
+    ex = acp_harness._build_acp_executor()
+    assert isinstance(ex, AcpExecutor)
+    assert ex._config.env_passthrough == ("XAI_API_KEY", "GROK_TOKEN")
+    # And it actually lands in the env the agent is spawned with.
+    assert ex._build_spawn_env().get("XAI_API_KEY") == "xai-secret"
+
+
 # ---------------------------------------------------------------------------
 # Hermetic end-to-end: drive a real fake ACP agent over stdio
 # ---------------------------------------------------------------------------
@@ -569,3 +587,104 @@ async def test_end_to_end_denied_permission(tmp_path: Path) -> None:
 
     # Turn still completes even though the tool was rejected.
     assert any(isinstance(e, TurnComplete) for e in events)
+
+
+# ---------------------------------------------------------------------------
+# Spawn env: the agent must actually receive credentials (#4281)
+# ---------------------------------------------------------------------------
+
+
+def test_spawn_env_keeps_credential_declared_on_the_agent() -> None:
+    """A name in the agent's own ``env_passthrough`` reaches the subprocess.
+
+    This is the config path a user actually has (an ``acp.agents:`` row).
+    Without it the agent starts unauthenticated and stalls during the
+    handshake, which surfaced as a turn that failed with no message at all.
+    """
+    ex = AcpExecutor(
+        AcpAgentConfig(command="agent stdio", name="Grok", env_passthrough=("XAI_API_KEY",))
+    )
+    with patch.dict(os.environ, {"XAI_API_KEY": "xai-secret"}, clear=False):
+        env = ex._build_spawn_env()
+    assert env.get("XAI_API_KEY") == "xai-secret"
+
+
+def test_spawn_env_keeps_credential_declared_on_the_spec() -> None:
+    """A spec-declared ``os_env.sandbox.env_passthrough`` name still works."""
+    os_env = OSEnvSpec(
+        type="caller_process",
+        cwd=None,
+        sandbox=OSEnvSandboxSpec(type="none", env_passthrough=["XAI_API_KEY"]),
+        fork=False,
+    )
+    ex = AcpExecutor(AcpAgentConfig(command="agent stdio", name="Grok"), os_env=os_env)
+    with patch.dict(os.environ, {"XAI_API_KEY": "xai-secret"}, clear=False):
+        env = ex._build_spawn_env()
+    assert env.get("XAI_API_KEY") == "xai-secret"
+
+
+def test_spawn_env_unions_agent_and_spec_declarations() -> None:
+    """Both sources apply; neither shadows the other."""
+    os_env = OSEnvSpec(
+        type="caller_process",
+        cwd=None,
+        sandbox=OSEnvSandboxSpec(type="none", env_passthrough=["FROM_SPEC"]),
+        fork=False,
+    )
+    ex = AcpExecutor(
+        AcpAgentConfig(command="agent stdio", name="A", env_passthrough=("FROM_AGENT",)),
+        os_env=os_env,
+    )
+    with patch.dict(os.environ, {"FROM_SPEC": "1", "FROM_AGENT": "2"}, clear=False):
+        env = ex._build_spawn_env()
+    assert env.get("FROM_SPEC") == "1"
+    assert env.get("FROM_AGENT") == "2"
+
+
+def test_spawn_env_still_excludes_undeclared_secret() -> None:
+    """Deny-by-default holds: an undeclared provider key is not handed over."""
+    ex = AcpExecutor(AcpAgentConfig(command="agent stdio", name="Grok"))
+    with patch.dict(os.environ, {"UNRELATED_API_KEY": "nope"}, clear=False):
+        env = ex._build_spawn_env()
+    assert "UNRELATED_API_KEY" not in env
+
+
+@pytest.mark.asyncio
+async def test_handshake_timeout_reports_a_non_blank_error(tmp_path: Path) -> None:
+    """An agent that never answers ``session/new`` yields a named error.
+
+    ``asyncio.TimeoutError`` has an empty ``str()``, so reporting the failure by
+    ``str(exc)`` produced the blank "inner executor error: " an operator can't
+    act on.
+    """
+    agent_path = tmp_path / "silent_agent.py"
+    agent_path.write_text(
+        "import sys, json\n"
+        "for line in sys.stdin:\n"
+        "    line = line.strip()\n"
+        "    if not line:\n"
+        "        continue\n"
+        "    msg = json.loads(line)\n"
+        "    if msg.get('method') == 'initialize':\n"
+        "        sys.stdout.write(json.dumps({'jsonrpc': '2.0', 'id': msg['id'],\n"
+        "            'result': {'protocolVersion': 1, 'agentCapabilities': {}}}) + '\\n')\n"
+        "        sys.stdout.flush()\n"
+        # session/new deliberately unanswered -> the handshake RPC times out.
+    )
+    command = shlex.join([sys.executable, str(agent_path)])
+
+    ex = AcpExecutor(AcpAgentConfig(command=command, name="Silent"))
+    errors = []
+    with patch.object(acp_executor_module, "_INIT_TIMEOUT_SECONDS", 1.0):
+        try:
+            async for ev in ex.run_turn([{"role": "user", "content": "hi"}], [], ""):
+                if isinstance(ev, ExecutorError):
+                    errors.append(ev)
+        finally:
+            await ex.close()
+
+    assert errors, "a stalled handshake must surface an ExecutorError"
+    assert errors[0].message.strip(), "the turn error must never be blank"
+    # Names the stalled call, not just the exception type.
+    assert "session/new" in errors[0].message
+    assert "Silent" in errors[0].message

tests/onboarding/test_acp_auth.py

@@ -62,3 +62,64 @@ def test_omnigent_mcp_requires_boolean(value: object) -> None:
                 }
             }
         )
+
+
+def test_env_passthrough_parses_and_round_trips() -> None:
+    """Declared names survive parse → persist, so the config is stable."""
+    entries = acp_agents(
+        {
+            "acp": {
+                "agents": [
+                    {
+                        "name": "Grok Build",
+                        "command": "grok agent stdio",
+                        "env_passthrough": ["XAI_API_KEY", "XAI_API_KEY", " GROK_TOKEN "],
+                    },
+                    {"name": "Goose", "command": "goose acp"},
+                ]
+            }
+        }
+    )
+
+    # De-duplicated, trimmed, order preserved; absent → empty.
+    assert entries[0].env_passthrough == ("XAI_API_KEY", "GROK_TOKEN")
+    assert entries[1].env_passthrough == ()
+    assert acp_agents_settings(entries) == {
+        "acp": {
+            "agents": [
+                {
+                    "name": "Grok Build",
+                    "command": "grok agent stdio",
+                    "env_passthrough": ["XAI_API_KEY", "GROK_TOKEN"],
+                },
+                {"name": "Goose", "command": "goose acp"},
+            ]
+        }
+    }
+
+
+def test_env_passthrough_accepts_a_bare_string() -> None:
+    entries = acp_agents(
+        {"acp": {"agents": [{"name": "A", "command": "a", "env_passthrough": "XAI_API_KEY"}]}}
+    )
+    assert entries[0].env_passthrough == ("XAI_API_KEY",)
+
+
+def test_env_passthrough_rejects_a_value_instead_of_a_name() -> None:
+    """``NAME=secret`` would put a credential in plaintext and not reach the agent."""
+    with pytest.raises(ValueError, match="NAMES, not values"):
+        acp_agents(
+            {
+                "acp": {
+                    "agents": [
+                        {"name": "A", "command": "a", "env_passthrough": ["XAI_API_KEY=sk-secret"]}
+                    ]
+                }
+            }
+        )
+
+
+@pytest.mark.parametrize("value", [5, {"a": 1}, [""], [None]])
+def test_env_passthrough_rejects_non_names(value: object) -> None:
+    with pytest.raises(ValueError, match="env_passthrough"):
+        acp_agents({"acp": {"agents": [{"name": "A", "command": "a", "env_passthrough": value}]}})

tests/runtime/test_acp_spawn_env.py

@@ -161,3 +161,41 @@ def test_embedded_omnigent_mcp_flag_forwarded() -> None:
 def test_malformed_embedded_agent_fails_loudly(acp_agent: object) -> None:
     with pytest.raises(ValueError, match="executor acp_agent"):
         _build_acp_spawn_env(_make_spec(harness="acp:helper", acp_agent=acp_agent))
+
+
+def test_env_passthrough_names_are_forwarded(_isolate_config: Path) -> None:
+    """Declared names reach the wrap so the agent can authenticate."""
+    _write_acp_config(
+        _isolate_config,
+        agents=[
+            {
+                "name": "Grok Build",
+                "command": "grok agent stdio",
+                "env_passthrough": ["XAI_API_KEY", "GROK_TOKEN"],
+            }
+        ],
+    )
+    env = _build_acp_spawn_env(_make_spec(harness="acp:grok-build"))
+    assert env["HARNESS_ACP_ENV_PASSTHROUGH"] == "XAI_API_KEY,GROK_TOKEN"
+
+
+def test_env_passthrough_absent_when_undeclared(_isolate_config: Path) -> None:
+    """No declaration writes no var, so the spawn env stays deny-by-default."""
+    _write_acp_config(_isolate_config)
+    env = _build_acp_spawn_env(_make_spec(harness="acp:goose"))
+    assert "HARNESS_ACP_ENV_PASSTHROUGH" not in env
+
+
+def test_embedded_agent_forwards_env_passthrough(_isolate_config: Path) -> None:
+    """A spec-embedded one-shot agent declares names the same way."""
+    _write_acp_config(_isolate_config, agents=[])
+    spec = _make_spec(
+        harness="acp:embedded",
+        acp_agent={
+            "name": "Embedded",
+            "command": "agent stdio",
+            "env_passthrough": ["XAI_API_KEY"],
+        },
+    )
+    env = _build_acp_spawn_env(spec)
+    assert env["HARNESS_ACP_ENV_PASSTHROUGH"] == "XAI_API_KEY"

tests/test_agent_spawn_env_canary.py

@@ -260,3 +260,23 @@ def test_declared_passthrough_tolerates_a_missing_chain():
 
     assert declared_passthrough(None) == ()
     assert declared_passthrough(_NoSandbox()) == ()
+
+
+def test_acp_agent_declaration_passes_only_what_it_names(hostile_env, monkeypatch):
+    """A generic-ACP agent's own ``env_passthrough`` is an allowlist, not a bypass.
+
+    The executor cannot infer which family an arbitrary agent authenticates
+    with, so the agent names its variables. Everything it does not name stays
+    withheld — a declaration must not reopen the whole environment.
+    """
+    from omnigent.inner.acp_executor import AcpAgentConfig, AcpExecutor
+
+    monkeypatch.setattr("os.environ", {**hostile_env, "XAI_API_KEY": "declared-and-wanted"})
+    ex = AcpExecutor(
+        AcpAgentConfig(command="agent stdio", name="Grok", env_passthrough=("XAI_API_KEY",))
+    )
+    env = ex._build_spawn_env()
+
+    assert env.get("XAI_API_KEY") == "declared-and-wanted"
+    for name in CANARY_SECRETS:
+        assert name not in env, name