feat(desktop): pool reset credits and use earliest expiry first

.github/CONTRIBUTING.md

@@ -335,6 +335,8 @@ These rules are intentionally lightweight. They don't require:
 uv run pytest tests/unit/test_proxy_api_responses_contract.py -q
 ```
 
+PostgreSQL tests use a dedicated disposable database named `codex_lb_test`. Set `CODEX_LB_DATABASE_URL` and `CODEX_LB_TEST_DATABASE_URL` to that same database before running tests. Migration tests reset its `public` schema; never use a service database.
+
 ## Release process
 
 Releases are automated via [release-please](https://github.com/googleapis/release-please):

.github/simplicity-budgets.toml

@@ -53,8 +53,8 @@ max_items = 5
 # K2 bridge: constantize-session-bridge-tunables fixed seven never-tuned
 # HTTP session bridge tunables; 96 -> 95 drop-bridge-recovery-modes deleted
 # the ambiguous-continuation recovery mode selector (fail-closed is the only
-# behaviour now).
-max = 95
+# behaviour now); 95 -> 96 adds the opt-in Desktop relay topology.
+max = 96
 
 [root_files]
 # Complete allowlist of tracked repository-root entries (files and

.github/workflows/ci.yml

@@ -521,16 +521,17 @@ jobs:
         env:
           POSTGRES_USER: codex_lb
           POSTGRES_PASSWORD: codex_lb
-          POSTGRES_DB: codex_lb
+          POSTGRES_DB: codex_lb_test
         ports:
           - 5432:5432
         options: >-
-          --health-cmd "pg_isready -U codex_lb -d codex_lb"
+          --health-cmd "pg_isready -U codex_lb -d codex_lb_test"
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
     env:
-      CODEX_LB_TEST_DATABASE_URL: postgresql+asyncpg://codex_lb:codex_lb@127.0.0.1:5432/codex_lb
+      CODEX_LB_DATABASE_URL: postgresql+asyncpg://codex_lb:codex_lb@127.0.0.1:5432/codex_lb_test
+      CODEX_LB_TEST_DATABASE_URL: postgresql+asyncpg://codex_lb:codex_lb@127.0.0.1:5432/codex_lb_test
       PYTHONFAULTHANDLER: "1"
 
     steps:
@@ -606,11 +607,11 @@ jobs:
         env:
           POSTGRES_USER: codex_lb
           POSTGRES_PASSWORD: codex_lb
-          POSTGRES_DB: codex_lb
+          POSTGRES_DB: codex_lb_test
         ports:
           - 5432:5432
         options: >-
-          --health-cmd "pg_isready -U codex_lb -d codex_lb"
+          --health-cmd "pg_isready -U codex_lb -d codex_lb_test"
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5

Makefile

@@ -1,7 +1,10 @@
 PYTEST_ARGS := -q -ra -o faulthandler_timeout=300 -o faulthandler_exit_on_timeout=true --timeout=180 --timeout-method=thread --durations=20
-POSTGRES_TEST_DATABASE_URL ?= postgresql+asyncpg://codex_lb:codex_lb@127.0.0.1:5432/codex_lb
+POSTGRES_TEST_DATABASE_URL ?= postgresql+asyncpg://codex_lb:codex_lb@127.0.0.1:5432/codex_lb_test
 INTEGRATION_CORE_SHARD_COUNT := 3
 POSTGRES_PYTEST_TARGETS := \
+	tests/integration/test_desktop_reset_safety.py::test_redemption_ledger_first_writer_wins_across_sessions \
+	tests/integration/test_desktop_reset_migration.py \
+	tests/integration/test_desktop_reset_invite_migration.py \
 	tests/integration/test_cost_backfill.py \
 	tests/integration/test_atomic_quota_warmup_claims.py \
 	tests/integration/test_report_rollup.py \

app/cli.py

@@ -28,6 +28,9 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
     )
     subparsers = parser.add_subparsers(dest="command")
 
+    desktop_relay = subparsers.add_parser("desktop-relay", help="Run the optional localhost:8000 Desktop relay.")
+    desktop_relay.add_argument("--lb-url", default="http://127.0.0.1:2455", help="Local LB HTTP(S) origin.")
+
     codex_sessions = subparsers.add_parser(
         "codex-sessions",
         help="Manage local Codex session metadata.",
@@ -91,6 +94,15 @@ def _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
 def main(argv: Sequence[str] | None = None) -> None:
     args = _parse_args(argv)
 
+    if args.command == "desktop-relay":
+        from app.modules.desktop_relay.api import run
+
+        try:
+            run(args.lb_url)
+        except ValueError as exc:
+            raise SystemExit(str(exc)) from exc
+        return
+
     if args.command == "codex-sessions":
         if args.codex_sessions_command == "retag":
             _run_codex_sessions_retag(args)
@@ -103,7 +115,9 @@ def main(argv: Sequence[str] | None = None) -> None:
     port = _parse_server_port(args.port)
     timeout_keep_alive = _parse_server_timeout_keep_alive(args.timeout_keep_alive)
     ws_max_size = _parse_server_ws_max_size(args.ws_max_size)
-    os.environ["PORT"] = str(port)
+    from app.core.config.settings import record_http_listener
+
+    record_http_listener(host=args.host, port=port, ssl_certfile=args.ssl_certfile, ssl_keyfile=args.ssl_keyfile)
 
     _run_server(
         "app.main:app",

app/core/clients/usage.py

@@ -362,7 +362,7 @@ def _usage_payload_or_raise(data: JsonObject, status: int) -> UsagePayload:
         )
         raise UsageFetchError(status, message, code=code)
     try:
-        return UsagePayload.model_validate(data)
+        return UsagePayload.from_upstream(data)
     except ValidationError as exc:
         logger.warning(
             "Usage fetch invalid payload request_id=%s",

app/core/config/settings.py

@@ -257,6 +257,32 @@ def _configured_http_port() -> int:
     return 2455
 
 
+def record_http_listener(*, host: str, port: int, ssl_certfile: str | None, ssl_keyfile: str | None) -> None:
+    # Preserve the CLI-selected listener in the process configuration that
+    # startup validation reads, including explicit CLI overrides of env values.
+    os.environ["PORT"] = str(port)
+    os.environ["HOST"] = host
+    os.environ["SSL_CERTFILE"] = ssl_certfile or ""
+    os.environ["SSL_KEYFILE"] = ssl_keyfile or ""
+
+
+def desktop_relay_lb_origin(mode: Literal["off", "loopback", "container"]) -> str | None:
+    if mode == "off":
+        return None
+    message = "Embedded Desktop relay requires a fixed main HTTP port from 1 to 65535, except 8000"
+    try:
+        port = int(os.getenv("PORT", "2455"))
+    except ValueError:
+        raise ValueError(message) from None
+    if not 1 <= port <= 65535 or port == 8000:
+        raise ValueError(message)
+    if os.getenv("HOST", "127.0.0.1") not in {"127.0.0.1", "0.0.0.0", "localhost"}:
+        raise ValueError("Embedded Desktop relay requires main HOST 127.0.0.1, localhost or 0.0.0.0")
+    if os.getenv("SSL_CERTFILE") or os.getenv("SSL_KEYFILE"):
+        raise ValueError("Embedded Desktop relay requires local HTTP; TLS listeners are unsupported")
+    return f"http://127.0.0.1:{port}"
+
+
 def _normalize_cidr_list(value: StringListInput, *, field_name: str, invalid_label: str) -> list[str]:
     if value is None:
         return []
@@ -291,6 +317,7 @@ class Settings(BaseSettings):
     )
 
     data_dir: Path = Field(default_factory=_default_home_dir)
+    desktop_relay_mode: Literal["off", "loopback", "container"] = "off"
     database_url: str = DEFAULT_DATABASE_URL
     # Pool timeout and recycle are fixed constants in ``app/db/session.py``;
     # the background-task engine always derives its pool sizing from the two

app/core/config/tiers.py

@@ -35,6 +35,7 @@
 # "may this value differ between two replicas / must it exist before the DB?".
 SETTING_TIERS: Final[dict[str, Tier]] = {
     "data_dir": "T0",
+    "desktop_relay_mode": "T1",
     "database_url": "T0",
     "database_pool_size": "T1",
     "database_max_overflow": "T1",

app/core/exceptions.py

@@ -49,6 +49,12 @@ class ProxyRateLimitError(AppError):
     error_type = "rate_limit_error"
 
 
+class ProxyConflictError(AppError):
+    status_code = 409
+    code = "conflict"
+    error_type = "invalid_request_error"
+
+
 class ProxyUpstreamError(AppError):
     status_code = 503
     code = "upstream_error"

app/core/handlers/exceptions.py

@@ -28,6 +28,7 @@
     DashboardUpstreamError,
     DashboardValidationError,
     ProxyAuthError,
+    ProxyConflictError,
     ProxyModelNotAllowed,
     ProxyRateLimitError,
     ProxyReasoningEffortNotAllowed,
@@ -57,6 +58,7 @@
 
 _OPENAI_EXCEPTION_TYPES: tuple[type[AppError], ...] = (
     ProxyAuthError,
+    ProxyConflictError,
     ProxyModelNotAllowed,
     ProxyReasoningEffortNotAllowed,
     ProxyRateLimitError,

app/core/usage/models.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
-from pydantic import BaseModel, ConfigDict
+from copy import deepcopy
+
+from pydantic import BaseModel, ConfigDict, PrivateAttr
+
+from app.core.types import JsonObject
 
 
 class UsageWindow(BaseModel):
@@ -44,6 +48,8 @@ class AdditionalRateLimitPayload(BaseModel):
 class UsagePayload(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
+    _raw_payload: JsonObject | None = PrivateAttr(default=None)
+
     plan_type: str | None = None
     workspace_id: str | None = None
     workspace_label: str | None = None
@@ -52,3 +58,14 @@ class UsagePayload(BaseModel):
     credits: CreditsPayload | None = None
     rate_limit_reset_credits: RateLimitResetCreditsPayload | None = None
     additional_rate_limits: list[AdditionalRateLimitPayload] | None = None
+
+    @classmethod
+    def from_upstream(cls, data: JsonObject) -> UsagePayload:
+        """Keep the caller envelope for Desktop without changing parsed serialization."""
+        payload = cls.model_validate(data)
+        payload._raw_payload = deepcopy(data)
+        return payload
+
+    @property
+    def raw_payload(self) -> JsonObject | None:
+        return self._raw_payload

app/db/alembic/versions/20260909_210000_desktop_reset_pool.py

@@ -0,0 +1,38 @@
+"""Persist explicit Desktop reset pooling and immutable redemption selection."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260909_210000_desktop_reset_pool"
+down_revision = "20260909_080000_dashboard_stream_bridge_budgets"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    columns = {column["name"] for column in inspector.get_columns("dashboard_settings")}
+    if "desktop_reset_pool_enabled" not in columns:
+        op.add_column(
+            "dashboard_settings",
+            sa.Column("desktop_reset_pool_enabled", sa.Boolean(), nullable=False, server_default=sa.false()),
+        )
+    if not inspector.has_table("desktop_reset_credit_redemptions"):
+        op.create_table(
+            "desktop_reset_credit_redemptions",
+            sa.Column("caller_account_id", sa.String(), primary_key=True),
+            sa.Column("redeem_request_id", sa.String(), primary_key=True),
+            sa.Column("owner_account_id", sa.String(), nullable=False),
+            sa.Column("owner_chatgpt_account_id", sa.String(), nullable=False),
+            sa.Column("credit_id", sa.String(), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        )
+
+
+def downgrade() -> None:
+    if op.get_bind().execute(sa.text("SELECT 1 FROM desktop_reset_credit_redemptions LIMIT 1")).first():
+        raise RuntimeError(
+            "Disable reset pooling and retain its redemption ledger; existing bindings cannot be discarded"
+        )
+    op.drop_table("desktop_reset_credit_redemptions")
+    op.drop_column("dashboard_settings", "desktop_reset_pool_enabled")

app/db/alembic/versions/20260910_010000_merge_desktop_reset_pool_heads.py

@@ -0,0 +1,21 @@
+"""Merge Desktop reset pooling with the main migration history."""
+
+from __future__ import annotations
+
+revision = "20260910_010000_merge_desktop_reset_pool_heads"
+down_revision = (
+    "20260909_210000_desktop_reset_pool",
+    "20260910_000000_request_logs_missing_cost_index",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Join both completed histories without changing their data."""
+    pass
+
+
+def downgrade() -> None:
+    """Restore both parent stamps without removing schema or bindings."""
+    pass

app/db/alembic/versions/20260910_020000_merge_reset_spool_heads.py

@@ -0,0 +1,21 @@
+"""Join Desktop reset and dashboard spool-retention migration histories."""
+
+from __future__ import annotations
+
+revision = "20260910_020000_merge_reset_spool_heads"
+down_revision = (
+    "20260910_010000_merge_desktop_reset_pool_heads",
+    "20260910_010000_dashboard_spool_retention",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Join existing schemas without changing their data."""
+    pass
+
+
+def downgrade() -> None:
+    """Restore both parent stamps without removing their schemas."""
+    pass

app/db/alembic/versions/20260910_030000_merge_reset_guest_heads.py

@@ -0,0 +1,21 @@
+"""Join reset pooling and guest-session generation without rewriting history."""
+
+from __future__ import annotations
+
+revision = "20260910_030000_merge_reset_guest_heads"
+down_revision = (
+    "20260910_020000_merge_reset_spool_heads",
+    "20260908_000000_add_guest_session_generation",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Join existing schemas while retaining stored authentication state."""
+    pass
+
+
+def downgrade() -> None:
+    """Restore both parent stamps without removing their schemas or data."""
+    pass

app/db/alembic/versions/20260910_040000_merge_reset_dashboard_user_heads.py

@@ -0,0 +1,21 @@
+"""Join reset pooling with dashboard users without rewriting published history."""
+
+from __future__ import annotations
+
+revision = "20260910_040000_merge_reset_dashboard_user_heads"
+down_revision = (
+    "20260910_030000_merge_reset_guest_heads",
+    "20260909_030000_add_audit_actor_columns",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Retain both parent schemas and their stored data."""
+    pass
+
+
+def downgrade() -> None:
+    """Restore both parent stamps without removing their schemas or data."""
+    pass

app/db/alembic/versions/20260910_050000_merge_reset_invite_heads.py

@@ -0,0 +1,21 @@
+"""Join reset pooling and dashboard invitations without rewriting history."""
+
+from __future__ import annotations
+
+revision = "20260910_050000_merge_reset_invite_heads"
+down_revision = (
+    "20260910_040000_merge_reset_dashboard_user_heads",
+    "20260909_040000_add_dashboard_user_invites",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """Retain both parent schemas and stored data."""
+    pass
+
+
+def downgrade() -> None:
+    """Restore the direct parent stamps without deleting data."""
+    pass