feat(proxy): durable http bridge ownership

app/db/alembic/versions/20260321_120000_add_http_bridge_leases.py

@@ -0,0 +1,74 @@
+"""add http_bridge_leases table
+
+Revision ID: 20260321_120000_add_http_bridge_leases
+Revises: 20260320_000000_add_request_log_requested_actual_tiers
+Create Date: 2026-03-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.engine import Connection
+
+# revision identifiers, used by Alembic.
+revision = "20260321_120000_add_http_bridge_leases"
+down_revision = "20260320_000000_add_request_log_requested_actual_tiers"
+branch_labels = None
+depends_on = None
+
+
+def _table_exists(connection: Connection, table_name: str) -> bool:
+    inspector = sa.inspect(connection)
+    return inspector.has_table(table_name)
+
+
+def _index_exists(connection: Connection, index_name: str, table_name: str) -> bool:
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(table_name):
+        return False
+    return any(index["name"] == index_name for index in inspector.get_indexes(table_name))
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if not _table_exists(bind, "http_bridge_leases"):
+        op.create_table(
+            "http_bridge_leases",
+            sa.Column("session_id", sa.String(), primary_key=True),
+            sa.Column("affinity_kind", sa.String(), nullable=False),
+            sa.Column("affinity_key", sa.String(), nullable=False),
+            sa.Column("api_key_scope", sa.String(), nullable=False, server_default=sa.text("''")),
+            sa.Column("owner_instance_id", sa.String(), nullable=False),
+            sa.Column("lease_expires_at", sa.DateTime(), nullable=False),
+            sa.Column("account_id", sa.String(), nullable=True),
+            sa.Column("request_model", sa.String(), nullable=True),
+            sa.Column("codex_session", sa.Boolean(), nullable=False, server_default=sa.false()),
+            sa.Column("idle_ttl_seconds", sa.Float(), nullable=False),
+            sa.Column("upstream_turn_state", sa.String(), nullable=True),
+            sa.Column("downstream_turn_state", sa.String(), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        )
+    if not _index_exists(bind, "ix_http_bridge_leases_owner_expires", "http_bridge_leases"):
+        op.create_index(
+            "ix_http_bridge_leases_owner_expires",
+            "http_bridge_leases",
+            ["owner_instance_id", "lease_expires_at"],
+        )
+    if not _index_exists(bind, "ix_http_bridge_leases_expires", "http_bridge_leases"):
+        op.create_index(
+            "ix_http_bridge_leases_expires",
+            "http_bridge_leases",
+            ["lease_expires_at"],
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if _table_exists(bind, "http_bridge_leases"):
+        if _index_exists(bind, "ix_http_bridge_leases_expires", "http_bridge_leases"):
+            op.drop_index("ix_http_bridge_leases_expires", table_name="http_bridge_leases")
+        if _index_exists(bind, "ix_http_bridge_leases_owner_expires", "http_bridge_leases"):
+            op.drop_index("ix_http_bridge_leases_owner_expires", table_name="http_bridge_leases")
+        op.drop_table("http_bridge_leases")

app/db/alembic/versions/20260322_000000_merge_http_bridge_lease_head.py

@@ -0,0 +1,25 @@
+"""merge http bridge lease head
+
+Revision ID: 20260322_000000_merge_http_bridge_lease_head
+Revises: 20260321_120000_add_http_bridge_leases, 20260321_210000_merge_request_log_tiers_and_dashboard_index_heads
+Create Date: 2026-03-22
+"""
+
+from __future__ import annotations
+
+# revision identifiers, used by Alembic.
+revision = "20260322_000000_merge_http_bridge_lease_head"
+down_revision = (
+    "20260321_120000_add_http_bridge_leases",
+    "20260321_210000_merge_request_log_tiers_and_dashboard_index_heads",
+)
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

app/db/alembic/versions/20260325_120000_enforce_http_bridge_lease_affinity_uniqueness.py

@@ -0,0 +1,73 @@
+"""enforce http bridge lease affinity uniqueness
+
+Revision ID: 20260325_120000_enforce_http_bridge_lease_affinity_uniqueness
+Revises: 20260322_000000_merge_http_bridge_lease_head
+Create Date: 2026-03-25
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.engine import Connection
+
+# revision identifiers, used by Alembic.
+revision = "20260325_120000_enforce_http_bridge_lease_affinity_uniqueness"
+down_revision = "20260322_000000_merge_http_bridge_lease_head"
+branch_labels = None
+depends_on = None
+
+
+def _table_exists(connection: Connection, table_name: str) -> bool:
+    inspector = sa.inspect(connection)
+    return inspector.has_table(table_name)
+
+
+def _index_exists(connection: Connection, index_name: str, table_name: str) -> bool:
+    inspector = sa.inspect(connection)
+    if not inspector.has_table(table_name):
+        return False
+    return any(index["name"] == index_name for index in inspector.get_indexes(table_name))
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if not _table_exists(bind, "http_bridge_leases"):
+        return
+    op.execute(
+        sa.text(
+            """
+            DELETE FROM http_bridge_leases
+            WHERE session_id IN (
+                SELECT session_id
+                FROM (
+                    SELECT
+                        session_id,
+                        ROW_NUMBER() OVER (
+                            PARTITION BY affinity_kind, affinity_key, api_key_scope
+                            ORDER BY lease_expires_at DESC, updated_at DESC, created_at DESC, session_id DESC
+                        ) AS row_num
+                    FROM http_bridge_leases
+                ) ranked_leases
+                WHERE ranked_leases.row_num > 1
+            )
+            """
+        )
+    )
+    if not _index_exists(bind, "ux_http_bridge_leases_affinity_scope", "http_bridge_leases"):
+        op.create_index(
+            "ux_http_bridge_leases_affinity_scope",
+            "http_bridge_leases",
+            ["affinity_kind", "affinity_key", "api_key_scope"],
+            unique=True,
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if _table_exists(bind, "http_bridge_leases") and _index_exists(
+        bind,
+        "ux_http_bridge_leases_affinity_scope",
+        "http_bridge_leases",
+    ):
+        op.drop_index("ux_http_bridge_leases_affinity_scope", table_name="http_bridge_leases")

app/db/models.py

@@ -157,6 +157,41 @@ class StickySession(Base):
     )
 
 
+class HttpBridgeLease(Base):
+    __tablename__ = "http_bridge_leases"
+    __table_args__ = (
+        Index(
+            "ux_http_bridge_leases_affinity_scope",
+            "affinity_kind",
+            "affinity_key",
+            "api_key_scope",
+            unique=True,
+        ),
+        Index("ix_http_bridge_leases_owner_expires", "owner_instance_id", "lease_expires_at"),
+        Index("ix_http_bridge_leases_expires", "lease_expires_at"),
+    )
+
+    session_id: Mapped[str] = mapped_column(String, primary_key=True)
+    affinity_kind: Mapped[str] = mapped_column(String, nullable=False)
+    affinity_key: Mapped[str] = mapped_column(String, nullable=False)
+    api_key_scope: Mapped[str] = mapped_column(String, nullable=False, default="", server_default=text("''"))
+    owner_instance_id: Mapped[str] = mapped_column(String, nullable=False)
+    lease_expires_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    account_id: Mapped[str | None] = mapped_column(String, nullable=True)
+    request_model: Mapped[str | None] = mapped_column(String, nullable=True)
+    codex_session: Mapped[bool] = mapped_column(Boolean, default=False, server_default=false(), nullable=False)
+    idle_ttl_seconds: Mapped[float] = mapped_column(Float, nullable=False)
+    upstream_turn_state: Mapped[str | None] = mapped_column(String, nullable=True)
+    downstream_turn_state: Mapped[str | None] = mapped_column(String, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+
 class DashboardSettings(Base):
     __tablename__ = "dashboard_settings"
 

app/dependencies.py

@@ -19,6 +19,7 @@
 from app.modules.firewall.repository import FirewallRepository
 from app.modules.firewall.service import FirewallService
 from app.modules.oauth.service import OauthService
+from app.modules.proxy.bridge_repository import HttpBridgeLeasesRepository
 from app.modules.proxy.repo_bundle import ProxyRepositories
 from app.modules.proxy.service import ProxyService
 from app.modules.proxy.sticky_repository import StickySessionsRepository
@@ -151,6 +152,7 @@ async def _proxy_repo_context() -> AsyncIterator[ProxyRepositories]:
             usage=UsageRepository(session),
             request_logs=RequestLogsRepository(session),
             sticky_sessions=StickySessionsRepository(session),
+            http_bridge_leases=HttpBridgeLeasesRepository(session),
             api_keys=ApiKeysRepository(session),
             additional_usage=AdditionalUsageRepository(session),
         )

app/modules/proxy/api.py

@@ -445,13 +445,9 @@ async def _stream_responses(
     rate_limit_headers = await context.service.rate_limit_headers()
     bridge_active = prefer_http_bridge and proxy_service_module.get_settings().http_responses_session_bridge_enabled
     downstream_turn_state = (
-        proxy_service_module.ensure_http_downstream_turn_state(request.headers) if bridge_active else None
-    )
-    turn_state_headers = (
-        proxy_service_module.build_downstream_turn_state_response_headers(downstream_turn_state)
-        if downstream_turn_state is not None
-        else {}
+        proxy_service_module.requested_http_downstream_turn_state(request.headers) if bridge_active else None
     )
+    turn_state_headers: dict[str, str] = {}
     payload.stream = True
     if prefer_http_bridge:
         stream = context.service.stream_http_responses(
@@ -464,6 +460,7 @@ async def _stream_responses(
             api_key_reservation=reservation,
             suppress_text_done_events=suppress_text_done_events,
             downstream_turn_state=downstream_turn_state,
+            response_headers_out=turn_state_headers,
         )
     else:
         stream = context.service.stream_responses(
@@ -482,15 +479,15 @@ async def _stream_responses(
         return StreamingResponse(
             _prepend_first(None, stream),
             media_type="text/event-stream",
-            headers={"Cache-Control": "no-cache", **rate_limit_headers},
+            headers={"Cache-Control": "no-cache", **turn_state_headers, **rate_limit_headers},
         )
     except ProxyResponseError as exc:
         await _release_reservation(reservation)
         return _logged_error_json_response(
             request,
             exc.status_code,
             exc.payload,
-            headers=rate_limit_headers,
+            headers={**turn_state_headers, **rate_limit_headers},
         )
     return StreamingResponse(
         _prepend_first(first, stream),
@@ -521,13 +518,9 @@ async def _collect_responses(
     rate_limit_headers = await context.service.rate_limit_headers()
     bridge_active = prefer_http_bridge and proxy_service_module.get_settings().http_responses_session_bridge_enabled
     downstream_turn_state = (
-        proxy_service_module.ensure_http_downstream_turn_state(request.headers) if bridge_active else None
-    )
-    turn_state_headers = (
-        proxy_service_module.build_downstream_turn_state_response_headers(downstream_turn_state)
-        if downstream_turn_state is not None
-        else {}
+        proxy_service_module.requested_http_downstream_turn_state(request.headers) if bridge_active else None
     )
+    turn_state_headers: dict[str, str] = {}
     payload.stream = True
     if prefer_http_bridge:
         stream = context.service.stream_http_responses(
@@ -540,6 +533,7 @@ async def _collect_responses(
             api_key_reservation=reservation,
             suppress_text_done_events=suppress_text_done_events,
             downstream_turn_state=downstream_turn_state,
+            response_headers_out=turn_state_headers,
         )
     else:
         stream = context.service.stream_responses(
@@ -561,7 +555,7 @@ async def _collect_responses(
             request,
             exc.status_code,
             error.model_dump(mode="json", exclude_none=True),
-            headers=rate_limit_headers,
+            headers={**turn_state_headers, **rate_limit_headers},
         )
     if isinstance(response_payload, OpenAIResponsePayload):
         if response_payload.status == "failed":