fix(proxy): keep hard continuity owners ahead of affinity

app/db/alembic/versions/20260814_000000_add_accounts_chatgpt_identity_index.py

@@ -0,0 +1,32 @@
+"""add account ChatGPT identity lookup index
+
+Revision ID: 20260814_000000_add_accounts_chatgpt_identity_index
+Revises: 20260806_030000_add_api_key_allowed_reasoning_efforts
+Create Date: 2026-08-14
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260814_000000_add_accounts_chatgpt_identity_index"
+down_revision = "20260806_030000_add_api_key_allowed_reasoning_efforts"
+branch_labels = None
+depends_on = None
+
+_INDEX = "idx_accounts_chatgpt_account_id"
+_TABLE = "accounts"
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        with op.get_context().autocommit_block():
+            op.execute(sa.text(f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {_INDEX} ON {_TABLE} (chatgpt_account_id)"))
+    else:
+        op.execute(sa.text(f"CREATE INDEX IF NOT EXISTS {_INDEX} ON {_TABLE} (chatgpt_account_id)"))
+
+
+def downgrade() -> None:
+    op.drop_index(_INDEX, table_name=_TABLE, if_exists=True)

app/db/models.py

@@ -2154,6 +2154,7 @@ class HttpBridgeRetryCircuit(Base):
     postgresql_where=text("delete_requested_at IS NOT NULL"),
     sqlite_where=text("delete_requested_at IS NOT NULL"),
 )
+Index("idx_accounts_chatgpt_account_id", Account.chatgpt_account_id)
 Index("idx_api_keys_name", ApiKey.name)
 Index("idx_logs_account_time", RequestLog.account_id, RequestLog.requested_at)
 Index("idx_logs_model_source_time", RequestLog.model_source_id, RequestLog.requested_at)

app/modules/proxy/_service/compact.py

@@ -731,12 +731,11 @@ async def _resolve_compact_turn_state_owner(
                 if owner_account_id == resolved_owner and session_identity is not None
             }
             if len(session_identities) > 1:
-                sources = ", ".join(source for source, _account_id, _session_id in owner_refs)
                 raise ProxyResponseError(
                     502,
                     openai_error(
                         "continuity_owner_conflict",
-                        f"Account-owned continuity sources conflict ({sources}); retry the logical turn.",
+                        "Turn-state owner sessions conflict; retry the logical turn.",
                         error_type="server_error",
                     ),
                 )

app/modules/proxy/_service/streaming/retry.py

@@ -1376,16 +1376,27 @@ async def _retry_account_model_rejection(
                         )
                         return
                     if require_preferred_account and preferred_account_id is not None:
+                        error_code = "previous_response_owner_unavailable"
                         message = "Previous response owner account is unavailable; retry later."
+                        reason = "owner_account_unavailable"
+                        upstream_error_code = "no_accounts"
+                        if selection.error_code == "continuity_owner_conflict":
+                            error_code = "continuity_owner_conflict"
+                            message = (
+                                selection.error_message
+                                or "Account-owned continuity sources conflict; retry the logical turn"
+                            )
+                            reason = "owner_conflict"
+                            upstream_error_code = selection.error_code
                         _record_continuity_fail_closed(
                             surface="http_stream",
-                            reason="owner_account_unavailable",
+                            reason=reason,
                             previous_response_id=payload.previous_response_id,
                             session_id=headers.get("x-codex-turn-state") or headers.get("session_id"),
-                            upstream_error_code="no_accounts",
+                            upstream_error_code=upstream_error_code,
                         )
                         event = response_failed_event(
-                            "previous_response_owner_unavailable",
+                            error_code,
                             message,
                             response_id=request_id,
                         )
@@ -1397,7 +1408,7 @@ async def _retry_account_model_rejection(
                             model=payload.model,
                             latency_ms=int((time.monotonic() - start) * 1000),
                             status="error",
-                            error_code="previous_response_owner_unavailable",
+                            error_code=error_code,
                             error_message=message,
                             reasoning_effort=payload.reasoning.effort if payload.reasoning else None,
                             transport=request_transport,
@@ -1516,16 +1527,27 @@ async def _retry_account_model_rejection(
                             request_id,
                         )
                     else:
+                        error_code = "previous_response_owner_unavailable"
                         message = "Previous response owner account is unavailable; retry later."
+                        reason = "owner_account_unavailable"
+                        upstream_error_code = "upstream_unavailable"
+                        if selection.error_code == "continuity_owner_conflict":
+                            error_code = "continuity_owner_conflict"
+                            message = (
+                                selection.error_message
+                                or "Account-owned continuity sources conflict; retry the logical turn"
+                            )
+                            reason = "owner_conflict"
+                            upstream_error_code = selection.error_code
                         _record_continuity_fail_closed(
                             surface="http_stream",
-                            reason="owner_account_unavailable",
+                            reason=reason,
                             previous_response_id=payload.previous_response_id,
                             session_id=headers.get("x-codex-turn-state") or headers.get("session_id"),
-                            upstream_error_code="upstream_unavailable",
+                            upstream_error_code=upstream_error_code,
                         )
                         event = response_failed_event(
-                            "previous_response_owner_unavailable",
+                            error_code,
                             message,
                             response_id=request_id,
                         )
@@ -1537,7 +1559,7 @@ async def _retry_account_model_rejection(
                             model=payload.model,
                             latency_ms=int((time.monotonic() - start) * 1000),
                             status="error",
-                            error_code="previous_response_owner_unavailable",
+                            error_code=error_code,
                             error_message=message,
                             reasoning_effort=payload.reasoning.effort if payload.reasoning else None,
                             transport=request_transport,

app/modules/usage/live_ingest.py

@@ -6,17 +6,22 @@
 import weakref
 from dataclasses import dataclass
 
+from sqlalchemy import select
+
 from app.core import usage as usage_core
 from app.core.config.settings import get_settings
 from app.core.usage.live_hub import register_live_usage_publisher
 from app.core.usage.live_snapshots import LiveRateLimitSnapshot, LiveUsageWindow
+from app.db.models import Account
 from app.db.session import get_background_session
 from app.modules.proxy.account_cache import get_account_selection_cache
 from app.modules.proxy.rate_limit_cache import get_rate_limit_headers_cache
 from app.modules.usage.repository import UsageRepository, UsageWindowWrite
 
 logger = logging.getLogger(__name__)
 
+_RESOLUTION_TTL_SECONDS = 300.0
+
 # Write-coalescing tuning (fixed; issue #1340 / PRINCIPLES.md P2). The
 # ingestor keeps both as constructor fields so tests can exercise queue
 # overflow and coalescing with small values.
@@ -108,6 +113,8 @@ def __init__(
         self._queue: asyncio.Queue[_QueuedSnapshot] = asyncio.Queue(maxsize=max(1, queue_size))
         self._write_min_interval_seconds = write_min_interval_seconds
         self._last_write: dict[str, tuple[tuple[object, ...], float]] = {}
+        self._resolution_cache: dict[str, tuple[str | None, float]] = {}
+        self._resolution_aliases: dict[str, tuple[str, str | None]] = {}
         self._consumer: asyncio.Task[None] | None = None
         self._dropped = 0
         self._last_cache_invalidation = 0.0
@@ -121,7 +128,13 @@ def publish(
         chatgpt_account_id: str | None = None,
     ) -> None:
         item = _QueuedSnapshot(account_id=account_id, chatgpt_account_id=chatgpt_account_id, snapshot=snapshot)
-        if account_id is not None and self._should_skip(account_id, snapshot):
+        coalesce_account_id = account_id
+        if account_id is not None:
+            alias = self._resolution_aliases.get(account_id)
+            if alias is not None and chatgpt_account_id is None:
+                alias_account_id, _alias_chatgpt_account_id = alias
+                coalesce_account_id = alias_account_id
+        if coalesce_account_id is not None and self._should_skip(coalesce_account_id, snapshot):
             return
         try:
             self._queue.put_nowait(item)
@@ -185,6 +198,15 @@ async def _run(self) -> None:
                 )
 
     async def _ingest(self, item: _QueuedSnapshot) -> None:
+        raw_account_id = item.account_id
+        account_id = await self._resolve_persisted_account_id(raw_account_id, item.chatgpt_account_id)
+        if account_id is None:
+            return
+        if raw_account_id and raw_account_id != account_id:
+            self._resolution_aliases[raw_account_id] = (account_id, item.chatgpt_account_id)
+        if self._should_skip(account_id, item.snapshot):
+            return
+
         snapshot = item.snapshot
         primary = snapshot.primary
         secondary = snapshot.secondary
@@ -281,6 +303,73 @@ async def _invalidate_caches_now(self) -> None:
         # values before the TTL expires.
         await get_rate_limit_headers_cache().invalidate()
 
+    async def _resolve_persisted_account_id(
+        self,
+        account_id: str | None,
+        chatgpt_account_id: str | None,
+    ) -> str | None:
+        if account_id is not None:
+            exact = await self._resolve_account_id_by_id(account_id)
+            if exact is not None:
+                if chatgpt_account_id:
+                    resolved = await self._resolve_account_id(chatgpt_account_id)
+                    if resolved is not None and exact != resolved:
+                        return None
+                return exact
+            resolved = await self._resolve_account_id(account_id)
+            if resolved is not None:
+                if chatgpt_account_id and chatgpt_account_id != account_id:
+                    chatgpt_resolved = await self._resolve_account_id(chatgpt_account_id)
+                    if chatgpt_resolved is not None and chatgpt_resolved != resolved:
+                        return None
+                return resolved
+        return await self._resolve_account_id(chatgpt_account_id)
+
+    async def _resolve_account_id_by_id(self, account_id: str | None) -> str | None:
+        if not account_id:
+            return None
+        cache_key = f"id:{account_id}"
+        cached = self._resolution_cache.get(cache_key)
+        now = time.monotonic()
+        if cached is not None and now - cached[1] < _RESOLUTION_TTL_SECONDS:
+            cached_account_id = cached[0]
+            if cached_account_id is None:
+                return None
+            if await self._account_id_exists(cached_account_id):
+                return cached_account_id
+            self._resolution_cache.pop(cache_key, None)
+        async with get_background_session() as session:
+            resolved = await session.scalar(select(Account.id).where(Account.id == account_id))
+        if not isinstance(resolved, str):
+            resolved = None
+        self._resolution_cache[cache_key] = (resolved, now)
+        return resolved
+
+    async def _resolve_account_id_by_account_id(self, account_id: str) -> str | None:
+        return await self._resolve_account_id_by_id(account_id)
+
+    async def _resolve_account_id(self, chatgpt_account_id: str | None) -> str | None:
+        if not chatgpt_account_id:
+            return None
+        cache_key = f"chatgpt:{chatgpt_account_id}"
+        now = time.monotonic()
+        async with get_background_session() as session:
+            rows = (
+                (await session.execute(select(Account.id).where(Account.chatgpt_account_id == chatgpt_account_id)))
+                .scalars()
+                .all()
+            )
+        # Ambiguous identities (multiple workspace slots) are dropped rather
+        # than guessed; the poller stays authoritative for them.
+        resolved = rows[0] if len(rows) == 1 else None
+        self._resolution_cache[cache_key] = (resolved, now)
+        return resolved
+
+    async def _account_id_exists(self, account_id: str) -> bool:
+        async with get_background_session() as session:
+            resolved = await session.scalar(select(Account.id).where(Account.id == account_id))
+        return isinstance(resolved, str)
+
 
 _ingestor: LiveUsageIngestor | None = None
 # Registrations a nested startup displaced, innermost-last. A stack rather

openspec/changes/normalize-live-usage-account-identity/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-14

openspec/changes/normalize-live-usage-account-identity/proposal.md

@@ -0,0 +1,35 @@
+## Why
+
+Live usage snapshots can arrive with either the internal codex-lb account ID
+or the upstream ChatGPT account identity. Some live paths include a
+workspace-local suffix on the internal ID, which does not necessarily identify
+the persisted codex-lb account row. Guessing by string shape can attribute one
+workspace slot's quota state to another slot and break routing decisions.
+
+## What Changes
+
+- Resolve supplied internal account IDs only when the exact persisted account
+  row exists.
+- Resolve suffixed or otherwise unknown internal IDs only through a unique
+  upstream ChatGPT account identity when that identity is also supplied.
+- Drop unresolved or ambiguous live snapshots rather than guessing.
+- Coalesce later snapshots under the resolved account ID after a raw ID has
+  been resolved.
+
+## Capabilities
+
+### New Capabilities
+
+None.
+
+### Modified Capabilities
+
+- `live-usage-ingestion`: live snapshot attribution uses explicit identity
+  evidence and preserves per-account write coalescing after normalization.
+
+## Impact
+
+- Code: live usage ingestor identity resolution and coalescing.
+- Tests: integration coverage for hub-published snapshots, unproven suffix
+  drops, and resolved-alias coalescing.
+- Configuration, schema, response shapes, and background polling are unchanged.

openspec/changes/normalize-live-usage-account-identity/specs/live-usage-ingestion/spec.md

@@ -0,0 +1,80 @@
+# live-usage-ingestion Delta
+
+## ADDED Requirements
+
+### Requirement: Live usage account attribution requires explicit identity proof
+
+Live usage ingestion MUST attribute a snapshot to a persisted account row only
+when the supplied internal account ID exactly matches that row, or when the
+snapshot also supplies a ChatGPT account identity that resolves to exactly one
+persisted account row. A workspace-local suffix, shard suffix, or other string
+shape on an unknown internal account ID MUST NOT by itself justify stripping or
+rewriting the ID.
+
+When an exact internal account ID and a ChatGPT account identity both resolve
+but identify different persisted account rows, the snapshot MUST be dropped
+without failing the proxied request. The ingestor MUST NOT let cached ChatGPT
+identity resolution attribute a snapshot to a row that no longer owns that
+ChatGPT identity, and it MUST revalidate cached exact internal account matches
+before using them for attribution.
+
+When neither identity path resolves to exactly one persisted account row, the
+snapshot MUST be dropped without failing the proxied request. After a raw
+snapshot identity resolves to a persisted account row, duplicate coalescing
+MUST use that resolved account ID so repeated live snapshots do not bypass the
+per-account write interval. When a later snapshot supplies a ChatGPT account
+identity, the ingestor MUST NOT skip that snapshot using a previously cached
+raw-ID alias before revalidating the supplied ChatGPT identity. ChatGPT identity
+lookups MUST use an indexed lookup path and only need to inspect enough rows to
+distinguish a unique match from ambiguity.
+
+#### Scenario: Hub-published snapshot resolves through ChatGPT identity
+
+- **GIVEN** a proxy path publishes a live usage snapshot with a workspace-suffixed internal ID
+- **AND** it also supplies a ChatGPT account identity that maps to exactly one persisted account
+- **WHEN** the live usage ingestor persists the snapshot
+- **THEN** the usage rows are written for the uniquely mapped persisted account
+
+#### Scenario: Unproven suffix is dropped
+
+- **GIVEN** a proxy path publishes a live usage snapshot with an internal ID that does not exactly match a persisted account row
+- **AND** no unique ChatGPT account identity is supplied for that snapshot
+- **WHEN** the live usage ingestor processes it
+- **THEN** no usage rows are written for a guessed prefix account
+- **AND** the proxied request is not failed
+
+#### Scenario: Normalized aliases are coalesced
+
+- **GIVEN** a raw snapshot identity has resolved to a persisted account row
+- **WHEN** the same raw identity publishes an unchanged snapshot inside the write coalescing interval
+- **THEN** the duplicate snapshot is skipped using the resolved account ID
+
+#### Scenario: Conflicting identities are dropped
+
+- **GIVEN** a proxy path publishes a live usage snapshot with an exact internal account ID for one persisted account
+- **AND** it also supplies a ChatGPT account identity that uniquely maps to a different persisted account
+- **WHEN** the live usage ingestor processes it
+- **THEN** no usage rows are written for either conflicting account
+
+#### Scenario: Moved ChatGPT identities are revalidated
+
+- **GIVEN** a ChatGPT account identity previously resolved to one persisted account
+- **AND** that ChatGPT account identity now belongs to a different persisted account
+- **WHEN** a later snapshot uses the same ChatGPT account identity with an unknown internal account ID
+- **THEN** the usage rows are written only for the current persisted account that owns the ChatGPT identity
+
+#### Scenario: Supplied ChatGPT identity bypasses stale raw alias coalescing
+
+- **GIVEN** a raw snapshot identity previously resolved to a persisted account row
+- **AND** an unchanged later snapshot supplies the same raw identity with a ChatGPT account identity
+- **WHEN** that ChatGPT account identity now resolves to a different persisted account row
+- **THEN** the later snapshot is not skipped by the stale raw identity alias
+- **AND** the usage rows are written only for the current persisted account that owns the ChatGPT identity
+
+#### Scenario: Cached exact account matches are revalidated
+
+- **GIVEN** an exact internal account ID previously resolved to a persisted account row
+- **AND** that persisted account row has since been deleted or rewritten
+- **WHEN** a later snapshot uses the same internal account ID
+- **THEN** the ingestor revalidates the cached match before attribution
+- **AND** the stale cached account ID is not used for a usage-history write