feat(proxy): support OAuth callers for Codex Live Voice

app/db/alembic/versions/20260804_000000_add_oauth_live_policies.py

@@ -0,0 +1,71 @@
+"""Add the global OAuth Live policy.
+
+Revision ID: 20260804_000000_add_oauth_live_policies
+Revises: 20260803_000000_merge_http_bridge_recovery_and_capability_lineage_heads
+Create Date: 2026-08-04
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260804_000000_add_oauth_live_policies"
+down_revision = "20260803_000000_merge_http_bridge_recovery_and_capability_lineage_heads"
+branch_labels = None
+depends_on = None
+
+_POLICY_TABLE = "oauth_live_global_policy"
+_ASSIGNMENTS_TABLE = "oauth_live_global_policy_accounts"
+_ALLOWED_ACCOUNT_INDEX = "ix_oauth_live_global_policy_accounts_allowed_account_id"
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table(_POLICY_TABLE):
+        op.create_table(
+            _POLICY_TABLE,
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("is_active", sa.Boolean(), server_default=sa.false(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.CheckConstraint("id = 1", name="ck_oauth_live_global_policy_singleton"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    inspector = sa.inspect(bind)
+    if not inspector.has_table(_ASSIGNMENTS_TABLE):
+        op.create_table(
+            _ASSIGNMENTS_TABLE,
+            sa.Column("policy_id", sa.Integer(), nullable=False),
+            sa.Column("allowed_account_id", sa.String(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+            sa.ForeignKeyConstraint(["policy_id"], [f"{_POLICY_TABLE}.id"], ondelete="CASCADE"),
+            sa.ForeignKeyConstraint(["allowed_account_id"], ["accounts.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("policy_id", "allowed_account_id"),
+        )
+    assignment_indexes = (
+        {str(index["name"]) for index in sa.inspect(bind).get_indexes(_ASSIGNMENTS_TABLE)}
+        if sa.inspect(bind).has_table(_ASSIGNMENTS_TABLE)
+        else set()
+    )
+    if _ALLOWED_ACCOUNT_INDEX not in assignment_indexes:
+        op.create_index(
+            _ALLOWED_ACCOUNT_INDEX,
+            _ASSIGNMENTS_TABLE,
+            ["allowed_account_id"],
+            unique=False,
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if inspector.has_table(_ASSIGNMENTS_TABLE):
+        indexes = {str(index["name"]) for index in inspector.get_indexes(_ASSIGNMENTS_TABLE)}
+        if _ALLOWED_ACCOUNT_INDEX in indexes:
+            op.drop_index(_ALLOWED_ACCOUNT_INDEX, table_name=_ASSIGNMENTS_TABLE)
+        op.drop_table(_ASSIGNMENTS_TABLE)
+    if sa.inspect(bind).has_table(_POLICY_TABLE):
+        op.drop_table(_POLICY_TABLE)

app/db/models.py

@@ -7,6 +7,7 @@
 from sqlalchemy import (
     BigInteger,
     Boolean,
+    CheckConstraint,
     DateTime,
     Float,
     ForeignKey,
@@ -1105,6 +1106,64 @@ class ApiKey(Base):
     )
 
 
+class OAuthLivePolicy(Base):
+    __tablename__ = "oauth_live_global_policy"
+    __table_args__ = (CheckConstraint("id = 1", name="ck_oauth_live_global_policy_singleton"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=False, server_default=false(), nullable=False)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+    allowed_accounts: Mapped[list["OAuthLivePolicyAccount"]] = relationship(
+        "OAuthLivePolicyAccount",
+        back_populates="policy",
+        cascade="all, delete-orphan",
+        passive_deletes=True,
+        lazy="selectin",
+    )
+
+
+class OAuthLivePolicyAccount(Base):
+    __tablename__ = "oauth_live_global_policy_accounts"
+    __table_args__ = (
+        Index(
+            "ix_oauth_live_global_policy_accounts_allowed_account_id",
+            "allowed_account_id",
+        ),
+    )
+
+    policy_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("oauth_live_global_policy.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    allowed_account_id: Mapped[str] = mapped_column(
+        String,
+        ForeignKey("accounts.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+
+    policy: Mapped["OAuthLivePolicy"] = relationship(
+        "OAuthLivePolicy",
+        back_populates="allowed_accounts",
+    )
+
+
 class ApiKeyAccountAssignment(Base):
     __tablename__ = "api_key_accounts"
 

app/dependencies.py

@@ -32,6 +32,8 @@
 from app.modules.model_sources.repository import ModelSourcesRepository
 from app.modules.model_sources.service import ModelSourcesService
 from app.modules.oauth.service import OauthService
+from app.modules.oauth_live.repository import OAuthLivePolicyRepository
+from app.modules.oauth_live.service import OAuthLivePolicyService
 from app.modules.proxy.capability_lineage_repository import CapabilityLineageRepository
 from app.modules.proxy.repo_bundle import ProxyRepositories
 from app.modules.proxy.service import ProxyService
@@ -74,6 +76,13 @@ class OauthContext:
     service: OauthService
 
 
+@dataclass(slots=True)
+class OAuthLivePolicyContext:
+    session: AsyncSession
+    repository: OAuthLivePolicyRepository
+    service: OAuthLivePolicyService
+
+
 @dataclass(slots=True)
 class DashboardAuthContext:
     session: AsyncSession
@@ -232,6 +241,17 @@ def get_oauth_context(
     return OauthContext(service=OauthService(accounts_repository, repo_factory=_accounts_repo_context))
 
 
+def get_oauth_live_policy_context(
+    session: AsyncSession = Depends(get_session),
+) -> OAuthLivePolicyContext:
+    repository = OAuthLivePolicyRepository(session)
+    return OAuthLivePolicyContext(
+        session=session,
+        repository=repository,
+        service=OAuthLivePolicyService(repository),
+    )
+
+
 def get_dashboard_auth_context(
     session: AsyncSession = Depends(get_session),
 ) -> DashboardAuthContext:

app/main.py

@@ -75,6 +75,7 @@
 from app.modules.health import api as health_api
 from app.modules.model_sources import api as model_sources_api
 from app.modules.oauth import api as oauth_api
+from app.modules.oauth_live import api as oauth_live_api
 from app.modules.proxy import api as proxy_api
 from app.modules.proxy.cap_partitioning import refresh_cap_partition
 from app.modules.proxy.durable_bridge_coordinator import DurableBridgeSessionCoordinator
@@ -708,6 +709,7 @@ def create_app() -> FastAPI:
     app.include_router(conversation_archive_api.router)
     app.include_router(runtime_api.router)
     app.include_router(oauth_api.router)
+    app.include_router(oauth_live_api.router)
     app.include_router(dashboard_auth_api.router)
     app.include_router(settings_api.router)
     app.include_router(firewall_api.router)

app/modules/oauth_live/__init__.py

@@ -0,0 +1 @@
+"""OAuth-authenticated Live Voice caller policy."""

app/modules/oauth_live/api.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Body, Depends, Request
+
+from app.core.audit.service import AuditService
+from app.core.auth.dependencies import (
+    require_dashboard_write_access,
+    set_dashboard_error_format,
+    validate_dashboard_session,
+)
+from app.core.exceptions import DashboardBadRequestError
+from app.dependencies import OAuthLivePolicyContext, get_oauth_live_policy_context
+from app.modules.oauth_live.schemas import OAuthLivePolicyResponse, OAuthLivePolicyUpdateRequest
+from app.modules.oauth_live.service import OAuthLivePolicyValidationError
+
+router = APIRouter(
+    prefix="/api/oauth-live-policy",
+    tags=["dashboard"],
+    dependencies=[Depends(validate_dashboard_session), Depends(set_dashboard_error_format)],
+)
+
+
+@router.get("", response_model=OAuthLivePolicyResponse)
+async def get_oauth_live_policy(
+    context: OAuthLivePolicyContext = Depends(get_oauth_live_policy_context),
+) -> OAuthLivePolicyResponse:
+    return await context.service.get_policy()
+
+
+@router.put("", response_model=OAuthLivePolicyResponse)
+async def update_oauth_live_policy(
+    request: Request,
+    payload: OAuthLivePolicyUpdateRequest = Body(...),
+    _write_access=Depends(require_dashboard_write_access),
+    context: OAuthLivePolicyContext = Depends(get_oauth_live_policy_context),
+) -> OAuthLivePolicyResponse:
+    try:
+        updated = await context.service.update_policy(payload)
+    except OAuthLivePolicyValidationError as exc:
+        raise DashboardBadRequestError(str(exc), code="invalid_oauth_live_policy") from exc
+    AuditService.log_async(
+        "oauth_live_policy_updated",
+        actor_ip=request.client.host if request.client else None,
+        details={
+            "is_active": updated.is_active,
+            "allowed_account_count": len(updated.allowed_account_ids),
+        },
+    )
+    return updated

app/modules/oauth_live/repository.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.db.models import Account, AccountStatus, OAuthLivePolicy, OAuthLivePolicyAccount
+
+GLOBAL_POLICY_ID = 1
+
+
+class OAuthLivePolicyRepository:
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
+    async def existing_account_ids(self, account_ids: list[str]) -> frozenset[str]:
+        if not account_ids:
+            return frozenset()
+        rows = await self._session.scalars(select(Account.id).where(Account.id.in_(account_ids)))
+        return frozenset(rows.all())
+
+    async def get_policy(self) -> OAuthLivePolicy | None:
+        result = await self._session.execute(
+            select(OAuthLivePolicy)
+            .options(selectinload(OAuthLivePolicy.allowed_accounts))
+            .where(OAuthLivePolicy.id == GLOBAL_POLICY_ID)
+        )
+        return result.scalar_one_or_none()
+
+    async def replace_policy(
+        self,
+        *,
+        is_active: bool,
+        allowed_account_ids: list[str],
+    ) -> OAuthLivePolicy:
+        policy = await self._session.get(OAuthLivePolicy, GLOBAL_POLICY_ID)
+        if policy is None:
+            policy = OAuthLivePolicy(id=GLOBAL_POLICY_ID, is_active=is_active)
+            self._session.add(policy)
+            await self._session.flush()
+        else:
+            policy.is_active = is_active
+            policy.updated_at = datetime.now(timezone.utc)
+
+        await self._session.execute(
+            delete(OAuthLivePolicyAccount).where(OAuthLivePolicyAccount.policy_id == GLOBAL_POLICY_ID)
+        )
+        self._session.add_all(
+            OAuthLivePolicyAccount(
+                policy_id=GLOBAL_POLICY_ID,
+                allowed_account_id=allowed_account_id,
+            )
+            for allowed_account_id in allowed_account_ids
+        )
+        await self._session.commit()
+        refreshed = await self.get_policy()
+        assert refreshed is not None
+        return refreshed
+
+    async def get_active_allowed_account_ids(self) -> frozenset[str]:
+        rows = await self._session.scalars(
+            select(OAuthLivePolicyAccount.allowed_account_id)
+            .join(
+                OAuthLivePolicy,
+                OAuthLivePolicy.id == OAuthLivePolicyAccount.policy_id,
+            )
+            .join(Account, Account.id == OAuthLivePolicyAccount.allowed_account_id)
+            .where(
+                OAuthLivePolicy.id == GLOBAL_POLICY_ID,
+                OAuthLivePolicy.is_active.is_(True),
+                Account.status == AccountStatus.ACTIVE,
+            )
+        )
+        return frozenset(rows.all())
+
+    async def rollback(self) -> None:
+        await self._session.rollback()

app/modules/oauth_live/schemas.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from app.modules.shared.schemas import DashboardModel
+
+
+class OAuthLivePolicyUpdateRequest(DashboardModel):
+    is_active: bool
+    allowed_account_ids: list[str]
+
+
+class OAuthLivePolicyResponse(DashboardModel):
+    is_active: bool
+    allowed_account_ids: list[str]
+    created_at: datetime | None
+    updated_at: datetime | None