diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 302a5e03..3e24da9f 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/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) diff --git a/backend/apps/service/analytics.py b/backend/apps/service/analytics.py index 1e938d2c..0cb49189 100644 --- a/backend/apps/service/analytics.py +++ b/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: diff --git a/backend/apps/service/analytics_agent_bridge.py b/backend/apps/service/analytics_agent_bridge.py new file mode 100644 index 00000000..f3081cd6 --- /dev/null +++ b/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), + ) diff --git a/backend/apps/service/analytics_frontend_bridge.py b/backend/apps/service/analytics_frontend_bridge.py new file mode 100644 index 00000000..64d446e7 --- /dev/null +++ b/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) diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py index 6b7da0d7..b5831a9c 100644 --- a/backend/apps/service/service.py +++ b/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")