[eric] analytics: convention pass: split into 3 modules (<300 each), pydantic models replace dicts, @typechecked, one-line comments

backend/apps/agents/core/ws_manager.py

@@ -95,8 +95,8 @@ class ConnectionManager:
         # ws.send_text directly, not this path, so reconnects don't double-count.
         if event == "agent:message":
             try:
-                from backend.apps.service.analytics import bridge_agent_message
-                bridge_agent_message(session_id, data.get("message") or {})
+                from backend.apps.service.analytics_agent_bridge import bridge_agent_message, BroadcastMessage
+                bridge_agent_message(session_id, BroadcastMessage.model_validate(data.get("message") or {}))
             except Exception:
                 logger.debug("agent:message analytics bridge failed", exc_info=True)
 

backend/apps/service/analytics.py

@@ -1,8 +1,9 @@
-"""swarm-analytics client singleton for the desktop backend.
+"""swarm-analytics client singleton + typed event wrappers for the desktop backend.
 
-One client per process. Bootstraps an install token on first use (persisted to
-settings) and reuses it forever. All failures are swallowed: analytics must
-never break the app. See ANALYTICS_OVERVIEW.md for the SDK contract.
+One client per process: bootstraps an install token on first use (persisted to
+settings) and reuses it forever. Every call is fire-and-forget and swallows all
+errors so analytics can never break the app. The agent-message and frontend-event
+bridges live in their own modules. See ANALYTICS_OVERVIEW.md for the SDK contract.
 """
 
 from __future__ import annotations
@@ -12,6 +13,8 @@ import os
 import platform
 from typing import Any, Optional
 
+from typeguard import typechecked
+
 from swarm_analytics import AnalyticsClient
 
 logger = logging.getLogger(__name__)
@@ -21,31 +24,30 @@ P_CLIENT: Optional[AnalyticsClient] = None
 # Env-overridable so prod points at the cloud edge; this default is the analytics service's own port, not the desktop's 8324.
 P_DEFAULT_ANALYTICS_URL = "http://127.0.0.1:6792"
 
+# Fired at most once per process; the renderer triggers it (the only tz/locale source that works for packaged + dev + OSS) so this guard enforces once-per-launch.
+P_OPENED_FIRED = False
 
+
+@typechecked
 def p_base_url() -> str:
     return os.environ.get("OPENSWARM_ANALYTICS_URL", P_DEFAULT_ANALYTICS_URL).rstrip("/")
 
 
+@typechecked
 def p_mode() -> str:
-    """Map the existing opt-out toggle onto the SDK mode.
-
-    logs.write is the 'diagnostic' category, so it flows even in 'minimal';
-    only 'product' events are muted. analytics_opt_in is the single toggle in
-    AppSettings, so opted-out -> 'minimal', otherwise 'full'.
-    """
+    # logs.write is diagnostic so it flows even in 'minimal'; only product events are muted.
     try:
         from backend.apps.settings.store import load_settings
-        s = load_settings()
-        if not getattr(s, "analytics_opt_in", True):
+        if not getattr(load_settings(), "analytics_opt_in", True):
             return "minimal"
     except Exception:
         pass
     return "full"
 
 
+@typechecked
 def get_analytics_client() -> Optional[AnalyticsClient]:
-    """Lazily bootstrap + cache the client. Returns None if setup fails
-    (e.g. offline first run) so callers can no-op safely."""
+    # Lazy bootstrap + cache; returns None (callers no-op) when setup fails, e.g. offline first run.
     global P_CLIENT
     if P_CLIENT is not None:
         return P_CLIENT
@@ -54,7 +56,7 @@ def get_analytics_client() -> Optional[AnalyticsClient]:
         s = load_settings()
         install_id = getattr(s, "installation_id", None)
         if not install_id:
-            return None  # main.py mints this pre-bind; bail defensively
+            return None
         base_url = p_base_url()
         token = getattr(s, "analytics_token", None)
         if not token:
@@ -68,6 +70,7 @@ def get_analytics_client() -> Optional[AnalyticsClient]:
     return P_CLIENT
 
 
+@typechecked
 def shutdown_analytics() -> None:
     global P_CLIENT
     if P_CLIENT is not None:
@@ -78,14 +81,7 @@ def shutdown_analytics() -> None:
             P_CLIENT = None
 
 
-# ---------------------------------------------------------------------------
-# Typed fire-and-forget wrappers. Each one resolves the singleton, no-ops when
-# the client is unavailable, and swallows every error (including the SDK's
-# synchronous pydantic.ValidationError) so a bad/missing analytics call can
-# never break a product code path. Call these from feature code, not the raw
-# client.
-# ---------------------------------------------------------------------------
-
+@typechecked
 def track_link_email(email: Optional[str]) -> None:
     if not email:
         return
@@ -98,9 +94,9 @@ def track_link_email(email: Optional[str]) -> None:
         logger.debug("analytics link_email failed: %s", e)
 
 
+@typechecked
 def track_agent_created(*, id: str, dashboard_id: Optional[str] = None) -> None:
-    """Name-free existence/dashboard event, fired at launch. The human-readable
-    title arrives later via track_agent_title once it's generated."""
+    # Name-free existence event at launch; the human-readable title arrives later via track_agent_title.
     c = get_analytics_client()
     if c is None:
         return
@@ -110,6 +106,7 @@ def track_agent_created(*, id: str, dashboard_id: Optional[str] = None) -> None:
         logger.debug("analytics agent.create failed: %s", e)
 
 
+@typechecked
 def track_agent_title(*, id: str, title: str) -> None:
     if not title:
         return
@@ -122,6 +119,7 @@ def track_agent_title(*, id: str, title: str) -> None:
         logger.debug("analytics agent.title failed: %s", e)
 
 
+@typechecked
 def track_agent_message(
     *,
     agent_id: str,
@@ -158,87 +156,9 @@ def track_agent_message(
         logger.debug("analytics agent.message failed: %s", e)
 
 
-def p_branch_version(session, message: dict) -> int:
-    """Edit marker for events.agent.message.branch_id.
-
-    Only the message that *created* a forked branch -- i.e. the actual edit -- gets
-    a non-zero version; replies and fresh turns typed on that branch reset to 0.
-    The edit is always the first `user` message on a forked branch (branches are
-    only ever born from agent_manager.edit_message), so a message that isn't that
-    first user message is "new" and scores 0. Repeated edits of the SAME user
-    message (siblings sharing a fork_point) are ranked 1, 2, ... by created_at."""
-    branch_str = message.get("branch_id") or "main"
-    branches = getattr(session, "branches", None) or {}
-    b = branches.get(branch_str)
-    fork_point = getattr(b, "fork_point_message_id", None) if b else None
-    if not fork_point:
-        return 0  # main / never-edited path
-    msg_id = message.get("id")
-    branch_user_msgs = [
-        m for m in (getattr(session, "messages", None) or [])
-        if getattr(m, "branch_id", None) == branch_str and getattr(m, "role", None) == "user"
-    ]
-    if not branch_user_msgs or getattr(branch_user_msgs[0], "id", None) != msg_id:
-        return 0  # a reply or a later new turn on this branch -> not an edit
-    siblings = sorted(
-        (x for x in branches.values()
-         if getattr(x, "fork_point_message_id", None) == fork_point),
-        key=lambda x: x.created_at,
-    )
-    for i, x in enumerate(siblings, start=1):
-        if x.id == branch_str:
-            return i
-    return 0
-
-
-def bridge_agent_message(session_id: str, message: dict) -> None:
-    """Re-emit a broadcast `agent:message` as the typed events.agent.message.
-
-    Called from ws_manager.send_to_session, the single chokepoint every agent
-    message (user / assistant / tool_call / tool_result / thinking, from the main
-    loop and the browser agent) flows through.
-
-    `seq` is the message's index in the session's persisted history
-    (session.messages). Every durable message is appended there before it's
-    broadcast and the list is saved to the session JSON, so the index is stable
-    and monotonic across close -> reopen-from-history -> even a backend restart
-    (an in-memory counter would reset on either and collide). Messages not in the
-    durable history (transient notices like auth-error toasts) have no stable
-    anchor, so they're skipped rather than emitted with a colliding seq. Full
-    content is forwarded. Best-effort: never raises into the broadcast path."""
-    if not isinstance(message, dict):
-        return
-    msg_id = message.get("id")
-    role = message.get("role")
-    if not msg_id or not role:
-        return
-    try:
-        from backend.apps.agents.agent_manager import agent_manager
-        sess = agent_manager.sessions.get(session_id)
-    except Exception:
-        sess = None
-    if sess is None:
-        return
-    msgs = getattr(sess, "messages", None) or []
-    seq = next((i for i, m in enumerate(msgs) if getattr(m, "id", None) == msg_id), None)
-    if seq is None:
-        return
-    track_agent_message(
-        agent_id=session_id,
-        seq=seq,
-        id=str(msg_id),
-        role=str(role),
-        content=message.get("content"),
-        parent_id=message.get("parent_id"),
-        branch_id=p_branch_version(sess, message),
-        provider=getattr(sess, "provider", None),
-        model=getattr(sess, "model", None),
-        thinking_level=getattr(sess, "thinking_level", None),
-    )
-
-
+@typechecked
 def track_dashboard_event(*, dashboard_id: str, action: str) -> None:
-    """action is one of: open, close, create, delete (validated by the SDK)."""
+    # action is one of: open, close, create, delete (validated by the SDK).
     c = get_analytics_client()
     if c is None:
         return
@@ -248,8 +168,9 @@ def track_dashboard_event(*, dashboard_id: str, action: str) -> None:
         logger.debug("analytics dashboard.event failed: %s", e)
 
 
+@typechecked
 def track_onboarding_step(*, step_id: str, status: str) -> None:
-    """status is one of: started, completed, abandoned (validated by the SDK)."""
+    # status is one of: started, completed, abandoned (validated by the SDK).
     c = get_analytics_client()
     if c is None:
         return
@@ -259,20 +180,9 @@ def track_onboarding_step(*, step_id: str, status: str) -> None:
         logger.debug("analytics onboarding.step failed: %s", e)
 
 
-# app_lifecycle.opened is fired at most once per backend process. The renderer
-# triggers it (so it carries the browser's canonical tz/locale, the only source
-# that works for packaged, dev, AND open-source runs), but a renderer can remount
-# or hard-reload many times against one long-lived backend -- especially in dev --
-# so this process-scoped guard is what actually enforces one event per app launch.
-P_OPENED_FIRED = False
-
-
+@typechecked
 def persist_client_env(*, timezone: Optional[str] = None, locale: Optional[str] = None) -> None:
-    """Store the renderer-reported tz/locale so the cloud envelope (stamped on
-    every submission via client.resolve_*) can use them on dev / open-source runs
-    where Electron's env injection never happens. Overwrites every launch, so a
-    user who changed timezone since last open reports the new one. Writes to disk
-    only when a value actually changed, to avoid settings churn each launch."""
+    # Store the renderer-reported tz/locale for the cloud envelope on dev/OSS runs; disk-write only when a value actually changed.
     tz = (timezone or "").strip() or None
     loc = (locale or "").strip() or None
     if tz is None and loc is None:
@@ -293,11 +203,8 @@ def persist_client_env(*, timezone: Optional[str] = None, locale: Optional[str]
         logger.debug("analytics persist_client_env failed: %s", e)
 
 
+@typechecked
 def track_app_opened(*, timezone: Optional[str] = None, locale: Optional[str] = None) -> None:
-    """Fire app_lifecycle.opened once per backend process. tz/locale come from the
-    renderer (browser Intl); os/version are filled in here. Falls back to the
-    shared resolver only if the caller passed nothing (defensive; the renderer
-    path always supplies both)."""
     global P_OPENED_FIRED
     if P_OPENED_FIRED:
         return
@@ -319,6 +226,7 @@ def track_app_opened(*, timezone: Optional[str] = None, locale: Optional[str] =
         logger.debug("analytics app_lifecycle.opened failed: %s", e)
 
 
+@typechecked
 def track_app_closed() -> None:
     c = get_analytics_client()
     if c is None:

backend/apps/service/analytics_agent_bridge.py

@@ -0,0 +1,84 @@
+"""Bridge a broadcast `agent:message` into the typed `events.agent.message`.
+
+Called from ws_manager.send_to_session, the single chokepoint every agent message
+flows through. Best-effort: never raises into the broadcast path.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Optional
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.agents.core.models import AgentSession
+from backend.apps.service.analytics import track_agent_message
+
+logger = logging.getLogger(__name__)
+
+
+class BroadcastMessage(BaseModel):
+    # An agent:message broadcast payload, validated at the WS boundary; extra fields ignored.
+    model_config = ConfigDict(validate_assignment=True, extra="ignore")
+    id: Optional[str] = None
+    role: Optional[str] = None
+    content: Any = None
+    parent_id: Optional[str] = None
+    branch_id: Optional[str] = None
+
+
+@typechecked
+def p_branch_version(session: AgentSession, message: BroadcastMessage) -> int:
+    # Edit marker for branch_id: only the message that CREATED a forked branch (the actual edit) scores non-zero; replies and new turns reset to 0.
+    branch_str = message.branch_id or "main"
+    branches = getattr(session, "branches", None) or {}
+    b = branches.get(branch_str)
+    fork_point = getattr(b, "fork_point_message_id", None) if b else None
+    if not fork_point:
+        return 0
+    branch_user_msgs = [
+        m for m in (getattr(session, "messages", None) or [])
+        if getattr(m, "branch_id", None) == branch_str and getattr(m, "role", None) == "user"
+    ]
+    if not branch_user_msgs or getattr(branch_user_msgs[0], "id", None) != message.id:
+        return 0
+    siblings = sorted(
+        (x for x in branches.values()
+         if getattr(x, "fork_point_message_id", None) == fork_point),
+        key=lambda x: x.created_at,
+    )
+    for i, x in enumerate(siblings, start=1):
+        if x.id == branch_str:
+            return i
+    return 0
+
+
+@typechecked
+def bridge_agent_message(session_id: str, message: BroadcastMessage) -> None:
+    # seq is the message's stable index in the persisted history (survives close -> reopen -> restart); transient messages with no anchor are skipped.
+    if not message.id or not message.role:
+        return
+    try:
+        from backend.apps.agents.agent_manager import agent_manager
+        sess = agent_manager.sessions.get(session_id)
+    except Exception:
+        sess = None
+    if sess is None:
+        return
+    msgs = getattr(sess, "messages", None) or []
+    seq = next((i for i, m in enumerate(msgs) if getattr(m, "id", None) == message.id), None)
+    if seq is None:
+        return
+    track_agent_message(
+        agent_id=session_id,
+        seq=seq,
+        id=str(message.id),
+        role=str(message.role),
+        content=message.content,
+        parent_id=message.parent_id,
+        branch_id=p_branch_version(sess, message),
+        provider=getattr(sess, "provider", None),
+        model=getattr(sess, "model", None),
+        thinking_level=getattr(sess, "thinking_level", None),
+    )

backend/apps/service/analytics_frontend_bridge.py

@@ -0,0 +1,59 @@
+"""Bridge frontend `report()` {s, a, p} events into typed product events.
+
+The frontend is browser-side and can't reach the analytics service directly, so
+onboarding/dashboard/app events arrive here as envelopes. Best-effort.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.service.analytics import (
+    persist_client_env,
+    track_app_opened,
+    track_dashboard_event,
+    track_onboarding_step,
+)
+
+# report() action -> SDK onboarding status; the timeout/error variants both count as abandoned.
+P_ONBOARDING_STATUS = {
+    "step_started": "started",
+    "step_completed": "completed",
+    "step_aborted": "abandoned",
+    "step_selector_timeout": "abandoned",
+    "step_error": "abandoned",
+}
+
+
+class FrontendEventProps(BaseModel):
+    model_config = ConfigDict(validate_assignment=True, extra="ignore")
+    dashboard_id: Optional[str] = None
+    step_id: Optional[str] = None
+    timezone: Optional[str] = None
+    locale: Optional[str] = None
+
+
+class FrontendEvent(BaseModel):
+    # A report() envelope {s, a, p}; extra fields ignored at the HTTP boundary.
+    model_config = ConfigDict(validate_assignment=True, extra="ignore")
+    s: Optional[str] = None
+    a: Optional[str] = None
+    p: FrontendEventProps = FrontendEventProps()
+
+
+@typechecked
+def bridge_frontend_event(event: FrontendEvent) -> None:
+    # Dashboard create/delete are NOT bridged here; those fire authoritatively from the dashboards routes, so bridging them too would double-count.
+    if event.s == "onboarding_v2":
+        status = P_ONBOARDING_STATUS.get(event.a or "")
+        if status and event.p.step_id:
+            track_onboarding_step(step_id=str(event.p.step_id), status=status)
+    elif event.s == "dashboard" and event.a in ("open", "close"):
+        if event.p.dashboard_id:
+            track_dashboard_event(dashboard_id=str(event.p.dashboard_id), action=str(event.a))
+    elif event.s == "app" and event.a == "opened":
+        persist_client_env(timezone=event.p.timezone, locale=event.p.locale)
+        track_app_opened(timezone=event.p.timezone, locale=event.p.locale)

backend/apps/service/service.py

@@ -443,47 +443,12 @@ async def service_status():
 # ---------------------------------------------------------------------------
 
 def p_bridge_to_analytics(item: dict) -> None:
-    """Re-emit frontend `report()` events through the typed swarm-analytics SDK.
-
-    The frontend is browser-side and can't reach the analytics service
-    directly, so onboarding steps and dashboard open/close arrive here as
-    {s, a, p} envelopes. We translate the ones we care about into product
-    events. Best-effort: never raises (the track_* wrappers swallow errors).
-    Dashboard create/delete are NOT bridged here; those fire authoritatively
-    from the dashboards routes, bridging them too would double-count.
-    """
-    s = item.get("s")
-    a = item.get("a")
-    p = item.get("p") or {}
-    if not isinstance(p, dict):
-        return
-    if s == "onboarding_v2":
-        status = {
-            "step_started": "started",
-            "step_completed": "completed",
-            "step_aborted": "abandoned",
-            "step_selector_timeout": "abandoned",
-            "step_error": "abandoned",
-        }.get(a)
-        step_id = p.get("step_id")
-        if status and step_id:
-            from backend.apps.service.analytics import track_onboarding_step
-            track_onboarding_step(step_id=str(step_id), status=status)
-    elif s == "dashboard" and a in ("open", "close"):
-        dashboard_id = p.get("dashboard_id")
-        if dashboard_id:
-            from backend.apps.service.analytics import track_dashboard_event
-            track_dashboard_event(dashboard_id=str(dashboard_id), action=a)
-    elif s == "app" and a == "opened":
-        # The renderer reports the browser's canonical IANA timezone + BCP 47
-        # locale on launch. Persist them (overwriting last launch, so a timezone
-        # switch is picked up) for the cloud envelope, then emit the once-per-
-        # process app_lifecycle.opened carrying those exact values.
-        tz = p.get("timezone") if isinstance(p.get("timezone"), str) else None
-        loc = p.get("locale") if isinstance(p.get("locale"), str) else None
-        from backend.apps.service.analytics import persist_client_env, track_app_opened
-        persist_client_env(timezone=tz, locale=loc)
-        track_app_opened(timezone=tz, locale=loc)
+    # Boundary adapter: validate the raw report() envelope into a typed event, hand it to the analytics bridge.
+    from backend.apps.service.analytics_frontend_bridge import bridge_frontend_event, FrontendEvent
+    try:
+        bridge_frontend_event(FrontendEvent.model_validate(item))
+    except Exception:
+        pass
 
 
 @service.router.post("/submit")