[eric] analytics: forward-port swarm-analytics ingest onto eric/dev (configurable base_url, lifecycle + frontend-event bridge)

backend/apps/service/analytics.py

@@ -0,0 +1,332 @@
+"""swarm-analytics client singleton 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.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import platform
+from typing import Any, Optional
+
+from swarm_analytics import AnalyticsClient
+
+logger = logging.getLogger(__name__)
+
+P_CLIENT: Optional[AnalyticsClient] = None
+
+# Where the SDK ships events. Configurable so local dev hits a local
+# product-analytics-v1 (its .env BACKEND_PORT) while prod points at the
+# cloud-hosted ingest via OPENSWARM_ANALYTICS_URL. NOT the desktop backend's
+# port (8324); the analytics service listens on 6792.
+P_DEFAULT_ANALYTICS_URL = "http://127.0.0.1:6792"
+
+
+def p_base_url() -> str:
+    return os.environ.get("OPENSWARM_ANALYTICS_URL", P_DEFAULT_ANALYTICS_URL).rstrip("/")
+
+
+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'.
+    """
+    try:
+        from backend.apps.settings.store import load_settings
+        s = load_settings()
+        if not getattr(s, "analytics_opt_in", True):
+            return "minimal"
+    except Exception:
+        pass
+    return "full"
+
+
+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."""
+    global P_CLIENT
+    if P_CLIENT is not None:
+        return P_CLIENT
+    try:
+        from backend.apps.settings.store import load_settings, save_settings
+        s = load_settings()
+        install_id = getattr(s, "installation_id", None)
+        if not install_id:
+            return None  # main.py mints this pre-bind; bail defensively
+        base_url = p_base_url()
+        token = getattr(s, "analytics_token", None)
+        if not token:
+            token = AnalyticsClient.register(base_url=base_url, install_id=install_id)
+            s.analytics_token = token
+            save_settings(s)
+        P_CLIENT = AnalyticsClient(base_url=base_url, token=token, mode=p_mode())
+    except Exception as e:
+        logger.debug("analytics setup failed (non-critical): %s", e)
+        return None
+    return P_CLIENT
+
+
+def shutdown_analytics() -> None:
+    global P_CLIENT
+    if P_CLIENT is not None:
+        try:
+            P_CLIENT.flush(timeout=2.0)
+            P_CLIENT.close()
+        finally:
+            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.
+# ---------------------------------------------------------------------------
+
+def track_link_email(email: Optional[str]) -> None:
+    if not email:
+        return
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.identify.link_email(email=email)
+    except Exception as e:
+        logger.debug("analytics link_email failed: %s", e)
+
+
+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."""
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.events.agent.create(id=id, dashboard_id=dashboard_id)
+    except Exception as e:
+        logger.debug("analytics agent.create failed: %s", e)
+
+
+def track_agent_title(*, id: str, title: str) -> None:
+    if not title:
+        return
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.events.agent.title(id=id, title=title)
+    except Exception as e:
+        logger.debug("analytics agent.title failed: %s", e)
+
+
+def track_agent_message(
+    *,
+    agent_id: str,
+    seq: int,
+    id: str,
+    role: str,
+    content: Any = None,
+    parent_id: Optional[str] = None,
+    branch_id: int = 0,
+    provider: Optional[str] = None,
+    model: Optional[str] = None,
+    thinking_level: Optional[str] = None,
+) -> None:
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        from swarm_analytics import AgentMessage
+        c.events.agent.message(
+            agent_id=agent_id,
+            seq=seq,
+            message=AgentMessage(
+                id=id,
+                role=role,
+                content=content,
+                parent_id=parent_id,
+                branch_id=branch_id,
+                provider=provider,
+                model=model,
+                thinking_level=thinking_level,
+            ),
+        )
+    except Exception as e:
+        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),
+    )
+
+
+def track_dashboard_event(*, dashboard_id: str, action: str) -> None:
+    """action is one of: open, close, create, delete (validated by the SDK)."""
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.events.dashboard.event(dashboard_id=dashboard_id, action=action)
+    except Exception as e:
+        logger.debug("analytics dashboard.event failed: %s", e)
+
+
+def track_onboarding_step(*, step_id: str, status: str) -> None:
+    """status is one of: started, completed, abandoned (validated by the SDK)."""
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.events.onboarding.step(step_id=step_id, status=status)
+    except Exception as e:
+        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
+
+
+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."""
+    tz = (timezone or "").strip() or None
+    loc = (locale or "").strip() or None
+    if tz is None and loc is None:
+        return
+    try:
+        from backend.apps.settings.store import load_settings, save_settings
+        s = load_settings()
+        changed = False
+        if tz and getattr(s, "timezone", None) != tz:
+            s.timezone = tz
+            changed = True
+        if loc and getattr(s, "locale", None) != loc:
+            s.locale = loc
+            changed = True
+        if changed:
+            save_settings(s)
+    except Exception as e:
+        logger.debug("analytics persist_client_env failed: %s", e)
+
+
+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
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        from backend.apps.service.version import APP_VERSION
+        from backend.apps.service.client import resolve_timezone, resolve_locale
+        c.events.app_lifecycle.opened(
+            os=platform.system(),
+            os_version=platform.release(),
+            app_version=APP_VERSION,
+            timezone=timezone if timezone is not None else resolve_timezone(),
+            locale=locale if locale is not None else resolve_locale(),
+        )
+        P_OPENED_FIRED = True
+    except Exception as e:
+        logger.debug("analytics app_lifecycle.opened failed: %s", e)
+
+
+def track_app_closed() -> None:
+    c = get_analytics_client()
+    if c is None:
+        return
+    try:
+        c.events.app_lifecycle.closed()
+    except Exception as e:
+        logger.debug("analytics app_lifecycle.closed failed: %s", e)

backend/apps/service/client.py

@@ -45,6 +45,49 @@ P_PATH_BY_KIND = {
 P_TIMEOUT_SECONDS = 5.0
 P_MAX_INFLIGHT = 16
 
+
+def resolve_timezone() -> str:
+    """Best-effort IANA timezone for analytics. Prefers the renderer-reported
+    value persisted in settings (the only source that works on dev / OSS where
+    Electron's env injection never runs), then the OS, then UTC."""
+    try:
+        from backend.apps.settings.store import load_settings
+        tz = getattr(load_settings(), "timezone", None)
+        if tz:
+            return tz
+    except Exception:
+        pass
+    try:
+        from tzlocal import get_localzone_name
+        name = get_localzone_name()
+        if name:
+            return name
+    except Exception:
+        pass
+    try:
+        return time.tzname[0] or "UTC"
+    except Exception:
+        return "UTC"
+
+
+def resolve_locale() -> str:
+    """Best-effort BCP-47 locale, settings-first then OS, defaulting to en-US."""
+    try:
+        from backend.apps.settings.store import load_settings
+        loc = getattr(load_settings(), "locale", None)
+        if loc:
+            return loc
+    except Exception:
+        pass
+    try:
+        import locale as _locale
+        code = _locale.getlocale()[0]
+        if code:
+            return code.replace("_", "-")
+    except Exception:
+        pass
+    return "en-US"
+
 test_sink: Optional[Any] = None
 install_id: Optional[str] = None
 p_user_id: Optional[str] = None

backend/apps/service/service.py

@@ -189,6 +189,18 @@ async def service_lifespan():
             id_props["subscription_expires"] = settings.openswarm_subscription_expires
 
         svc.sync({"identity": id_props})
+
+        # swarm-analytics: bootstrap the client (registers + persists a token on
+        # first run), prove the pipe with one diagnostic log write, and link the
+        # user's email. All best-effort; the wrappers swallow every error.
+        from backend.apps.service.analytics import get_analytics_client, track_link_email
+        analytics_client = get_analytics_client()
+        if analytics_client is not None:
+            try:
+                analytics_client.logs.write(tag="app", subtag="backend_started", data={"app_version": APP_VERSION})
+            except Exception:
+                pass
+        track_link_email(getattr(settings, "user_email", None))
     except Exception as e:
         logger.debug(f"Service startup event failed (non-critical): {e}")
 
@@ -239,6 +251,15 @@ async def service_lifespan():
     except Exception:
         pass
 
+    # swarm-analytics: fire the app-closed event and flush+close the client so
+    # buffered events land before the process exits. Best-effort.
+    try:
+        from backend.apps.service.analytics import track_app_closed, shutdown_analytics
+        track_app_closed()
+        shutdown_analytics()
+    except Exception:
+        pass
+
     logger.info("Service shut down")
 
 
@@ -424,6 +445,50 @@ async def service_status():
 # Frontend event endpoints
 # ---------------------------------------------------------------------------
 
+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)
+
+
 @service.router.post("/submit")
 async def post_submit(body=Body(...)):
     """Accepts three body shapes for backward compatibility:
@@ -453,6 +518,7 @@ async def post_submit(body=Body(...)):
             if isinstance(item, dict):
                 if any(k in item for k in ("s", "a", "p")):
                     svc.sync(item)
+                    p_bridge_to_analytics(item)
                     continue
                 kind = item.get("kind") or ""
                 payload = item.get("payload") or {}
@@ -465,6 +531,7 @@ async def post_submit(body=Body(...)):
     # Shape 1: frontend `report()`; flat {s, a, p, ...}
     if any(k in body for k in ("s", "a", "p")):
         svc.sync(body)
+        p_bridge_to_analytics(body)
         return {"ok": True}
     # Shape 2: legacy {kind, payload}
     kind = body.get("kind") or ""
@@ -488,11 +555,13 @@ async def post_event(body: dict):
     if not action:
         action = "fired"
 
-    svc.sync({
+    envelope = {
         "s": str(surface)[:64],
         "a": str(action)[:64],
         "p": body.get("props") or body.get("properties") or {},
-    })
+    }
+    svc.sync(envelope)
+    p_bridge_to_analytics(envelope)
     return {"ok": True}
 
 

backend/apps/settings/models.py

@@ -65,6 +65,11 @@ class AppSettings(BaseModel):
     dismissed_mcp_suggestions: dict[str, str] = Field(default_factory=dict)
     analytics_opt_in: bool = True
     installation_id: Optional[str] = None
+    # Minted once by the analytics SDK's register() and reused forever; server-owned.
+    analytics_token: Optional[str] = None
+    # Renderer-reported browser Intl values, stamped on analytics submissions; server-owned.
+    timezone: Optional[str] = None
+    locale: Optional[str] = None
     first_opened_at: Optional[str] = None
     connection_mode: str = "own_key"
     openswarm_bearer_token: Optional[str] = None

backend/apps/settings/settings.py

@@ -139,6 +139,9 @@ SERVER_OWNED_FIELDS = (
     "user_id",
     "signin_method",
     "installation_id",
+    "analytics_token",
+    "timezone",
+    "locale",
     "claude_subscription_token",
     "openai_subscription_token",
     "gemini_subscription_token",
@@ -244,7 +247,7 @@ async def apply_settings_update(body: AppSettings, protect_fields: set[str] | No
 
     secret_keys = {"anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
                    "claude_subscription_token", "openai_subscription_token", "gemini_subscription_token",
-                   "openswarm_bearer_token", "free_trial_token", "installation_id"}
+                   "openswarm_bearer_token", "free_trial_token", "installation_id", "analytics_token"}
     safe = {k: v for k, v in body.model_dump().items() if k not in secret_keys}
     p_sync(safe)
 

backend/requirements.txt

@@ -16,6 +16,10 @@ python-dotenv==1.1.1
 Pillow==12.2.0
 httpx==0.28.1
 trafilatura==2.0.0
+# swarm-analytics: typed client for the product-analytics ingest service.
+# Validates payloads against the server schema locally; all calls are
+# fire-and-forget and swallow errors so analytics can never break the app.
+swarm-analytics==0.1.1
 # tzlocal: dev-mode fallback for resolving the user's IANA timezone when
 # Electron's OPENSWARM_TIMEZONE env var isn't set (i.e. `bash run.sh`).
 # Packaged builds get the env var directly so this is a safety net.

frontend/src/app/Main.tsx

@@ -78,7 +78,7 @@ if (typeof window !== 'undefined') {
   if (ric) ric(prefetchAll, { timeout: 1500 });
   else window.setTimeout(prefetchAll, 500);
 }
-import { report, getSessionTraceState, getRecentActions } from '@/shared/serviceClient';
+import { report, reportAppOpened, getSessionTraceState, getRecentActions } from '@/shared/serviceClient';
 import { useRouteTracker } from '@/shared/hooks/useRouteTracker';
 import { useDeepLink } from '@/shared/hooks/useDeepLink';
 import { useWindowFocus } from '@/shared/hooks/useWindowFocus';
@@ -223,6 +223,10 @@ const SettingsLoader: React.FC<{ children: React.ReactNode }> = ({ children }) =
   useEffect(() => {
     dispatch(fetchSettings());
     dispatch(fetchModels());
+    // Report the app launch with the browser's canonical tz/locale so the backend
+    // can emit analytics app_lifecycle.opened with values that work in packaged,
+    // dev, and open-source builds. Guarded once per page load; backend dedupes per process.
+    reportAppOpened();
     // Connected subscriptions live in their own slice; without this the dashboard
     // (and the onboarding gate) think no model is connected until the user opens
     // Settings > Models, so a fresh launch shows a false "connect a model" empty

frontend/src/shared/serviceClient.ts

@@ -87,6 +87,31 @@ export function report(
   sync({ s: surface, a: action, p: props || {} }, opts);
 }
 
+let _openedSent = false;
+
+/**
+ * Report the app launch with the browser's canonical timezone + locale (the
+ * Intl API gives the same values Electron does, but works in dev and the
+ * open-source build too, where Electron's env injection never runs). The backend
+ * persists these and emits analytics `app_lifecycle.opened` from them.
+ *
+ * Guarded so a remount won't re-send within one page load; the backend also
+ * dedupes per process, so a hard reload can't double-count an app launch.
+ */
+export function reportAppOpened(): void {
+  if (_openedSent) return;
+  _openedSent = true;
+  let timezone = '';
+  let locale = '';
+  try {
+    timezone = Intl.DateTimeFormat().resolvedOptions().timeZone || '';
+  } catch { /* leave empty; backend resolver/fallback handles it */ }
+  try {
+    locale = (typeof navigator !== 'undefined' && navigator.language) || '';
+  } catch { /* leave empty */ }
+  report('app', 'opened', { timezone, locale }, { immediate: true });
+}
+
 export function getSessionTraceState(): {
   appStartTs: number;
   lastTs: number;
@@ -99,5 +124,5 @@ export function getSessionTraceState(): {
   };
 }
 
-const serviceClient = { sync, report, getSessionTraceState, getRecentActions };
+const serviceClient = { sync, report, reportAppOpened, getSessionTraceState, getRecentActions };
 export default serviceClient;