diff --git a/backend/apps/service/analytics.py b/backend/apps/service/analytics.py new file mode 100644 index 00000000..e92c435f --- /dev/null +++ b/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) diff --git a/backend/apps/service/client.py b/backend/apps/service/client.py index cfb58655..376d127f 100644 --- a/backend/apps/service/client.py +++ b/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 diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py index 1f5ec87e..7983f589 100644 --- a/backend/apps/service/service.py +++ b/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} diff --git a/backend/apps/settings/models.py b/backend/apps/settings/models.py index 39e016c3..307d6471 100644 --- a/backend/apps/settings/models.py +++ b/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 diff --git a/backend/apps/settings/settings.py b/backend/apps/settings/settings.py index d135d086..0a560e27 100644 --- a/backend/apps/settings/settings.py +++ b/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) diff --git a/backend/requirements.txt b/backend/requirements.txt index 61b0595c..5e6ebb43 100644 --- a/backend/requirements.txt +++ b/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. diff --git a/frontend/src/app/Main.tsx b/frontend/src/app/Main.tsx index 7e4cb477..64963801 100644 --- a/frontend/src/app/Main.tsx +++ b/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 diff --git a/frontend/src/shared/serviceClient.ts b/frontend/src/shared/serviceClient.ts index fee2b254..780e3830 100644 --- a/frontend/src/shared/serviceClient.ts +++ b/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;