fix(usage): normalize live ingest account ids

app/core/clients/proxy.py

@@ -2688,7 +2688,7 @@ async def stream_responses(
                 publish_live_usage(
                     parse_rate_limit_event_text(event_block),
                     account_id=codex_lb_account_id,
-                    chatgpt_account_id=None if codex_lb_account_id else account_id,
+                    chatgpt_account_id=account_id,
                 )
             yield event_block
 
@@ -2859,7 +2859,7 @@ async def _stream_via_http_attempt(
                     publish_live_usage(
                         parse_rate_limit_headers(getattr(raw_resp, "headers", None)),
                         account_id=codex_lb_account_id,
-                        chatgpt_account_id=None if codex_lb_account_id else account_id,
+                        chatgpt_account_id=account_id,
                     )
                 if resp.status >= 400:
                     if raise_for_status:
@@ -2953,7 +2953,7 @@ async def _stream_via_http_attempt(
                 publish_live_usage(
                     parse_rate_limit_headers(getattr(resp, "headers", None)),
                     account_id=codex_lb_account_id,
-                    chatgpt_account_id=None if codex_lb_account_id else account_id,
+                    chatgpt_account_id=account_id,
                 )
             if resp.status >= 400:
                 if raise_for_status:

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_000000_add_anonymous_telemetry
+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_000000_add_anonymous_telemetry"
+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

@@ -2113,6 +2113,7 @@ class HttpBridgeRetryCircuit(Base):
     postgresql_include=["used_percent", "reset_at", "window_minutes", "id"],
 )
 Index("idx_accounts_email", Account.email)
+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/http_bridge/upstream_events.py

@@ -1337,6 +1337,7 @@ async def _relay_http_bridge_upstream_messages(
                         publish_live_usage(
                             parse_rate_limit_event_text(message.text),
                             account_id=session.account.id,
+                            chatgpt_account_id=getattr(session.account, "chatgpt_account_id", None),
                         )
                     await self._process_http_bridge_upstream_text(session, message.text)
                     if await self._retire_http_bridge_after_drain_if_ready(session):

app/modules/usage/live_ingest.py

@@ -68,7 +68,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_cache: dict[str, tuple[str, 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
@@ -82,7 +83,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)
@@ -142,10 +149,21 @@ async def _run(self) -> None:
 
     async def _ingest(self, item: _QueuedSnapshot) -> None:
         account_id = item.account_id
+        raw_account_id = account_id
         if account_id is None:
-            account_id = await self._resolve_account_id(item.chatgpt_account_id)
+            account_id = await self._resolve_account_id_by_chatgpt_account_id(item.chatgpt_account_id)
+        else:
+            resolved_account_id = await self._resolve_account_id(
+                chatgpt_account_id=item.chatgpt_account_id,
+                account_id=account_id,
+            )
+            if resolved_account_id is None:
+                return
+            account_id = resolved_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
 
@@ -236,25 +254,62 @@ async def _invalidate_caches_now(self) -> None:
         # values before the TTL expires.
         await get_rate_limit_headers_cache().invalidate()
 
-    async def _resolve_account_id(self, chatgpt_account_id: str | None) -> str | None:
+    async def _resolve_account_id(
+        self,
+        chatgpt_account_id: str | None,
+        *,
+        account_id: str | None = None,
+    ) -> str | None:
+        if account_id:
+            if chatgpt_account_id:
+                resolved = await self._resolve_account_id_by_chatgpt_account_id(chatgpt_account_id)
+                if resolved is not None:
+                    exact = await self._resolve_account_id_by_account_id(account_id)
+                    if exact is not None and exact != resolved:
+                        return None
+                    return resolved
+            return await self._resolve_account_id_by_account_id(account_id)
+        return await self._resolve_account_id_by_chatgpt_account_id(chatgpt_account_id)
+
+    async def _resolve_account_id_by_chatgpt_account_id(self, chatgpt_account_id: str | None) -> str | None:
         if not chatgpt_account_id:
             return None
-        cached = self._resolution_cache.get(chatgpt_account_id)
-        now = time.monotonic()
-        if cached is not None and now - cached[1] < _RESOLUTION_TTL_SECONDS:
-            return cached[0]
         async with get_background_session() as session:
             rows = (
-                (await session.execute(select(Account.id).where(Account.chatgpt_account_id == chatgpt_account_id)))
+                (
+                    await session.execute(
+                        select(Account.id).where(Account.chatgpt_account_id == chatgpt_account_id).limit(2)
+                    )
+                )
                 .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[chatgpt_account_id] = (resolved, now)
         return resolved
 
+    async def _resolve_account_id_by_account_id(self, account_id: str) -> str | None:
+        cache_key = f"account:{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 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.execute(select(Account.id).where(Account.id == account_id))).scalar_one_or_none()
+        if resolved is not 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.execute(select(Account.id).where(Account.id == account_id))).scalar_one_or_none()
+        return resolved is not None
+
 
 _ingestor: LiveUsageIngestor | None = None
 

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

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

@@ -0,0 +1,22 @@
+## 1. Implementation
+
+- [x] 1.1 Require exact account-row proof for direct internal account IDs.
+- [x] 1.2 Resolve unknown or suffixed internal IDs only through a unique supplied ChatGPT account identity.
+- [x] 1.3 Drop unresolved or ambiguous snapshots without failing the serving path.
+- [x] 1.4 Coalesce duplicate snapshots under the normalized account ID after resolution.
+- [x] 1.5 Revalidate cached exact account IDs and bypass stale raw aliases when a ChatGPT identity is supplied.
+- [x] 1.6 Add an indexed, bounded ChatGPT identity lookup path.
+
+## 2. Regression Coverage
+
+- [x] 2.1 Cover the proxy hub path that publishes both a raw internal ID and a ChatGPT identity.
+- [x] 2.2 Cover that an unproven suffixed internal ID is dropped instead of guessed.
+- [x] 2.3 Cover duplicate coalescing after a raw ID is resolved to the stored account.
+- [x] 2.4 Cover moved ChatGPT identities, conflicting supplied identities, and deleted cached exact IDs.
+
+## 3. Validation
+
+- [x] 3.1 Run focused live usage ingest tests.
+- [x] 3.2 Run Ruff on changed Python files.
+- [x] 3.3 Run strict OpenSpec validation.
+- [x] 3.4 Run migration policy and schema drift validation.