[feat] Session control milestone 2: live events

.gitleaksignore

@@ -158,6 +158,8 @@ ad74134f522cde71f860cb59b6363a8fdf0a64c6:ee/setup_agenta_web.sh:generic-api-key:
 590578c803d94d8ccb1a6ca977471f3d44b43fc3:hosting/helm/oss/templates/config/app-configmap.yaml:generic-api-key:45
 1d8f08b267675726441fcaaae24572bb635c5eac:api/oss/src/utils/env.py:generic-api-key:53
 55f27e52327062382beb299b162f94895268d766:web/oss/public/__ENV.js:generic-api-key:1
+012ae6318c880d19944ffe1ff51740da0613cd8a:services/runner/tests/unit/server.test.ts:generic-api-key:586
+e22c629b3113e522f4fdd918a9b0971e62145056:services/runner/tests/unit/server.test.ts:generic-api-key:889
 c98a5da1a33d2c0986e3c66329eaa5237fbccf3d:hosting/docker-compose/ee/aws/docker-compose.oss.prod.yml:generic-api-key:73
 bf0cd42bffc2581b1df6f56fa6e4b20ff9b68c33:hosting/docker-compose/ee/aws/docker-compose.oss.aws.yml:generic-api-key:61
 52cd40cefd3121eea2e21205e8208712b093529a:core/hosting/docker-compose/ee/docker-compose.dev.yml:generic-api-key:18

api/ee/src/middlewares/throttling.py

@@ -6,6 +6,7 @@
 
 from oss.src.utils.caching import get_cache, set_cache
 from oss.src.utils.logging import get_module_logger
+from oss.src.middlewares.auth import SECRET_RESOLVE_GRANT, request_has_grant
 from oss.src.utils.throttling import Algorithm, check_throttles
 
 from ee.src.core.access.entitlements.types import (
@@ -40,6 +41,14 @@ def _normalize_path(request: Request) -> str:
     return path
 
 
+def _is_runner_record_ingest(request: Request, method: str, path: str) -> bool:
+    return (
+        method == Method.POST.value
+        and path == "/sessions/records/ingest"
+        and request_has_grant(request, SECRET_RESOLVE_GRANT)
+    )
+
+
 def _matches_endpoint(
     method: str,
     path: str,
@@ -168,6 +177,11 @@ async def throttling_middleware(request: Request, call_next):
     if hasattr(request.state, "admin") and request.state.admin:
         return await call_next(request)
 
+    method = request.method.lower()
+    path = _normalize_path(request)
+    if _is_runner_record_ingest(request, method, path):
+        return await call_next(request)
+
     organization_id = (
         request.state.organization_id
         if hasattr(request.state, "organization_id")
@@ -221,10 +235,6 @@ async def throttling_middleware(request: Request, call_next):
     if not throttles:
         return await call_next(request)
 
-    method = request.method.lower()
-
-    path = _normalize_path(request)
-
     # log.debug(
     #     "[throttling] START", org=organization_id, plan=plan, method=method, path=path
     # )

api/ee/tests/pytest/unit/test_throttling.py

@@ -0,0 +1,101 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import Request, Response
+
+from ee.src.core.access.entitlements.types import (
+    Bucket,
+    Category,
+    Mode,
+    Throttle,
+    Tracker,
+)
+from ee.src.middlewares.throttling import throttling_middleware
+from oss.src.middlewares.auth import SECRET_RESOLVE_GRANT
+
+
+def _request(path: str, *, grants: tuple[str, ...] = ()) -> Request:
+    request = Request(
+        {
+            "type": "http",
+            "method": "POST",
+            "path": path,
+            "root_path": "/api" if path.startswith("/api/") else "",
+            "headers": [],
+        }
+    )
+    request.state.organization_id = "organization-1"
+    request.state.token_grants = grants
+    return request
+
+
+async def test_runner_record_ingest_bypasses_plan_throttle():
+    request = _request(
+        "/api/sessions/records/ingest",
+        grants=(SECRET_RESOLVE_GRANT,),
+    )
+    call_next = AsyncMock(return_value=Response(status_code=204))
+
+    with (
+        patch(
+            "ee.src.middlewares.throttling._get_plan", new_callable=AsyncMock
+        ) as get_plan,
+        patch(
+            "ee.src.middlewares.throttling.check_throttles",
+            new_callable=AsyncMock,
+        ) as check_throttles,
+    ):
+        response = await throttling_middleware(request, call_next)
+
+    assert response.status_code == 204
+    call_next.assert_awaited_once_with(request)
+    get_plan.assert_not_awaited()
+    check_throttles.assert_not_awaited()
+
+
+@pytest.mark.parametrize(
+    ("path", "grants"),
+    [
+        ("/sessions/records/ingest", ()),
+        ("/sessions/query", (SECRET_RESOLVE_GRANT,)),
+    ],
+)
+async def test_throttle_still_counts_browser_ingest_and_other_runner_routes(
+    path: str,
+    grants: tuple[str, ...],
+):
+    request = _request(path, grants=grants)
+    call_next = AsyncMock(return_value=Response(status_code=204))
+    standard = Throttle(
+        categories=[Category.STANDARD],
+        mode=Mode.INCLUDE,
+        bucket=Bucket(capacity=10, rate=10),
+    )
+    allowed = SimpleNamespace(
+        allow=True,
+        tokens_remaining=9,
+        retry_after_seconds=0,
+    )
+
+    with (
+        patch(
+            "ee.src.middlewares.throttling._get_plan",
+            new_callable=AsyncMock,
+            return_value="test-plan",
+        ) as get_plan,
+        patch(
+            "ee.src.middlewares.throttling.get_plan_entitlements",
+            return_value={Tracker.THROTTLES: [standard]},
+        ),
+        patch(
+            "ee.src.middlewares.throttling.check_throttles",
+            new_callable=AsyncMock,
+            return_value=[allowed],
+        ) as check_throttles,
+    ):
+        response = await throttling_middleware(request, call_next)
+
+    assert response.status_code == 204
+    get_plan.assert_awaited_once_with("organization-1")
+    check_throttles.assert_awaited_once()

api/entrypoints/worker_streams.py

@@ -27,6 +27,10 @@
 from oss.src.core.secrets.services import VaultService
 from oss.src.core.sessions.interactions.service import SessionInteractionsService
 from oss.src.core.sessions.records.service import RecordsService
+from oss.src.core.sessions.records.streaming import (
+    LIVE_FRAME_STREAM_NAME,
+    RECORD_STREAM_NAME,
+)
 from oss.src.core.tracing.service import TracingService
 from oss.src.dbs.postgres.events.dao import EventsDAO
 from oss.src.dbs.postgres.secrets.dao import SecretsDAO
@@ -37,6 +41,7 @@
 from oss.src.dbs.redis.sessions.watch import SessionsWatchPublisher
 from oss.src.tasks.asyncio.events.worker import EventsWorker
 from oss.src.tasks.asyncio.sessions.records_worker import RecordsWorker
+from oss.src.tasks.asyncio.sessions.live_relay_worker import LiveRelayWorker
 from oss.src.tasks.asyncio.shared.consumer import StreamConsumer
 from oss.src.tasks.asyncio.tracing.worker import TracingWorker
 from oss.src.tasks.asyncio.webhooks.dispatcher import WebhooksDispatcher
@@ -85,7 +90,7 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
     return RecordsWorker(
         service=RecordsService(records_dao=RecordsDAO()),
         redis_client=redis_client,
-        stream_name="streams:records",
+        stream_name=RECORD_STREAM_NAME,
         consumer_group="worker-records",
         # M3 live relay: post-append change notifications on the durable plane,
         # reusing this process's durable connection.
@@ -103,6 +108,14 @@ async def _build_records_worker(redis_client: Redis) -> StreamConsumer:
     )
 
 
+async def _build_live_relay_worker(redis_client: Redis) -> StreamConsumer:
+    return LiveRelayWorker(
+        redis_client=redis_client,
+        stream_name=LIVE_FRAME_STREAM_NAME,
+        consumer_group="worker-session-live-relay",
+    )
+
+
 async def _build_events_worker(redis_client: Redis) -> StreamConsumer:
     events_service = EventsService(events_dao=EventsDAO())
 
@@ -136,6 +149,22 @@ async def _build_events_worker(redis_client: Redis) -> StreamConsumer:
     )
 
 
+async def _initialize_consumer(consumer: StreamConsumer) -> None:
+    await consumer.create_consumer_group()
+    removed = await prune_idle_consumers(
+        url=env.redis.uri_durable,
+        queue_name=consumer.stream_name,
+        consumer_group_name=consumer.consumer_group,
+        keep=consumer.consumer_name,
+    )
+    if removed:
+        log.info(
+            "[STREAMS] Pruned idle consumers",
+            stream=consumer.stream_name,
+            removed=removed,
+        )
+
+
 async def main_async() -> int:
     try:
         streams = _selected_streams()
@@ -165,19 +194,19 @@ async def main_async() -> int:
         ]
 
         for consumer in consumers:
-            await consumer.create_consumer_group()
-            removed = await prune_idle_consumers(
-                url=env.redis.uri_durable,
-                queue_name=consumer.stream_name,
-                consumer_group_name=consumer.consumer_group,
-                keep=consumer.consumer_name,
-            )
-            if removed:
-                log.info(
-                    "[STREAMS] Pruned idle consumers",
-                    stream=consumer.stream_name,
-                    removed=removed,
+            await _initialize_consumer(consumer)
+
+        if env.sessions.shared_reader and "records" in streams:
+            try:
+                live_relay = await _build_live_relay_worker(redis_client)
+                await _initialize_consumer(live_relay)
+            except Exception:
+                log.error(
+                    "[STREAMS] Live relay disabled after initialization failure",
+                    exc_info=True,
                 )
+            else:
+                consumers.append(live_relay)
 
         log.info("[STREAMS] Starting worker-streams", selected=streams)
 

api/oss/databases/postgres/migrations/tracing_oss/versions/oss000000006_add_session_sequence_cursors.py

@@ -0,0 +1,63 @@
+"""add session sequence cursors
+
+Revision ID: oss000000006
+Revises: oss000000005
+Create Date: 2026-09-04 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "oss000000006"
+down_revision: Union[str, None] = "oss000000005"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "session_sequence_cursors",
+        sa.Column("project_id", sa.UUID(), nullable=False),
+        sa.Column("session_id", sa.String(), nullable=False),
+        sa.Column("latest_sequence", sa.BigInteger(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=True,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=True,
+        ),
+        sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
+        sa.Column("created_by_id", sa.UUID(), nullable=True),
+        sa.Column("updated_by_id", sa.UUID(), nullable=True),
+        sa.Column("deleted_by_id", sa.UUID(), nullable=True),
+        sa.PrimaryKeyConstraint("project_id", "session_id"),
+    )
+    op.add_column("records", sa.Column("sequence", sa.BigInteger(), nullable=True))
+    with op.get_context().autocommit_block():
+        op.create_index(
+            "ux_records_session_id_sequence",
+            "records",
+            ["project_id", "session_id", "sequence"],
+            unique=True,
+            postgresql_concurrently=True,
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.drop_index(
+            "ux_records_session_id_sequence",
+            table_name="records",
+            postgresql_concurrently=True,
+        )
+    op.drop_column("records", "sequence")
+    op.drop_table("session_sequence_cursors")