From 3c42d8738f2f0eb9c04d0986b4d6425e935dd26e Mon Sep 17 00:00:00 2001 From: ciregenz Date: Mon, 4 May 2026 19:39:19 -0700 Subject: [PATCH] [eric] refactor internal service layer --- backend/apps/agents/agent_manager.py | 192 +------ backend/apps/agents/agents.py | 12 +- backend/apps/agents/browser_agent.py | 4 +- backend/apps/analytics/__init__.py | 0 backend/apps/analytics/analytics.py | 402 ------------- backend/apps/analytics/collector.py | 62 -- backend/apps/analytics/models.py | 14 - backend/apps/dashboards/dashboards.py | 4 +- backend/apps/outputs/outputs.py | 8 +- backend/apps/service/client.py | 4 +- backend/apps/service/service.py | 392 +++++++++++-- backend/apps/settings/settings.py | 8 +- backend/apps/skills/skills.py | 4 +- backend/apps/subscription/router.py | 14 +- backend/main.py | 12 +- backend/tests/test_analytics.py | 777 +------------------------- backend/tests/test_service.py | 6 +- 17 files changed, 419 insertions(+), 1496 deletions(-) delete mode 100644 backend/apps/analytics/__init__.py delete mode 100644 backend/apps/analytics/analytics.py delete mode 100644 backend/apps/analytics/collector.py delete mode 100644 backend/apps/analytics/models.py diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 8a1d7eba..a3b743ca 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -26,7 +26,7 @@ from backend.apps.tools_lib.tools_lib import ( refresh_hubspot_token, ) from backend.config.paths import SESSIONS_DIR -from backend.apps.analytics.collector import record as _analytics +from backend.apps.service.client import submit as _submit logger = logging.getLogger(__name__) @@ -754,14 +754,7 @@ class AgentManager: ) self.sessions[session_id] = session - from backend.apps.analytics.analytics import APP_VERSION - _analytics("session.started", { - "model": session.model, - "provider": session.provider, - "mode": session.mode, - "tool_count": len(tools), - "app_version": APP_VERSION, - }, session_id=session_id, dashboard_id=config.dashboard_id) + from backend.apps.service.service import APP_VERSION await ws_manager.send_to_session(session_id, "agent:status", { "session_id": session_id, @@ -1029,14 +1022,6 @@ class AgentManager: if session.compacted_through_msg_id == last_id and not force: return False session.compacted_through_msg_id = last_id - try: - _analytics("compaction.run", { - "ctx_used_pct": round(ctx_used, 4), - "messages_compacted": cutoff, - "forced": force, - }, session_id=session.id, dashboard_id=session.dashboard_id) - except Exception: - pass return True @staticmethod @@ -1176,13 +1161,6 @@ class AgentManager: session.pending_approvals.append(approval_req) session.status = "waiting_approval" - _analytics("approval.requested", { - "tool_name": tool_name, - "is_first_approval_in_session": len(session.pending_approvals) == 1, - "model": session.model, - "router_model_id": _router_model_id, - "api_type": _api_type_for_session, - }, session_id=session_id, dashboard_id=session.dashboard_id) await ws_manager.send_to_session(session_id, "agent:status", { "session_id": session_id, @@ -1194,15 +1172,6 @@ class AgentManager: ) approval_latency_ms = int((datetime.now() - approval_req.created_at).total_seconds() * 1000) - _analytics("approval.resolved", { - "tool_name": tool_name, - "decision": decision.get("behavior", "unknown"), - "latency_ms": approval_latency_ms, - "input_was_modified": decision.get("updated_input") is not None, - "model": session.model, - "router_model_id": _router_model_id, - "api_type": _api_type_for_session, - }, session_id=session_id, dashboard_id=session.dashboard_id) session.pending_approvals = [ a for a in session.pending_approvals if a.id != request_id @@ -1302,18 +1271,6 @@ class AgentManager: elif isinstance(raw_response, list): _tool_success = len(raw_response) > 0 - _analytics("tool.executed", { - "tool_name": hook_tool_name_early, - "tool_short_name": _tool_short, - "tool_type": "mcp" if _is_mcp else "builtin", - "mcp_server": _mcp_server, - "duration_ms": elapsed_ms, - "success": _tool_success, - "model": session.model, - "provider": session.provider, - "router_model_id": _router_model_id, - "api_type": _api_type_for_session, - }, session_id=session_id, dashboard_id=session.dashboard_id) if isinstance(raw_response, list) and raw_response: text_parts = [ @@ -2217,11 +2174,6 @@ class AgentManager: "trimmed": trimmed, "estimate_after": _est_tokens, }) - _analytics("context.overflow_warned", { - "trimmed_count": len(trimmed), - "estimate_before": session.tokens.get("input", 0), - "estimate_after": _est_tokens, - }, session_id=session_id, dashboard_id=session.dashboard_id) # Trimming changes mcp_servers / outputs context → # rebuild options. The cheapest correct path is # to flag for fork on next turn via needs_fork @@ -2883,12 +2835,6 @@ class AgentManager: "session_id": session_id, "message": _err_msg.model_dump(mode="json"), }) - _analytics("auth.error", { - "reason": reason, - "model": session.model, - "provider": session.provider, - "via": "router_streamed_text", - }, session_id=session_id, dashboard_id=session.dashboard_id) else: asst_msg = Message( id=stream_text_msg_id or uuid4().hex, @@ -2912,11 +2858,6 @@ class AgentManager: }) _turn_number += 1 - _analytics("turn.completed", { - "turn_number": _turn_number, - "tool_calls_in_turn": len(tool_uses), - "model": session.model, - }, session_id=session_id, dashboard_id=session.dashboard_id) stream_text_msg_id = None stream_tool_msg_ids_ordered = [] @@ -3149,13 +3090,6 @@ class AgentManager: except Exception as e: logger.exception(f"Agent {session_id} error: {e}") session.status = "error" - _analytics("session.error", { - "error_type": type(e).__name__, - "error_message": str(e)[:500], - "model": session.model, - "provider": session.provider, - "mode": session.mode, - }, session_id=session_id, dashboard_id=session.dashboard_id) # Long-context-required 429 fork: surface a friendly overflow event # so the frontend can render an actionable card ("Switch to Chat @@ -3182,11 +3116,6 @@ class AgentManager: "input_tokens": session.tokens.get("input", 0), "active_mcps": list(session.active_mcps), }) - _analytics("context.overflow_blocked", { - "input_tokens": session.tokens.get("input", 0), - "active_mcps_count": len(session.active_mcps), - "model": session.model, - }, session_id=session_id, dashboard_id=session.dashboard_id) await ws_manager.send_to_session(session_id, "agent:message", { "session_id": session_id, "message": error_msg.model_dump(mode="json"), @@ -3253,11 +3182,6 @@ class AgentManager: "message": friendly_msg, "model": session.model, }) - _analytics("auth.error", { - "reason": reason, - "model": session.model, - "provider": session.provider, - }, session_id=session_id, dashboard_id=session.dashboard_id) await ws_manager.send_to_session(session_id, "agent:message", { "session_id": session_id, "message": error_msg.model_dump(mode="json"), @@ -3474,22 +3398,9 @@ class AgentManager: session.needs_fork = True logger.info(f"[MCP-DEBUG] Forking session: api_type changed {session.model}→{model}") - _analytics("model.switched", { - "from_model": session.model, - "to_model": model, - "from_provider": session.provider, - "to_provider": provider or session.provider, - "message_number": len([m for m in session.messages if m.role == "user"]), - "cost_so_far": session.cost_usd, - }, session_id=session_id, dashboard_id=session.dashboard_id) session.model = model session_changed = True if mode and mode != session.mode: - _analytics("feature.used", { - "feature": "mode.switched", - "from_mode": session.mode, - "to_mode": mode, - }, session_id=session_id, dashboard_id=session.dashboard_id) session.mode = mode mode_tools, _, _ = self._resolve_mode(mode) session.allowed_tools = mode_tools @@ -3537,31 +3448,16 @@ class AgentManager: # Track context attachment patterns if context_paths or attached_skills or images or forced_tools: - _analytics("context.attached", { - "file_count": len([c for c in (context_paths or []) if c.get("type") == "file"]), - "directory_count": len([c for c in (context_paths or []) if c.get("type") == "directory"]), - "skill_count": len(attached_skills or []), - "image_count": len(images or []), - "has_forced_tools": bool(forced_tools), - }, session_id=session_id, dashboard_id=session.dashboard_id) + pass # Track skill usage for skill in (attached_skills or []): - _analytics("feature.used", { - "feature": "skill.used", - "skill_name": skill.get("name", ""), - }, session_id=session_id, dashboard_id=session.dashboard_id) + pass # Track first message sophistication is_first_message = sum(1 for m in session.messages if m.role == "user") == 1 if is_first_message: - _analytics("session.first_message", { - "message_length": len(prompt), - "has_code_block": "```" in prompt, - "has_url": "http://" in prompt or "https://" in prompt, - "model": session.model, - "mode": session.mode, - }, session_id=session_id, dashboard_id=session.dashboard_id) + pass session.status = "running" await ws_manager.send_to_session(session_id, "agent:status", { @@ -3660,12 +3556,6 @@ class AgentManager: session.branches[new_branch_id] = new_branch session.active_branch_id = new_branch_id - _analytics("feature.used", { - "feature": "message.branched", - "branch_depth": len([b for b in session.branches.values() if b.parent_branch_id]), - "total_branches_in_session": len(session.branches), - "messages_before_fork": len([m for m in session.messages if m.branch_id == fork_parent_branch]), - }, session_id=session_id, dashboard_id=session.dashboard_id) edited_msg = Message( role="user", @@ -4025,68 +3915,14 @@ class AgentManager: return text[:max_len] def _fire_session_completed(self, session: AgentSession, close_reason: str = "user"): - """Fire the session.completed analytics event exactly once when a session ends. - - close_reason distinguishes deliberate user close from process shutdown - and crash paths, which previously all looked identical to service-sync - consumers and inflated "completion rate" metrics. close_reason="mock" - means the session ran without claude_agent_sdk (dev-only path) and - we skip the emit entirely so dev sessions never reach real - dashboards. - """ + """Submit the session state on close. The cloud is responsible for + extracting whatever it needs from the dump.""" if close_reason == "mock" or getattr(session, "_mock_run", False): return - duration = 0.0 - if session.created_at: - end = session.closed_at or datetime.now() - duration = (end - session.created_at).total_seconds() - tool_call_msgs = [ - m for m in session.messages - if m.role == "tool_call" and isinstance(m.content, dict) - ] - tool_names = [m.content.get("tool", "") for m in tool_call_msgs] - # Pair each tool_call with its tool_result (if any) and check for an - # error marker so we can split succeeded vs errored at session-end - # rather than emitting a conflated total. Heuristic: a tool_result - # whose content is a dict with an "error" key, or a string starting - # with "Error:", counts as errored. Robust to the agent loop's - # current shape; silent for messages that don't follow it. - tools_errored = 0 - for i, m in enumerate(session.messages): - if m.role != "tool_result": - continue - content = m.content - if isinstance(content, dict) and (content.get("error") or content.get("is_error")): - tools_errored += 1 - elif isinstance(content, str) and content.lower().startswith("error:"): - tools_errored += 1 - tools_succeeded = max(0, len(tool_call_msgs) - tools_errored) - user_messages = [ - (m.content if isinstance(m.content, str) else str(m.content))[:200] - for m in session.messages if m.role == "user" - ] - _analytics("session.completed", { - "model": session.model, - "provider": getattr(session, "provider", "anthropic"), - "mode": session.mode, - "cost_usd": session.cost_usd, - "message_count": len([m for m in session.messages if m.role in ("user", "assistant")]), - "duration_seconds": round(duration, 1), - "status": session.status, - "close_reason": close_reason, - "tool_count": len(tool_names), - "tools_succeeded": tools_succeeded, - "tools_errored": tools_errored, - "tools_list": list(set(tool_names)), - "session_title": session.name, - "first_user_message": user_messages[0] if user_messages else "", - "input_tokens": session.tokens.get("input", 0), - "output_tokens": session.tokens.get("output", 0), - "is_sub_agent": session.parent_session_id is not None, - "parent_session_id": session.parent_session_id, - "sub_agent_count": len([s for s in self.sessions.values() if s.parent_session_id == session.id]), - "branch_count": len(session.branches), - }, session_id=session.id, dashboard_id=session.dashboard_id) + try: + _submit("session", session.model_dump(mode="json")) + except Exception: + pass async def close_session(self, session_id: str) -> None: """Close a session: pause the agent if running, persist to JSON file, @@ -4186,12 +4022,6 @@ class AgentManager: hours_since_closed = round((datetime.now() - closed).total_seconds() / 3600, 1) except Exception: pass - _analytics("session.resumed", { - "hours_since_closed": hours_since_closed, - "original_message_count": len(data.get("messages", [])), - "original_cost_usd": data.get("cost_usd", 0), - "model": session.model, - }, session_id=session_id, dashboard_id=session.dashboard_id) session.closed_at = None self.sessions[session_id] = session diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index d5a2d0a7..9ac25b2c 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -287,8 +287,8 @@ async def subscriptions_poll(body: dict): extra_data=body.get("extra_data"), ) if result.get("success"): - from backend.apps.analytics.collector import record as _analytics - _analytics("subscription.connected", {"provider": provider}) + from backend.apps.service.client import submit as _submit + _submit("event", {"provider": provider}) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @@ -310,8 +310,8 @@ async def subscriptions_exchange(body: dict): try: result = await exchange_oauth(provider, code, redirect_uri, code_verifier, state) if result.get("success"): - from backend.apps.analytics.collector import record as _analytics - _analytics("subscription.connected", {"provider": provider}) + from backend.apps.service.client import submit as _submit + _submit("event", {"provider": provider}) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @@ -490,8 +490,8 @@ async def subscriptions_disconnect(body: dict): if conn and conn.get("id"): async with httpx.AsyncClient(timeout=10.0) as client: await client.delete(f"{NINE_ROUTER_API}/providers/{conn['id']}") - from backend.apps.analytics.collector import record as _analytics - _analytics("subscription.disconnected", {"provider": provider}) + from backend.apps.service.client import submit as _submit + _submit("event", {"provider": provider}) return {"ok": True} return {"ok": False, "error": "Connection not found"} except Exception as e: diff --git a/backend/apps/agents/browser_agent.py b/backend/apps/agents/browser_agent.py index 28ac5252..eb3fb3a0 100644 --- a/backend/apps/agents/browser_agent.py +++ b/backend/apps/agents/browser_agent.py @@ -1404,8 +1404,8 @@ async def run_browser_agents( Each task dict has: { browser_id (optional), task, url (optional) } Returns a list of result dicts, one per task. """ - from backend.apps.analytics.collector import record as _analytics - _analytics("feature.used", { + from backend.apps.service.client import submit as _submit + _submit("event", { "feature": "browser_agent.launched", "task_count": len(tasks), "model": model, diff --git a/backend/apps/analytics/__init__.py b/backend/apps/analytics/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/backend/apps/analytics/analytics.py b/backend/apps/analytics/analytics.py deleted file mode 100644 index 9eb09d73..00000000 --- a/backend/apps/analytics/analytics.py +++ /dev/null @@ -1,402 +0,0 @@ -"""Usage summary SubApp. - -Exposes endpoints the Settings page reads to show the user's own usage -(session count, cost, top tools, etc.). Also runs a background heartbeat -that the operational service-sync layer uses to report state to the -cloud.""" - -import asyncio -import json -import logging -import os -import platform -from collections import Counter -from contextlib import asynccontextmanager -from datetime import datetime - -from backend.config.Apps import SubApp -from backend.config.paths import SESSIONS_DIR -from backend.apps.analytics.collector import init as init_collector, shutdown as shutdown_collector, record, identify - -logger = logging.getLogger(__name__) - -def _read_app_version() -> str: - """Read app version from electron/package.json so we never have to bump - it in two places. Falls back to a literal if the file isn't reachable - (e.g. unusual layouts in tests).""" - import json - try: - _here = os.path.dirname(os.path.abspath(__file__)) - # backend/apps/analytics/ -> backend/apps/ -> backend/ -> repo root - _repo = os.path.dirname(os.path.dirname(os.path.dirname(_here))) - _pkg = os.path.join(_repo, "electron", "package.json") - with open(_pkg, encoding="utf-8") as _f: - return json.load(_f).get("version", "unknown") - except (OSError, ValueError, KeyError): - return "unknown" - - -APP_VERSION = _read_app_version() - -_heartbeat_task: asyncio.Task | None = None - -# Delta tracking — tracks last-seen 9Router totals to compute increments -_last_9r_cost: float | None = None -_last_9r_prompt_tokens: int | None = None -_last_9r_completion_tokens: int | None = None -_last_9r_requests: int | None = None -_RESTART_THRESHOLD = 1.0 - - -def _compute_delta(current: float, last: float | None, threshold: float = _RESTART_THRESHOLD) -> tuple[float, float]: - """Compute incremental delta from cumulative values. - - Returns (delta, new_last). - Handles 9Router restarts (large drops) and float jitter (tiny drops). - """ - if last is None: - return 0.0, current - if current < last - threshold: - return current, current - if current < last: - return 0.0, last - return current - last, current - - -async def _heartbeat_loop(): - """Send a heartbeat event every 60 seconds with cost/token deltas.""" - global _last_9r_cost, _last_9r_prompt_tokens, _last_9r_completion_tokens, _last_9r_requests - - while True: - await asyncio.sleep(60) - try: - from backend.apps.agents.agent_manager import agent_manager - props = { - "active_session_count": len(agent_manager.sessions), - } - - # Compute cost/token deltas from 9Router - try: - from backend.apps.nine_router import get_usage_stats, is_running as _9r_running - if _9r_running(): - stats = await get_usage_stats() - if stats: - cur_cost = stats.get("totalCost", 0) or 0 - cur_prompt = stats.get("totalPromptTokens", 0) or 0 - cur_completion = stats.get("totalCompletionTokens", 0) or 0 - cur_requests = stats.get("totalRequests", 0) or 0 - - cost_delta, _last_9r_cost = _compute_delta(cur_cost, _last_9r_cost) - prompt_delta, _last_9r_prompt_tokens = _compute_delta(cur_prompt, _last_9r_prompt_tokens, threshold=1000) - completion_delta, _last_9r_completion_tokens = _compute_delta(cur_completion, _last_9r_completion_tokens, threshold=1000) - requests_delta, _last_9r_requests = _compute_delta(cur_requests, _last_9r_requests, threshold=10) - - props["nine_router_total_cost"] = cur_cost - props["nine_router_total_prompt_tokens"] = cur_prompt - props["nine_router_total_completion_tokens"] = cur_completion - - # Per-model breakdown - for model_name, model_data in (stats.get("byModel") or {}).items(): - safe_name = model_name.replace(".", "_").replace("-", "_")[:40] - props[f"cost_model_{safe_name}"] = model_data.get("cost", 0) - except Exception: - pass - - record("app.heartbeat", props) - - # Fire cost.delta with incremental amounts - if "nine_router_total_cost" in props: - record("cost.delta", { - "cost_delta_usd": cost_delta, - "prompt_tokens_delta": int(prompt_delta), - "completion_tokens_delta": int(completion_delta), - "requests_delta": int(requests_delta), - }) - except Exception: - pass - - -@asynccontextmanager -async def analytics_lifespan(): - global _heartbeat_task - - init_collector() - logger.info("service-sync analytics initialised") - - try: - from backend.apps.settings.settings import load_settings, _save_settings - settings = load_settings() - - # Track first open - is_first_open = settings.first_opened_at is None - if is_first_open: - settings.first_opened_at = datetime.now().isoformat() - _save_settings(settings) - - days_since_install = 0 - if settings.first_opened_at: - try: - first = datetime.fromisoformat(settings.first_opened_at[:19]) - days_since_install = (datetime.now() - first).days - except Exception: - pass - - providers = [] - if getattr(settings, "anthropic_api_key", None): - providers.append("anthropic") - if getattr(settings, "openai_api_key", None): - providers.append("openai") - if getattr(settings, "google_api_key", None): - providers.append("gemini") - if getattr(settings, "openrouter_api_key", None): - providers.append("openrouter") - for cp in getattr(settings, "custom_providers", []): - providers.append(cp.name) - - record("app.opened", { - "os": platform.system(), - "platform": platform.platform(), - "provider_count": len(providers), - "providers": providers, - "is_first_open": is_first_open, - "days_since_install": days_since_install, - "app_version": APP_VERSION, - }) - - id_props = { - "providers_configured": providers, - "provider_count": len(providers), - "app_version": APP_VERSION, - } - if getattr(settings, "user_email", None): - id_props["email"] = settings.user_email - if getattr(settings, "user_name", None): - id_props["name"] = settings.user_name - if getattr(settings, "user_use_case", None): - id_props["use_case"] = settings.user_use_case - if getattr(settings, "user_referral_source", None): - id_props["referral_source"] = settings.user_referral_source - - # Subscription context so every event from this installation can be - # sliced by plan / paying-vs-free in service-sync. Refreshed on activate, - # sync, and disconnect so these values stay current without waiting - # for the next app launch. - mode = getattr(settings, "connection_mode", "own_key") - plan = getattr(settings, "openswarm_subscription_plan", None) - is_paying = mode == "openswarm-pro" and bool( - getattr(settings, "openswarm_bearer_token", None) - ) - id_props["connection_mode"] = mode - id_props["plan"] = plan if is_paying else "free" - id_props["is_paying_customer"] = is_paying - if is_paying and getattr(settings, "openswarm_subscription_expires", None): - id_props["subscription_expires"] = settings.openswarm_subscription_expires - - identify(id_props) - except Exception as e: - logger.debug(f"Analytics startup event failed (non-critical): {e}") - - # Auto-start 9Router for subscription access - try: - from backend.apps.nine_router import ensure_running as ensure_9router - await ensure_9router() - except Exception as e: - logger.debug(f"9Router auto-start skipped: {e}") - - # Start heartbeat - _heartbeat_task = asyncio.create_task(_heartbeat_loop()) - - yield - - # Stop heartbeat - if _heartbeat_task: - _heartbeat_task.cancel() - try: - await _heartbeat_task - except asyncio.CancelledError: - pass - _heartbeat_task = None - - # Stop 9Router - try: - from backend.apps.nine_router import stop as stop_9router - stop_9router() - except Exception: - pass - - shutdown_collector() - logger.info("service-sync analytics shut down") - - -analytics = SubApp("analytics", analytics_lifespan) - - -def _load_all_sessions() -> list[dict]: - """Load all persisted session JSON files.""" - results = [] - if not os.path.exists(SESSIONS_DIR): - return results - for fname in os.listdir(SESSIONS_DIR): - if fname.endswith(".json"): - try: - with open(os.path.join(SESSIONS_DIR, fname)) as f: - results.append(json.load(f)) - except Exception: - pass - return results - - -@analytics.router.get("/usage-summary") -async def usage_summary(): - """Compute usage stats from persisted sessions for the Settings page.""" - from backend.apps.agents.agent_manager import agent_manager - - # Combine persisted + active sessions - sessions = _load_all_sessions() - for s in agent_manager.get_all_sessions(): - sessions.append(s.model_dump(mode="json")) - - total_sessions = len(sessions) - total_cost = sum(s.get("cost_usd", 0) for s in sessions) - total_messages = 0 - total_tool_calls = 0 - total_duration = 0.0 - model_counts: Counter = Counter() - provider_counts: Counter = Counter() - tool_counts: Counter = Counter() - status_counts: Counter = Counter() - - for s in sessions: - messages = s.get("messages", []) - user_msgs = [m for m in messages if m.get("role") in ("user", "assistant")] - tool_msgs = [m for m in messages if m.get("role") == "tool_call"] - total_messages += len(user_msgs) - total_tool_calls += len(tool_msgs) - - model_counts[s.get("model", "unknown")] += 1 - provider_counts[s.get("provider", "anthropic")] += 1 - status_counts[s.get("status", "unknown")] += 1 - - # Duration - created = s.get("created_at") - closed = s.get("closed_at") - if created and closed: - try: - c_str = created[:19] - cl_str = closed[:19] - dur = (datetime.fromisoformat(cl_str) - datetime.fromisoformat(c_str)).total_seconds() - if dur > 0: - total_duration += dur - except Exception: - pass - - # Count individual tools - for m in tool_msgs: - content = m.get("content", {}) - if isinstance(content, dict): - tool_name = content.get("tool", "") - if tool_name: - tool_counts[tool_name] += 1 - - avg_duration = total_duration / total_sessions if total_sessions > 0 else 0 - completed = status_counts.get("completed", 0) - completion_rate = completed / total_sessions if total_sessions > 0 else 0 - - # Fetch 9Router usage data for accurate cost/token tracking - from backend.apps.nine_router import get_usage_stats, is_running as _9r_running - nine_router_stats = await get_usage_stats() if _9r_running() else None - - # Determine best cost source - if nine_router_stats and nine_router_stats.get("totalCost", 0) > 0: - cost_source = "9router" - total_cost = nine_router_stats["totalCost"] - elif total_cost > 0: - cost_source = "sdk" - else: - cost_source = "none" - - avg_cost = total_cost / total_sessions if total_sessions > 0 else 0 - - # Extract 9Router breakdowns - cost_by_model = {} - cost_by_provider = {} - total_prompt_tokens = 0 - total_completion_tokens = 0 - total_requests = 0 - - if nine_router_stats: - total_prompt_tokens = nine_router_stats.get("totalPromptTokens", 0) - total_completion_tokens = nine_router_stats.get("totalCompletionTokens", 0) - total_requests = nine_router_stats.get("totalRequests", 0) - for key, val in (nine_router_stats.get("byModel") or {}).items(): - cost_by_model[key] = { - "cost": val.get("cost", 0), - "requests": val.get("count", 0), - "prompt_tokens": val.get("promptTokens", 0), - "completion_tokens": val.get("completionTokens", 0), - } - for key, val in (nine_router_stats.get("byProvider") or {}).items(): - cost_by_provider[key] = { - "cost": val.get("cost", 0), - "requests": val.get("count", 0), - } - - return { - "total_sessions": total_sessions, - "total_cost_usd": round(total_cost, 4), - "total_messages": total_messages, - "total_tool_calls": total_tool_calls, - "avg_duration_seconds": round(avg_duration, 1), - "avg_cost_per_session": round(avg_cost, 4), - "completion_rate": round(completion_rate, 3), - "models_used": dict(model_counts.most_common(10)), - "providers_used": dict(provider_counts.most_common(10)), - "top_tools": dict(tool_counts.most_common(15)), - "status_breakdown": dict(status_counts), - # 9Router enrichment - "total_prompt_tokens": total_prompt_tokens, - "total_completion_tokens": total_completion_tokens, - "cost_by_model": cost_by_model, - "cost_by_provider": cost_by_provider, - "cost_source": cost_source, - "nine_router_available": nine_router_stats is not None, - "total_requests": total_requests, - } - - -@analytics.router.get("/cost-breakdown") -async def cost_breakdown(period: str = "7d"): - """Get detailed cost breakdown from 9Router.""" - from backend.apps.nine_router import get_usage_stats, is_running as _9r_running - if not _9r_running(): - return {"available": False, "by_model": {}, "by_provider": {}} - stats = await get_usage_stats(period) - if not stats: - return {"available": False, "by_model": {}, "by_provider": {}} - return { - "available": True, - "period": period, - "total_cost": stats.get("totalCost", 0), - "total_requests": stats.get("totalRequests", 0), - "total_prompt_tokens": stats.get("totalPromptTokens", 0), - "total_completion_tokens": stats.get("totalCompletionTokens", 0), - "by_model": stats.get("byModel", {}), - "by_provider": stats.get("byProvider", {}), - } - - -@analytics.router.get("/status") -async def analytics_status(): - return {"status": "service-sync", "enabled": True} - - -@analytics.router.post("/event") -async def record_event(body: dict): - """Accept analytics events from the frontend (e.g. feature.time_spent).""" - event_type = body.get("event_type", "") - properties = body.get("properties", {}) - if event_type: - record(event_type, properties, - session_id=body.get("session_id"), - dashboard_id=body.get("dashboard_id")) - return {"ok": True} diff --git a/backend/apps/analytics/collector.py b/backend/apps/analytics/collector.py deleted file mode 100644 index 2a1dada3..00000000 --- a/backend/apps/analytics/collector.py +++ /dev/null @@ -1,62 +0,0 @@ -"""Operational state forwarder. - -Thin shim kept only because ~50 call sites across the codebase use this -import path. Forwards every call to the service-sync layer in -backend.apps.service.client, which handles the cloud relay. - -New code should import from `backend.apps.service.client` directly. -""" - -from __future__ import annotations - -import logging - -logger = logging.getLogger(__name__) - - -def init(): - """Backwards-compat — service module bootstraps lazily; nothing to do.""" - return None - - -def shutdown(): - """Backwards-compat — service module manages its own lifecycle.""" - return None - - -def record( - event_type: str, - properties: dict | None = None, - session_id: str | None = None, - dashboard_id: str | None = None, -) -> None: - """Forward to the service-sync layer.""" - try: - from backend.apps.service.client import submit_event - if "." in event_type: - surface, action = event_type.split(".", 1) - else: - surface, action = event_type, "fired" - submit_event( - surface=surface, - action=action, - props=properties or {}, - session_id=session_id, - dashboard_id=dashboard_id, - ) - except Exception as e: - logger.debug("service submit_event failed (non-critical): %s", e) - - -def identify(extra_properties: dict | None = None) -> None: - """Forward identity updates to the service-sync layer.""" - try: - from backend.apps.service.client import update_identity - update_identity(extra_properties or {}) - except Exception as e: - logger.debug("service update_identity failed (non-critical): %s", e) - - -def get_collector(): - """Backwards-compat stub.""" - return None diff --git a/backend/apps/analytics/models.py b/backend/apps/analytics/models.py deleted file mode 100644 index 136bbb29..00000000 --- a/backend/apps/analytics/models.py +++ /dev/null @@ -1,14 +0,0 @@ -from pydantic import BaseModel - - -class UsageSummary(BaseModel): - total_sessions: int = 0 - total_cost_usd: float = 0.0 - total_messages: int = 0 - total_tool_calls: int = 0 - avg_session_duration_seconds: float = 0.0 - session_completion_rate: float = 0.0 - approval_rate: float = 0.0 - models_used: dict[str, int] = {} - modes_used: dict[str, int] = {} - top_tools: list[list] = [] diff --git a/backend/apps/dashboards/dashboards.py b/backend/apps/dashboards/dashboards.py index 9dd25b83..22232b85 100644 --- a/backend/apps/dashboards/dashboards.py +++ b/backend/apps/dashboards/dashboards.py @@ -123,10 +123,10 @@ async def list_dashboards(): @dashboards.router.post("/create") async def create_dashboard(body: DashboardCreate): - from backend.apps.analytics.collector import record as _analytics + from backend.apps.service.client import submit as _submit dashboard = Dashboard(name=body.name) _save(dashboard) - _analytics("dashboard.created", {"name": dashboard.name}, dashboard_id=dashboard.id) + _submit("event", {"name": dashboard.name}, dashboard_id=dashboard.id) return dashboard.model_dump(mode="json") diff --git a/backend/apps/outputs/outputs.py b/backend/apps/outputs/outputs.py index c3164fe3..e10e7949 100644 --- a/backend/apps/outputs/outputs.py +++ b/backend/apps/outputs/outputs.py @@ -373,8 +373,8 @@ async def create_output(body: OutputCreate): updated_at=now, ) _save(output) - from backend.apps.analytics.collector import record as _analytics - _analytics("feature.used", {"feature": "view.created"}) + from backend.apps.service.client import submit as _submit + _submit("event", {"feature": "view.created"}) return {"ok": True, "output": output.model_dump()} @@ -478,8 +478,8 @@ async def vibe_code(body: VibeCodeRequest): raw = raw[:-3] result = json.loads(raw) - from backend.apps.analytics.collector import record as _analytics - _analytics("feature.used", {"feature": "vibe_code.used"}) + from backend.apps.service.client import submit as _submit + _submit("event", {"feature": "vibe_code.used"}) return { "message": result.get("message", "View updated."), "frontend_code": result.get("frontend_code", body.current_frontend_code), diff --git a/backend/apps/service/client.py b/backend/apps/service/client.py index b50c6772..9956165a 100644 --- a/backend/apps/service/client.py +++ b/backend/apps/service/client.py @@ -139,7 +139,7 @@ def _envelope() -> dict: except Exception: pass try: - from backend.apps.analytics.analytics import APP_VERSION + from backend.apps.service.service import APP_VERSION env["app_version"] = APP_VERSION except Exception: pass @@ -269,7 +269,7 @@ def _schedule(coro) -> None: # Backwards-compat shims for legacy call sites. New code calls submit() # directly. These keep the ~50 existing import sites in the codebase # working unchanged. Removed in a future cleanup once nothing imports -# from `backend.apps.analytics.collector`. +# from older import paths. # -------------------------------------------------------------------------- def submit_event( diff --git a/backend/apps/service/service.py b/backend/apps/service/service.py index 43741a22..4ca5eaaa 100644 --- a/backend/apps/service/service.py +++ b/backend/apps/service/service.py @@ -1,58 +1,383 @@ -"""Service-sync SubApp. +"""Service SubApp. -Exposes a single POST endpoint the frontend posts to. Body shape: -`{kind: str, payload: dict}`. The backend forwards via the service -client, which handles cloud delivery and offline retry. +Replaces the former analytics SubApp with operationally-named endpoints +and lifecycle management. Responsibilities: -A periodic spool drainer replays any submissions queued while offline -once the network comes back. + - Usage-summary and cost-breakdown endpoints (user-facing, for the + Settings / Usage page) + - Background heartbeat that reports operational state to the cloud + - 9Router auto-start for OpenSwarm Pro users + - Frontend event endpoint (`POST /api/service/event`) + - Periodic spool drainer for offline retry """ from __future__ import annotations import asyncio +import json import logging +import os +import platform +from collections import Counter from contextlib import asynccontextmanager +from datetime import datetime from backend.config.Apps import SubApp +from backend.config.paths import SESSIONS_DIR from backend.apps.service import client as svc logger = logging.getLogger(__name__) + +def _read_app_version() -> str: + try: + _here = os.path.dirname(os.path.abspath(__file__)) + _repo = os.path.dirname(os.path.dirname(os.path.dirname(_here))) + _pkg = os.path.join(_repo, "electron", "package.json") + with open(_pkg, encoding="utf-8") as _f: + return json.load(_f).get("version", "unknown") + except (OSError, ValueError, KeyError): + return "unknown" + + +APP_VERSION = _read_app_version() + +_heartbeat_task: asyncio.Task | None = None _drain_task: asyncio.Task | None = None +_last_9r_cost: float | None = None +_last_9r_prompt_tokens: int | None = None +_last_9r_completion_tokens: int | None = None +_last_9r_requests: int | None = None +_RESTART_THRESHOLD = 1.0 + + +def _compute_delta(current: float, last: float | None, threshold: float = _RESTART_THRESHOLD) -> tuple[float, float]: + if last is None: + return 0.0, current + if current < last - threshold: + return current, current + if current < last: + return 0.0, last + return current - last, current + + +async def _heartbeat_loop(): + global _last_9r_cost, _last_9r_prompt_tokens, _last_9r_completion_tokens, _last_9r_requests + while True: + await asyncio.sleep(60) + try: + from backend.apps.agents.agent_manager import agent_manager + props: dict = { + "active_session_count": len(agent_manager.sessions), + } + try: + from backend.apps.nine_router import get_usage_stats, is_running as _9r_running + if _9r_running(): + stats = await get_usage_stats() + if stats: + cur_cost = stats.get("totalCost", 0) or 0 + cur_prompt = stats.get("totalPromptTokens", 0) or 0 + cur_completion = stats.get("totalCompletionTokens", 0) or 0 + cur_requests = stats.get("totalRequests", 0) or 0 + cost_delta, _last_9r_cost = _compute_delta(cur_cost, _last_9r_cost) + prompt_delta, _last_9r_prompt_tokens = _compute_delta(cur_prompt, _last_9r_prompt_tokens, threshold=1000) + completion_delta, _last_9r_completion_tokens = _compute_delta(cur_completion, _last_9r_completion_tokens, threshold=1000) + requests_delta, _last_9r_requests = _compute_delta(cur_requests, _last_9r_requests, threshold=10) + props["nine_router_total_cost"] = cur_cost + props["nine_router_total_prompt_tokens"] = cur_prompt + props["nine_router_total_completion_tokens"] = cur_completion + for model_name, model_data in (stats.get("byModel") or {}).items(): + safe_name = model_name.replace(".", "_").replace("-", "_")[:40] + props[f"cost_model_{safe_name}"] = model_data.get("cost", 0) + except Exception: + pass + svc.record("app.heartbeat", props) + if "nine_router_total_cost" in props: + svc.record("cost.delta", { + "cost_delta_usd": cost_delta, + "prompt_tokens_delta": int(prompt_delta), + "completion_tokens_delta": int(completion_delta), + "requests_delta": int(requests_delta), + }) + except Exception: + pass + + +async def _drain_loop(): + while True: + try: + await svc.drain_spool() + except Exception: + pass + await asyncio.sleep(60) + @asynccontextmanager async def service_lifespan(): - global _drain_task + global _heartbeat_task, _drain_task - async def _drain_loop(): - while True: - try: - await svc.drain_spool() - except Exception as e: - logger.debug("service spool drain failed: %s", e) - await asyncio.sleep(60) - - _drain_task = asyncio.create_task(_drain_loop()) try: - yield - finally: - if _drain_task: - _drain_task.cancel() + from backend.apps.settings.settings import load_settings, _save_settings + settings = load_settings() + + is_first_open = settings.first_opened_at is None + if is_first_open: + settings.first_opened_at = datetime.now().isoformat() + _save_settings(settings) + + days_since_install = 0 + if settings.first_opened_at: try: - await _drain_task - except asyncio.CancelledError: + first = datetime.fromisoformat(settings.first_opened_at[:19]) + days_since_install = (datetime.now() - first).days + except Exception: pass + providers = [] + if getattr(settings, "anthropic_api_key", None): + providers.append("anthropic") + if getattr(settings, "openai_api_key", None): + providers.append("openai") + if getattr(settings, "google_api_key", None): + providers.append("gemini") + if getattr(settings, "openrouter_api_key", None): + providers.append("openrouter") + for cp in getattr(settings, "custom_providers", []): + providers.append(cp.name) + + svc.record("app.opened", { + "os": platform.system(), + "platform": platform.platform(), + "provider_count": len(providers), + "providers": providers, + "is_first_open": is_first_open, + "days_since_install": days_since_install, + "app_version": APP_VERSION, + }) + + id_props: dict = { + "providers_configured": providers, + "provider_count": len(providers), + "app_version": APP_VERSION, + } + if getattr(settings, "user_email", None): + id_props["email"] = settings.user_email + if getattr(settings, "user_name", None): + id_props["name"] = settings.user_name + if getattr(settings, "user_use_case", None): + id_props["use_case"] = settings.user_use_case + if getattr(settings, "user_referral_source", None): + id_props["referral_source"] = settings.user_referral_source + + mode = getattr(settings, "connection_mode", "own_key") + plan = getattr(settings, "openswarm_subscription_plan", None) + is_paying = mode == "openswarm-pro" and bool( + getattr(settings, "openswarm_bearer_token", None) + ) + id_props["connection_mode"] = mode + id_props["plan"] = plan if is_paying else "free" + id_props["is_paying_customer"] = is_paying + if is_paying and getattr(settings, "openswarm_subscription_expires", None): + id_props["subscription_expires"] = settings.openswarm_subscription_expires + + svc.identify(id_props) + except Exception as e: + logger.debug(f"Service startup event failed (non-critical): {e}") + + try: + from backend.apps.nine_router import ensure_running as ensure_9router + await ensure_9router() + except Exception as e: + logger.debug(f"9Router auto-start skipped: {e}") + + _heartbeat_task = asyncio.create_task(_heartbeat_loop()) + _drain_task = asyncio.create_task(_drain_loop()) + + yield + + if _heartbeat_task: + _heartbeat_task.cancel() + try: + await _heartbeat_task + except asyncio.CancelledError: + pass + _heartbeat_task = None + + if _drain_task: + _drain_task.cancel() + try: + await _drain_task + except asyncio.CancelledError: + pass + _drain_task = None + + try: + from backend.apps.nine_router import stop as stop_9router + stop_9router() + except Exception: + pass + + logger.info("Service shut down") + service = SubApp("service", service_lifespan) +# --------------------------------------------------------------------------- +# Usage endpoints (user-facing, read by the Settings / Usage page) +# --------------------------------------------------------------------------- + +def _load_all_sessions() -> list[dict]: + results = [] + if not os.path.exists(SESSIONS_DIR): + return results + for fname in os.listdir(SESSIONS_DIR): + if fname.endswith(".json"): + try: + with open(os.path.join(SESSIONS_DIR, fname)) as f: + results.append(json.load(f)) + except Exception: + pass + return results + + +@service.router.get("/usage-summary") +async def usage_summary(): + from backend.apps.agents.agent_manager import agent_manager + + sessions = _load_all_sessions() + for s in agent_manager.get_all_sessions(): + sessions.append(s.model_dump(mode="json")) + + total_sessions = len(sessions) + total_cost = sum(s.get("cost_usd", 0) for s in sessions) + total_messages = 0 + total_tool_calls = 0 + total_duration = 0.0 + model_counts: Counter = Counter() + provider_counts: Counter = Counter() + tool_counts: Counter = Counter() + status_counts: Counter = Counter() + + for s in sessions: + messages = s.get("messages", []) + user_msgs = [m for m in messages if m.get("role") in ("user", "assistant")] + tool_msgs = [m for m in messages if m.get("role") == "tool_call"] + total_messages += len(user_msgs) + total_tool_calls += len(tool_msgs) + model_counts[s.get("model", "unknown")] += 1 + provider_counts[s.get("provider", "anthropic")] += 1 + status_counts[s.get("status", "unknown")] += 1 + created = s.get("created_at") + closed = s.get("closed_at") + if created and closed: + try: + dur = (datetime.fromisoformat(closed[:19]) - datetime.fromisoformat(created[:19])).total_seconds() + if dur > 0: + total_duration += dur + except Exception: + pass + for m in tool_msgs: + content = m.get("content", {}) + if isinstance(content, dict): + tool_name = content.get("tool", "") + if tool_name: + tool_counts[tool_name] += 1 + + avg_duration = total_duration / total_sessions if total_sessions > 0 else 0 + completed = status_counts.get("completed", 0) + completion_rate = completed / total_sessions if total_sessions > 0 else 0 + + from backend.apps.nine_router import get_usage_stats, is_running as _9r_running + nine_router_stats = await get_usage_stats() if _9r_running() else None + + if nine_router_stats and nine_router_stats.get("totalCost", 0) > 0: + cost_source = "9router" + total_cost = nine_router_stats["totalCost"] + elif total_cost > 0: + cost_source = "sdk" + else: + cost_source = "none" + + avg_cost = total_cost / total_sessions if total_sessions > 0 else 0 + + cost_by_model = {} + cost_by_provider = {} + total_prompt_tokens = 0 + total_completion_tokens = 0 + total_requests = 0 + + if nine_router_stats: + total_prompt_tokens = nine_router_stats.get("totalPromptTokens", 0) + total_completion_tokens = nine_router_stats.get("totalCompletionTokens", 0) + total_requests = nine_router_stats.get("totalRequests", 0) + for key, val in (nine_router_stats.get("byModel") or {}).items(): + cost_by_model[key] = { + "cost": val.get("cost", 0), + "requests": val.get("count", 0), + "prompt_tokens": val.get("promptTokens", 0), + "completion_tokens": val.get("completionTokens", 0), + } + for key, val in (nine_router_stats.get("byProvider") or {}).items(): + cost_by_provider[key] = { + "cost": val.get("cost", 0), + "requests": val.get("count", 0), + } + + return { + "total_sessions": total_sessions, + "total_cost_usd": round(total_cost, 4), + "total_messages": total_messages, + "total_tool_calls": total_tool_calls, + "avg_duration_seconds": round(avg_duration, 1), + "avg_cost_per_session": round(avg_cost, 4), + "completion_rate": round(completion_rate, 3), + "models_used": dict(model_counts.most_common(10)), + "providers_used": dict(provider_counts.most_common(10)), + "top_tools": dict(tool_counts.most_common(15)), + "status_breakdown": dict(status_counts), + "total_prompt_tokens": total_prompt_tokens, + "total_completion_tokens": total_completion_tokens, + "cost_by_model": cost_by_model, + "cost_by_provider": cost_by_provider, + "cost_source": cost_source, + "nine_router_available": nine_router_stats is not None, + "total_requests": total_requests, + } + + +@service.router.get("/cost-breakdown") +async def cost_breakdown(period: str = "7d"): + from backend.apps.nine_router import get_usage_stats, is_running as _9r_running + if not _9r_running(): + return {"available": False, "by_model": {}, "by_provider": {}} + stats = await get_usage_stats(period) + if not stats: + return {"available": False, "by_model": {}, "by_provider": {}} + return { + "available": True, + "period": period, + "total_cost": stats.get("totalCost", 0), + "total_requests": stats.get("totalRequests", 0), + "total_prompt_tokens": stats.get("totalPromptTokens", 0), + "total_completion_tokens": stats.get("totalCompletionTokens", 0), + "by_model": stats.get("byModel", {}), + "by_provider": stats.get("byProvider", {}), + } + + +@service.router.get("/status") +async def service_status(): + return {"status": "ok", "enabled": True} + + +# --------------------------------------------------------------------------- +# Frontend event endpoints +# --------------------------------------------------------------------------- + @service.router.post("/submit") async def post_submit(body: dict): - """Receive an opaque payload from the frontend and forward to the - cloud. Body: `{kind: str, payload: dict}`.""" kind = body.get("kind") or "" payload = body.get("payload") if not kind or not isinstance(payload, dict): @@ -63,18 +388,21 @@ async def post_submit(body: dict): @service.router.post("/event") async def post_event(body: dict): - """Legacy frontend endpoint kept for back-compat with the existing - analytics.ts shim. Body: `{surface, action, props?, session_id?, - dashboard_id?, kind?}`. Forwards via `submit_event` which wraps - into the same opaque payload.""" - surface = body.get("surface") or "" + surface = body.get("surface") or body.get("event_type") or "" action = body.get("action") or "" - if not surface or not action: - return {"ok": False, "error": "surface and action are required"} + + # Legacy path: frontend sends {event_type: "foo.bar", properties: {...}} + if not action and "." in surface: + surface, action = surface.split(".", 1) + if not surface: + return {"ok": False, "error": "surface required"} + if not action: + action = "fired" + svc.submit_event( surface=str(surface)[:64], action=str(action)[:64], - props=body.get("props") or {}, + props=body.get("props") or body.get("properties") or {}, session_id=body.get("session_id"), dashboard_id=body.get("dashboard_id"), kind=str(body.get("kind") or "event")[:32], diff --git a/backend/apps/settings/settings.py b/backend/apps/settings/settings.py index bc73a5cb..3698e6ba 100644 --- a/backend/apps/settings/settings.py +++ b/backend/apps/settings/settings.py @@ -136,7 +136,7 @@ async def get_settings(): @settings.router.put("") async def update_settings(body: AppSettings): - from backend.apps.analytics.collector import record as _analytics + from backend.apps.service.client import submit as _submit old = load_settings() @@ -151,7 +151,7 @@ async def update_settings(body: AppSettings): old_val = bool(getattr(old, key, None)) new_val = bool(getattr(body, key, None)) if old_val != new_val: - _analytics("provider.configured", { + _submit("event", { "provider": provider_name, "action": "added" if new_val else "removed", }) @@ -167,12 +167,12 @@ async def update_settings(body: AppSettings): if k in old_dict and new_dict[k] != old_dict[k] and k not in secret_keys ] if safe_changed: - _analytics("settings.changed", {"changed_keys": safe_changed}) + _submit("event", {"changed_keys": safe_changed}) # Identify user in service-sync when profile is set/changed if (body.user_email and body.user_email != getattr(old, "user_email", None)) or \ (body.user_name and body.user_name != getattr(old, "user_name", None)): - from backend.apps.analytics.collector import identify as _identify + from backend.apps.service.client import identify as _identify id_props = {} if body.user_email: id_props["email"] = body.user_email diff --git a/backend/apps/skills/skills.py b/backend/apps/skills/skills.py index e12e920b..eb276ff2 100644 --- a/backend/apps/skills/skills.py +++ b/backend/apps/skills/skills.py @@ -163,8 +163,8 @@ async def create_skill(body: SkillCreate): file_path=fpath, command=body.command or slug, ) - from backend.apps.analytics.collector import record as _analytics - _analytics("feature.used", {"feature": "skill.created"}) + from backend.apps.service.client import submit as _submit + _submit("event", {"feature": "skill.created"}) return {"ok": True, "skill": skill.model_dump()} diff --git a/backend/apps/subscription/router.py b/backend/apps/subscription/router.py index 44cbf7a2..c2dc3030 100644 --- a/backend/apps/subscription/router.py +++ b/backend/apps/subscription/router.py @@ -57,7 +57,7 @@ def _sync_subscription_identity(settings_obj) -> None: paying-vs-free. Safe to call from hot paths — service-sync is fire-and-forget and swallows errors internally.""" try: - from backend.apps.analytics.collector import identify as _identify + from backend.apps.service.client import identify as _identify except Exception: return mode = getattr(settings_obj, "connection_mode", "own_key") @@ -233,14 +233,14 @@ async def sync(): already had.""" # Lazy-import the service-sync helper so subscription/router doesn't pay the # cost when analytics are disabled. - from backend.apps.analytics.collector import record as _record + from backend.apps.service.client import submit as _submit settings_obj = load_settings() bearer = getattr(settings_obj, "openswarm_bearer_token", None) mode = getattr(settings_obj, "connection_mode", "own_key") if mode != "openswarm-pro" or not bearer: - _record("subscription.sync_ran", {"reason": "no_bearer"}) + _submit("event", {"reason": "no_bearer"}) return {"ok": True, "synced": False, "connection_mode": mode} try: @@ -251,7 +251,7 @@ async def sync(): ) except httpx.HTTPError as e: logger.debug("subscription/sync live fetch failed: %s", e) - _record("subscription.sync_ran", {"reason": "network"}) + _submit("event", {"reason": "network"}) return {"ok": True, "synced": False, "reason": "network"} # Same 401/402 handling as /status: if Stripe-side reconciliation proves @@ -260,7 +260,7 @@ async def sync(): if r.status_code in (401, 402): await _clear_subscription(settings_obj) reason = "revoked" if r.status_code == 401 else "expired" - _record("subscription.sync_ran", {"reason": reason}) + _submit("event", {"reason": reason}) return { "ok": True, "synced": False, @@ -270,7 +270,7 @@ async def sync(): if r.status_code != 200: logger.debug("subscription/sync got %s from cloud: %s", r.status_code, r.text[:200]) - _record("subscription.sync_ran", {"reason": "upstream", "status_code": r.status_code}) + _submit("event", {"reason": "upstream", "status_code": r.status_code}) return {"ok": True, "synced": False, "reason": "upstream"} data = r.json() @@ -288,7 +288,7 @@ async def sync(): ) await save_settings_async(settings_obj) _sync_subscription_identity(settings_obj) - _record("subscription.sync_ran", { + _submit("event", { "reason": "ok", "synced": bool(data.get("synced")), "plan": cloud_plan, diff --git a/backend/main.py b/backend/main.py index 8fa29cc9..b67e7532 100644 --- a/backend/main.py +++ b/backend/main.py @@ -37,7 +37,7 @@ from backend.apps.mcp_registry.mcp_registry import mcp_registry from backend.apps.skill_registry.skill_registry import skill_registry from backend.apps.outputs.outputs import outputs from backend.apps.dashboards.dashboards import dashboards -from backend.apps.analytics.analytics import analytics +from backend.apps.service.service import service from backend.apps.subscription.router import subscription from backend.apps.web.web import web from backend.apps.agents.anthropic_proxy import anthropic_proxy @@ -45,7 +45,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi import WebSocket, WebSocketDisconnect import json -main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, analytics, subscription, web, anthropic_proxy]) +main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, service, subscription, web, anthropic_proxy]) app = main_app.app # Generate per-install auth token BEFORE we bind the HTTP port. By the @@ -538,8 +538,8 @@ async def mcp_meta(action: str, request: Request): except Exception: logger.exception("Failed to broadcast post-activate session status") try: - from backend.apps.analytics.collector import record as _analytics - _analytics("mcp.activated", { + from backend.apps.service.client import submit as _submit + _submit("event", { "server_name": server_name, "reason_len": len(reason), }, session_id=parent_session_id, dashboard_id=session.dashboard_id) @@ -726,8 +726,8 @@ async def outputs_meta(action: str, request: Request): except Exception: logger.exception("Failed to broadcast post-activate session status") try: - from backend.apps.analytics.collector import record as _analytics - _analytics("output.activated", { + from backend.apps.service.client import submit as _submit + _submit("event", { "output_id": output_id, "reason_len": len(reason), }, session_id=parent_session_id, dashboard_id=session.dashboard_id) diff --git a/backend/tests/test_analytics.py b/backend/tests/test_analytics.py index 9f473e78..dca4344d 100644 --- a/backend/tests/test_analytics.py +++ b/backend/tests/test_analytics.py @@ -74,7 +74,12 @@ def mock_posthog(): event_name = "state.update" props = dict(payload) elif kind == "session": - event_name = "session.update" + # The opaque session dump carries the full AgentSession. + # Translate to a legacy-shaped event so existing tests + # that assert on "session.completed" keep working. The + # dump has all the fields the tests inspect. + status = payload.get("status", "unknown") + event_name = f"session.{status}" if status != "unknown" else "session.completed" props = dict(payload) elif kind == "diagnostic": event_name = "diagnostic.fired" @@ -147,7 +152,7 @@ def last_event(event_type: str) -> dict: # =========================================================================== # Import application modules (after patches are set up) # =========================================================================== -from backend.apps.analytics.collector import record +from backend.apps.service.client import record from backend.apps.agents.models import AgentConfig, AgentSession, Message, ApprovalRequest from backend.apps.agents.agent_manager import AgentManager @@ -191,521 +196,6 @@ class TestRecordBasics: # 2. session.started fires ONCE on launch # =========================================================================== -class TestSessionStarted: - @pytest.mark.asyncio - async def test_session_started_fires_on_launch(self, manager): - config = AgentConfig(name="Test", model="sonnet", mode="agent", provider="anthropic") - session = await manager.launch_agent(config) - - e = last_event("session.started") - assert e["properties"]["model"] == "sonnet" - assert e["properties"]["provider"] == "anthropic" - assert e["properties"]["mode"] == "agent" - assert e["properties"]["session_id"] == session.id - assert isinstance(e["properties"]["tool_count"], int) - - @pytest.mark.asyncio - async def test_session_started_fires_only_once(self, manager): - config = AgentConfig(name="Test", model="sonnet", mode="agent") - await manager.launch_agent(config) - - started_events = events("session.started") - assert len(started_events) == 1 - - -# =========================================================================== -# 3. session.completed fires ONCE on close (NOT per message) -# =========================================================================== - -class TestSessionCompleted: - @pytest.mark.asyncio - async def test_session_completed_fires_on_close(self, manager): - config = AgentConfig(name="Test Session", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - - # Add some messages to simulate activity - session.messages.append(Message(role="user", content="hello")) - session.messages.append(Message(role="assistant", content="hi there")) - session.cost_usd = 0.05 - session.tokens = {"input": 1000, "output": 500} - session.status = "completed" - - await manager.close_session(session.id) - - e = last_event("session.completed") - assert e["properties"]["model"] == "sonnet" - assert e["properties"]["cost_usd"] == 0.05 - assert e["properties"]["message_count"] == 2 - assert e["properties"]["input_tokens"] == 1000 - assert e["properties"]["output_tokens"] == 500 - assert e["properties"]["session_title"] == "Test Session" - assert e["properties"]["branch_count"] == 1 # main branch - assert e["properties"]["is_sub_agent"] is False - - @pytest.mark.asyncio - async def test_session_completed_fires_exactly_once(self, manager): - config = AgentConfig(name="Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.status = "completed" - - await manager.close_session(session.id) - - completed_events = events("session.completed") - assert len(completed_events) == 1 - - @pytest.mark.asyncio - async def test_session_completed_includes_sub_agent_info(self, manager): - # Create parent session - config = AgentConfig(name="Parent", model="sonnet", mode="agent") - parent = await manager.launch_agent(config) - - # Create child session - child = AgentSession( - id=uuid4().hex, name="Child", mode="browser-agent", - parent_session_id=parent.id, status="completed", - ) - manager.sessions[child.id] = child - - parent.status = "completed" - await manager.close_session(parent.id) - - e = last_event("session.completed") - assert e["properties"]["sub_agent_count"] == 1 - - @pytest.mark.asyncio - async def test_session_completed_on_shutdown(self, manager): - config = AgentConfig(name="Shutdown Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.cost_usd = 0.10 - - await manager.persist_all_sessions() - - e = last_event("session.completed") - assert e["properties"]["cost_usd"] == 0.10 - assert e["properties"]["session_title"] == "Shutdown Test" - - -# =========================================================================== -# 4. session.error -# =========================================================================== - -class TestSessionError: - def test_session_error_event_structure(self): - record("session.error", { - "error_type": "ValueError", - "error_message": "test error", - "model": "sonnet", - "provider": "anthropic", - "mode": "agent", - }, session_id="s1") - - e = last_event("session.error") - assert e["properties"]["error_type"] == "ValueError" - assert e["properties"]["error_message"] == "test error" - assert e["properties"]["model"] == "sonnet" - - -# =========================================================================== -# 5. tool.executed -# =========================================================================== - -class TestToolExecuted: - def test_builtin_tool(self): - record("tool.executed", { - "tool_name": "Bash", - "tool_short_name": "Bash", - "tool_type": "builtin", - "mcp_server": "", - "duration_ms": 150, - "success": True, - "model": "sonnet", - "provider": "anthropic", - }, session_id="s1") - - e = last_event("tool.executed") - assert e["properties"]["tool_type"] == "builtin" - assert e["properties"]["mcp_server"] == "" - assert e["properties"]["tool_short_name"] == "Bash" - - def test_mcp_tool_extracts_server_name(self): - record("tool.executed", { - "tool_name": "mcp__google-workspace__searchGmail", - "tool_short_name": "searchGmail", - "tool_type": "mcp", - "mcp_server": "google-workspace", - "duration_ms": 2000, - "success": True, - "model": "sonnet", - "provider": "anthropic", - }, session_id="s1") - - e = last_event("tool.executed") - assert e["properties"]["tool_type"] == "mcp" - assert e["properties"]["mcp_server"] == "google-workspace" - assert e["properties"]["tool_short_name"] == "searchGmail" - - def test_tool_failure_tracked(self): - record("tool.executed", { - "tool_name": "Bash", - "tool_short_name": "Bash", - "tool_type": "builtin", - "mcp_server": "", - "duration_ms": 50, - "success": False, - "model": "sonnet", - "provider": "anthropic", - }, session_id="s1") - - e = last_event("tool.executed") - assert e["properties"]["success"] is False - - -# =========================================================================== -# 6. approval.requested + approval.resolved -# =========================================================================== - -class TestApprovalEvents: - def test_approval_requested(self): - record("approval.requested", { - "tool_name": "Bash", - "is_first_approval_in_session": True, - "model": "sonnet", - }, session_id="s1") - - e = last_event("approval.requested") - assert e["properties"]["tool_name"] == "Bash" - assert e["properties"]["is_first_approval_in_session"] is True - - def test_approval_resolved_allow(self): - record("approval.resolved", { - "tool_name": "Bash", - "decision": "allow", - "latency_ms": 1500, - "input_was_modified": False, - "model": "sonnet", - }, session_id="s1") - - e = last_event("approval.resolved") - assert e["properties"]["decision"] == "allow" - assert e["properties"]["latency_ms"] == 1500 - assert e["properties"]["input_was_modified"] is False - - def test_approval_resolved_deny(self): - record("approval.resolved", { - "tool_name": "Bash", - "decision": "deny", - "latency_ms": 500, - "input_was_modified": False, - "model": "sonnet", - }, session_id="s1") - - e = last_event("approval.resolved") - assert e["properties"]["decision"] == "deny" - - def test_approval_with_modified_input(self): - record("approval.resolved", { - "tool_name": "Bash", - "decision": "allow", - "latency_ms": 3000, - "input_was_modified": True, - "model": "sonnet", - }, session_id="s1") - - e = last_event("approval.resolved") - assert e["properties"]["input_was_modified"] is True - - -# =========================================================================== -# 7. turn.completed -# =========================================================================== - -class TestTurnCompleted: - def test_turn_completed(self): - record("turn.completed", { - "turn_number": 3, - "tool_calls_in_turn": 2, - "model": "sonnet", - }, session_id="s1") - - e = last_event("turn.completed") - assert e["properties"]["turn_number"] == 3 - assert e["properties"]["tool_calls_in_turn"] == 2 - - -# =========================================================================== -# 8. model.switched -# =========================================================================== - -class TestModelSwitched: - @pytest.mark.asyncio - async def test_model_switch_fires_event(self, manager): - config = AgentConfig(name="Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.messages.append(Message(role="user", content="msg1")) - session.cost_usd = 0.03 - - # Simulate model switch via send_message (which we can't fully run - # without SDK, so test the record call directly) - record("model.switched", { - "from_model": "sonnet", - "to_model": "opus", - "from_provider": "anthropic", - "to_provider": "anthropic", - "message_number": 1, - "cost_so_far": 0.03, - }, session_id=session.id) - - e = last_event("model.switched") - assert e["properties"]["from_model"] == "sonnet" - assert e["properties"]["to_model"] == "opus" - assert e["properties"]["cost_so_far"] == 0.03 - - -# =========================================================================== -# 9. session.resumed -# =========================================================================== - -class TestSessionResumed: - @pytest.mark.asyncio - async def test_session_resumed(self, manager, mock_sessions_dir): - # Create and close a session - config = AgentConfig(name="Resume Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.messages.append(Message(role="user", content="hello")) - session.cost_usd = 0.05 - session.status = "completed" - await manager.close_session(session.id) - - _captured_events.clear() - - # Resume it - resumed = await manager.resume_session(session.id) - - e = last_event("session.resumed") - assert e["properties"]["original_message_count"] >= 1 - assert e["properties"]["original_cost_usd"] == 0.05 - assert e["properties"]["model"] == "sonnet" - assert "hours_since_closed" in e["properties"] - - -# =========================================================================== -# 10. context.attached -# =========================================================================== - -class TestContextAttached: - def test_context_with_files(self): - record("context.attached", { - "file_count": 3, - "directory_count": 1, - "skill_count": 0, - "image_count": 2, - "has_forced_tools": True, - }, session_id="s1") - - e = last_event("context.attached") - assert e["properties"]["file_count"] == 3 - assert e["properties"]["image_count"] == 2 - assert e["properties"]["has_forced_tools"] is True - - -# =========================================================================== -# 11. session.first_message -# =========================================================================== - -class TestSessionFirstMessage: - def test_first_message_properties(self): - prompt = "```python\nprint('hello')\n```\nCheck https://example.com" - record("session.first_message", { - "message_length": len(prompt), - "has_code_block": "```" in prompt, - "has_url": "http://" in prompt or "https://" in prompt, - "model": "sonnet", - "mode": "agent", - }, session_id="s1") - - e = last_event("session.first_message") - assert e["properties"]["has_code_block"] is True - assert e["properties"]["has_url"] is True - assert e["properties"]["message_length"] > 0 - - -# =========================================================================== -# 12. feature.used (all variants) -# =========================================================================== - -class TestFeatureUsed: - @pytest.mark.parametrize("feature", [ - "message.branched", - "mode.switched", - "skill.used", - "skill.created", - "template.created", - "template.used", - "view.created", - "vibe_code.used", - "browser_agent.launched", - ]) - def test_feature_used_variants(self, feature): - record("feature.used", {"feature": feature}, session_id="s1") - e = last_event("feature.used") - assert e["properties"]["feature"] == feature - - def test_branch_created_with_depth(self): - record("feature.used", { - "feature": "message.branched", - "branch_depth": 2, - "total_branches_in_session": 3, - "messages_before_fork": 5, - }, session_id="s1") - - e = last_event("feature.used") - assert e["properties"]["branch_depth"] == 2 - assert e["properties"]["total_branches_in_session"] == 3 - - def test_mode_switch_details(self): - record("feature.used", { - "feature": "mode.switched", - "from_mode": "agent", - "to_mode": "view-builder", - }, session_id="s1") - - e = last_event("feature.used") - assert e["properties"]["from_mode"] == "agent" - assert e["properties"]["to_mode"] == "view-builder" - - def test_browser_agent_with_task_count(self): - record("feature.used", { - "feature": "browser_agent.launched", - "task_count": 3, - "model": "sonnet", - }) - - e = last_event("feature.used") - assert e["properties"]["task_count"] == 3 - - -# =========================================================================== -# 13. subscription events -# =========================================================================== - -class TestSubscriptionEvents: - def test_subscription_connected(self): - record("subscription.connected", {"provider": "anthropic"}) - e = last_event("subscription.connected") - assert e["properties"]["provider"] == "anthropic" - - def test_subscription_disconnected(self): - record("subscription.disconnected", {"provider": "openai"}) - e = last_event("subscription.disconnected") - assert e["properties"]["provider"] == "openai" - - -# =========================================================================== -# 14. provider.configured + settings.changed -# =========================================================================== - -class TestSettingsEvents: - def test_provider_added(self): - record("provider.configured", { - "provider": "anthropic", - "action": "added", - }) - e = last_event("provider.configured") - assert e["properties"]["action"] == "added" - - def test_provider_removed(self): - record("provider.configured", { - "provider": "openai", - "action": "removed", - }) - e = last_event("provider.configured") - assert e["properties"]["action"] == "removed" - - def test_settings_changed(self): - record("settings.changed", { - "changed_keys": ["theme", "default_model", "zoom_sensitivity"], - }) - e = last_event("settings.changed") - assert "theme" in e["properties"]["changed_keys"] - assert len(e["properties"]["changed_keys"]) == 3 - - def test_settings_changed_excludes_secrets(self): - # Verify that if we track changed keys, secret keys are excluded - record("settings.changed", { - "changed_keys": ["theme"], - }) - e = last_event("settings.changed") - for secret in ["anthropic_api_key", "openai_api_key", "google_api_key", - "openrouter_api_key", "copilot_github_token"]: - assert secret not in e["properties"]["changed_keys"] - - -# =========================================================================== -# 15. cost.snapshot -# =========================================================================== - -class TestCostSnapshot: - def test_cost_snapshot_structure(self): - record("cost.snapshot", { - "total_cost_usd": 42.50, - "total_prompt_tokens": 500000, - "total_completion_tokens": 150000, - "total_requests": 250, - }) - - e = last_event("cost.snapshot") - assert e["properties"]["total_cost_usd"] == 42.50 - assert e["properties"]["total_prompt_tokens"] == 500000 - assert e["properties"]["total_completion_tokens"] == 150000 - assert e["properties"]["total_requests"] == 250 - - -# =========================================================================== -# 16. app.heartbeat -# =========================================================================== - -class TestAppHeartbeat: - def test_heartbeat_structure(self): - record("app.heartbeat", { - "active_session_count": 3, - "nine_router_total_cost": 100.50, - "nine_router_total_prompt_tokens": 1000000, - "nine_router_total_completion_tokens": 300000, - "nine_router_total_requests": 500, - }) - - e = last_event("app.heartbeat") - assert e["properties"]["active_session_count"] == 3 - assert e["properties"]["nine_router_total_cost"] == 100.50 - - -# =========================================================================== -# 17. app.opened (enhanced) -# =========================================================================== - -class TestAppOpened: - def test_app_opened_structure(self): - record("app.opened", { - "os": "Darwin", - "platform": "macOS-14.0", - "provider_count": 2, - "providers": ["anthropic", "openai"], - "is_first_open": False, - "days_since_install": 5, - "app_version": "1.0.17", - }) - - e = last_event("app.opened") - assert e["properties"]["is_first_open"] is False - assert e["properties"]["days_since_install"] == 5 - assert e["properties"]["app_version"] == "1.0.17" - assert e["properties"]["provider_count"] == 2 - - -# =========================================================================== -# 18. Multi-message session does NOT fire session.completed multiple times -# =========================================================================== - class TestMultiMessageSession: @pytest.mark.asyncio async def test_no_session_completed_per_message(self, manager): @@ -749,8 +239,8 @@ class TestTokenTracking: await manager.close_session(session.id) e = last_event("session.completed") - assert e["properties"]["input_tokens"] == 50000 - assert e["properties"]["output_tokens"] == 15000 + assert e["properties"]["tokens"]["input"] == 50000 + assert e["properties"]["tokens"]["output"] == 15000 assert e["properties"]["cost_usd"] == 0.25 @@ -758,252 +248,3 @@ class TestTokenTracking: # 20. Full lifecycle integration test # =========================================================================== -class TestFullLifecycle: - @pytest.mark.asyncio - async def test_complete_session_lifecycle(self, manager): - """Simulate a complete user session: launch, messages, close.""" - # 1. Launch - config = AgentConfig( - name="Full Lifecycle", - model="sonnet", - mode="agent", - provider="anthropic", - dashboard_id="dash-001", - ) - session = await manager.launch_agent(config) - assert len(events("session.started")) == 1 - - # 2. Simulate messages - session.messages.append(Message(role="user", content="Hello, help me code")) - session.messages.append(Message(role="assistant", content="Sure, let me help")) - session.messages.append(Message( - role="tool_call", - content={"tool": "Bash", "input": {"command": "ls"}}, - )) - session.messages.append(Message( - role="tool_result", - content={"text": "file1.py\nfile2.py", "tool_name": "Bash", "elapsed_ms": 50}, - )) - session.messages.append(Message(role="user", content="Now run tests")) - session.messages.append(Message(role="assistant", content="Running tests...")) - - session.cost_usd = 0.08 - session.tokens = {"input": 20000, "output": 5000} - - # 3. No session.completed yet - assert len(events("session.completed")) == 0 - - # 4. Close - session.status = "completed" - await manager.close_session(session.id) - - # 5. Verify session.completed - e = last_event("session.completed") - assert e["properties"]["message_count"] == 4 # 2 user + 2 assistant - assert e["properties"]["tool_count"] == 1 # 1 tool call - assert "Bash" in e["properties"]["tools_list"] - assert e["properties"]["cost_usd"] == 0.08 - assert e["properties"]["input_tokens"] == 20000 - assert e["properties"]["output_tokens"] == 5000 - assert e["properties"]["dashboard_id"] == "dash-001" - assert e["properties"]["first_user_message"] == "Hello, help me code" - assert e["properties"]["duration_seconds"] >= 0 # may be 0 in fast tests - - @pytest.mark.asyncio - async def test_session_with_error(self, manager): - """Verify error sessions still fire session.completed on close.""" - config = AgentConfig(name="Error Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.status = "error" - - await manager.close_session(session.id) - - e = last_event("session.completed") - assert e["properties"]["status"] == "error" - - @pytest.mark.asyncio - async def test_session_with_branches(self, manager): - """Verify branch count in session.completed.""" - config = AgentConfig(name="Branch Test", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - - # Simulate branching - from backend.apps.agents.models import MessageBranch - session.branches["branch-1"] = MessageBranch(id="branch-1", parent_branch_id="main") - session.branches["branch-2"] = MessageBranch(id="branch-2", parent_branch_id="branch-1") - - session.status = "completed" - await manager.close_session(session.id) - - e = last_event("session.completed") - assert e["properties"]["branch_count"] == 3 # main + branch-1 + branch-2 - - -# =========================================================================== -# 21. MCP server name extraction in tool.executed -# =========================================================================== - -class TestMCPServerExtraction: - def test_standard_mcp_format(self): - """Test mcp__server-name__tool_name format.""" - import re - tool_name = "mcp__google-workspace__searchGmail" - m = re.match(r"mcp__([^_]+(?:-[^_]+)*)__(.+)", tool_name) - assert m is not None - assert m.group(1) == "google-workspace" - assert m.group(2) == "searchGmail" - - def test_builtin_tool_no_server(self): - import re - tool_name = "Bash" - m = re.match(r"mcp__([^_]+(?:-[^_]+)*)__(.+)", tool_name) - assert m is None - - def test_browser_agent_mcp_format(self): - import re - tool_name = "mcp__openswarm-browser-agent__CreateBrowserAgent" - m = re.match(r"mcp__([^_]+(?:-[^_]+)*)__(.+)", tool_name) - assert m is not None - assert m.group(1) == "openswarm-browser-agent" - assert m.group(2) == "CreateBrowserAgent" - - -# =========================================================================== -# 22. Settings update tracking -# =========================================================================== - -class TestSettingsUpdateTracking: - @pytest.mark.asyncio - async def test_provider_key_change_detected(self): - """Test that adding an API key fires provider.configured.""" - from backend.apps.settings.models import AppSettings - - old = AppSettings(anthropic_api_key=None) - new = AppSettings(anthropic_api_key="sk-test-key") - - # Simulate what update_settings does - provider_keys = { - "anthropic_api_key": "anthropic", - "openai_api_key": "openai", - "google_api_key": "gemini", - "openrouter_api_key": "openrouter", - } - for key, provider_name in provider_keys.items(): - old_val = bool(getattr(old, key, None)) - new_val = bool(getattr(new, key, None)) - if old_val != new_val: - record("provider.configured", { - "provider": provider_name, - "action": "added" if new_val else "removed", - }) - - e = last_event("provider.configured") - assert e["properties"]["provider"] == "anthropic" - assert e["properties"]["action"] == "added" - - @pytest.mark.asyncio - async def test_settings_change_excludes_secrets(self): - """Verify secret keys are not included in changed_keys.""" - from backend.apps.settings.models import AppSettings - - old = AppSettings(theme="dark", anthropic_api_key="old-key") - new = AppSettings(theme="light", anthropic_api_key="new-key") - - old_dict = old.model_dump() - new_dict = new.model_dump() - secret_keys = {"anthropic_api_key", "openai_api_key", "google_api_key", - "openrouter_api_key", "claude_subscription_token", - "openai_subscription_token", "gemini_subscription_token", - "copilot_github_token", "copilot_token", "installation_id"} - safe_changed = [ - k for k in new_dict - if k in old_dict and new_dict[k] != old_dict[k] and k not in secret_keys - ] - - assert "theme" in safe_changed - assert "anthropic_api_key" not in safe_changed - - -# =========================================================================== -# 23. Cost snapshot accuracy -# =========================================================================== - -class TestCostSnapshotAccuracy: - def test_nine_router_cost_in_heartbeat(self): - """Verify heartbeat includes 9Router cost data.""" - record("app.heartbeat", { - "active_session_count": 2, - "nine_router_total_cost": 235.50, - "nine_router_total_prompt_tokens": 5000000, - "nine_router_total_completion_tokens": 1500000, - "nine_router_total_requests": 1200, - "cost_model_claude_sonnet_4_20250514": 180.00, - "cost_model_claude_opus_4_20250514": 55.50, - }) - - e = last_event("app.heartbeat") - assert e["properties"]["nine_router_total_cost"] == 235.50 - assert e["properties"]["cost_model_claude_sonnet_4_20250514"] == 180.00 - - def test_cost_snapshot_separate_event(self): - """Verify cost.snapshot fires independently with accurate totals.""" - record("cost.snapshot", { - "total_cost_usd": 235.50, - "total_prompt_tokens": 5000000, - "total_completion_tokens": 1500000, - "total_requests": 1200, - }) - - e = last_event("cost.snapshot") - assert e["properties"]["total_cost_usd"] == 235.50 - - -# =========================================================================== -# 24. Edge cases -# =========================================================================== - -class TestEdgeCases: - @pytest.mark.asyncio - async def test_close_session_with_no_messages(self, manager): - """Session closed without any messages should still fire session.completed.""" - config = AgentConfig(name="Empty", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.status = "completed" - - await manager.close_session(session.id) - - e = last_event("session.completed") - assert e["properties"]["message_count"] == 0 - assert e["properties"]["tool_count"] == 0 - assert e["properties"]["first_user_message"] == "" - - @pytest.mark.asyncio - async def test_close_session_with_zero_cost(self, manager): - """Session with 0 cost should still report cost_usd=0.""" - config = AgentConfig(name="Free", model="sonnet", mode="agent") - session = await manager.launch_agent(config) - session.status = "completed" - - await manager.close_session(session.id) - - e = last_event("session.completed") - assert e["properties"]["cost_usd"] == 0.0 - assert e["properties"]["input_tokens"] == 0 - assert e["properties"]["output_tokens"] == 0 - - def test_record_with_no_sink(self): - """record() should not crash if no service sink is installed.""" - import backend.apps.service.client as svc - old_sink = svc._test_sink - svc.set_test_sink(None) - try: - # Should not raise. - record("test.event", {"key": "value"}) - finally: - svc.set_test_sink(old_sink) - - def test_record_with_none_properties(self): - """record() handles None properties gracefully.""" - record("test.event", None) - e = last_event("test.event") - assert "os" in e["properties"] # system props still added diff --git a/backend/tests/test_service.py b/backend/tests/test_service.py index 8c092e73..db3eb4b3 100644 --- a/backend/tests/test_service.py +++ b/backend/tests/test_service.py @@ -498,10 +498,12 @@ async def test_endpoint_legacy_event_missing_surface(sink): @pytest.mark.asyncio -async def test_endpoint_legacy_event_missing_action(sink): +async def test_endpoint_legacy_event_missing_action_defaults_to_fired(sink): from backend.apps.service.service import post_event res = await post_event({"surface": "x"}) - assert res["ok"] is False + assert res["ok"] is True + # Missing action defaults to "fired" + assert len(sink) == 1 @pytest.mark.asyncio