diff --git a/AGENTS.md b/AGENTS.md
new file mode 100644
index 0000000..c3dffca
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1,10 @@
+# Zenith repository guidance
+
+- Codex ACP is subscription-only by default. Keep its managed `CODEX_HOME`
+  ChatGPT-only with login shells and shell snapshots disabled; never restore ambient
+  credential inheritance or the legacy danger-full-access command overrides.
+- API-billed ACP tasks require both an explicit task `billing.api_grant` request and
+  an exact, unexpired, non-revoked operator grant from `ZENITH_API_GRANTS_FILE`.
+  Never store credential material in task JSON, receipts, prompts, or grant records.
+- Keep ACP and Zenith MCP child environments allowlisted. Only the exact Codex ACP
+  child for an authorized API task may receive `OPENAI_API_KEY`.
diff --git a/README.md b/README.md
index 8a3009f..0bdea23 100644
--- a/README.md
+++ b/README.md
@@ -76,6 +76,62 @@ npm install -g @agentclientprotocol/codex-acp
 command -v codex-acp
 ```
 
+**Codex authentication and API billing**
+
+Codex ACP tasks default to ChatGPT subscription authentication. Zenith creates a
+dedicated managed home at `$ZENITH_HOME/codex-subscription` (override with
+`ZENITH_CODEX_SUBSCRIPTION_HOME`) with ChatGPT-only login, login shells disabled,
+shell snapshots disabled, and no inherited shell environment. Authenticate that
+home once before the first Codex ACP mission:
+
+```bash
+CODEX_HOME="${ZENITH_CODEX_SUBSCRIPTION_HOME:-$HOME/.zenith/codex-subscription}" codex login
+```
+
+Ambient API keys are not forwarded to ACP agents or their Zenith MCP helpers. An
+API-billed task must contain an explicit non-secret request:
+
+```yaml
+billing:
+  mode: api
+  api_grant:
+    grant_id: issue-123-image
+    api_project: isolated-image-project
+    max_usd: "5.00"
+    expires_at: "2026-09-04T18:00:00Z"
+```
+
+That request does not authorize itself. The operator must also set
+`ZENITH_API_GRANTS_FILE` to a private (`0600`), operator-owned JSON registry whose
+record exactly matches the Zenith project, mission, task, provider, API project,
+budget, and expiry. The credential lives in a separate private one-line file:
+
+```json
+{
+  "version": 1,
+  "grants": [{
+    "grant_id": "issue-123-image",
+    "zenith_project_id": "20260904T120000Z-example",
+    "mission_id": "mission-001",
+    "task_id": "w-image",
+    "provider": "codex",
+    "api_project": "isolated-image-project",
+    "max_usd": "5.00",
+    "issued_at": "2026-09-04T12:00:00Z",
+    "expires_at": "2026-09-04T18:00:00Z",
+    "approved_by": "operator@example",
+    "credential_file": "/secure/zenith/issue-123-image.key",
+    "revoked": false
+  }]
+}
+```
+
+Only the matching ACP child receives `OPENAI_API_KEY`; the MCP helper does not.
+Authorization receipts without credentials are written under the mission's
+`.zenith-runtime/billing-receipts/`. `max_usd` is an authorization ceiling checked
+by Zenith, not a live usage meter, so the referenced OpenAI project/service account
+must have the corresponding provider-side budget and restrictions.
+
 **Initialize a workspace**
 
 Initialize the project workspace Zenith should operate on. This is your target app/repo, not the Zenith source checkout:
diff --git a/zenith/src/zenith_harness/acp_auth.py b/zenith/src/zenith_harness/acp_auth.py
new file mode 100644
index 0000000..0865c2c
--- /dev/null
+++ b/zenith/src/zenith_harness/acp_auth.py
@@ -0,0 +1,375 @@
+"""Fail-closed authentication contexts for ACP child processes.
+
+Task JSON can request API billing, but only a private operator registry can
+authorize it.  No credential is ever stored in tasks.json or an audit receipt.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import stat
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import TYPE_CHECKING, Literal
+
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, SecretStr, model_validator
+
+from .models import API_GRANT_ID_REGEX, Task
+
+if TYPE_CHECKING:
+    from .config import HarnessConfig
+
+
+SUBSCRIPTION_CODEX_CONFIG = """forced_login_method = "chatgpt"
+allow_login_shell = false
+
+[features]
+shell_snapshot = false
+
+[shell_environment_policy]
+inherit = "none"
+ignore_default_excludes = false
+"""
+
+API_CODEX_CONFIG = """allow_login_shell = false
+
+[features]
+shell_snapshot = false
+
+[shell_environment_policy]
+inherit = "none"
+ignore_default_excludes = false
+"""
+
+_SAFE_ENV_NAMES = frozenset(
+    {
+        "COLORTERM",
+        "FORCE_COLOR",
+        "HOME",
+        "LANG",
+        "LANGUAGE",
+        "LOGNAME",
+        "NO_COLOR",
+        "PATH",
+        "SHELL",
+        "SSL_CERT_DIR",
+        "SSL_CERT_FILE",
+        "TERM",
+        "TMP",
+        "TMPDIR",
+        "TEMP",
+        "TZ",
+        "USER",
+        # Adapter binary location, not a credential.
+        "CODEX_PATH",
+        # Test/smoke metadata. These are paths and opaque ids, not secrets.
+        "ZENITH_HANDOFF_PATH",
+        "ZENITH_NODE_ID",
+        "ZENITH_NODE_TYPE",
+    }
+)
+
+
+class ACPAuthError(RuntimeError):
+    """Authentication policy or operator grant failed closed."""
+
+
+class OperatorApiGrant(BaseModel):
+    """One operator-owned grant record. The credential remains in a private file."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    grant_id: str = Field(pattern=API_GRANT_ID_REGEX.pattern)
+    zenith_project_id: str = Field(min_length=1)
+    mission_id: str = Field(min_length=1)
+    task_id: str = Field(min_length=1)
+    provider: Literal["codex"] = "codex"
+    api_project: str = Field(min_length=1)
+    max_usd: Decimal = Field(gt=0)
+    issued_at: AwareDatetime
+    expires_at: AwareDatetime
+    approved_by: str = Field(min_length=1)
+    credential_file: Path
+    revoked: bool = False
+
+    @model_validator(mode="after")
+    def validate_window(self) -> OperatorApiGrant:
+        if self.expires_at <= self.issued_at:
+            raise ValueError("expires_at must be later than issued_at")
+        return self
+
+
+class OperatorApiGrantRegistry(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    version: Literal[1] = 1
+    grants: list[OperatorApiGrant] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def reject_duplicate_ids(self) -> OperatorApiGrantRegistry:
+        ids = [grant.grant_id for grant in self.grants]
+        if len(ids) != len(set(ids)):
+            raise ValueError("operator grant ids must be unique")
+        return self
+
+
+@dataclass(frozen=True)
+class ResolvedApiGrant:
+    grant_id: str
+    api_project: str
+    max_usd: Decimal
+    expires_at: datetime
+    approved_by: str
+    credential: SecretStr
+    registry_sha256: str
+
+
+@dataclass(frozen=True)
+class ACPAuthContext:
+    mode: Literal["subscription", "api"]
+    codex_home: Path | None = None
+    api_grant: ResolvedApiGrant | None = None
+
+
+def sanitized_process_env(source: dict[str, str] | None = None) -> dict[str, str]:
+    """Return the small non-secret environment shared with ACP/MCP children."""
+
+    ambient = source if source is not None else os.environ
+    return {
+        name: value
+        for name, value in ambient.items()
+        if name in _SAFE_ENV_NAMES or name.startswith("LC_")
+    }
+
+
+def prepare_acp_auth_context(
+    *,
+    config: HarnessConfig,
+    provider,
+    task: Task | None,
+    project_id: str,
+    mission_id: str,
+    now: datetime | None = None,
+) -> ACPAuthContext:
+    """Resolve a task request to subscription auth or an operator API grant."""
+
+    provider_name = getattr(provider, "name", None)
+    billing = task.billing if task is not None else None
+    wants_api = billing is not None and billing.mode == "api"
+
+    if provider_name != "codex":
+        if wants_api:
+            raise ACPAuthError("OpenAI API grants are only valid for the codex provider")
+        return ACPAuthContext(mode="subscription")
+
+    if not wants_api:
+        home = config.resolved_codex_subscription_home
+        _ensure_managed_codex_home(home, SUBSCRIPTION_CODEX_CONFIG, subscription=True)
+        return ACPAuthContext(mode="subscription", codex_home=home)
+
+    assert task is not None and billing is not None and billing.api_grant is not None
+    if config.api_grants_file is None:
+        raise ACPAuthError(
+            "task requests API billing but ZENITH_API_GRANTS_FILE is not configured"
+        )
+    grant = _resolve_operator_grant(
+        grants_file=config.api_grants_file,
+        task=task,
+        project_id=project_id,
+        mission_id=mission_id,
+        now=now or datetime.now(UTC),
+    )
+    home = config.harness_home / "codex-api" / grant.grant_id
+    _ensure_managed_codex_home(home, API_CODEX_CONFIG, subscription=False)
+    return ACPAuthContext(mode="api", codex_home=home, api_grant=grant)
+
+
+def build_acp_subprocess_env(provider, auth: ACPAuthContext | None = None) -> dict[str, str]:
+    """Build a sanitized ACP environment and inject only an authorized task key."""
+
+    env = sanitized_process_env()
+    if getattr(provider, "name", None) != "codex":
+        if auth is not None and auth.mode == "api":
+            raise ACPAuthError("API auth context cannot be used with a non-codex provider")
+        return env
+    if auth is None or auth.codex_home is None:
+        raise ACPAuthError("codex ACP launch requires an explicit auth context")
+
+    home = str(auth.codex_home)
+    env["HOME"] = home
+    env["CODEX_HOME"] = home
+    if auth.mode == "api":
+        if auth.api_grant is None:
+            raise ACPAuthError("API auth context is missing its resolved operator grant")
+        env["OPENAI_API_KEY"] = auth.api_grant.credential.get_secret_value()
+    return env
+
+
+def api_grant_receipt(auth: ACPAuthContext) -> dict[str, str] | None:
+    """Return non-secret audit fields for an authorized API launch."""
+
+    grant = auth.api_grant
+    if grant is None:
+        return None
+    return {
+        "billing_mode": "api",
+        "grant_id": grant.grant_id,
+        "api_project": grant.api_project,
+        "max_usd": str(grant.max_usd),
+        "expires_at": grant.expires_at.isoformat(),
+        "approved_by": grant.approved_by,
+        "registry_sha256": grant.registry_sha256,
+    }
+
+
+def _resolve_operator_grant(
+    *,
+    grants_file: Path,
+    task: Task,
+    project_id: str,
+    mission_id: str,
+    now: datetime,
+) -> ResolvedApiGrant:
+    request = task.billing.api_grant
+    assert request is not None
+    registry_bytes = _read_private_file(grants_file, label="operator API grant registry")
+    try:
+        registry = OperatorApiGrantRegistry.model_validate_json(registry_bytes)
+    except Exception as exc:  # noqa: BLE001
+        raise ACPAuthError(f"operator API grant registry is invalid: {exc}") from exc
+
+    matching = [grant for grant in registry.grants if grant.grant_id == request.grant_id]
+    if len(matching) != 1:
+        raise ACPAuthError("requested API grant is not present in the operator registry")
+    grant = matching[0]
+    expected = {
+        "zenith_project_id": project_id,
+        "mission_id": mission_id,
+        "task_id": task.id,
+        "provider": "codex",
+        "api_project": request.api_project,
+        "max_usd": request.max_usd,
+        "expires_at": request.expires_at,
+    }
+    actual = {
+        "zenith_project_id": grant.zenith_project_id,
+        "mission_id": grant.mission_id,
+        "task_id": grant.task_id,
+        "provider": grant.provider,
+        "api_project": grant.api_project,
+        "max_usd": grant.max_usd,
+        "expires_at": grant.expires_at,
+    }
+    if actual != expected:
+        raise ACPAuthError("task API request does not exactly match its operator grant")
+    if grant.revoked:
+        raise ACPAuthError("operator API grant is revoked")
+    if now < grant.issued_at:
+        raise ACPAuthError("operator API grant is not active yet")
+    if now >= grant.expires_at:
+        raise ACPAuthError("operator API grant has expired")
+
+    credential_path = grant.credential_file
+    if not credential_path.is_absolute():
+        credential_path = grants_file.parent / credential_path
+    credential_bytes = _read_private_file(credential_path, label="API credential")
+    try:
+        credential_text = credential_bytes.decode("utf-8").rstrip("\r\n")
+    except UnicodeDecodeError as exc:
+        raise ACPAuthError("API credential is not UTF-8 text") from exc
+    if not credential_text or "\n" in credential_text or "\r" in credential_text:
+        raise ACPAuthError("API credential must contain exactly one non-empty line")
+
+    return ResolvedApiGrant(
+        grant_id=grant.grant_id,
+        api_project=grant.api_project,
+        max_usd=grant.max_usd,
+        expires_at=grant.expires_at,
+        approved_by=grant.approved_by,
+        credential=SecretStr(credential_text),
+        registry_sha256=hashlib.sha256(registry_bytes).hexdigest(),
+    )
+
+
+def _ensure_managed_codex_home(home: Path, config_text: str, *, subscription: bool) -> None:
+    home.mkdir(parents=True, exist_ok=True, mode=0o700)
+    home_stat = home.lstat()
+    if stat.S_ISLNK(home_stat.st_mode) or not stat.S_ISDIR(home_stat.st_mode):
+        raise ACPAuthError(f"managed Codex home is not a real directory: {home}")
+    if home_stat.st_uid != os.geteuid():
+        raise ACPAuthError(f"managed Codex home is not owned by the current user: {home}")
+    os.chmod(home, 0o700)
+
+    config_path = home / "config.toml"
+    if config_path.exists():
+        config_bytes = _read_private_file(config_path, label="managed Codex config")
+        try:
+            existing_config = config_bytes.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise ACPAuthError("managed Codex config is not UTF-8 text") from exc
+        if existing_config != config_text:
+            raise ACPAuthError(f"managed Codex config differs from the required profile: {config_path}")
+    else:
+        tmp_path = home / f".config.toml.{os.getpid()}.tmp"
+        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
+        try:
+            descriptor = os.open(tmp_path, flags, 0o600)
+        except OSError as exc:
+            raise ACPAuthError(f"cannot create managed Codex config: {config_path}") from exc
+        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+            handle.write(config_text)
+        os.replace(tmp_path, config_path)
+    os.chmod(config_path, 0o600)
+
+    snapshots = home / "shell_snapshots"
+    if snapshots.exists():
+        snapshot_stat = snapshots.lstat()
+        if stat.S_ISLNK(snapshot_stat.st_mode) or not stat.S_ISDIR(snapshot_stat.st_mode):
+            raise ACPAuthError(f"managed Codex shell_snapshots path is unsafe: {snapshots}")
+        if any(snapshots.iterdir()):
+            raise ACPAuthError(f"managed Codex home contains forbidden shell snapshots: {snapshots}")
+
+    auth_path = home / "auth.json"
+    if not auth_path.exists():
+        return
+    auth_bytes = _read_private_file(auth_path, label="Codex auth cache")
+    try:
+        auth_payload = json.loads(auth_bytes)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ACPAuthError("Codex auth cache is invalid") from exc
+    if not isinstance(auth_payload, dict):
+        raise ACPAuthError("Codex auth cache must contain a JSON object")
+    if subscription:
+        if auth_payload.get("auth_mode") != "chatgpt":
+            raise ACPAuthError("subscription Codex home is not authenticated with ChatGPT")
+        if auth_payload.get("OPENAI_API_KEY") not in (None, ""):
+            raise ACPAuthError("subscription Codex home contains an API key")
+    else:
+        raise ACPAuthError("API Codex homes must not contain a persistent auth cache")
+
+
+def _read_private_file(path: Path, *, label: str) -> bytes:
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except FileNotFoundError as exc:
+        raise ACPAuthError(f"{label} is missing: {path}") from exc
+    except OSError as exc:
+        raise ACPAuthError(f"{label} cannot be opened safely: {path}") from exc
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            raise ACPAuthError(f"{label} must be a regular non-symlink file: {path}")
+        if metadata.st_uid != os.geteuid():
+            raise ACPAuthError(f"{label} must be owned by the current user: {path}")
+        if stat.S_IMODE(metadata.st_mode) & 0o077:
+            raise ACPAuthError(f"{label} permissions must not allow group/other access: {path}")
+        with os.fdopen(descriptor, "rb") as handle:
+            descriptor = -1
+            return handle.read()
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
diff --git a/zenith/src/zenith_harness/acp_runner.py b/zenith/src/zenith_harness/acp_runner.py
index a7a71a1..52a0664 100644
--- a/zenith/src/zenith_harness/acp_runner.py
+++ b/zenith/src/zenith_harness/acp_runner.py
@@ -8,11 +8,20 @@
 import socket
 import subprocess
 import sys
+from contextlib import suppress
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Awaitable, Callable, ClassVar, Coroutine, Literal
 
 from .assets import AssetLoader
+from .acp_auth import (
+    ACPAuthContext,
+    api_grant_receipt,
+    build_acp_subprocess_env,
+    prepare_acp_auth_context,
+    sanitized_process_env,
+)
 from .config import HarnessConfig
 from .dispatcher import DispatchRequest, NodeHandoff
 from .models import (
@@ -92,52 +101,82 @@ class ACPError(Exception):
 def _augment_acp_command(
     command: str, provider, reasoning_effort: str | None = None
 ) -> str:
-    """Append provider-specific config flags to the ACP launch command.
-
-    For codex-acp this is the no-ask, no-sandbox combo — equivalent to
-    `codex --dangerously-bypass-approvals-and-sandbox`, which codex-acp
-    does not expose as a flag but accepts via `-c` overrides.
+    """Return the adapter command unchanged.
 
-    `reasoning_effort` is the per-role override from
-    ZENITH_<ROLE>_REASONING_EFFORT (validated against
-    `config.VALID_REASONING_EFFORTS` at discovery); None keeps the
-    historical "xhigh" default.
-
-    For hermes the command is passed through unchanged.
+    Codex security settings live in the dedicated managed CODEX_HOME.  Appending
+    CLI-style ``-c`` flags to ``codex-acp`` is both unenforced by the adapter and
+    risks selecting the legacy danger-full-access sandbox path.
     """
-    name = getattr(provider, "name", None)
-    if name == "codex":
-        effort = reasoning_effort or "xhigh"
-        return (
-            command
-            + ' -c sandbox_mode="danger-full-access"'
-            + ' -c approval_policy="never"'
-            + f' -c model_reasoning_effort="{effort}"'
-        )
-    # hermes: no-op
+    del provider, reasoning_effort
     return command
 
 
-def _acp_subprocess_env(provider) -> dict[str, str]:
-    """Build the env handed to an ACP-agent subprocess.
+def _acp_subprocess_env(
+    provider,
+    auth: ACPAuthContext | None = None,
+    reasoning_effort: str | None = None,
+) -> dict[str, str]:
+    """Build the allowlisted environment handed to an ACP subprocess."""
 
-    For codex we preserve PATH so node-based ACP adapters can launch via
-    `/usr/bin/env node`, and pass sandbox-disable hints through env. The
-    command line also receives `sandbox_mode="danger-full-access"` in
-    `_augment_acp_command`.
-
-    For hermes the env is passed through unchanged.
-    """
-    env = os.environ.copy()
-    name = getattr(provider, "name", None)
-    if name == "codex":
-        # Env-var hints — harmless if codex ignores them.
-        env["CODEX_SANDBOX"] = "danger-full-access"
-        env["CODEX_DISABLE_SANDBOX"] = "1"
-    # hermes: no special env needed
+    env = build_acp_subprocess_env(provider, auth)
+    if getattr(provider, "name", None) == "codex":
+        env["CODEX_CONFIG"] = json.dumps(
+            {"model_reasoning_effort": reasoning_effort or "xhigh"}
+        )
     return env
 
 
+def _write_api_authorization_receipt(
+    *,
+    path: Path,
+    auth: ACPAuthContext,
+    project_id: str,
+    mission_id: str,
+    task_id: str,
+    provider_name: str,
+    started_at: str,
+    status: str,
+    finished_at: str | None = None,
+    exit_code: int | None = None,
+) -> None:
+    grant_fields = api_grant_receipt(auth)
+    if grant_fields is None:
+        return
+    payload: dict[str, Any] = {
+        "version": 1,
+        "project_id": project_id,
+        "mission_id": mission_id,
+        "task_id": task_id,
+        "provider": provider_name,
+        "started_at": started_at,
+        "status": status,
+        **grant_fields,
+    }
+    if finished_at is not None:
+        payload["finished_at"] = finished_at
+    if exit_code is not None:
+        payload["exit_code"] = exit_code
+    atomic_write_json(path, payload)
+
+
+async def _enforce_api_grant_expiry(
+    process: asyncio.subprocess.Process,
+    auth: ACPAuthContext,
+    expired: asyncio.Event,
+) -> None:
+    """Terminate the only key-bearing process when its operator grant expires."""
+
+    grant = auth.api_grant
+    if grant is None:
+        return
+    delay = max(0.0, (grant.expires_at - datetime.now(UTC)).total_seconds())
+    await asyncio.sleep(delay)
+    expired.set()
+    if process.returncode is None:
+        with suppress(OSError, ProcessLookupError):
+            process.terminate()
+
+
 _NOT_FOUND = object()
 
 
@@ -581,10 +620,34 @@ async def run_node(
             role_config.worker_reasoning_effort,
         )
 
+        auth_context = prepare_acp_auth_context(
+            config=role_config,
+            provider=role_config.worker_provider,
+            task=task,
+            project_id=project_id,
+            mission_id=mission_id,
+        )
+
         workspace_dir = str(Path(cwd).expanduser().resolve() if cwd else store.workspace_dir(project_id))
         project_bucket = str(store.zenith_dir(project_id))
         handoff_path = store.attempt_path(project_id, mission_id, spawn_ts, task.id)
         handoff_path.parent.mkdir(parents=True, exist_ok=True)
+        receipt_path = (
+            store.mission_runtime_dir(project_id, mission_id)
+            / "billing-receipts"
+            / f"{spawn_ts}__{task.id}.json"
+        )
+        receipt_started_at = datetime.now(UTC).isoformat()
+        _write_api_authorization_receipt(
+            path=receipt_path,
+            auth=auth_context,
+            project_id=project_id,
+            mission_id=mission_id,
+            task_id=task.id,
+            provider_name=role_config.worker_provider.name,
+            started_at=receipt_started_at,
+            status="authorized",
+        )
 
         # 0) For claude-agent-acp: drop a project-level settings.json that
         #    overrides the user's global ~/.claude/settings.json. Adapter
@@ -609,6 +672,17 @@ async def run_node(
             if mcp_process.returncode is None:
                 mcp_process.terminate()
             await _close_subprocess(mcp_process, timeout=5)
+            _write_api_authorization_receipt(
+                path=receipt_path,
+                auth=auth_context,
+                project_id=project_id,
+                mission_id=mission_id,
+                task_id=task.id,
+                provider_name=role_config.worker_provider.name,
+                started_at=receipt_started_at,
+                status="mcp_start_failed",
+                finished_at=datetime.now(UTC).isoformat(),
+            )
             return self._synthesize_missing_handoff(
                 task, summary="Worker MCP server failed to start"
             )
@@ -632,14 +706,36 @@ async def run_node(
         )
 
         # 3) Spawn the ACP agent.
-        process = await asyncio.create_subprocess_shell(
-            acp_command,
-            stdin=asyncio.subprocess.PIPE,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-            cwd=workspace_dir,
-            env=_acp_subprocess_env(role_config.worker_provider),
-            limit=SUBPROCESS_STREAM_LIMIT,
+        try:
+            process = await asyncio.create_subprocess_shell(
+                acp_command,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                cwd=workspace_dir,
+                env=_acp_subprocess_env(
+                    role_config.worker_provider,
+                    auth_context,
+                    role_config.worker_reasoning_effort,
+                ),
+                limit=SUBPROCESS_STREAM_LIMIT,
+            )
+        except Exception:  # noqa: BLE001
+            _write_api_authorization_receipt(
+                path=receipt_path,
+                auth=auth_context,
+                project_id=project_id,
+                mission_id=mission_id,
+                task_id=task.id,
+                provider_name=role_config.worker_provider.name,
+                started_at=receipt_started_at,
+                status="acp_start_failed",
+                finished_at=datetime.now(UTC).isoformat(),
+            )
+            raise
+        grant_expired = asyncio.Event()
+        grant_expiry_task = asyncio.create_task(
+            _enforce_api_grant_expiry(process, auth_context, grant_expired)
         )
         progress_tracker = ACPProgressTracker(callback=progress_callback)
         client = ACPClient(
@@ -690,6 +786,9 @@ async def run_node(
             session_error = str(exc)
             logger.error("ACP session failed for node %s: %s", task.id, exc)
         finally:
+            grant_expiry_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await grant_expiry_task
             await progress_tracker.flush()
             if process.returncode is None:
                 try:
@@ -715,6 +814,18 @@ async def run_node(
                 except OSError:
                     pass
             await _close_subprocess(mcp_process, timeout=5)
+            _write_api_authorization_receipt(
+                path=receipt_path,
+                auth=auth_context,
+                project_id=project_id,
+                mission_id=mission_id,
+                task_id=task.id,
+                provider_name=role_config.worker_provider.name,
+                started_at=receipt_started_at,
+                status="grant_expired" if grant_expired.is_set() else "finished",
+                finished_at=datetime.now(UTC).isoformat(),
+                exit_code=worker_exit_code,
+            )
 
         # 5) Parse and return.
         if handoff_path.exists():
@@ -751,6 +862,14 @@ async def run_terminal_review(
             role_config.worker_reasoning_effort,
         )
 
+        auth_context = prepare_acp_auth_context(
+            config=role_config,
+            provider=role_config.worker_provider,
+            task=None,
+            project_id=project_id,
+            mission_id=mission_id,
+        )
+
         workspace_dir = str(store.workspace_dir(project_id))
         project_bucket = str(store.zenith_dir(project_id))
         report_path = store.terminal_review_path(project_id, mission_id, spawn_ts)
@@ -796,7 +915,11 @@ async def run_terminal_review(
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             cwd=workspace_dir,
-            env=_acp_subprocess_env(role_config.worker_provider),
+            env=_acp_subprocess_env(
+                role_config.worker_provider,
+                auth_context,
+                role_config.worker_reasoning_effort,
+            ),
             limit=SUBPROCESS_STREAM_LIMIT,
         )
         tracker = ACPProgressTracker(callback=progress_callback)
@@ -895,7 +1018,7 @@ async def _start_worker_mcp_server(
             "--port",
             str(mcp_port),
         ]
-        env = os.environ.copy()
+        env = sanitized_process_env()
         env["ZENITH_HOME"] = str(self.config.harness_home)
         env["ZENITH_PROJECT_ID"] = project_id
         env["ZENITH_MISSION_ID"] = mission_id
@@ -933,7 +1056,8 @@ async def _start_terminal_reviewer_mcp(
             "--port",
             str(mcp_port),
         ]
-        env = os.environ.copy()
+        env = sanitized_process_env()
+        env["ZENITH_HOME"] = str(self.config.harness_home)
         env["ZENITH_PROJECT_ID"] = project_id
         env["ZENITH_MISSION_ID"] = mission_id
         env["ZENITH_TERMINAL_REVIEW_PATH"] = report_path
diff --git a/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md b/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md
index 22f6589..1aff76a 100644
--- a/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md
+++ b/zenith/src/zenith_harness/bundled/prompts/orchestrator/system_prompt.md
@@ -368,6 +368,15 @@ Every orchestrator tool returns an envelope with `projectId`, `state`, `projectR
 
 - `tasks`: list of task objects.
 - task fields include `id`, `type`, `body`, `targets`, `skill`, and `depends_on`.
+- every task defaults to `billing: {mode: subscription, api_grant: null}`. Keep this
+  default unless the user explicitly authorizes API-billed work while confirming
+  the plan. Never infer API permission from tool availability, ambient credentials,
+  prior tasks, or the agent's own judgment.
+- an API-billed work/validate task must set `billing.mode: api` and include
+  `api_grant` with `grant_id`, `api_project`, `max_usd`, and timezone-aware
+  `expires_at`. This is only a request; runtime also requires an exact external
+  operator grant bound to the project, mission, task, provider, budget, and expiry.
+- gates and the terminal reviewer are always subscription-only.
 - `type`: `work`, `validate`, or `gate`.
 - `work` and `validate` tasks require non-empty `body` and a `skill`.
 - `gate` tasks require `skill: null`, empty `body`, and one or more `targets`.
diff --git a/zenith/src/zenith_harness/cli.py b/zenith/src/zenith_harness/cli.py
index 22a8ce7..d50baea 100644
--- a/zenith/src/zenith_harness/cli.py
+++ b/zenith/src/zenith_harness/cli.py
@@ -34,6 +34,8 @@
     "GLM_API_KEY",
     "GLM_BASE_URL",
     "MAX_THINKING_TOKENS",
+    "ZENITH_API_GRANTS_FILE",
+    "ZENITH_CODEX_SUBSCRIPTION_HOME",
     "ZENITH_WORKER_REASONING_EFFORT",
     "ZENITH_VALIDATOR_REASONING_EFFORT",
     "ZENITH_TERMINAL_REVIEWER_REASONING_EFFORT",
diff --git a/zenith/src/zenith_harness/config.py b/zenith/src/zenith_harness/config.py
index 22adcc0..e538d6c 100644
--- a/zenith/src/zenith_harness/config.py
+++ b/zenith/src/zenith_harness/config.py
@@ -15,7 +15,7 @@
 DEFAULT_MAX_PARALLEL_NODES = 4
 
 # codex-acp `model_reasoning_effort` values. Also a safety allowlist: the
-# resolved value is spliced into a shell command line by acp_runner. Codex's
+# resolved value is serialized into the adapter's CODEX_CONFIG. Codex's
 # "ultra" is deliberately excluded: it is not a reasoning tier (codex
 # downgrades the request to "max" on the wire) but a switch to proactive
 # multi-agent mode — a lane spawning its own agent swarm inside a harness
@@ -72,6 +72,8 @@ class HarnessConfig:
     terminal_reviewer_provider_name: str | None
     terminal_reviewer_acp_command: str | None
     max_parallel_nodes: int = DEFAULT_MAX_PARALLEL_NODES
+    codex_subscription_home: Path | None = None
+    api_grants_file: Path | None = None
     # Per-role reasoning effort for providers whose ACP command accepts one
     # (codex today). None means the provider default ("xhigh" for codex).
     worker_reasoning_effort: str | None = None
@@ -109,6 +111,12 @@ def discover(cls) -> HarnessConfig:
         terminal_reviewer_acp_command = os.environ.get(
             "ZENITH_TERMINAL_REVIEWER_ACP_COMMAND"
         )
+        codex_subscription_home = _resolve_optional_path(
+            os.environ.get("ZENITH_CODEX_SUBSCRIPTION_HOME")
+        )
+        api_grants_file = _resolve_optional_path(
+            os.environ.get("ZENITH_API_GRANTS_FILE")
+        )
         return cls(
             bundled_dir=_bundled_dir(),
             harness_home=harness_home,
@@ -123,6 +131,8 @@ def discover(cls) -> HarnessConfig:
             max_parallel_nodes=_resolve_max_parallel(
                 os.environ.get("ZENITH_MAX_PARALLEL_NODES")
             ),
+            codex_subscription_home=codex_subscription_home,
+            api_grants_file=api_grants_file,
             worker_reasoning_effort=_resolve_reasoning_effort(
                 os.environ.get("ZENITH_WORKER_REASONING_EFFORT"),
                 env_var="ZENITH_WORKER_REASONING_EFFORT",
@@ -182,6 +192,10 @@ def resolved_terminal_reviewer_acp_command(self) -> str | None:
             or self.resolved_validator_acp_command
         )
 
+    @property
+    def resolved_codex_subscription_home(self) -> Path:
+        return self.codex_subscription_home or self.harness_home / "codex-subscription"
+
     @property
     def provider_selection(self) -> ProviderSelection:
         return ProviderSelection(
diff --git a/zenith/src/zenith_harness/models.py b/zenith/src/zenith_harness/models.py
index 2ca8521..fb8069f 100644
--- a/zenith/src/zenith_harness/models.py
+++ b/zenith/src/zenith_harness/models.py
@@ -14,9 +14,10 @@
 from __future__ import annotations
 
 import re
+from decimal import Decimal
 from typing import Annotated, Any, Literal, Union
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, model_validator
 
 # ---------------------------------------------------------------------------
 # Identifier conventions
@@ -25,11 +26,13 @@
 ASSERTION_ID_REGEX = re.compile(r"^[A-Z][A-Z0-9-]+$")
 TASK_ID_REGEX = re.compile(r"^[A-Za-z][A-Za-z0-9_-]*$")
 SKILL_NAME_REGEX = re.compile(r"^[a-z][a-z0-9_-]*$")
+API_GRANT_ID_REGEX = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
 
 
 TaskType = Literal["work", "validate", "gate"]
 TaskStatus = Literal["pending", "running", "cleared", "failed", "superseded"]
 AssertionStatus = Literal["pending", "passed", "failed"]
+BillingMode = Literal["subscription", "api"]
 
 
 # ---------------------------------------------------------------------------
@@ -37,6 +40,49 @@
 # ---------------------------------------------------------------------------
 
 
+class ApiGrantRequest(BaseModel):
+    """Non-secret API authorization requested by one task.
+
+    This request never authorizes itself.  The runtime must match it to an
+    operator-owned grant registry before any credential is injected.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    grant_id: str = Field(
+        pattern=API_GRANT_ID_REGEX.pattern,
+        description="Operator-issued grant id; contains no credential material.",
+    )
+    api_project: str = Field(
+        min_length=1,
+        description="Exact provider project the credential is restricted to.",
+    )
+    max_usd: Decimal = Field(
+        gt=0,
+        description="Maximum authorized spend recorded for this task.",
+    )
+    expires_at: AwareDatetime = Field(
+        description="Timezone-aware expiry copied from the operator grant.",
+    )
+
+
+class BillingPolicy(BaseModel):
+    """Task billing mode. Subscription is the fail-closed default."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    mode: BillingMode = "subscription"
+    api_grant: ApiGrantRequest | None = None
+
+    @model_validator(mode="after")
+    def validate_mode_and_grant(self) -> BillingPolicy:
+        if self.mode == "subscription" and self.api_grant is not None:
+            raise ValueError("subscription billing must not include api_grant")
+        if self.mode == "api" and self.api_grant is None:
+            raise ValueError("api billing requires api_grant")
+        return self
+
+
 class Task(BaseModel):
     """A single mission task.
 
@@ -85,6 +131,13 @@ class Task(BaseModel):
             "`type == 'gate'`, not from the dep itself."
         ),
     )
+    billing: BillingPolicy = Field(
+        default_factory=BillingPolicy,
+        description=(
+            "Credential policy for this task. Defaults to subscription. API mode "
+            "is only a request and requires an exact external operator grant."
+        ),
+    )
 
 
 class TaskList(BaseModel):
diff --git a/zenith/src/zenith_harness/task_validation.py b/zenith/src/zenith_harness/task_validation.py
index 407eb50..07275e2 100644
--- a/zenith/src/zenith_harness/task_validation.py
+++ b/zenith/src/zenith_harness/task_validation.py
@@ -106,6 +106,8 @@ def check_task_shape(tl: TaskList) -> list[ValidationError]:
                 errors.append(ValidationError("gate_with_body", task.id))
             if not task.targets:
                 errors.append(ValidationError("empty_targets", task.id))
+            if task.billing.mode != "subscription":
+                errors.append(ValidationError("gate_with_api_billing", task.id))
         else:
             if not task.skill:
                 errors.append(
diff --git a/zenith/tests/test_acp_auth.py b/zenith/tests/test_acp_auth.py
new file mode 100644
index 0000000..8d829b8
--- /dev/null
+++ b/zenith/tests/test_acp_auth.py
@@ -0,0 +1,437 @@
+"""ACP authentication and explicit API-grant security tests."""
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from pathlib import Path
+
+import pytest
+from pydantic import SecretStr
+
+from zenith_harness.acp_auth import (
+    ACPAuthContext,
+    ACPAuthError,
+    ResolvedApiGrant,
+    SUBSCRIPTION_CODEX_CONFIG,
+    api_grant_receipt,
+    build_acp_subprocess_env,
+    prepare_acp_auth_context,
+    sanitized_process_env,
+)
+from zenith_harness.acp_runner import (
+    ACPNodeRunner,
+    _enforce_api_grant_expiry,
+    _write_api_authorization_receipt,
+)
+from zenith_harness.assets import AssetLoader
+from zenith_harness.config import HarnessConfig
+from zenith_harness.models import ApiGrantRequest, BillingPolicy, Task
+from zenith_harness.providers import PROVIDERS
+
+
+def _config(harness_home: Path, grants_file: Path | None = None) -> HarnessConfig:
+    bundled = Path(__file__).resolve().parents[1] / "src" / "zenith_harness" / "bundled"
+    return HarnessConfig(
+        bundled_dir=bundled,
+        harness_home=harness_home,
+        projects_dir=harness_home / "projects",
+        orchestrator_provider_name="codex",
+        worker_provider_name="codex",
+        worker_acp_command="codex-acp",
+        validator_provider_name=None,
+        validator_acp_command=None,
+        terminal_reviewer_provider_name=None,
+        terminal_reviewer_acp_command=None,
+        api_grants_file=grants_file,
+    )
+
+
+def _api_task(*, expires_at: datetime, max_usd: str = "5.00") -> Task:
+    return Task(
+        id="w-api",
+        type="work",
+        body="explicitly approved API operation",
+        targets=["VAL-API"],
+        skill="api-worker",
+        billing=BillingPolicy(
+            mode="api",
+            api_grant=ApiGrantRequest(
+                grant_id="grant-001",
+                api_project="isolated-api-project",
+                max_usd=Decimal(max_usd),
+                expires_at=expires_at,
+            ),
+        ),
+    )
+
+
+def _write_private(path: Path, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+    path.chmod(0o600)
+
+
+def _write_registry(
+    path: Path,
+    credential_file: Path,
+    *,
+    issued_at: datetime,
+    expires_at: datetime,
+    max_usd: str = "5.00",
+    revoked: bool = False,
+) -> None:
+    payload = {
+        "version": 1,
+        "grants": [
+            {
+                "grant_id": "grant-001",
+                "zenith_project_id": "project-001",
+                "mission_id": "mission-001",
+                "task_id": "w-api",
+                "provider": "codex",
+                "api_project": "isolated-api-project",
+                "max_usd": max_usd,
+                "issued_at": issued_at.isoformat(),
+                "expires_at": expires_at.isoformat(),
+                "approved_by": "operator@example",
+                "credential_file": str(credential_file),
+                "revoked": revoked,
+            }
+        ],
+    }
+    _write_private(path, json.dumps(payload))
+
+
+def test_subscription_is_default_and_strips_all_ambient_credentials(
+    monkeypatch: pytest.MonkeyPatch, harness_home: Path
+) -> None:
+    sentinels = {
+        "OPENAI_API_KEY": "sentinel-openai",
+        "CODEX_API_KEY": "sentinel-codex",
+        "CODEX_ACCESS_TOKEN": "sentinel-access",
+        "ANTHROPIC_API_KEY": "sentinel-anthropic",
+        "DEFAULT_AUTH_REQUEST": "sentinel-auth-request",
+        "SOME_OTHER_SECRET": "sentinel-secret",
+    }
+    for name, value in sentinels.items():
+        monkeypatch.setenv(name, value)
+
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+    context = prepare_acp_auth_context(
+        config=_config(harness_home),
+        provider=PROVIDERS["codex"],
+        task=task,
+        project_id="project-001",
+        mission_id="mission-001",
+    )
+    env = build_acp_subprocess_env(PROVIDERS["codex"], context)
+
+    assert task.billing.mode == "subscription"
+    assert context.mode == "subscription"
+    assert env["HOME"] == str(harness_home / "codex-subscription")
+    assert env["CODEX_HOME"] == env["HOME"]
+    assert all(name not in env for name in sentinels)
+    assert "CODEX_SANDBOX" not in env
+    assert "CODEX_DISABLE_SANDBOX" not in env
+    assert context.codex_home is not None
+    assert (context.codex_home / "config.toml").read_text() == SUBSCRIPTION_CODEX_CONFIG
+
+
+def test_subscription_profile_rejects_persistent_api_auth(harness_home: Path) -> None:
+    home = harness_home / "codex-subscription"
+    home.mkdir()
+    _write_private(
+        home / "auth.json",
+        json.dumps({"auth_mode": "apikey", "OPENAI_API_KEY": "must-not-survive"}),
+    )
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+
+    with pytest.raises(ACPAuthError, match="not authenticated with ChatGPT"):
+        prepare_acp_auth_context(
+            config=_config(harness_home),
+            provider=PROVIDERS["codex"],
+            task=task,
+            project_id="project-001",
+            mission_id="mission-001",
+        )
+
+
+def test_ambient_old_snapshot_is_not_reused(
+    monkeypatch: pytest.MonkeyPatch, harness_home: Path, tmp_path: Path
+) -> None:
+    ambient_home = tmp_path / "ambient-codex"
+    snapshot = ambient_home / "shell_snapshots" / "old.sh"
+    snapshot.parent.mkdir(parents=True)
+    snapshot.write_text("export OPENAI_API_KEY=stale-key\n", encoding="utf-8")
+    monkeypatch.setenv("CODEX_HOME", str(ambient_home))
+    monkeypatch.setenv("OPENAI_API_KEY", "stale-key")
+
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+    context = prepare_acp_auth_context(
+        config=_config(harness_home),
+        provider=PROVIDERS["codex"],
+        task=task,
+        project_id="project-001",
+        mission_id="mission-001",
+    )
+    env = build_acp_subprocess_env(PROVIDERS["codex"], context)
+
+    assert env["CODEX_HOME"] != str(ambient_home)
+    assert "OPENAI_API_KEY" not in env
+    assert context.codex_home is not None
+    assert not (context.codex_home / "shell_snapshots").exists()
+
+
+def test_managed_subscription_home_rejects_new_snapshots(harness_home: Path) -> None:
+    snapshots = harness_home / "codex-subscription" / "shell_snapshots"
+    snapshots.mkdir(parents=True)
+    (snapshots / "unexpected.sh").write_text("export TOKEN=unexpected\n", encoding="utf-8")
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+
+    with pytest.raises(ACPAuthError, match="forbidden shell snapshots"):
+        prepare_acp_auth_context(
+            config=_config(harness_home),
+            provider=PROVIDERS["codex"],
+            task=task,
+            project_id="project-001",
+            mission_id="mission-001",
+        )
+
+
+def test_managed_subscription_config_cannot_be_overridden(harness_home: Path) -> None:
+    home = harness_home / "codex-subscription"
+    home.mkdir()
+    _write_private(home / "config.toml", 'forced_login_method = "api"\n')
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+
+    with pytest.raises(ACPAuthError, match="differs from the required profile"):
+        prepare_acp_auth_context(
+            config=_config(harness_home),
+            provider=PROVIDERS["codex"],
+            task=task,
+            project_id="project-001",
+            mission_id="mission-001",
+        )
+
+
+def test_api_request_without_operator_registry_fails_closed(harness_home: Path) -> None:
+    now = datetime.now(UTC)
+    with pytest.raises(ACPAuthError, match="ZENITH_API_GRANTS_FILE"):
+        prepare_acp_auth_context(
+            config=_config(harness_home),
+            provider=PROVIDERS["codex"],
+            task=_api_task(expires_at=now + timedelta(hours=1)),
+            project_id="project-001",
+            mission_id="mission-001",
+            now=now,
+        )
+
+
+def test_exact_operator_grant_injects_only_scoped_key_and_emits_safe_receipt(
+    monkeypatch: pytest.MonkeyPatch, harness_home: Path, tmp_path: Path
+) -> None:
+    now = datetime.now(UTC)
+    expires_at = now + timedelta(hours=1)
+    key = "test-explicit-api-key"
+    credential_file = tmp_path / "openai.key"
+    grants_file = tmp_path / "grants.json"
+    _write_private(credential_file, key + "\n")
+    _write_registry(
+        grants_file,
+        credential_file,
+        issued_at=now - timedelta(minutes=1),
+        expires_at=expires_at,
+    )
+    monkeypatch.setenv("OPENAI_API_KEY", "wrong-ambient-key")
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "wrong-provider-key")
+
+    context = prepare_acp_auth_context(
+        config=_config(harness_home, grants_file),
+        provider=PROVIDERS["codex"],
+        task=_api_task(expires_at=expires_at),
+        project_id="project-001",
+        mission_id="mission-001",
+        now=now,
+    )
+    env = build_acp_subprocess_env(PROVIDERS["codex"], context)
+    receipt = api_grant_receipt(context)
+
+    assert context.mode == "api"
+    assert env["OPENAI_API_KEY"] == key
+    assert env["CODEX_HOME"] == str(harness_home / "codex-api" / "grant-001")
+    assert "ANTHROPIC_API_KEY" not in env
+    assert key not in json.dumps(receipt)
+    assert receipt == {
+        "billing_mode": "api",
+        "grant_id": "grant-001",
+        "api_project": "isolated-api-project",
+        "max_usd": "5.00",
+        "expires_at": expires_at.isoformat(),
+        "approved_by": "operator@example",
+        "registry_sha256": receipt["registry_sha256"],
+    }
+
+    receipt_path = tmp_path / "receipt.json"
+    _write_api_authorization_receipt(
+        path=receipt_path,
+        auth=context,
+        project_id="project-001",
+        mission_id="mission-001",
+        task_id="w-api",
+        provider_name="codex",
+        started_at=now.isoformat(),
+        status="finished",
+        finished_at=(now + timedelta(minutes=1)).isoformat(),
+        exit_code=0,
+    )
+    receipt_text = receipt_path.read_text(encoding="utf-8")
+    assert key not in receipt_text
+    assert json.loads(receipt_text)["grant_id"] == "grant-001"
+
+
+def test_running_api_process_is_terminated_at_grant_expiry() -> None:
+    class FakeProcess:
+        returncode: int | None = None
+        terminated = False
+
+        def terminate(self) -> None:
+            self.terminated = True
+
+    process = FakeProcess()
+    expired = asyncio.Event()
+    context = ACPAuthContext(
+        mode="api",
+        codex_home=Path("/tmp/codex-api-test"),
+        api_grant=ResolvedApiGrant(
+            grant_id="grant-001",
+            api_project="isolated-api-project",
+            max_usd=Decimal("5.00"),
+            expires_at=datetime.now(UTC) + timedelta(milliseconds=5),
+            approved_by="operator@example",
+            credential=SecretStr("test-key"),
+            registry_sha256="0" * 64,
+        ),
+    )
+
+    asyncio.run(_enforce_api_grant_expiry(process, context, expired))  # type: ignore[arg-type]
+
+    assert expired.is_set()
+    assert process.terminated is True
+
+
+@pytest.mark.parametrize("failure", ["expired", "revoked", "budget-mismatch"])
+def test_invalid_operator_grants_fail_closed(
+    failure: str, harness_home: Path, tmp_path: Path
+) -> None:
+    now = datetime.now(UTC)
+    request_expiry = now + timedelta(hours=1)
+    grant_expiry = request_expiry
+    revoked = False
+    request_budget = "5.00"
+    if failure == "expired":
+        request_expiry = now - timedelta(minutes=1)
+        grant_expiry = request_expiry
+    elif failure == "revoked":
+        revoked = True
+    else:
+        request_budget = "6.00"
+
+    credential_file = tmp_path / "openai.key"
+    grants_file = tmp_path / "grants.json"
+    _write_private(credential_file, "test-key\n")
+    _write_registry(
+        grants_file,
+        credential_file,
+        issued_at=now - timedelta(hours=2),
+        expires_at=grant_expiry,
+        revoked=revoked,
+    )
+
+    with pytest.raises(ACPAuthError):
+        prepare_acp_auth_context(
+            config=_config(harness_home, grants_file),
+            provider=PROVIDERS["codex"],
+            task=_api_task(expires_at=request_expiry, max_usd=request_budget),
+            project_id="project-001",
+            mission_id="mission-001",
+            now=now,
+        )
+
+
+def test_world_readable_grant_registry_is_rejected(harness_home: Path, tmp_path: Path) -> None:
+    now = datetime.now(UTC)
+    credential_file = tmp_path / "openai.key"
+    grants_file = tmp_path / "grants.json"
+    _write_private(credential_file, "test-key\n")
+    _write_registry(
+        grants_file,
+        credential_file,
+        issued_at=now - timedelta(minutes=1),
+        expires_at=now + timedelta(hours=1),
+    )
+    grants_file.chmod(0o644)
+
+    with pytest.raises(ACPAuthError, match="permissions"):
+        prepare_acp_auth_context(
+            config=_config(harness_home, grants_file),
+            provider=PROVIDERS["codex"],
+            task=_api_task(expires_at=now + timedelta(hours=1)),
+            project_id="project-001",
+            mission_id="mission-001",
+            now=now,
+        )
+
+
+def test_non_codex_provider_cannot_consume_openai_api_grant(harness_home: Path) -> None:
+    now = datetime.now(UTC)
+    with pytest.raises(ACPAuthError, match="only valid for the codex provider"):
+        prepare_acp_auth_context(
+            config=_config(harness_home),
+            provider=PROVIDERS["claude"],
+            task=_api_task(expires_at=now + timedelta(hours=1)),
+            project_id="project-001",
+            mission_id="mission-001",
+            now=now,
+        )
+
+
+def test_mcp_environment_uses_same_secret_allowlist(
+    monkeypatch: pytest.MonkeyPatch, harness_home: Path
+) -> None:
+    monkeypatch.setenv("OPENAI_API_KEY", "must-not-reach-mcp")
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-reach-mcp")
+    captured: dict[str, str] = {}
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        captured.update(kwargs["env"])
+        return object()
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
+    config = _config(harness_home)
+    runner = ACPNodeRunner(config=config, loader=AssetLoader(config))
+    task = Task(id="w1", type="work", body="work", targets=["VAL-1"], skill="s")
+    asyncio.run(
+        runner._start_worker_mcp_server(
+            task=task,
+            project_id="project-001",
+            mission_id="mission-001",
+            handoff_path="/tmp/handoff.json",
+            workspace_dir="/tmp",
+            mcp_port=12345,
+        )
+    )
+
+    assert "OPENAI_API_KEY" not in captured
+    assert "ANTHROPIC_API_KEY" not in captured
+    assert captured["ZENITH_NODE_ID"] == "w1"
+
+
+def test_sanitizer_never_copies_unknown_secret_names(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("A_NEW_VENDOR_SECRET", "sentinel")
+    monkeypatch.setenv("PATH", "/safe/bin")
+
+    env = sanitized_process_env()
+    assert env["PATH"] == "/safe/bin"
+    assert "A_NEW_VENDOR_SECRET" not in env
diff --git a/zenith/tests/test_acp_runner.py b/zenith/tests/test_acp_runner.py
index 412ea69..c3793ad 100644
--- a/zenith/tests/test_acp_runner.py
+++ b/zenith/tests/test_acp_runner.py
@@ -21,6 +21,7 @@
     _acp_subprocess_env,
     _augment_acp_command,
 )
+from zenith_harness.acp_auth import ACPAuthContext
 from zenith_harness.providers import PROVIDERS
 from zenith_harness.assets import AssetLoader
 from zenith_harness.config import HarnessConfig
@@ -147,21 +148,38 @@ def test_synthesize_missing_handoff_records_failure(
     assert handoff_path.exists()
 
 
-def test_augment_acp_command_codex_appends_bypass_flags():
+def test_augment_acp_command_codex_does_not_append_bypass_flags():
     out = _augment_acp_command("codex-acp", PROVIDERS["codex"])
-    assert 'sandbox_mode="danger-full-access"' in out
-    assert 'approval_policy="never"' in out
-    assert 'model_reasoning_effort="xhigh"' in out
-    assert out.startswith("codex-acp ")
+    assert out == "codex-acp"
 
 
 def test_augment_acp_command_codex_reasoning_effort_override():
     out = _augment_acp_command("codex-acp", PROVIDERS["codex"], reasoning_effort="medium")
-    assert 'model_reasoning_effort="medium"' in out
-    assert "xhigh" not in out
-    # The bypass flags are effort-independent.
-    assert 'sandbox_mode="danger-full-access"' in out
-    assert 'approval_policy="never"' in out
+    assert out == "codex-acp"
+
+
+def test_codex_acp_reasoning_effort_uses_sanitized_adapter_config(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    monkeypatch.setenv(
+        "CODEX_CONFIG",
+        json.dumps(
+            {
+                "model_reasoning_effort": "max",
+                "sandbox_mode": "danger-full-access",
+                "approval_policy": "never",
+            }
+        ),
+    )
+
+    env = _acp_subprocess_env(
+        PROVIDERS["codex"],
+        ACPAuthContext(mode="subscription", codex_home=tmp_path / "codex-home"),
+        reasoning_effort="medium",
+    )
+
+    assert json.loads(env["CODEX_CONFIG"]) == {"model_reasoning_effort": "medium"}
 
 
 def test_augment_acp_command_claude_untouched():
@@ -183,11 +201,14 @@ def test_codex_acp_env_preserves_node_path_when_bwrap_is_present(
         path.chmod(0o755)
     monkeypatch.setenv("PATH", str(bin_dir))
 
-    env = _acp_subprocess_env(PROVIDERS["codex"])
+    env = _acp_subprocess_env(
+        PROVIDERS["codex"],
+        ACPAuthContext(mode="subscription", codex_home=tmp_path / "codex-home"),
+    )
 
     assert str(bin_dir) in env["PATH"].split(os.pathsep)
-    assert env["CODEX_SANDBOX"] == "danger-full-access"
-    assert env["CODEX_DISABLE_SANDBOX"] == "1"
+    assert "CODEX_SANDBOX" not in env
+    assert "CODEX_DISABLE_SANDBOX" not in env
 
 
 def test_attempt_path_naming(config: HarnessConfig, project_setup):
diff --git a/zenith/tests/test_acp_sandbox.py b/zenith/tests/test_acp_sandbox.py
index 7ec2593..d4b47b9 100644
--- a/zenith/tests/test_acp_sandbox.py
+++ b/zenith/tests/test_acp_sandbox.py
@@ -4,11 +4,12 @@
 import os
 from pathlib import Path
 
+from zenith_harness.acp_auth import ACPAuthContext
 from zenith_harness.acp_runner import _acp_subprocess_env
 from zenith_harness.providers import PROVIDERS
 
 
-def test_claude_env_unchanged() -> None:
+def test_claude_env_is_sanitized() -> None:
     env = _acp_subprocess_env(PROVIDERS["claude"])
     assert env.get("PATH", "") == os.environ.get("PATH", "")
     # Claude provider must NOT receive codex-specific hints.
@@ -30,7 +31,10 @@ def test_codex_preserves_path_when_bwrap_is_present(tmp_path: Path, monkeypatch)
     new_path = f"{fake_bwrap_dir}{os.pathsep}{other_dir}"
     monkeypatch.setenv("PATH", new_path)
 
-    env = _acp_subprocess_env(PROVIDERS["codex"])
+    env = _acp_subprocess_env(
+        PROVIDERS["codex"],
+        ACPAuthContext(mode="subscription", codex_home=tmp_path / "codex-home"),
+    )
     parts = env["PATH"].split(os.pathsep)
     assert str(fake_bwrap_dir) in parts
     assert str(other_dir) in parts
@@ -40,14 +44,17 @@ def test_codex_with_no_bwrap_on_path_unchanged(monkeypatch, tmp_path: Path) -> N
     only_other = tmp_path / "other"
     only_other.mkdir()
     monkeypatch.setenv("PATH", str(only_other))
-    env = _acp_subprocess_env(PROVIDERS["codex"])
+    env = _acp_subprocess_env(
+        PROVIDERS["codex"],
+        ACPAuthContext(mode="subscription", codex_home=tmp_path / "codex-home"),
+    )
     assert env["PATH"] == str(only_other)
 
 
-def test_codex_sets_env_var_hints() -> None:
-    """Belt-and-suspenders hints: codex versions that respect either env
-    var will skip bwrap regardless of PATH state.
-    """
-    env = _acp_subprocess_env(PROVIDERS["codex"])
-    assert env.get("CODEX_SANDBOX") == "danger-full-access"
-    assert env.get("CODEX_DISABLE_SANDBOX") == "1"
+def test_codex_does_not_set_sandbox_bypass_hints(tmp_path: Path) -> None:
+    env = _acp_subprocess_env(
+        PROVIDERS["codex"],
+        ACPAuthContext(mode="subscription", codex_home=tmp_path / "codex-home"),
+    )
+    assert "CODEX_SANDBOX" not in env
+    assert "CODEX_DISABLE_SANDBOX" not in env
diff --git a/zenith/tests/test_cli.py b/zenith/tests/test_cli.py
index d5da40a..cad1a09 100644
--- a/zenith/tests/test_cli.py
+++ b/zenith/tests/test_cli.py
@@ -141,6 +141,30 @@ def test_codex_init_writes_reasoning_effort_env(
         assert server_env["ZENITH_VALIDATOR_REASONING_EFFORT"] == "medium"
         assert server_env["ZENITH_TERMINAL_REVIEWER_REASONING_EFFORT"] == "low"
 
+    def test_codex_init_forwards_auth_registry_paths_without_api_key(
+        self,
+        runner: CliRunner,
+        workspace: Path,
+        env: dict[str, str],
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        subscription_home = workspace / "subscription-auth"
+        grants_file = workspace / "api-grants.json"
+        monkeypatch.setenv("ZENITH_CODEX_SUBSCRIPTION_HOME", str(subscription_home))
+        monkeypatch.setenv("ZENITH_API_GRANTS_FILE", str(grants_file))
+        monkeypatch.setenv("OPENAI_API_KEY", "must-not-enter-host-config")
+
+        r = runner.invoke(cli, ["init", "--workspace-dir", str(workspace), "--agent", "codex"])
+        assert r.exit_code == 0, r.output
+
+        config = tomllib.loads(
+            (workspace / ".codex" / "config.toml").read_text(encoding="utf-8")
+        )
+        server_env = config["mcp_servers"]["zenith"]["env"]
+        assert server_env["ZENITH_CODEX_SUBSCRIPTION_HOME"] == str(subscription_home)
+        assert server_env["ZENITH_API_GRANTS_FILE"] == str(grants_file)
+        assert "OPENAI_API_KEY" not in server_env
+
     def test_codex_init_escapes_quoted_acp_commands(
         self, runner: CliRunner, workspace: Path, env: dict[str, str]
     ) -> None:
diff --git a/zenith/tests/test_config.py b/zenith/tests/test_config.py
index 83c9c49..1364853 100644
--- a/zenith/tests/test_config.py
+++ b/zenith/tests/test_config.py
@@ -31,6 +31,21 @@ def test_discover_defaults_to_four_parallel_nodes(
     config = HarnessConfig.discover()
 
     assert config.max_parallel_nodes == 4
+    assert config.resolved_codex_subscription_home == harness_home / "codex-subscription"
+    assert config.api_grants_file is None
+
+
+def test_discover_explicit_codex_auth_paths(monkeypatch, harness_home: Path) -> None:
+    subscription_home = harness_home / "subscription-auth"
+    grants_file = harness_home / "grants.json"
+    monkeypatch.setenv("ZENITH_HOME", str(harness_home))
+    monkeypatch.setenv("ZENITH_CODEX_SUBSCRIPTION_HOME", str(subscription_home))
+    monkeypatch.setenv("ZENITH_API_GRANTS_FILE", str(grants_file))
+
+    config = HarnessConfig.discover()
+
+    assert config.resolved_codex_subscription_home == subscription_home
+    assert config.api_grants_file == grants_file
 
 
 def test_discover_explicit_one_uses_serial_parallelism(
@@ -98,7 +113,7 @@ def test_discover_invalid_reasoning_effort_rejected(
     monkeypatch.setenv("ZENITH_HOME", str(harness_home))
     monkeypatch.delenv("ZENITH_PROJECT_BUCKET_DIR", raising=False)
     _clear_effort_env(monkeypatch)
-    # Not silently ignored: the value lands in a shell command line, and a
+    # Not silently ignored: the value lands in the ACP adapter config, and a
     # typo'd downgrade would silently keep spending xhigh.
     monkeypatch.setenv("ZENITH_VALIDATOR_REASONING_EFFORT", "extra-high")
 
diff --git a/zenith/tests/test_models.py b/zenith/tests/test_models.py
index ea2fe87..94daa95 100644
--- a/zenith/tests/test_models.py
+++ b/zenith/tests/test_models.py
@@ -5,7 +5,9 @@
 from pydantic import ValidationError as PydanticValidationError
 
 from zenith_harness.models import (
+    ApiGrantRequest,
     AttentionNeeded,
+    BillingPolicy,
     Decision,
     Draft,
     Envelope,
@@ -27,6 +29,44 @@ def test_minimal_work(self) -> None:
         assert t.skill == "api-contract-worker"
         assert t.depends_on == []
         assert t.auto_merge is True
+        assert t.billing.mode == "subscription"
+        assert t.billing.api_grant is None
+
+    def test_api_billing_requires_explicit_grant(self) -> None:
+        with pytest.raises(PydanticValidationError):
+            BillingPolicy(mode="api")
+
+    def test_subscription_rejects_api_grant(self) -> None:
+        from datetime import UTC, datetime, timedelta
+
+        request = ApiGrantRequest(
+            grant_id="grant-001",
+            api_project="project",
+            max_usd="1.00",
+            expires_at=datetime.now(UTC) + timedelta(hours=1),
+        )
+        with pytest.raises(PydanticValidationError):
+            BillingPolicy(mode="subscription", api_grant=request)
+
+    def test_task_rejects_inline_api_key(self) -> None:
+        with pytest.raises(PydanticValidationError):
+            Task(
+                id="w1",
+                type="work",
+                body="work",
+                targets=["VAL-001"],
+                skill="s",
+                billing={
+                    "mode": "api",
+                    "api_grant": {
+                        "grant_id": "grant-001",
+                        "api_project": "project",
+                        "max_usd": "1.00",
+                        "expires_at": "2026-09-04T18:00:00Z",
+                        "api_key": "must-never-be-accepted",
+                    },
+                },
+            )
 
     def test_work_accepts_legacy_auto_merge_field(self) -> None:
         t = Task(
diff --git a/zenith/tests/test_task_validation.py b/zenith/tests/test_task_validation.py
index d8c11ef..b73869a 100644
--- a/zenith/tests/test_task_validation.py
+++ b/zenith/tests/test_task_validation.py
@@ -1,9 +1,10 @@
 """Submit-time task-list validation. See `specs/task_list/PRODUCT.md`."""
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
-from zenith_harness.models import Task, TaskList
+from zenith_harness.models import ApiGrantRequest, BillingPolicy, Task, TaskList
 from zenith_harness.task_validation import (
     check_acyclic,
     check_coverage,
@@ -116,6 +117,24 @@ def test_empty_targets_gate_rejected(self) -> None:
         errs = check_task_shape(tl)
         assert any(e.code == "empty_targets" for e in errs)
 
+    def test_gate_with_api_billing_rejected(self) -> None:
+        task = _task("g1", "gate", ["X"])
+        task = task.model_copy(
+            update={
+                "billing": BillingPolicy(
+                    mode="api",
+                    api_grant=ApiGrantRequest(
+                        grant_id="grant-001",
+                        api_project="project",
+                        max_usd="1.00",
+                        expires_at=datetime.now(UTC) + timedelta(hours=1),
+                    ),
+                )
+            }
+        )
+        errs = check_task_shape(TaskList(tasks=[task]))
+        assert any(e.code == "gate_with_api_billing" for e in errs)
+
 
 class TestDepsResolve:
     def test_dep_unknown_task(self) -> None: