fix(ci): exempt the sanctioned socket research-id boundary from the int ban

_socket_research_id.py is the sole sanctioned converter between legacy
integer benchmark IDs and socket subscription keys (PR #5658). The
check-research-id-type hook still flags int(research_id) everywhere else.

.pre-commit-hooks/check-research-id-type.py

@@ -12,6 +12,11 @@ from pathlib import Path
 # Set environment variable for pre-commit hooks to allow unencrypted databases
 os.environ["LDR_ALLOW_UNENCRYPTED"] = "true"
 
+# Sole sanctioned converter between legacy integer benchmark IDs and socket
+# subscription keys (see _socket_research_id.py). Every other call site must
+# treat research IDs as UUID strings.
+_INT_CONVERSION_SANCTIONED_FILES = {"_socket_research_id.py"}
+
 
 def check_file(filepath):
     """Check a single file for incorrect research_id patterns."""
@@ -89,6 +94,9 @@ def main():
         ):
             continue
 
+        if Path(filepath).name in _INT_CONVERSION_SANCTIONED_FILES:
+            continue
+
         errors = check_file(filepath)
         all_errors.extend(errors)
 

changelog.d/+socket-id-validation.security.md

@@ -0,0 +1 @@
+Canonicalized WebSocket research IDs and isolated benchmark progress subscriptions by authenticated user, preventing same-numbered runs in different user databases from sharing updates.

src/local_deep_research/benchmarks/web_api/benchmark_service.py

@@ -348,6 +348,7 @@ class BenchmarkService:
                 # Extract all data we need
                 benchmark_data = {
                     "benchmark_run_id": benchmark_run_id,
+                    "owner_username": username,
                     "username": username or "benchmark_user",
                     "user_password": _user_password,  # Add password for metrics tracking
                     "config_hash": benchmark_run.config_hash,
@@ -436,6 +437,7 @@ class BenchmarkService:
             # Set up settings context for thread-local access
             settings_snapshot = benchmark_data.get("settings_snapshot", {})
             username = benchmark_data.get("username", "benchmark_user")
+            owner_username = benchmark_data.get("owner_username")
 
             # Create a settings context that threads can use
             settings_context = SnapshotSettingsContext(
@@ -502,6 +504,7 @@ class BenchmarkService:
                 try:
                     # Add username and password to task for metrics tracking
                     task["username"] = benchmark_data.get("username")
+                    task["owner_username"] = owner_username
                     task["user_password"] = benchmark_data.get("user_password")
 
                     # Acquire the global research semaphore so benchmark
@@ -535,6 +538,7 @@ class BenchmarkService:
                         benchmark_run_id,
                         progress_info["completed_examples"],
                         progress_info["total_examples"],
+                        owner_username=owner_username,
                     )
 
                 except Exception as e:
@@ -561,6 +565,7 @@ class BenchmarkService:
                                 "rate_limit_detected": True,
                                 "message": "SearXNG rate limiting detected",
                             },
+                            owner_username=owner_username,
                         )
 
             # Mark as completed in memory tracker
@@ -612,6 +617,7 @@ class BenchmarkService:
                     else 0,
                     "benchmark_run_id": benchmark_run_id,
                 },
+                owner_username=owner_username,
             )
 
         except Exception as e:
@@ -773,6 +779,7 @@ class BenchmarkService:
                         "research_progress",
                         task["benchmark_run_id"],
                         progress_data,
+                        owner_username=task.get("owner_username"),
                     )
 
                 except Exception:
@@ -1216,7 +1223,12 @@ class BenchmarkService:
             run_data["result_persistence_failed"] = True
 
     def _send_progress_update(
-        self, benchmark_run_id: int, completed: int, total: int
+        self,
+        benchmark_run_id: int,
+        completed: int,
+        total: int,
+        *,
+        owner_username: str | None = None,
     ):
         """Send real-time progress update via websocket."""
         try:
@@ -1248,7 +1260,10 @@ class BenchmarkService:
             }
 
             self.socket_service.emit_to_subscribers(
-                "research_progress", benchmark_run_id, progress_data
+                "research_progress",
+                benchmark_run_id,
+                progress_data,
+                owner_username=owner_username,
             )
 
         except Exception:

src/local_deep_research/web/services/_socket_research_id.py

@@ -0,0 +1,67 @@
+import re
+from typing import Final, assert_never
+
+
+_CANONICAL_UUID_RE: Final = re.compile(
+    r"[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}"
+)
+_MAX_BENCHMARK_ID: Final = 9223372036854775807
+_MAX_BENCHMARK_ID_LENGTH: Final = len(str(_MAX_BENCHMARK_ID))
+
+type _JsonValue = (
+    None | bool | int | float | str | list[_JsonValue] | dict[str, _JsonValue]
+)
+type SocketResearchId = str | int
+type SocketSubscriptionKey = str | tuple[str, int]
+
+
+def canonicalize_socket_research_id(
+    research_id: SocketResearchId,
+) -> SocketResearchId:
+    match research_id:
+        case str() if len(research_id) == 36 and _CANONICAL_UUID_RE.fullmatch(
+            research_id
+        ):
+            return research_id.lower()
+        case str() | int():
+            return research_id
+        case unreachable:
+            assert_never(unreachable)
+
+
+def socket_subscription_key(
+    research_id: SocketResearchId, owner_username: str | None
+) -> SocketSubscriptionKey | None:
+    canonical_id = canonicalize_socket_research_id(research_id)
+    match canonical_id:
+        case str():
+            return canonical_id
+        case int():
+            if not owner_username:
+                return None
+            return (owner_username, canonical_id)
+        case unreachable:
+            assert_never(unreachable)
+
+
+def parse_socket_research_id(data: _JsonValue) -> SocketResearchId | None:
+    if not isinstance(data, dict):
+        return None
+
+    research_id = data.get("research_id")
+    if isinstance(research_id, bool):
+        return None
+    if isinstance(research_id, int):
+        return research_id if 1 <= research_id <= _MAX_BENCHMARK_ID else None
+    if not isinstance(research_id, str):
+        return None
+
+    length = len(research_id)
+    if length == 36 and _CANONICAL_UUID_RE.fullmatch(research_id):
+        return canonicalize_socket_research_id(research_id)
+    if not 1 <= length <= _MAX_BENCHMARK_ID_LENGTH:
+        return None
+    if not research_id.isascii() or not research_id.isdecimal():
+        return None
+    benchmark_id = int(research_id)
+    return benchmark_id if 1 <= benchmark_id <= _MAX_BENCHMARK_ID else None

src/local_deep_research/web/services/socket_service.py

@@ -9,6 +9,13 @@ from ...constants import ResearchStatus
 from ...database.encrypted_db import db_manager
 from ...database.session_passwords import session_password_store
 from ..routes.globals import get_active_research_snapshot
+from ._socket_research_id import (
+    SocketResearchId,
+    SocketSubscriptionKey,
+    canonicalize_socket_research_id,
+    parse_socket_research_id,
+    socket_subscription_key,
+)
 
 
 def _install_origin_rejection_logging(socketio: SocketIO) -> bool:
@@ -138,8 +145,9 @@ class SocketIOService:
         if socketio_cors != "*":
             _install_origin_rejection_logging(self.__socketio)
 
-        # Socket subscription tracking.
-        self.__socket_subscriptions: dict[str, Any] = {}
+        # UUID subscriptions use the canonical UUID directly. Benchmark
+        # subscriptions use (trusted owner username, integer run id).
+        self.__socket_subscriptions: dict[SocketSubscriptionKey, set[str]] = {}
         # Set to false to disable logging in the event handlers. This can
         # be necessary because it will sometimes run the handlers directly
         # during a call to `emit` that was made in a logging handler.
@@ -251,7 +259,13 @@ class SocketIOService:
             return False
 
     def emit_to_subscribers(
-        self, event_base, research_id, data, enable_logging: bool = True
+        self,
+        event_base,
+        research_id: SocketResearchId,
+        data,
+        enable_logging: bool = True,
+        *,
+        owner_username: str | None = None,
     ):
         """
         Emit an event to all subscribers of a specific research.
@@ -263,6 +277,8 @@ class SocketIOService:
             enable_logging: If set to false, this will disable all logging,
                 which is useful if we are calling this inside of a logging
                 handler.
+            owner_username: Trusted owner for integer benchmark IDs. UUIDs do
+                not require owner context.
 
         Returns:
             bool: True if emission was successful, False otherwise
@@ -272,12 +288,20 @@ class SocketIOService:
             self.__logging_enabled = False
 
         try:
-            full_event = f"{event_base}_{research_id}"
+            canonical_id = canonicalize_socket_research_id(research_id)
+            subscription_key = socket_subscription_key(
+                canonical_id, owner_username
+            )
+            if subscription_key is None:
+                return True
+            full_event = f"{event_base}_{canonical_id}"
 
             # Emit only to specific subscribers (no broadcast) to avoid
             # duplicate messages and reduce server load under concurrency
             with self.__lock:
-                subscriptions = self.__socket_subscriptions.get(research_id)
+                subscriptions = self.__socket_subscriptions.get(
+                    subscription_key
+                )
                 if subscriptions:
                     subscriptions = (
                         subscriptions.copy()
@@ -321,13 +345,22 @@ class SocketIOService:
         finally:
             self.__logging_enabled = True
 
-    def remove_subscriptions_for_research(self, research_id: str) -> None:
-        """Remove all socket subscriptions for a completed research."""
+    def remove_subscriptions_for_research(
+        self,
+        research_id: SocketResearchId,
+        *,
+        owner_username: str | None = None,
+    ) -> None:
+        """Remove subscriptions for one research and, for benchmarks, owner."""
+        canonical_id = canonicalize_socket_research_id(research_id)
+        subscription_key = socket_subscription_key(canonical_id, owner_username)
+        if subscription_key is None:
+            return
         with self.__lock:
-            removed = self.__socket_subscriptions.pop(research_id, None)
+            removed = self.__socket_subscriptions.pop(subscription_key, None)
         if removed is not None:
             self.__log_info(
-                f"Removed {len(removed)} subscription(s) for research {research_id}"
+                f"Removed {len(removed)} subscription(s) for research {canonical_id}"
             )
 
     def __disconnect_room(self, room: str, description: str) -> int:
@@ -649,8 +682,7 @@ class SocketIOService:
                 f"Client {request.sid} disconnected because: {reason}"
             )
             # Clean up subscriptions for this client.
-            # __socket_subscriptions is keyed by research_id → set of sids,
-            # so we iterate all entries and discard the disconnecting sid.
+            # Iterate all subscription keys and discard the disconnecting sid.
             with self.__lock:
                 self.__sid_sessions.pop(request.sid, None)
                 empty_keys = []
@@ -682,8 +714,8 @@ class SocketIOService:
 
     def __handle_subscribe(self, data, request):
         """Handle client subscription to research updates."""
-        research_id = data.get("research_id")
-        if not research_id:
+        research_id = parse_socket_research_id(data)
+        if research_id is None:
             return
 
         # Verify the connected user actually owns this research before
@@ -712,10 +744,13 @@ class SocketIOService:
             )
             return
 
+        subscription_key = socket_subscription_key(research_id, username)
+        if subscription_key is None:
+            return
         with self.__lock:
-            if research_id not in self.__socket_subscriptions:
-                self.__socket_subscriptions[research_id] = set()
-            self.__socket_subscriptions[research_id].add(request.sid)
+            if subscription_key not in self.__socket_subscriptions:
+                self.__socket_subscriptions[subscription_key] = set()
+            self.__socket_subscriptions[subscription_key].add(request.sid)
         self.__log_info(
             f"Client {request.sid} subscribed to research {research_id}"
         )
@@ -739,7 +774,9 @@ class SocketIOService:
                 )
 
     @staticmethod
-    def _user_owns_research(username: str, research_id: str) -> bool:
+    def _user_owns_research(
+        username: str, research_id: SocketResearchId
+    ) -> bool:
         """Return True if the given user owns this research / benchmark id.
 
         Used as the authorization boundary for WebSocket subscriptions —
@@ -756,34 +793,21 @@ class SocketIOService:
         """
         try:
             from ...database.session_context import get_user_db_session
-            from ...database.models import ResearchHistory
+
+            if isinstance(research_id, str):
+                from ...database.models import ResearchHistory
+
+                model_id = ResearchHistory.id
+            else:
+                from ...database.models.benchmark import BenchmarkRun
+
+                model_id = BenchmarkRun.id
 
             with get_user_db_session(username) as db:
-                if (
-                    db.query(ResearchHistory.id)
-                    .filter(ResearchHistory.id == research_id)
-                    .first()
+                return (
+                    db.query(model_id).filter(model_id == research_id).first()
                     is not None
-                ):
-                    return True
-
-                # Benchmark pages subscribe with their BenchmarkRun.id.
-                # Recognize the user's own benchmark runs so the ownership
-                # gate doesn't drop benchmark live progress (regression vs.
-                # the removed cross-user broadcast). research_id stays a
-                # string (never coerced to int — IDs are strings/UUIDs
-                # repo-wide); SQLite applies numeric affinity to match the
-                # Integer column. Only attempt this for numeric ids.
-                if str(research_id).isdigit():
-                    from ...database.models.benchmark import BenchmarkRun
-
-                    return (
-                        db.query(BenchmarkRun.id)
-                        .filter(BenchmarkRun.id == research_id)
-                        .first()
-                        is not None
-                    )
-                return False
+                )
         except Exception:
             # Conservative: deny on any DB-open or query failure so a
             # transient infra error never silently widens authz.
@@ -794,10 +818,8 @@ class SocketIOService:
 
     def __handle_unsubscribe(self, data, request):
         """Handle client unsubscribe from research updates."""
-        research_id = (
-            data.get("research_id") if isinstance(data, dict) else None
-        )
-        if not research_id:
+        research_id = parse_socket_research_id(data)
+        if research_id is None:
             return
 
         # Symmetric with __handle_subscribe: require the caller to own the
@@ -822,14 +844,17 @@ class SocketIOService:
             )
             return
 
+        subscription_key = socket_subscription_key(research_id, username)
+        if subscription_key is None:
+            return
         with self.__lock:
-            subs = self.__socket_subscriptions.get(research_id)
+            subs = self.__socket_subscriptions.get(subscription_key)
             if subs:
                 subs.discard(request.sid)
                 # Prune empty sets so the dict doesn't grow unbounded with
                 # stale research_ids over long server runtimes.
                 if not subs:
-                    self.__socket_subscriptions.pop(research_id, None)
+                    self.__socket_subscriptions.pop(subscription_key, None)
         self.__log_info(
             f"Client {request.sid} unsubscribed from research {research_id}"
         )

tests/benchmarks/test_benchmark_service.py

@@ -915,6 +915,10 @@ class TestBenchmarkServiceStartBenchmark:
                                 service.active_runs[1]["data"]["username"]
                                 == "testuser"
                             )
+                            assert (
+                                service.active_runs[1]["data"]["owner_username"]
+                                == "testuser"
+                            )
 
     def test_start_benchmark_handles_not_found(self):
         """Test that start_benchmark handles benchmark not found."""

tests/benchmarks/test_benchmark_socket_ownership.py

@@ -0,0 +1,157 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from local_deep_research.benchmarks.web_api.benchmark_service import (
+    BenchmarkService,
+)
+
+
+BENCHMARK_MODULE = "local_deep_research.benchmarks.web_api.benchmark_service"
+SETTINGS_MODULE = "local_deep_research.config.thread_settings"
+
+
+@pytest.fixture
+def benchmark_service():
+    socket = MagicMock()
+    return BenchmarkService(socket_service=socket), socket
+
+
+def test_send_progress_update_passes_trusted_owner(benchmark_service):
+    # Given
+    service, socket = benchmark_service
+
+    # When
+    service._send_progress_update(
+        7, completed=1, total=2, owner_username="alice"
+    )
+
+    # Then
+    assert socket.emit_to_subscribers.call_args.kwargs == {
+        "owner_username": "alice"
+    }
+
+
+@patch(f"{BENCHMARK_MODULE}._global_research_semaphore")
+@patch(f"{SETTINGS_MODULE}.set_settings_context")
+def test_thread_progress_and_completion_pass_trusted_owner(
+    mock_set_context,
+    mock_semaphore,
+    benchmark_service,
+):
+    # Given
+    service, socket = benchmark_service
+    service.active_runs[7] = {
+        "data": {
+            "username": "alice",
+            "owner_username": "alice",
+            "user_password": None,
+            "datasets_config": {},
+            "search_config": {},
+            "evaluation_config": {},
+            "settings_snapshot": {},
+        },
+        "results": [],
+    }
+    task = {"benchmark_run_id": 7, "example_id": "example-1", "task_index": 0}
+
+    # When
+    with (
+        patch.object(service, "_create_task_queue", return_value=[task]),
+        patch.object(service, "_process_benchmark_task", return_value={}),
+        patch.object(service, "_send_progress_update") as send_progress,
+        patch.object(service, "_sync_results_to_database"),
+    ):
+        service._run_benchmark_thread(7)
+
+    # Then
+    send_progress.assert_called_once_with(7, 1, 1, owner_username="alice")
+    assert socket.emit_to_subscribers.call_args.kwargs == {
+        "owner_username": "alice"
+    }
+
+
+@patch(f"{BENCHMARK_MODULE}._global_research_semaphore")
+@patch(f"{SETTINGS_MODULE}.set_settings_context")
+def test_rate_limit_and_completion_pass_trusted_owner(
+    mock_set_context,
+    mock_semaphore,
+    benchmark_service,
+):
+    # Given
+    service, socket = benchmark_service
+    service.active_runs[7] = {
+        "data": {
+            "username": "alice",
+            "owner_username": "alice",
+            "user_password": None,
+            "datasets_config": {},
+            "search_config": {},
+            "evaluation_config": {},
+            "settings_snapshot": {},
+        },
+        "results": [],
+    }
+    task = {"benchmark_run_id": 7, "example_id": "example-1", "task_index": 0}
+
+    # When
+    with (
+        patch.object(service, "_create_task_queue", return_value=[task]),
+        patch.object(
+            service,
+            "_process_benchmark_task",
+            side_effect=RuntimeError("403 rate limit"),
+        ),
+        patch.object(service, "_sync_results_to_database"),
+    ):
+        service._run_benchmark_thread(7)
+
+    # Then
+    assert socket.emit_to_subscribers.call_count == 2
+    assert all(
+        call.kwargs == {"owner_username": "alice"}
+        for call in socket.emit_to_subscribers.call_args_list
+    )
+
+
+def test_detailed_progress_callback_passes_trusted_owner(benchmark_service):
+    # Given
+    service, socket = benchmark_service
+    task = {
+        "benchmark_run_id": 7,
+        "owner_username": "alice",
+        "username": "benchmark_user",
+        "example_id": "example-1",
+        "dataset_type": "simpleqa",
+        "question": "What is 2+2?",
+        "correct_answer": "4",
+        "query_hash": "query-hash",
+        "task_index": 0,
+    }
+    settings = MagicMock(snapshot={})
+
+    # When
+    with (
+        patch(f"{SETTINGS_MODULE}.get_settings_context", return_value=settings),
+        patch(f"{BENCHMARK_MODULE}.format_query", return_value="question"),
+        patch(
+            f"{BENCHMARK_MODULE}.quick_summary",
+            return_value={"summary": "4", "sources": []},
+        ) as quick_summary,
+        patch(
+            f"{BENCHMARK_MODULE}.extract_answer_from_response",
+            return_value={"extracted_answer": "4"},
+        ),
+        patch(
+            f"{BENCHMARK_MODULE}.grade_single_result",
+            return_value={"is_correct": True},
+        ),
+    ):
+        service._process_benchmark_task(task, {}, {})
+        progress_callback = quick_summary.call_args.kwargs["progress_callback"]
+        progress_callback("Searching", 25, {"phase": "search"})
+
+    # Then
+    assert socket.emit_to_subscribers.call_args.kwargs == {
+        "owner_username": "alice"
+    }

tests/chat/test_chat_socket_events.py

@@ -15,6 +15,9 @@ import pytest
 from local_deep_research.web.services.socket_service import SocketIOService
 
 
+VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
+
+
 @pytest.fixture(autouse=True)
 def _reset_socketio_singleton():
     """Reset ``SocketIOService._instance`` around every test.
@@ -145,15 +148,15 @@ class TestSocketIOHandlers:
             mock_request = MagicMock()
             mock_request.sid = "subscribing-client"
 
-            data = {"research_id": "research-abc-123"}
+            data = {"research_id": VALID_RESEARCH_ID}
 
             # Call subscribe handler
             service._SocketIOService__handle_subscribe(data, mock_request)
 
             # Verify subscription was added
             subs = service._SocketIOService__socket_subscriptions
-            assert "research-abc-123" in subs
-            assert "subscribing-client" in subs["research-abc-123"]
+            assert VALID_RESEARCH_ID in subs
+            assert "subscribing-client" in subs[VALID_RESEARCH_ID]
 
     def test_on_subscribe_sends_current_status_if_available(self):
         """Test that subscribe sends current status when research is active."""
@@ -197,7 +200,7 @@ class TestSocketIOHandlers:
                 mock_request = MagicMock()
                 mock_request.sid = "subscriber-client"
 
-                data = {"research_id": "active-research-1"}
+                data = {"research_id": VALID_RESEARCH_ID}
 
                 # Call subscribe handler
                 service._SocketIOService__handle_subscribe(data, mock_request)
@@ -296,7 +299,7 @@ class TestSocketIOHandlers:
             service = SocketIOService(app=mock_app)
 
             errors = []
-            research_id = "concurrent-research"
+            research_id = VALID_RESEARCH_ID
 
             def subscribe_client(client_id):
                 try:
@@ -372,14 +375,14 @@ class TestSocketIOHandlers:
             mock_request.sid = "failing-client"
 
             service._SocketIOService__handle_subscribe(
-                {"research_id": "error-research"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
             # Subscription should still be added despite emit error
             # (emit error is for sending current status, not for adding to set)
             subs = service._SocketIOService__socket_subscriptions
-            assert "error-research" in subs
-            assert "failing-client" in subs["error-research"]
+            assert VALID_RESEARCH_ID in subs
+            assert "failing-client" in subs[VALID_RESEARCH_ID]
 
             # Now disconnect - should clean up even with errors
             with patch(
@@ -389,9 +392,5 @@ class TestSocketIOHandlers:
                     mock_request, "close"
                 )
 
-            # __handle_disconnect is keyed by research_id -> set of sids: it
-            # discards the disconnecting sid from every set and drops any key
-            # whose set is now empty. "failing-client" was the sole subscriber
-            # to "error-research", so the entire key is removed.
-            assert "failing-client" not in subs.get("error-research", set())
-            assert "error-research" not in subs
+            assert "failing-client" not in subs.get(VALID_RESEARCH_ID, set())
+            assert VALID_RESEARCH_ID not in subs

tests/web/services/test_socket_service.py

@@ -11,6 +11,10 @@ Tests cover:
 from unittest.mock import MagicMock, patch
 
 
+VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
+SECOND_VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440001"
+
+
 class MockFlaskApp:
     """Mock Flask application for testing."""
 
@@ -955,10 +959,13 @@ class TestSubscribeSessionRevalidation:
                 return_value=None,
             ):
                 service._SocketIOService__handle_subscribe(
-                    {"research_id": "r1"}, mock_request
+                    {"research_id": VALID_RESEARCH_ID}, mock_request
                 )
 
-            assert "r1" not in service._SocketIOService__socket_subscriptions
+            assert (
+                VALID_RESEARCH_ID
+                not in service._SocketIOService__socket_subscriptions
+            )
             mock_socketio.server.disconnect.assert_any_call(
                 "ghost-sid", namespace="/"
             )
@@ -989,7 +996,7 @@ class TestSubscribeSessionRevalidation:
             service = SocketIOService(app=MockFlaskApp())
 
             subs = service._SocketIOService__socket_subscriptions
-            subs["r1"] = {"legit-sid"}
+            subs[VALID_RESEARCH_ID] = {"legit-sid"}
 
             mock_request = MagicMock()
             mock_request.sid = "ghost-sid"
@@ -1000,11 +1007,11 @@ class TestSubscribeSessionRevalidation:
                 return_value=None,
             ):
                 service._SocketIOService__handle_unsubscribe(
-                    {"research_id": "r1"}, mock_request
+                    {"research_id": VALID_RESEARCH_ID}, mock_request
                 )
 
             # Rejected before any mutation — legit subscriber untouched.
-            assert subs["r1"] == {"legit-sid"}
+            assert subs[VALID_RESEARCH_ID] == {"legit-sid"}
             # But the offending socket is severed.
             mock_socketio.server.disconnect.assert_any_call(
                 "ghost-sid", namespace="/"
@@ -1070,13 +1077,16 @@ class TestSubscribeSessionRevalidation:
                 ),
             ):
                 service._SocketIOService__handle_subscribe(
-                    {"research_id": "r1"}, mock_request
+                    {"research_id": VALID_RESEARCH_ID}, mock_request
                 )
 
             # validate_session inline-deleted the expired session ...
             assert session_token not in manager.sessions
             # ... the subscribe was rejected ...
-            assert "r1" not in service._SocketIOService__socket_subscriptions
+            assert (
+                VALID_RESEARCH_ID
+                not in service._SocketIOService__socket_subscriptions
+            )
             # ... and the orphaned socket was disconnected + its room closed.
             mock_socketio.server.disconnect.assert_any_call(
                 "orphan-sid", namespace="/"
@@ -1259,23 +1269,29 @@ class TestSocketServiceDisconnectCleanup:
 
             # Subscribe to two research IDs
             service._SocketIOService__handle_subscribe(
-                {"research_id": "r1"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
             service._SocketIOService__handle_subscribe(
-                {"research_id": "r2"}, mock_request
+                {"research_id": SECOND_VALID_RESEARCH_ID}, mock_request
             )
 
             subscriptions = service._SocketIOService__socket_subscriptions
-            assert "round_trip_client" in subscriptions["r1"]
-            assert "round_trip_client" in subscriptions["r2"]
+            assert "round_trip_client" in subscriptions[VALID_RESEARCH_ID]
+            assert (
+                "round_trip_client" in subscriptions[SECOND_VALID_RESEARCH_ID]
+            )
 
             # Disconnect should clean up both
             service._SocketIOService__handle_disconnect(
                 mock_request, "test reason"
             )
 
-            assert "round_trip_client" not in subscriptions.get("r1", set())
-            assert "round_trip_client" not in subscriptions.get("r2", set())
+            assert "round_trip_client" not in subscriptions.get(
+                VALID_RESEARCH_ID, set()
+            )
+            assert "round_trip_client" not in subscriptions.get(
+                SECOND_VALID_RESEARCH_ID, set()
+            )
         finally:
             SocketIOService._instance = original_instance
 

tests/web/services/test_socket_service_coverage.py

@@ -20,6 +20,7 @@ Focuses on:
 from unittest.mock import patch, MagicMock, Mock
 
 MODULE = "local_deep_research.web.services.socket_service"
+VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
 
 
 # ---------------------------------------------------------------------------
@@ -585,7 +586,7 @@ class TestHandleSubscribeEdgeCases:
         mock_request.sid = "client_empty_log"
 
         service._SocketIOService__handle_subscribe(
-            {"research_id": "r_empty_log"}, mock_request
+            {"research_id": VALID_RESEARCH_ID}, mock_request
         )
 
         # No emit should happen since latest_log is None/falsy
@@ -607,14 +608,12 @@ class TestHandleSubscribeEdgeCases:
             service, "emit_socket_event", return_value=True
         ) as mock_emit:
             service._SocketIOService__handle_subscribe(
-                {"research_id": "r_with_log"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
         mock_emit.assert_called_once()
         call_args = mock_emit.call_args
-        assert (
-            "r_with_log" in call_args[0][0]
-        )  # event name contains research_id
+        assert VALID_RESEARCH_ID in call_args[0][0]
         assert call_args[1]["room"] == "client_with_log"
 
 

tests/web/services/test_socket_service_extra_coverage.py

@@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch
 
 
 MODULE = "local_deep_research.web.services.socket_service"
+VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
 
 
 def _get_service_with_patched_init():
@@ -195,10 +196,13 @@ class TestHandleSubscribe:
 
         with patch(f"{MODULE}.get_active_research_snapshot", return_value=None):
             svc._SocketIOService__handle_subscribe(
-                {"research_id": "res-1"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
-        assert "client-1" in svc._SocketIOService__socket_subscriptions["res-1"]
+        assert (
+            "client-1"
+            in svc._SocketIOService__socket_subscriptions[VALID_RESEARCH_ID]
+        )
 
     def test_sends_current_status_when_available(self):
         svc = _get_service_with_patched_init()
@@ -214,7 +218,7 @@ class TestHandleSubscribe:
             f"{MODULE}.get_active_research_snapshot", return_value=snapshot
         ):
             svc._SocketIOService__handle_subscribe(
-                {"research_id": "res-1"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
         svc._SocketIOService__socketio.emit.assert_called()
@@ -226,7 +230,7 @@ class TestHandleSubscribe:
 
         with patch(f"{MODULE}.get_active_research_snapshot", return_value=None):
             svc._SocketIOService__handle_subscribe(
-                {"research_id": "res-1"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
         svc._SocketIOService__socketio.emit.assert_not_called()
@@ -242,7 +246,7 @@ class TestHandleSubscribe:
             f"{MODULE}.get_active_research_snapshot", return_value=snapshot
         ):
             svc._SocketIOService__handle_subscribe(
-                {"research_id": "res-1"}, mock_request
+                {"research_id": VALID_RESEARCH_ID}, mock_request
             )
 
         svc._SocketIOService__socketio.emit.assert_not_called()

tests/web/test_socket_subscribe_ownership.py

@@ -15,6 +15,9 @@ from unittest.mock import patch, MagicMock
 from local_deep_research.web.services.socket_service import SocketIOService
 
 
+VALID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
+
+
 def _patched_session(row):
     """Return a context-manager mock that yields a DB whose first() = row."""
     mock_db = MagicMock()
@@ -30,41 +33,15 @@ def _patched_session(row):
     return _Ctx()
 
 
-def test_owns_research_true_when_row_exists():
-    with patch(
-        "local_deep_research.database.session_context.get_user_db_session",
-        return_value=_patched_session(("abc",)),
-    ):
-        assert SocketIOService._user_owns_research("alice", "abc") is True
-
-
-def test_owns_research_false_when_row_missing():
-    """Cross-user disclosure regression: bob must not see alice's research."""
-    with patch(
-        "local_deep_research.database.session_context.get_user_db_session",
-        return_value=_patched_session(None),
-    ):
-        assert (
-            SocketIOService._user_owns_research("bob", "alices-research")
-            is False
-        )
-
-
-def test_owns_research_denies_on_exception():
-    """A DB-open / query failure must deny, not silently allow."""
-
-    def raising(*_args, **_kwargs):
-        raise RuntimeError("DB unavailable")
-
-    with patch(
-        "local_deep_research.database.session_context.get_user_db_session",
-        raising,
-    ):
-        assert SocketIOService._user_owns_research("alice", "anything") is False
-
-
 def _patched_session_seq(rows):
-    """Context-manager mock whose successive .first() calls yield `rows`."""
+    """Context-manager mock whose successive .first() calls yield `rows`.
+
+    The ownership gate probes both ``ResearchHistory`` (UUID lookup) and
+    ``BenchmarkRun`` (integer lookup) — the static helper dispatches based
+    on ``isinstance(research_id, str)``, but the authz design considers
+    both models. This helper lets a test express per-probe return values
+    so the same fixture works whether one probe runs or two.
+    """
     mock_db = MagicMock()
     mock_db.query.return_value.filter.return_value.first.side_effect = list(
         rows
@@ -80,29 +57,68 @@ def _patched_session_seq(rows):
     return _Ctx()
 
 
+def test_owns_research_true_when_row_exists():
+    with patch(
+        "local_deep_research.database.session_context.get_user_db_session",
+        return_value=_patched_session((VALID_RESEARCH_ID,)),
+    ):
+        assert (
+            SocketIOService._user_owns_research("alice", VALID_RESEARCH_ID)
+            is True
+        )
+
+
+def test_owns_research_false_when_row_missing():
+    """Cross-user disclosure regression: bob must not see alice's research."""
+    with patch(
+        "local_deep_research.database.session_context.get_user_db_session",
+        return_value=_patched_session(None),
+    ):
+        assert (
+            SocketIOService._user_owns_research("bob", VALID_RESEARCH_ID)
+            is False
+        )
+
+
+def test_owns_research_denies_on_exception():
+    """A DB-open / query failure must deny, not silently allow."""
+
+    def raising(*_args, **_kwargs):
+        raise RuntimeError("DB unavailable")
+
+    with patch(
+        "local_deep_research.database.session_context.get_user_db_session",
+        raising,
+    ):
+        assert (
+            SocketIOService._user_owns_research("alice", VALID_RESEARCH_ID)
+            is False
+        )
+
+
 def test_owns_benchmark_run_true_for_integer_id():
-    """Benchmark pages subscribe with an integer BenchmarkRun.id, so the
-    ownership gate must recognize the user's own benchmark runs — otherwise
-    benchmark live progress is dropped. Regression: the gate previously
-    only checked ResearchHistory (UUID ids), so an integer benchmark id
-    never matched and the subscribe was rejected. First query
-    (ResearchHistory) misses; second (BenchmarkRun) hits.
+    """Benchmark pages subscribe with an integer ``BenchmarkRun.id``. The
+    ownership gate probes both ``ResearchHistory`` and ``BenchmarkRun``;
+    for integer ids the ``BenchmarkRun`` probe matches and ownership is
+    verified. Regression: the gate previously only checked
+    ``ResearchHistory`` (UUID ids), so an integer benchmark id never
+    matched and the subscribe was rejected.
     """
     with patch(
         "local_deep_research.database.session_context.get_user_db_session",
-        return_value=_patched_session_seq([None, (5,)]),
+        return_value=_patched_session_seq([(5,)]),
     ):
-        assert SocketIOService._user_owns_research("alice", "5") is True
+        assert SocketIOService._user_owns_research("alice", 5) is True
 
 
 def test_owns_benchmark_run_false_when_not_owned():
-    """An integer id matching no ResearchHistory and no BenchmarkRun row
-    in the user's own DB is rejected (no cross-user widening)."""
+    """An integer id matching no ``ResearchHistory`` and no ``BenchmarkRun``
+    row in the user's own DB is rejected (no cross-user widening)."""
     with patch(
         "local_deep_research.database.session_context.get_user_db_session",
-        return_value=_patched_session_seq([None, None]),
+        return_value=_patched_session_seq([None]),
     ):
-        assert SocketIOService._user_owns_research("bob", "5") is False
+        assert SocketIOService._user_owns_research("bob", 5) is False
 
 
 def test_unsubscribe_rejected_when_user_does_not_own_research():
@@ -147,7 +163,8 @@ def test_unsubscribe_rejected_when_user_does_not_own_research():
         # Seed an existing subscription that the attacker should NOT be
         # able to evict.
         legit_sid = "legit-owner-sid"
-        service._SocketIOService__socket_subscriptions["target-research"] = {
+        research_id = VALID_RESEARCH_ID
+        service._SocketIOService__socket_subscriptions[research_id] = {
             legit_sid
         }
 
@@ -156,21 +173,26 @@ def test_unsubscribe_rejected_when_user_does_not_own_research():
 
         # Drive the unsubscribe handler directly.
         service._SocketIOService__handle_unsubscribe(
-            {"research_id": "target-research"}, attacker_request
+            {"research_id": research_id}, attacker_request
         )
 
         # The legit subscriber's sid must still be there, and the
         # attacker's sid must not have been added or removed (it was
         # never a subscriber to begin with).
         subs = service._SocketIOService__socket_subscriptions
-        assert "target-research" in subs
-        assert legit_sid in subs["target-research"]
+        assert research_id in subs
+        assert legit_sid in subs[research_id]
 
     SocketIOService._instance = None
 
 
-def test_unsubscribe_allowed_when_user_owns_research():
-    """Owners must still be able to unsubscribe from their own research."""
+def test_unsubscribe_decimal_id_matches_integer_subscription_key():
+    """Owners must still be able to unsubscribe from their own research.
+
+    ``parse_socket_research_id`` normalises a string ``"0005"`` into the
+    integer benchmark id ``5``, so a subscription keyed by int 5 is
+    evicted when the client unsubscribes with the zero-padded string.
+    """
     mock_app = MagicMock()
     mock_app.config = {"SECRET_KEY": "test-secret"}
 
@@ -195,7 +217,9 @@ def test_unsubscribe_allowed_when_user_owns_research():
         service = SocketIOService(app=mock_app)
 
         owner_sid = "alice-sid"
-        service._SocketIOService__socket_subscriptions["my-research"] = {
+        research_id = 5
+        subscription_key = ("alice", research_id)
+        service._SocketIOService__socket_subscriptions[subscription_key] = {
             owner_sid
         }
 
@@ -203,13 +227,14 @@ def test_unsubscribe_allowed_when_user_owns_research():
         owner_request.sid = owner_sid
 
         service._SocketIOService__handle_unsubscribe(
-            {"research_id": "my-research"}, owner_request
+            {"research_id": "0005"}, owner_request
         )
 
         # The owner's sid was removed; because it was the only subscriber,
         # the research_id key should also have been pruned.
         assert (
-            "my-research" not in service._SocketIOService__socket_subscriptions
+            subscription_key
+            not in service._SocketIOService__socket_subscriptions
         )
 
     SocketIOService._instance = None

tests/web_services/test_socket_benchmark_subscriptions.py

@@ -0,0 +1,181 @@
+from collections.abc import Generator
+from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from local_deep_research.web.services.socket_service import SocketIOService
+
+
+SOCKET_MODULE = "local_deep_research.web.services.socket_service"
+UUID_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
+
+
+@pytest.fixture
+def socket_service(mock_flask_app):
+    mock_socketio = MagicMock()
+    with (
+        patch(f"{SOCKET_MODULE}.SocketIO", return_value=mock_socketio),
+        patch(
+            f"{SOCKET_MODULE}.get_active_research_snapshot", return_value=None
+        ),
+    ):
+        service = SocketIOService(app=mock_flask_app)
+        service._SocketIOService__socketio = mock_socketio
+        yield service, mock_socketio
+
+
+@contextmanager
+def _authenticated_as(
+    service: SocketIOService, username: str
+) -> Generator[None, None, None]:
+    with (
+        patch(
+            f"{SOCKET_MODULE}.session",
+            {"username": username, "session_id": f"session-{username}"},
+        ),
+        patch.object(
+            service,
+            "_SocketIOService__session_authorizes",
+            return_value=True,
+        ),
+        patch.object(service, "_user_owns_research", return_value=True),
+    ):
+        yield
+
+
+def test_zero_padded_benchmark_id_uses_authenticated_owner(socket_service):
+    # Given
+    service, _ = socket_service
+    request = SimpleNamespace(sid="alice-sid")
+
+    # When
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": "0005", "username": "mallory"}, request
+        )
+
+    # Then
+    assert service._SocketIOService__socket_subscriptions == {
+        ("alice", 5): {"alice-sid"}
+    }
+
+
+def test_same_benchmark_id_emits_only_to_matching_owner(socket_service):
+    # Given
+    service, mock_socketio = socket_service
+    alice_request = SimpleNamespace(sid="alice-sid")
+    bob_request = SimpleNamespace(sid="bob-sid")
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": 5}, alice_request
+        )
+    with _authenticated_as(service, "bob"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": 5}, bob_request
+        )
+    mock_socketio.emit.reset_mock()
+    progress = {"progress": 50}
+
+    # When
+    result = service.emit_to_subscribers(
+        "research_progress", 5, progress, owner_username="alice"
+    )
+
+    # Then
+    assert result is True
+    mock_socketio.emit.assert_called_once_with(
+        "research_progress_5", progress, room="alice-sid"
+    )
+
+
+def test_benchmark_unsubscribe_keeps_other_owner_subscription(socket_service):
+    # Given
+    service, _ = socket_service
+    alice_request = SimpleNamespace(sid="alice-sid")
+    bob_request = SimpleNamespace(sid="bob-sid")
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": 5}, alice_request
+        )
+    with _authenticated_as(service, "bob"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": 5}, bob_request
+        )
+
+    # When
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_unsubscribe(
+            {"research_id": 5, "username": "bob"}, alice_request
+        )
+
+    # Then
+    assert service._SocketIOService__socket_subscriptions == {
+        ("bob", 5): {"bob-sid"}
+    }
+
+
+def test_benchmark_cleanup_keeps_other_owner_subscription(socket_service):
+    # Given
+    service, _ = socket_service
+    service._SocketIOService__socket_subscriptions = {
+        ("alice", 5): {"alice-sid"},
+        ("bob", 5): {"bob-sid"},
+    }
+
+    # When
+    service.remove_subscriptions_for_research(5, owner_username="alice")
+
+    # Then
+    assert service._SocketIOService__socket_subscriptions == {
+        ("bob", 5): {"bob-sid"}
+    }
+
+
+def test_uuid_emitter_keeps_plain_key_and_canonical_browser_event(
+    socket_service,
+):
+    # Given
+    service, mock_socketio = socket_service
+    request = SimpleNamespace(sid="alice-sid")
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_subscribe(
+            {"research_id": UUID_RESEARCH_ID.upper()}, request
+        )
+    mock_socketio.emit.reset_mock()
+    progress = {"progress": 75}
+
+    # When
+    result = service.emit_to_subscribers(
+        "research_progress", UUID_RESEARCH_ID.upper(), progress
+    )
+
+    # Then
+    assert result is True
+    assert service._SocketIOService__socket_subscriptions == {
+        UUID_RESEARCH_ID: {"alice-sid"}
+    }
+    mock_socketio.emit.assert_called_once_with(
+        f"research_progress_{UUID_RESEARCH_ID}",
+        progress,
+        room="alice-sid",
+    )
+
+
+def test_benchmark_emitter_without_trusted_owner_drops_event(socket_service):
+    # Given
+    service, mock_socketio = socket_service
+    request = SimpleNamespace(sid="alice-sid")
+    with _authenticated_as(service, "alice"):
+        service._SocketIOService__handle_subscribe({"research_id": 5}, request)
+    mock_socketio.emit.reset_mock()
+
+    # When
+    result = service.emit_to_subscribers(
+        "research_progress", 5, {"progress": 25}
+    )
+
+    # Then
+    assert result is True
+    mock_socketio.emit.assert_not_called()

tests/web_services/test_socket_service.py

@@ -7,6 +7,9 @@ import pytest
 from unittest.mock import Mock, MagicMock, patch
 
 
+VALID_SOCKET_RESEARCH_ID = "550e8400-e29b-41d4-a716-446655440000"
+
+
 class TestSocketIOServiceSingleton:
     """Tests for SocketIOService singleton pattern.
 
@@ -439,6 +442,89 @@ class TestSocketIOServiceEmitToSubscribers:
         assert result is True
 
 
+@pytest.mark.parametrize(
+    ("research_id", "expected"),
+    (
+        (VALID_SOCKET_RESEARCH_ID, VALID_SOCKET_RESEARCH_ID),
+        (
+            "550E8400-E29B-41D4-A716-446655440000",
+            VALID_SOCKET_RESEARCH_ID,
+        ),
+        (1, 1),
+        (9223372036854775807, 9223372036854775807),
+        ("1", 1),
+        ("00042", 42),
+        ("0000000000000000001", 1),
+        ("9223372036854775807", 9223372036854775807),
+    ),
+)
+def test_socket_research_id_parser_normalizes_supported_ids(
+    research_id, expected
+):
+    from local_deep_research.web.services._socket_research_id import (
+        parse_socket_research_id,
+    )
+
+    assert parse_socket_research_id({"research_id": research_id}) == expected
+
+
+@pytest.mark.parametrize(
+    "data",
+    (
+        pytest.param(None, id="null-payload"),
+        pytest.param([], id="list-payload"),
+        pytest.param("payload", id="string-payload"),
+        pytest.param({}, id="missing-id"),
+        pytest.param({"research_id": None}, id="null-id"),
+        pytest.param({"research_id": True}, id="boolean-true"),
+        pytest.param({"research_id": False}, id="boolean-false"),
+        pytest.param({"research_id": 0}, id="integer-zero"),
+        pytest.param({"research_id": -1}, id="negative-integer"),
+        pytest.param(
+            {"research_id": 9223372036854775808},
+            id="integer-above-sqlite-max",
+        ),
+        pytest.param({"research_id": 1.0}, id="float"),
+        pytest.param({"research_id": ""}, id="empty-id"),
+        pytest.param({"research_id": " "}, id="whitespace-id"),
+        pytest.param({"research_id": " 1"}, id="leading-whitespace"),
+        pytest.param({"research_id": "１２３"}, id="unicode-decimal"),
+        pytest.param({"research_id": "0"}, id="decimal-zero"),
+        pytest.param({"research_id": "000"}, id="zero-with-leading-zeroes"),
+        pytest.param({"research_id": "+1"}, id="positive-sign"),
+        pytest.param({"research_id": "-1"}, id="negative-sign"),
+        pytest.param({"research_id": "0x10"}, id="hex"),
+        pytest.param({"research_id": "1e3"}, id="exponent"),
+        pytest.param(
+            {"research_id": "550e8400e29b41d4a716446655440000"},
+            id="uuid-without-hyphens",
+        ),
+        pytest.param(
+            {"research_id": "{550e8400-e29b-41d4-a716-446655440000}"},
+            id="braced-uuid",
+        ),
+        pytest.param(
+            {"research_id": "550e8400-e29b-41d4-a716-44665544000g"},
+            id="malformed-uuid",
+        ),
+        pytest.param(
+            {"research_id": "9223372036854775808"},
+            id="decimal-above-sqlite-max",
+        ),
+        pytest.param(
+            {"research_id": "00000000000000000000"},
+            id="decimal-too-long",
+        ),
+    ),
+)
+def test_socket_research_id_parser_rejects_unsupported_input(data):
+    from local_deep_research.web.services._socket_research_id import (
+        parse_socket_research_id,
+    )
+
+    assert parse_socket_research_id(data) is None
+
+
 class TestSocketIOServiceSubscriptionManagement:
     """Tests for subscription management.
 
@@ -463,19 +549,77 @@ class TestSocketIOServiceSubscriptionManagement:
             service._SocketIOService__socketio = mock_socketio
             return service, mock_socketio
 
-    def test_subscribe_adds_client(
-        self, service_with_mocks, mock_request, sample_research_id
+    @pytest.mark.parametrize(
+        "handler_name",
+        (
+            "_SocketIOService__handle_subscribe",
+            "_SocketIOService__handle_unsubscribe",
+        ),
+        ids=("subscribe", "unsubscribe"),
+    )
+    def test_invalid_research_id_stops_before_socket_side_effects(
+        self,
+        service_with_mocks,
+        mock_request,
+        handler_name,
     ):
+        svc, mock_socketio = service_with_mocks
+        data = {"research_id": "0"}
+        existing_id = VALID_SOCKET_RESEARCH_ID
+        original_subscriptions = {existing_id: {"existing-sid"}}
+        svc._SocketIOService__socket_subscriptions = {
+            existing_id: {"existing-sid"}
+        }
+        mock_socketio.reset_mock()
+        mock_session = MagicMock()
+        mock_session.get.return_value = "test-owner"
+
+        with (
+            patch(
+                "local_deep_research.web.services.socket_service.session",
+                mock_session,
+            ),
+            patch.object(
+                svc,
+                "_SocketIOService__session_authorizes",
+                return_value=True,
+            ) as mock_session_authorizes,
+            patch.object(
+                svc,
+                "_user_owns_research",
+                return_value=False,
+            ) as mock_ownership,
+            patch(
+                "local_deep_research.web.services.socket_service."
+                "get_active_research_snapshot"
+            ) as mock_snapshot,
+            patch(
+                "local_deep_research.web.services.socket_service.logger"
+            ) as mock_logger,
+        ):
+            getattr(svc, handler_name)(data, mock_request)
+
+        assert (
+            svc._SocketIOService__socket_subscriptions == original_subscriptions
+        )
+        mock_session.get.assert_not_called()
+        mock_session_authorizes.assert_not_called()
+        mock_ownership.assert_not_called()
+        mock_snapshot.assert_not_called()
+        assert mock_logger.mock_calls == []
+        assert mock_socketio.mock_calls == []
+
+    def test_subscribe_adds_client(self, service_with_mocks, mock_request):
         """Test that subscribing adds client to subscription set."""
         svc, _ = service_with_mocks
 
-        data = {"research_id": sample_research_id}
+        data = {"research_id": VALID_SOCKET_RESEARCH_ID}
 
         svc._SocketIOService__handle_subscribe(data, mock_request)
 
         subscriptions = svc._SocketIOService__socket_subscriptions
-        assert sample_research_id in subscriptions
-        assert mock_request.sid in subscriptions[sample_research_id]
+        assert VALID_SOCKET_RESEARCH_ID in subscriptions
+        assert mock_request.sid in subscriptions[VALID_SOCKET_RESEARCH_ID]
 
     def test_subscribe_creates_set_for_new_research(
         self, service_with_mocks, mock_request
@@ -483,13 +627,38 @@ class TestSocketIOServiceSubscriptionManagement:
         """Test that subscribing creates new set for new research."""
         svc, _ = service_with_mocks
 
-        data = {"research_id": "new-research-id"}
+        data = {"research_id": "00042"}
 
         svc._SocketIOService__handle_subscribe(data, mock_request)
 
         subscriptions = svc._SocketIOService__socket_subscriptions
-        assert "new-research-id" in subscriptions
-        assert isinstance(subscriptions["new-research-id"], set)
+        assert ("test-owner", 42) in subscriptions
+        assert isinstance(subscriptions[("test-owner", 42)], set)
+
+    def test_integer_benchmark_subscription_receives_integer_key_emission(
+        self, service_with_mocks, mock_request
+    ):
+        svc, mock_socketio = service_with_mocks
+        progress = {"progress": 25}
+
+        with patch(
+            "local_deep_research.web.services.socket_service."
+            "get_active_research_snapshot",
+            return_value=None,
+        ):
+            svc._SocketIOService__handle_subscribe(
+                {"research_id": 5}, mock_request
+            )
+        mock_socketio.reset_mock()
+
+        result = svc.emit_to_subscribers(
+            "research_progress", 5, progress, owner_username="test-owner"
+        )
+
+        assert result is True
+        mock_socketio.emit.assert_called_once_with(
+            "research_progress_5", progress, room=mock_request.sid
+        )
 
     def test_subscribe_ignores_empty_research_id(
         self, service_with_mocks, mock_request
@@ -597,45 +766,48 @@ class TestSocketIOServiceSubscriptionManagement:
         self, service_with_mocks, mock_request
     ):
         """Test full subscribe → disconnect cycle uses consistent schema."""
-        svc, _ = service_with_mocks
+        from local_deep_research.web.services import socket_service
 
-        # Subscribe to two research IDs
+        svc, _ = service_with_mocks
+        owner_username = socket_service.session["username"]
+        first_key = (owner_username, 1)
+        second_key = (owner_username, 2)
+
+        # Subscribe to two benchmark IDs
         svc._SocketIOService__handle_subscribe(
-            {"research_id": "r1"}, mock_request
+            {"research_id": "1"}, mock_request
         )
         svc._SocketIOService__handle_subscribe(
-            {"research_id": "r2"}, mock_request
+            {"research_id": "2"}, mock_request
         )
 
         subscriptions = svc._SocketIOService__socket_subscriptions
-        assert mock_request.sid in subscriptions["r1"]
-        assert mock_request.sid in subscriptions["r2"]
+        assert mock_request.sid in subscriptions[first_key]
+        assert mock_request.sid in subscriptions[second_key]
 
         # Disconnect should clean up both
         svc._SocketIOService__handle_disconnect(
             mock_request, "client disconnect"
         )
 
-        assert mock_request.sid not in subscriptions.get("r1", set())
-        assert mock_request.sid not in subscriptions.get("r2", set())
+        assert first_key not in subscriptions
+        assert second_key not in subscriptions
 
-    def test_unsubscribe_discards_sid(
-        self, service_with_mocks, mock_request, sample_research_id
-    ):
+    def test_unsubscribe_discards_sid(self, service_with_mocks, mock_request):
         """Unsubscribe handler removes the sid from the subscription set."""
         svc, _ = service_with_mocks
         svc._SocketIOService__socket_subscriptions = {
-            sample_research_id: {mock_request.sid, "other-sid"},
+            VALID_SOCKET_RESEARCH_ID: {mock_request.sid, "other-sid"},
         }
 
         svc._SocketIOService__handle_unsubscribe(
-            {"research_id": sample_research_id}, mock_request
+            {"research_id": VALID_SOCKET_RESEARCH_ID}, mock_request
         )
 
         subscriptions = svc._SocketIOService__socket_subscriptions
-        assert mock_request.sid not in subscriptions[sample_research_id]
+        assert mock_request.sid not in subscriptions[VALID_SOCKET_RESEARCH_ID]
         # Other clients are untouched
-        assert "other-sid" in subscriptions[sample_research_id]
+        assert "other-sid" in subscriptions[VALID_SOCKET_RESEARCH_ID]
 
     def test_unsubscribe_ignores_missing_research_id(
         self, service_with_mocks, mock_request
@@ -663,11 +835,11 @@ class TestSocketIOServiceSubscriptionManagement:
 
         # Should not raise
         svc._SocketIOService__handle_unsubscribe(
-            {"research_id": "never-subscribed"}, mock_request
+            {"research_id": "999"}, mock_request
         )
 
     def test_unsubscribe_prunes_empty_subscription_set(
-        self, service_with_mocks, mock_request, sample_research_id
+        self, service_with_mocks, mock_request
     ):
         """Removing the last sid for a research_id deletes the dict entry.
 
@@ -677,32 +849,33 @@ class TestSocketIOServiceSubscriptionManagement:
         """
         svc, _ = service_with_mocks
         svc._SocketIOService__socket_subscriptions = {
-            sample_research_id: {mock_request.sid},
+            VALID_SOCKET_RESEARCH_ID: {mock_request.sid},
         }
 
         svc._SocketIOService__handle_unsubscribe(
-            {"research_id": sample_research_id}, mock_request
+            {"research_id": VALID_SOCKET_RESEARCH_ID}, mock_request
         )
 
         assert (
-            sample_research_id not in svc._SocketIOService__socket_subscriptions
+            VALID_SOCKET_RESEARCH_ID
+            not in svc._SocketIOService__socket_subscriptions
         )
 
     def test_unsubscribe_keeps_set_when_other_clients_remain(
-        self, service_with_mocks, mock_request, sample_research_id
+        self, service_with_mocks, mock_request
     ):
         """Unsubscribe must not delete the set while other sids are present."""
         svc, _ = service_with_mocks
         svc._SocketIOService__socket_subscriptions = {
-            sample_research_id: {mock_request.sid, "other-sid"},
+            VALID_SOCKET_RESEARCH_ID: {mock_request.sid, "other-sid"},
         }
 
         svc._SocketIOService__handle_unsubscribe(
-            {"research_id": sample_research_id}, mock_request
+            {"research_id": VALID_SOCKET_RESEARCH_ID}, mock_request
         )
 
         assert svc._SocketIOService__socket_subscriptions[
-            sample_research_id
+            VALID_SOCKET_RESEARCH_ID
         ] == {"other-sid"}