perf+refactor: DB hot-path index, dedup, classifier split, history bound

backend/apps/connectors/auth.py

@@ -4,21 +4,15 @@
 legacy shared bearer token fallback, and routes principal checks for the
 MCP chat-surface bridge.
 """
-import hmac
 import time
 
-from django.conf import settings
 from django.core.cache import cache
 from rest_framework.authentication import BaseAuthentication
 from rest_framework.exceptions import AuthenticationFailed
-from rest_framework.permissions import BasePermission
 
 from apps.connectors.crypto import verify_signature
 from apps.connectors.models import ConnectorKey
-
-
-def _get_token() -> str:
-    return getattr(settings, "ORC_CONNECTOR_TOKEN", "")
+from apps.core.auth import BearerTokenPermission
 
 
 class ConnectorSignatureAuthentication(BaseAuthentication):
@@ -110,7 +104,7 @@ def authenticate_header(self, request):
         return "Bearer"
 
 
-class HasConnectorToken(BasePermission):
+class HasConnectorToken(BearerTokenPermission):
     """Bearer token gate for the connector bridge endpoints.
 
     Accepts requests that are EITHER:
@@ -121,20 +115,13 @@ class HasConnectorToken(BasePermission):
     is not signature-authenticated.
     """
 
+    settings_attr = "ORC_CONNECTOR_TOKEN"
+    unconfigured_flag = "_connector_token_unconfigured"
+
     def has_permission(self, request, view) -> bool:
         # If the request is already signature-authenticated, allow it through.
         if isinstance(request.successful_authenticator, ConnectorSignatureAuthentication):
             return True
 
         # Legacy shared-token path (DEPRECATED — migrate to Ed25519).
-        expected = _get_token()
-        if not expected:
-            request._connector_token_unconfigured = True
-            return False
-
-        auth_header = request.META.get("HTTP_AUTHORIZATION", "")
-        if not auth_header.lower().startswith("bearer "):
-            return False
-
-        provided = auth_header[len("bearer "):].strip()
-        return hmac.compare_digest(provided.encode(), expected.encode())
+        return super().has_permission(request, view)

backend/apps/core/auth.py

@@ -0,0 +1,36 @@
+"""Shared bearer-token permission base for ORC apps.
+
+Plain-Python utility — no models, not in INSTALLED_APPS.
+"""
+import hmac
+
+from django.conf import settings
+from rest_framework.permissions import BasePermission
+
+
+class BearerTokenPermission(BasePermission):
+    """Base bearer-token gate parameterised by settings attribute and request flag.
+
+    Subclasses must set:
+      settings_attr  — name of the Django settings attribute holding the expected token
+      unconfigured_flag — name of the request attribute set when the token is empty
+    """
+
+    settings_attr: str
+    unconfigured_flag: str
+
+    def _get_token(self) -> str:
+        return getattr(settings, self.settings_attr, "")
+
+    def has_permission(self, request, view) -> bool:
+        expected = self._get_token()
+        if not expected:
+            setattr(request, self.unconfigured_flag, True)
+            return False
+
+        auth_header = request.META.get("HTTP_AUTHORIZATION", "")
+        if not auth_header.lower().startswith("bearer "):
+            return False
+
+        provided = auth_header[len("bearer "):].strip()
+        return hmac.compare_digest(provided.encode(), expected.encode())

backend/apps/core/html.py

@@ -0,0 +1,9 @@
+"""Shared HTML utilities for ORC apps.
+
+Plain-Python utility — no Django imports, no models.
+"""
+
+
+def _esc(s: str) -> str:
+    """Minimal HTML escape: &, <, > → named entities."""
+    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")

backend/apps/gateway/auth.py

@@ -4,15 +4,9 @@
 via the MESSAGING_GATEWAY_TOKEN setting.
 """
 
-import hmac
-
-from django.conf import settings
 from rest_framework.authentication import BaseAuthentication
-from rest_framework.permissions import BasePermission
-
 
-def _get_token() -> str:
-    return getattr(settings, "MESSAGING_GATEWAY_TOKEN", "")
+from apps.core.auth import BearerTokenPermission
 
 
 class GatewayBearerAuthentication(BaseAuthentication):
@@ -30,22 +24,12 @@ def authenticate_header(self, request):
         return "Bearer"
 
 
-class HasGatewayToken(BasePermission):
+class HasGatewayToken(BearerTokenPermission):
     """Bearer token gate for the messaging gateway endpoints.
 
     Returns False (→ 503) if MESSAGING_GATEWAY_TOKEN is unset.
     Returns False (→ 401) if the header is absent or wrong.
     """
 
-    def has_permission(self, request, view) -> bool:
-        expected = _get_token()
-        if not expected:
-            request._gateway_token_unconfigured = True
-            return False
-
-        auth_header = request.META.get("HTTP_AUTHORIZATION", "")
-        if not auth_header.lower().startswith("bearer "):
-            return False
-
-        provided = auth_header[len("bearer "):].strip()
-        return hmac.compare_digest(provided.encode(), expected.encode())
+    settings_attr = "MESSAGING_GATEWAY_TOKEN"
+    unconfigured_flag = "_gateway_token_unconfigured"

backend/apps/hostlink/tests/test_headless.py

@@ -11,7 +11,6 @@
 from asgiref.sync import sync_to_async
 from django.test import override_settings
 
-from apps.hostlink import consumers
 from apps.hostlink.consumers import HostDaemonConsumer
 from apps.hosts.models import Host
 from apps.threads.models import Thread

backend/apps/hostlink/tests/test_pty_reconcile.py

@@ -32,7 +32,7 @@ def _make_pty_thread(host, session_name, *, status=Thread.StatusChoices.RUNNING)
     )
     return Thread.objects.create(
         external_session_ref=session_name,
-        name=f"orc-run: test",
+        name="orc-run: test",
         runtime="pty",
         runtime_mode=Thread.RuntimeModeChoices.PTY,
         host=host,

backend/apps/observe/delivery.py

@@ -177,6 +177,7 @@ async def _deliver_turn_once(thread, parsed, msg, *, forum_chat_id, api=None) ->
                 disable_notification=False,
             )
         except Exception:
+            log.debug("user HTML send failed, falling back to plain text", exc_info=True)
             label = settings.TELEGRAM_USER_LABEL
             plain = f"{label}: {parsed['text'][:3900]}"
             await api.send_message(
@@ -201,6 +202,7 @@ async def _deliver_turn_once(thread, parsed, msg, *, forum_chat_id, api=None) ->
                 disable_notification=True,
             )
         except Exception:
+            log.debug("assistant HTML send failed (all mode), falling back to plain text", exc_info=True)
             label = settings.TELEGRAM_ASSISTANT_LABEL
             plain = f"{label}: {parsed['text'][:3900]}"
             await api.send_message(
@@ -226,6 +228,7 @@ async def _deliver_turn_once(thread, parsed, msg, *, forum_chat_id, api=None) ->
                 disable_notification=True,
             )
         except Exception:
+            log.debug("new digest HTML send failed, falling back to plain text", exc_info=True)
             label = settings.TELEGRAM_ASSISTANT_LABEL
             plain = f"{label}: {parsed['text'][:3900]}"
             new_id = await api.send_message(
@@ -260,6 +263,7 @@ async def _deliver_turn_once(thread, parsed, msg, *, forum_chat_id, api=None) ->
                     disable_notification=True,
                 )
             except Exception:
+                log.debug("fresh digest HTML send failed after stale edit, falling back to plain text", exc_info=True)
                 plain = f"{label}: {parsed['text'][:3900]}"
                 new_id = await api.send_message(
                     forum_chat_id,

backend/apps/observe/formatting.py

@@ -11,6 +11,8 @@
 
 import re
 
+from apps.core.html import _esc
+
 _FENCE_RE = re.compile(r"```[^\n]*\n(.*?)```", re.DOTALL)
 _INLINE_CODE_RE = re.compile(r"`([^`]+)`")
 _BOLD_RE = re.compile(r"\*\*(.+?)\*\*")
@@ -19,10 +21,6 @@
 _HEADING_RE = re.compile(r"^#{1,6}\s+(.*)$", re.MULTILINE)
 
 
-def _esc(s: str) -> str:
-    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
-
-
 def _convert_text(text: str) -> str:
     """Convert a non-code Markdown segment to Telegram HTML (escaped first)."""
     out = _esc(text)

backend/apps/observe/management/commands/run_session_observer.py

@@ -4,6 +4,7 @@
 deduplicates them, and forwards each turn to the active messaging recipient.
 """
 import asyncio
+import logging
 import time
 from pathlib import Path
 
@@ -17,6 +18,12 @@
 )
 from apps.observe.runtimes import get_runtime_adapter, iter_runtime_files
 
+log = logging.getLogger(__name__)
+
+# Backoff: sleep this many seconds after N consecutive failures.
+_BACKOFF_THRESHOLD = 5
+_BACKOFF_SLEEP = 30
+
 
 def _resolve_runtimes() -> list[str]:
     """Return the list of runtimes to observe.
@@ -67,11 +74,14 @@ async def _run(self):
         # the original single-runtime behaviour — dedup is by uuid within a provider.
         offsets: dict[tuple[str, Path], int] = {}
         seen: dict[str, set] = {rt: set() for rt in runtimes}
+        # Maximum UUIDs retained per runtime to keep the dedup set bounded.
+        _seen_max = 5000
         # Per-file remembered session id (for runtimes whose turn lines lack one).
         file_states: dict[tuple[str, Path], dict] = {}
         # Per-DB poll state for sqlite-based adapters (tracks last_msg_id, etc.).
         sqlite_states: dict[tuple[str, Path], dict] = {}
         last_selected: dict[str, int | None] = dict.fromkeys(runtimes)
+        consecutive_errors = 0
 
         async def on_turn(thread, p, msg):
             if routing.active_recipient():
@@ -126,6 +136,22 @@ async def on_turn(thread, p, msg):
                                 provider=provider,
                                 file_state=file_states.setdefault(key, {}),
                             )
-            except Exception as exc:  # noqa: BLE001
-                self.stderr.write(f"observer scan error: {exc}")
+                            # Cap the dedup set to avoid unbounded growth.
+                            if len(seen[provider]) > _seen_max:
+                                # Discard oldest half; set has no order so we
+                                # convert to list and keep the second half.
+                                s = list(seen[provider])
+                                seen[provider] = set(s[len(s) // 2 :])
+            except Exception:  # noqa: BLE001
+                consecutive_errors += 1
+                log.exception("observer scan error (consecutive=%d)", consecutive_errors)
+                if consecutive_errors >= _BACKOFF_THRESHOLD:
+                    self.stderr.write(
+                        f"observer: {consecutive_errors} consecutive errors — "
+                        f"backing off {_BACKOFF_SLEEP}s"
+                    )
+                    await asyncio.sleep(_BACKOFF_SLEEP)
+                    continue
+            else:
+                consecutive_errors = 0
             await asyncio.sleep(2)

backend/apps/observe/runtimes/__init__.py

@@ -10,6 +10,38 @@ class UnknownRuntimeError(Exception):
     """Raised when no observe runtime adapter is registered for a provider."""
 
 
+def _cwd_to_repo(cwd: str) -> str:
+    """Extract the repository name from a working-directory path.
+
+    Cross-platform: normalises backslashes so a Windows path observed on
+    macOS/Linux still yields the rightmost path component.
+    """
+    return cwd.rstrip("/\\").replace("\\", "/").rsplit("/", 1)[-1]
+
+
+class JsonlScanMixin:
+    """Mixin that provides a concrete scan_file_meta for JSONL-based adapters."""
+
+    def scan_file_meta(self, path: str) -> dict:
+        """Iterate every line of a JSONL session file and merge session metadata.
+
+        The session_id is stripped from the merged result (it belongs on the
+        individual turn, not the file-level summary).
+        Returns {} on any OSError (missing file, permission denied, etc.).
+        """
+        merged: dict = {}
+        try:
+            with open(path, encoding="utf-8") as f:
+                for line in f:
+                    m = self.extract_session_meta(line)  # type: ignore[attr-defined]
+                    m.pop("session_id", None)
+                    if m:
+                        merged.update(m)
+        except OSError:
+            return {}
+        return merged
+
+
 @runtime_checkable
 class RuntimeAdapter(Protocol):
     provider: str

backend/apps/observe/runtimes/aider.py

@@ -5,7 +5,7 @@
 """
 import os
 
-from apps.observe.runtimes import register_runtime_adapter
+from apps.observe.runtimes import JsonlScanMixin, register_runtime_adapter
 
 # TODO-VERIFY: Aider Markdown chat-history format is loosely documented and
 # varies by version. The following constants isolate every assumption so they
@@ -30,7 +30,7 @@
 
 
 @register_runtime_adapter
-class AiderAdapter:
+class AiderAdapter(JsonlScanMixin):
     provider = "aider"
     source_kind = "file"
     default_root_env = "OBSERVE_AIDER_PROJECTS_DIR"
@@ -67,16 +67,3 @@ def parse_turn(self, raw: str) -> dict | None:
 
     def extract_session_meta(self, raw: str) -> dict:
         return {}
-
-    def scan_file_meta(self, path: str) -> dict:
-        merged: dict = {}
-        try:
-            with open(path, encoding="utf-8") as f:
-                for line in f:
-                    m = self.extract_session_meta(line)
-                    m.pop("session_id", None)
-                    if m:
-                        merged.update(m)
-        except OSError:
-            return {}
-        return merged

backend/apps/observe/runtimes/claude_code.py

@@ -6,11 +6,11 @@
 import json
 import os
 
-from apps.observe.runtimes import register_runtime_adapter
+from apps.observe.runtimes import JsonlScanMixin, _cwd_to_repo, register_runtime_adapter
 
 
 @register_runtime_adapter
-class ClaudeCodeAdapter:
+class ClaudeCodeAdapter(JsonlScanMixin):
     provider = "claude_code"
     source_kind = "file"
     default_root_env = "OBSERVE_CLAUDE_PROJECTS_DIR"
@@ -71,7 +71,7 @@ def extract_session_meta(self, raw: str) -> dict:
         if isinstance(cwd, str):
             # Cross-platform basename: split on both separators so a Windows cwd
             # (c:\Users\...\repo) yields the repo name when observed on macOS/Linux too.
-            meta["repo"] = cwd.rstrip("/\\").replace("\\", "/").rsplit("/", 1)[-1]
+            meta["repo"] = _cwd_to_repo(cwd)
         branch = obj.get("gitBranch")
         if isinstance(branch, str):
             meta["branch"] = branch
@@ -80,16 +80,3 @@ def extract_session_meta(self, raw: str) -> dict:
             if isinstance(title, str):
                 meta["title"] = title
         return meta
-
-    def scan_file_meta(self, path: str) -> dict:
-        merged: dict = {}
-        try:
-            with open(path, encoding="utf-8") as f:
-                for line in f:
-                    m = self.extract_session_meta(line)
-                    m.pop("session_id", None)
-                    if m:
-                        merged.update(m)
-        except OSError:
-            return {}
-        return merged