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

PRD 0030: Deduplicate egress token resolution across backends

• Status: Draft
• 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:

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:

# 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:

# 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).