complete(prd): mark PRD 0030 active

Extract egress_resolve_token_values_with_provider into bot_bottle/egress.py.
Both docker and smolmachines launch paths now call the shared function
instead of duplicating the forward_host_credentials / CODEX_HOST_CREDENTIAL_TOKEN_REF
resolution block.

Also fixes the host_env: object annotation on smolmachines._resolve_token_env
to the correct dict[str, str].

Closes #118.

bot_bottle/backend/docker/launch.py

@@ -42,11 +42,7 @@ from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
-from ...codex_auth import codex_host_access_token
-from ...egress import (
-    CODEX_HOST_CREDENTIAL_TOKEN_REF,
-    egress_resolve_token_values,
-)
+from ...egress import egress_resolve_token_values_with_provider
 from ...log import info
 from . import network as network_mod
 from . import util as docker_mod
@@ -180,18 +176,12 @@ def launch(
         # Step 7: compose up. Token values + the OAuth placeholder
         # flow through subprocess env; the compose file holds only
         # bare names for the secret-carrying entries.
-        token_values: dict[str, str] = {}
-        if plan.egress_plan.routes:
-            token_values = egress_resolve_token_values(
-                plan.egress_plan.token_env_map, dict(os.environ),
+        bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+        token_values = egress_resolve_token_values_with_provider(
+            plan.egress_plan.token_env_map,
+            bottle.agent_provider.forward_host_credentials,
+            dict(os.environ),
         )
-            if plan.spec.manifest.bottle_for(
-                plan.spec.agent_name,
-            ).agent_provider.forward_host_credentials:
-                access_token = codex_host_access_token(dict(os.environ))
-                for token_env, token_ref in plan.egress_plan.token_env_map.items():
-                    if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
-                        token_values[token_env] = access_token
         compose_env: dict[str, str] = {
             **os.environ,
             **plan.forwarded_env,

bot_bottle/backend/smolmachines/launch.py

@@ -26,11 +26,9 @@ from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
 
-from ...codex_auth import codex_host_access_token
 from ...egress import (
-    CODEX_HOST_CREDENTIAL_TOKEN_REF,
     EGRESS_ROUTES_IN_CONTAINER,
-    egress_resolve_token_values,
+    egress_resolve_token_values_with_provider,
 )
 from ...pipelock import (
     PIPELOCK_CA_CERT_IN_CONTAINER,
@@ -146,7 +144,7 @@ def launch(
         # spec's ports_to_publish list expands depending on which
         # daemons the agent needs to reach from the smolvm guest.
         bundle_spec = _bundle_launch_spec(plan, network, loopback_ip)
-        token_env = _resolve_token_env(plan, os.environ)
+        token_env = _resolve_token_env(plan, dict(os.environ))
         _bundle.ensure_bundle_image(bundle_spec.image)
         _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
         stack.callback(_bundle.stop_bundle, plan.slug)
@@ -420,24 +418,17 @@ def _bundle_launch_spec(
 
 
 def _resolve_token_env(
-    plan: SmolmachinesBottlePlan, host_env: object
+    plan: SmolmachinesBottlePlan, host_env: dict[str, str],
 ) -> dict[str, str]:
     """Resolve the egress token env-var values from the host's
     environ so they reach the bundle's process env via docker's
     `-e NAME` inheritance. Empty when no routes declare auth."""
-    ep = plan.egress_plan
-    if not ep.routes:
-        return {}
-    env = dict(host_env)
-    token_values = egress_resolve_token_values(ep.token_env_map, env)
-    if plan.spec.manifest.bottle_for(
-        plan.spec.agent_name,
-    ).agent_provider.forward_host_credentials:
-        access_token = codex_host_access_token(env)
-        for token_env, token_ref in ep.token_env_map.items():
-            if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
-                token_values[token_env] = access_token
-    return token_values
+    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    return egress_resolve_token_values_with_provider(
+        plan.egress_plan.token_env_map,
+        bottle.agent_provider.forward_host_credentials,
+        host_env,
+    )
 
 
 def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:

bot_bottle/egress.py

@@ -29,6 +29,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from .codex_auth import codex_host_access_token
 from .log import die
 
 if TYPE_CHECKING:
@@ -360,6 +361,31 @@ def egress_resolve_token_values(
     return out
 
 
+def egress_resolve_token_values_with_provider(
+    token_env_map: dict[str, str],
+    forward_host_credentials: bool,
+    host_env: dict[str, str],
+) -> dict[str, str]:
+    """Resolve all egress token env-var values, including the optional
+    Codex host credential slot.
+
+    Combines `egress_resolve_token_values` (manifest-declared token refs)
+    with the `forward_host_credentials` path (Codex ChatGPT bearer).
+    Returns an empty dict when `token_env_map` is empty.
+
+    Pure function: `host_env` is passed in so tests can use a sealed
+    mapping without touching `os.environ`."""
+    if not token_env_map:
+        return {}
+    token_values = egress_resolve_token_values(token_env_map, host_env)
+    if forward_host_credentials:
+        access_token = codex_host_access_token(host_env)
+        for token_env, token_ref in token_env_map.items():
+            if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
+                token_values[token_env] = access_token
+    return token_values
+
+
 class Egress(ABC):
     """The per-bottle egress proxy. Encapsulates the host-side prepare
     (route lift + routes.yaml render + token-env-map derivation); the
@@ -403,6 +429,7 @@ __all__ = [
     "egress_manifest_routes",
     "egress_render_routes",
     "egress_resolve_token_values",
+    "egress_resolve_token_values_with_provider",
     "egress_routes_for_bottle",
     "egress_token_env_map",
 ]

docs/prds/0030-deduplicate-egress-token-resolution.md

@@ -0,0 +1,202 @@
+# PRD 0030: Deduplicate egress token resolution across backends
+
+- **Status:** Active
+- **Author:** didericis-claude
+- **Created:** 2026-06-02
+- **Issue:** #118
+
+## Summary
+
+Extract the duplicated egress token resolution block — which calls
+`egress_resolve_token_values` and then slots in the Codex host
+credential — into a single shared function in `bot_bottle/egress.py`,
+and have both backends call it.
+
+## Problem
+
+The same logic block appears verbatim in two places:
+
+- `bot_bottle/backend/docker/launch.py` (~line 183): inline inside
+  `launch()`, building `token_values` before `compose_up`.
+- `bot_bottle/backend/smolmachines/launch.py` (~line 422): the private
+  `_resolve_token_env` helper, called before `_bundle.start_bundle`.
+
+Both blocks:
+1. Short-circuit to `{}` when there are no egress routes.
+2. Call `egress_resolve_token_values(token_env_map, host_env)` to
+   resolve ordinary manifest-declared token refs.
+3. Check `agent_provider.forward_host_credentials` and, when true,
+   call `codex_host_access_token` and slot the result into any
+   `token_env` whose `token_ref` is `CODEX_HOST_CREDENTIAL_TOKEN_REF`.
+
+The duplication means any change to step 3 — new provider credential
+type, token refresh logic, a new sentinel ref — must be applied twice.
+PRD 0029, which introduced `forward_host_credentials`, already had to
+wire both backends; the next change will too. This is a near-certain
+future sync bug.
+
+## Goals / Success Criteria
+
+- The `forward_host_credentials` resolution logic exists in exactly one
+  place in the codebase.
+- Both `docker/launch.py` and `smolmachines/launch.py` call that
+  shared function.
+- No behaviour change for either backend.
+- The shared function is unit-testable without a plan or backend
+  dependency.
+
+## Non-goals
+
+- Changes to token resolution semantics. This is a pure refactor.
+- Adding support for any new credential type or provider.
+- Consolidating any other backend differences beyond this one block.
+
+## Design
+
+### New function: `egress_resolve_token_values_with_provider`
+
+Add to `bot_bottle/egress.py`:
+
+```python
+def egress_resolve_token_values_with_provider(
+    token_env_map: dict[str, str],
+    forward_host_credentials: bool,
+    host_env: dict[str, str],
+) -> dict[str, str]:
+    """Resolve all egress token env-var values from the host environ.
+
+    Combines the manifest-declared token refs (via
+    `egress_resolve_token_values`) with the optional Codex host
+    credential slot (when `forward_host_credentials` is True). Returns
+    an empty dict when `token_env_map` is empty.
+
+    Pure function: `host_env` is passed in so callers can pass a sealed
+    mapping in tests without touching `os.environ`. The Codex access
+    token is read lazily only when the flag is set and a matching slot
+    exists, so callers that don't use `forward_host_credentials` pay no
+    cost."""
+    if not token_env_map:
+        return {}
+    token_values = egress_resolve_token_values(token_env_map, host_env)
+    if forward_host_credentials:
+        from .codex_auth import codex_host_access_token
+        access_token = codex_host_access_token(host_env)
+        for token_env, token_ref in token_env_map.items():
+            if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
+                token_values[token_env] = access_token
+    return token_values
+```
+
+The import of `codex_auth` is deferred inside the branch to avoid a
+circular import — `egress.py` is imported by `pipelock.py` which
+`codex_auth.py` does not import, but `codex_auth.py` imports from
+`egress.py` transitively via `agent_provider`. The lazy import sidesteps
+the cycle without restructuring the module graph.
+
+### Backend call sites
+
+**`docker/launch.py`** — replace the inline block:
+
+```python
+# before
+token_values: dict[str, str] = {}
+if plan.egress_plan.routes:
+    token_values = egress_resolve_token_values(
+        plan.egress_plan.token_env_map, dict(os.environ),
+    )
+    if plan.spec.manifest.bottle_for(
+        plan.spec.agent_name,
+    ).agent_provider.forward_host_credentials:
+        access_token = codex_host_access_token(dict(os.environ))
+        for token_env, token_ref in plan.egress_plan.token_env_map.items():
+            if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
+                token_values[token_env] = access_token
+
+# after
+bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+token_values = egress_resolve_token_values_with_provider(
+    plan.egress_plan.token_env_map,
+    bottle.agent_provider.forward_host_credentials,
+    dict(os.environ),
+)
+```
+
+**`smolmachines/launch.py`** — replace `_resolve_token_env`:
+
+```python
+# before
+def _resolve_token_env(
+    plan: SmolmachinesBottlePlan, host_env: object
+) -> dict[str, str]:
+    ep = plan.egress_plan
+    if not ep.routes:
+        return {}
+    env = dict(host_env)
+    token_values = egress_resolve_token_values(ep.token_env_map, env)
+    if plan.spec.manifest.bottle_for(
+        plan.spec.agent_name,
+    ).agent_provider.forward_host_credentials:
+        access_token = codex_host_access_token(env)
+        for token_env, token_ref in ep.token_env_map.items():
+            if token_ref == CODEX_HOST_CREDENTIAL_TOKEN_REF:
+                token_values[token_env] = access_token
+    return token_values
+
+# after
+def _resolve_token_env(
+    plan: SmolmachinesBottlePlan, host_env: dict[str, str],
+) -> dict[str, str]:
+    bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+    return egress_resolve_token_values_with_provider(
+        plan.egress_plan.token_env_map,
+        bottle.agent_provider.forward_host_credentials,
+        host_env,
+    )
+```
+
+Note: the `host_env: object` annotation on the smolmachines version is
+also corrected to `dict[str, str]` — the annotation was wrong; the
+function body immediately called `dict(host_env)` treating it as a
+mapping.
+
+### Imports
+
+- `docker/launch.py`: drop `codex_host_access_token` and
+  `CODEX_HOST_CREDENTIAL_TOKEN_REF` imports; add
+  `egress_resolve_token_values_with_provider`.
+- `smolmachines/launch.py`: same drops; add
+  `egress_resolve_token_values_with_provider`.
+- `egress.py`: add `egress_resolve_token_values_with_provider` to
+  `__all__`.
+
+## Implementation chunks
+
+1. **PRD (this commit).** Sets the design.
+2. **Shared function.** Add `egress_resolve_token_values_with_provider`
+   to `bot_bottle/egress.py` with unit tests covering: no routes →
+   empty dict, routes with no `forward_host_credentials` → manifest
+   tokens only, routes with `forward_host_credentials` + matching ref →
+   Codex token slotted in, routes with `forward_host_credentials` + no
+   matching ref → manifest tokens only (Codex call still made but no
+   slot to fill).
+3. **Docker backend.** Replace the inline block in `docker/launch.py`.
+4. **Smolmachines backend.** Replace `_resolve_token_env` in
+   `smolmachines/launch.py`; fix the `host_env: object` annotation.
+5. **Cleanup.** Remove now-unused imports from both backends.
+
+## Testing strategy
+
+- **Unit (must):** the four cases above in a new test in
+  `tests/unit/test_egress.py`, using a sealed `host_env` dict so tests
+  don't touch `os.environ` or the filesystem.
+- **No integration changes needed:** both backends' existing integration
+  tests exercise the full launch path; a behaviour regression would
+  surface as a token-injection failure in those tests.
+
+## References
+
+- Issue #118: Deduplicate egress token resolution across backends.
+- Issue #117: Complexity hotspots in launch, egress, and auth paths
+  (source of the finding).
+- PRD 0029: Provider auth credentials through egress (introduced the
+  duplicated block).

tests/unit/test_egress.py

@@ -2,6 +2,7 @@
 resolution (PRD 0017)."""
 
 import unittest
+import unittest.mock
 
 from bot_bottle.egress import (
     CODEX_HOST_CREDENTIAL_TOKEN_REF,
@@ -9,6 +10,7 @@ from bot_bottle.egress import (
     egress_manifest_routes,
     egress_render_routes,
     egress_resolve_token_values,
+    egress_resolve_token_values_with_provider,
     egress_routes_for_bottle,
     egress_token_env_map,
 )
@@ -349,5 +351,64 @@ class TestResolveTokenValues(unittest.TestCase):
         self.assertEqual({}, out)
 
 
+class TestResolveTokenValuesWithProvider(unittest.TestCase):
+    def test_empty_map_returns_empty(self):
+        out = egress_resolve_token_values_with_provider({}, False, {})
+        self.assertEqual({}, out)
+
+    def test_empty_map_with_forward_credentials_returns_empty(self):
+        # forward_host_credentials=True but no slots → no codex call needed.
+        out = egress_resolve_token_values_with_provider({}, True, {})
+        self.assertEqual({}, out)
+
+    def test_manifest_tokens_resolved_without_forward_credentials(self):
+        out = egress_resolve_token_values_with_provider(
+            {"EGRESS_TOKEN_0": "GH_PAT"},
+            False,
+            {"GH_PAT": "ghp_secret"},
+        )
+        self.assertEqual({"EGRESS_TOKEN_0": "ghp_secret"}, out)
+
+    def test_codex_token_slotted_in_when_forward_credentials_and_matching_ref(self):
+        with unittest.mock.patch(
+            "bot_bottle.egress.codex_host_access_token",
+            return_value="codex-access-token",
+        ):
+            out = egress_resolve_token_values_with_provider(
+                {"EGRESS_TOKEN_0": CODEX_HOST_CREDENTIAL_TOKEN_REF},
+                True,
+                {},
+            )
+        self.assertEqual({"EGRESS_TOKEN_0": "codex-access-token"}, out)
+
+    def test_codex_token_not_slotted_when_no_matching_ref(self):
+        # forward_host_credentials=True but no CODEX_HOST_CREDENTIAL_TOKEN_REF
+        # slot in the map → manifest tokens only; Codex token is fetched but
+        # nothing to slot it into.
+        with unittest.mock.patch(
+            "bot_bottle.egress.codex_host_access_token",
+            return_value="codex-access-token",
+        ):
+            out = egress_resolve_token_values_with_provider(
+                {"EGRESS_TOKEN_0": "GH_PAT"},
+                True,
+                {"GH_PAT": "ghp_secret"},
+            )
+        self.assertEqual({"EGRESS_TOKEN_0": "ghp_secret"}, out)
+
+    def test_codex_not_called_when_forward_credentials_false(self):
+        called = []
+        with unittest.mock.patch(
+            "bot_bottle.egress.codex_host_access_token",
+            side_effect=lambda *_: called.append(1) or "tok",
+        ):
+            egress_resolve_token_values_with_provider(
+                {"EGRESS_TOKEN_0": "GH_PAT"},
+                False,
+                {"GH_PAT": "ghp_secret"},
+            )
+        self.assertEqual([], called)
+
+
 if __name__ == "__main__":
     unittest.main()