[eric] delete miscelaneous scrapped code

.gitignore

@@ -60,3 +60,4 @@ htmlcov/
 Thumbs.db
 ehthumbs.db
 desktop.ini
+frontend/tsconfig.tsbuildinfo

backend/apps/agents/agent_manager.py

@@ -2413,6 +2413,19 @@ class AgentManager:
                         pass
                 if _turn_started_ts is not None:
                     _turn_total_ms = int((time.time() - _turn_started_ts) * 1000)
+                    # Accumulate into session-level "agent active time" and
+                    # the per-model breakdown so a session that spans
+                    # multiple turns reports the total wall-clock time the
+                    # agent was running. Per-model bucket uses the model
+                    # active *now* (model can be switched mid-turn but the
+                    # current value is the right attribution for the work
+                    # just produced).
+                    try:
+                        session.agent_active_ms = int(getattr(session, "agent_active_ms", 0) or 0) + _turn_total_ms
+                        m = session.model or "unknown"
+                        session.time_per_model[m] = int(session.time_per_model.get(m, 0)) + _turn_total_ms
+                    except Exception:
+                        pass
                 if _turn_thinking_msg_id is None:
                     _turn_thinking_msg_id = uuid4().hex
                 # Combined token total for the pill — input + output for
@@ -4015,7 +4028,7 @@ class AgentManager:
         """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 PostHog
+        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

backend/apps/agents/models.py

@@ -89,6 +89,15 @@ class AgentSession(BaseModel):
     closed_at: Optional[datetime] = None
     cost_usd: float = 0.0
     tokens: dict[str, int] = Field(default_factory=lambda: {"input": 0, "output": 0})
+    # Total wall-clock ms the agent spent in `status="running"`. Accumulates
+    # across turns; persists across resume. Used by the session-close
+    # report so we can report "agent active time" alongside total session
+    # duration. Off by default so legacy sessions deserialize cleanly.
+    agent_active_ms: int = 0
+    # Accumulated wall-clock ms spent on each model. Updated when the
+    # active model changes (model switch) or on close. Lets dashboards
+    # answer "how long did each model run?" without inferring from turns.
+    time_per_model: dict[str, int] = Field(default_factory=dict)
     messages: list[Message] = Field(default_factory=list)
     pending_approvals: list[ApprovalRequest] = Field(default_factory=list)
     branches: dict[str, "MessageBranch"] = Field(default_factory=lambda: {"main": MessageBranch(id="main")})

backend/apps/analytics/analytics.py

@@ -1,4 +1,9 @@
-"""Analytics SubApp: PostHog for product analytics + local usage summary from session data."""
+"""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
@@ -116,7 +121,7 @@ async def analytics_lifespan():
     global _heartbeat_task
 
     init_collector()
-    logger.info("PostHog analytics initialised")
+    logger.info("service-sync analytics initialised")
 
     try:
         from backend.apps.settings.settings import load_settings, _save_settings
@@ -173,7 +178,7 @@ async def analytics_lifespan():
             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 PostHog. Refreshed on activate,
+        # 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")
@@ -220,7 +225,7 @@ async def analytics_lifespan():
         pass
 
     shutdown_collector()
-    logger.info("PostHog analytics shut down")
+    logger.info("service-sync analytics shut down")
 
 
 analytics = SubApp("analytics", analytics_lifespan)
@@ -382,7 +387,7 @@ async def cost_breakdown(period: str = "7d"):
 
 @analytics.router.get("/status")
 async def analytics_status():
-    return {"status": "posthog", "enabled": True}
+    return {"status": "service-sync", "enabled": True}
 
 
 @analytics.router.post("/event")

backend/apps/analytics/collector.py

@@ -1,75 +1,27 @@
-"""PostHog-only analytics collector.
+"""Operational state forwarder.
 
-All events go directly to PostHog. No local SQLite storage.
+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.
 
-Usage from any module:
-    from backend.apps.analytics.collector import record
-    record("session.started", {"model": "opus"}, session_id="abc123")
+New code should import from `backend.apps.service.client` directly.
 """
 
-import logging
-import platform
-from uuid import uuid4
+from __future__ import annotations
 
-from posthog import Posthog
+import logging
 
 logger = logging.getLogger(__name__)
 
-POSTHOG_API_KEY = "phc_KdVLvAdjCuHeacFoDm1CM1Gb23XikewRqlX67Mj6TNB"
-POSTHOG_HOST = "https://us.i.posthog.com"
-
-_posthog: Posthog | None = None
-_installation_id: str | None = None
-
 
 def init():
-    """Initialise PostHog. Called once at app startup."""
-    global _posthog
-    if _posthog is None:
-        _posthog = Posthog(
-            project_api_key=POSTHOG_API_KEY,
-            host=POSTHOG_HOST,
-        )
-    return _posthog
+    """Backwards-compat — service module bootstraps lazily; nothing to do."""
+    return None
 
 
 def shutdown():
-    """Flush and close. Called at app shutdown."""
-    global _posthog
-    if _posthog:
-        try:
-            _posthog.shutdown()
-        except Exception:
-            pass
-        _posthog = None
-
-
-def _get_installation_id() -> str:
-    """Get or create a stable anonymous installation ID."""
-    global _installation_id
-    if _installation_id:
-        return _installation_id
-    try:
-        from backend.apps.settings.settings import load_settings, _save_settings
-        settings = load_settings()
-        iid = getattr(settings, "installation_id", None)
-        if not iid:
-            iid = uuid4().hex
-            settings.installation_id = iid
-            _save_settings(settings)
-        _installation_id = iid
-    except Exception:
-        _installation_id = uuid4().hex
-    return _installation_id
-
-
-def _is_opted_in() -> bool:
-    """Check if user has opted in to analytics."""
-    try:
-        from backend.apps.settings.settings import load_settings
-        return getattr(load_settings(), "analytics_opt_in", True)
-    except Exception:
-        return True
+    """Backwards-compat — service module manages its own lifecycle."""
+    return None
 
 
 def record(
@@ -77,47 +29,34 @@ def record(
     properties: dict | None = None,
     session_id: str | None = None,
     dashboard_id: str | None = None,
-):
-    """Record an analytics event to PostHog."""
-    if not _posthog:
-        return
-
-    props = {**(properties or {})}
-    if session_id:
-        props["session_id"] = session_id
-    if dashboard_id:
-        props["dashboard_id"] = dashboard_id
-    props["os"] = platform.system()
-    props["platform"] = platform.platform()
-
+) -> None:
+    """Forward to the service-sync layer."""
     try:
-        _posthog.capture(
-            event_type,
-            distinct_id=_get_installation_id(),
-            properties=props,
+        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(f"PostHog capture failed (non-critical): {e}")
+        logger.debug("service submit_event failed (non-critical): %s", e)
 
 
-def identify(extra_properties: dict | None = None):
-    """Set person properties on the current installation's PostHog profile."""
-    if not _posthog:
-        return
-
+def identify(extra_properties: dict | None = None) -> None:
+    """Forward identity updates to the service-sync layer."""
     try:
-        _posthog.set(
-            distinct_id=_get_installation_id(),
-            properties={
-                "os": platform.system(),
-                "platform": platform.platform(),
-                **(extra_properties or {}),
-            },
-        )
+        from backend.apps.service.client import update_identity
+        update_identity(extra_properties or {})
     except Exception as e:
-        logger.debug(f"PostHog identify failed (non-critical): {e}")
+        logger.debug("service update_identity failed (non-critical): %s", e)
 
 
 def get_collector():
-    """Backward compat — returns None since we no longer have a local collector."""
+    """Backwards-compat stub."""
     return None

backend/apps/service/buffer.py

@@ -0,0 +1,138 @@
+"""Bounded SQLite spool for offline operational submissions.
+
+When the desktop is offline (laptop closed, no internet, cloud unreachable),
+the service-sync layer can't reach `api.openswarm.com`. Rather than drop
+data on the floor, we spool submissions to a small SQLite file and replay
+them on the next online tick. The spool is bounded — when full, the oldest
+entries are dropped — so it can never balloon to a problem.
+
+Single file, single table, single thread guarded by a sqlite3 connection's
+implicit lock. No concurrency model beyond "don't write from two processes
+at once."
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import sqlite3
+import threading
+from contextlib import contextmanager
+from typing import Iterator, Optional
+
+logger = logging.getLogger(__name__)
+
+# Cap the spool at 50 MB on disk. SQLite's overhead means the actual ceiling
+# on retained payloads is somewhat smaller, which is fine — this is a
+# best-effort cushion, not a guaranteed retention window.
+_MAX_BYTES = 50 * 1024 * 1024
+
+# Trim 25% when we cross the cap so we don't trim on every insert.
+_TRIM_TARGET_FRACTION = 0.75
+
+_lock = threading.Lock()
+
+
+@contextmanager
+def _conn(spool_path: str) -> Iterator[sqlite3.Connection]:
+    """Open a connection that auto-commits and ensures the table exists.
+    Caller holds `_lock` for the duration of the context."""
+    os.makedirs(os.path.dirname(spool_path), exist_ok=True)
+    c = sqlite3.connect(spool_path, isolation_level=None, timeout=5.0)
+    try:
+        c.execute(
+            "CREATE TABLE IF NOT EXISTS spool ("
+            "  id INTEGER PRIMARY KEY AUTOINCREMENT,"
+            "  kind TEXT NOT NULL,"
+            "  payload TEXT NOT NULL,"
+            "  created_at REAL NOT NULL"
+            ")"
+        )
+        yield c
+    finally:
+        c.close()
+
+
+def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
+    """Append a submission to the spool. Drops the oldest if the spool is
+    over the byte cap."""
+    body = json.dumps(payload, separators=(",", ":"), default=str)
+    with _lock, _conn(spool_path) as c:
+        c.execute(
+            "INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
+            (kind, body, now),
+        )
+        # Cheap size check — only run trim when stat says we're over.
+        try:
+            size = os.path.getsize(spool_path)
+        except OSError:
+            size = 0
+        if size > _MAX_BYTES:
+            target = int(_MAX_BYTES * _TRIM_TARGET_FRACTION)
+            # Delete oldest rows until we're back under target. Use a
+            # reasonable batch size so we don't block forever.
+            for _ in range(64):
+                row = c.execute("SELECT id FROM spool ORDER BY id ASC LIMIT 1").fetchone()
+                if not row:
+                    break
+                c.execute("DELETE FROM spool WHERE id = ?", (row[0],))
+                try:
+                    new_size = os.path.getsize(spool_path)
+                except OSError:
+                    new_size = 0
+                if new_size <= target:
+                    break
+            # VACUUM is expensive; only run if we still appear oversized after
+            # trimming, otherwise free pages get reused on next insert.
+            try:
+                if os.path.getsize(spool_path) > _MAX_BYTES:
+                    c.execute("VACUUM")
+            except (OSError, sqlite3.DatabaseError):
+                pass
+
+
+def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
+    """Read up to `batch_size` oldest entries. Returns (id, kind, payload)
+    triples; caller is responsible for calling `acknowledge(ids)` once the
+    cloud accepts them."""
+    if not os.path.exists(spool_path):
+        return []
+    with _lock, _conn(spool_path) as c:
+        rows = c.execute(
+            "SELECT id, kind, payload FROM spool ORDER BY id ASC LIMIT ?",
+            (batch_size,),
+        ).fetchall()
+    out: list[tuple[int, str, dict]] = []
+    for rid, kind, body in rows:
+        try:
+            out.append((rid, kind, json.loads(body)))
+        except json.JSONDecodeError:
+            # Corrupt row — discard so it doesn't block draining behind it.
+            with _lock, _conn(spool_path) as c:
+                c.execute("DELETE FROM spool WHERE id = ?", (rid,))
+            logger.warning("Dropped corrupt spool row id=%s", rid)
+    return out
+
+
+def acknowledge(spool_path: str, ids: list[int]) -> None:
+    """Remove rows the cloud has accepted."""
+    if not ids:
+        return
+    with _lock, _conn(spool_path) as c:
+        c.executemany("DELETE FROM spool WHERE id = ?", [(i,) for i in ids])
+
+
+def count(spool_path: str) -> int:
+    """Return the number of pending entries. Used for tests + debug UI."""
+    if not os.path.exists(spool_path):
+        return 0
+    with _lock, _conn(spool_path) as c:
+        row = c.execute("SELECT COUNT(*) FROM spool").fetchone()
+    return int(row[0]) if row else 0
+
+
+def clear(spool_path: str) -> None:
+    """Delete all pending entries. Tests + manual reset only."""
+    with _lock, _conn(spool_path) as c:
+        c.execute("DELETE FROM spool")

backend/apps/service/client.py

@@ -0,0 +1,330 @@
+"""Operational state forwarder.
+
+Single public surface: `submit(kind, payload)`. The desktop hands off
+opaque payload dicts; the cloud at api.openswarm.com is responsible for
+parsing and routing them. The desktop has no schema knowledge.
+
+Three `kind` values are accepted — they're the routing primitive the
+cloud needs to send the payload to the right backend handler. The shape
+of `payload` is opaque from the desktop's perspective; the cloud knows
+how to read it.
+
+  - "state":      lightweight periodic ping
+  - "session":    full session dump on close
+  - "diagnostic": error / bug-report context
+
+Submissions that fail to deliver get spooled to a small SQLite file and
+replayed on the next online tick. Bounded to 50 MB.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import platform
+import time
+from typing import Any, Optional
+from uuid import uuid4
+
+import httpx
+
+from backend.apps.service import buffer
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_BASE = "https://api.openswarm.com"
+_PATH_BY_KIND = {
+    "state": "/api/service/state",
+    "session": "/api/service/sync",
+    "diagnostic": "/api/service/diagnostics",
+    "event": "/api/service/event",
+}
+
+_TIMEOUT_SECONDS = 5.0
+_MAX_INFLIGHT = 16
+
+_test_sink: Optional[Any] = None
+_install_id: Optional[str] = None
+_user_id: Optional[str] = None
+_inflight = 0
+_inflight_lock = asyncio.Lock()
+_drain_lock = asyncio.Lock()
+
+
+def _spool_path() -> str:
+    try:
+        from backend.config.paths import SETTINGS_DIR
+        return os.path.join(SETTINGS_DIR, "service_spool.db")
+    except Exception:
+        return os.path.expanduser("~/.openswarm/data/service_spool.db")
+
+
+def set_test_sink(fn: Optional[Any]) -> None:
+    """Test seam — receives every submission instead of the network."""
+    global _test_sink
+    _test_sink = fn
+
+
+def _get_install_id() -> str:
+    global _install_id
+    if _install_id:
+        return _install_id
+    try:
+        from backend.apps.settings.settings import load_settings, _save_settings
+        s = load_settings()
+        iid = getattr(s, "installation_id", None)
+        if not iid:
+            iid = uuid4().hex
+            s.installation_id = iid
+            _save_settings(s)
+        _install_id = iid
+    except Exception:
+        _install_id = uuid4().hex
+    return _install_id
+
+
+def _get_user_id() -> Optional[str]:
+    global _user_id
+    if _user_id:
+        return _user_id
+    try:
+        from backend.apps.settings.settings import load_settings
+        s = load_settings()
+        return getattr(s, "user_email", None) or None
+    except Exception:
+        return None
+
+
+def set_user_id(uid: Optional[str]) -> None:
+    global _user_id
+    _user_id = uid or None
+
+
+def _is_enabled(kind: str) -> bool:
+    """Honour user opt-out. Diagnostic always flows (errors block usability);
+    state + session honour the toggle."""
+    if kind == "diagnostic":
+        return True
+    try:
+        from backend.apps.settings.settings import load_settings
+        s = load_settings()
+        mode = getattr(s, "service_diagnostics_mode", None)
+        if mode == "minimal":
+            return False
+        if mode is None:
+            return bool(getattr(s, "analytics_opt_in", True))
+        return True
+    except Exception:
+        return True
+
+
+def _envelope() -> dict:
+    """Identity + environment metadata stamped on every submission."""
+    env: dict[str, Any] = {"install_id": _get_install_id()}
+    uid = _get_user_id()
+    if uid:
+        env["user_id"] = uid
+    try:
+        env["os"] = platform.system()
+        env["os_version"] = platform.release()
+        env["device_type"] = "desktop"
+    except Exception:
+        pass
+    try:
+        import datetime as _dt
+        local_tz = _dt.datetime.now().astimezone().tzinfo
+        if local_tz:
+            env["timezone"] = str(local_tz)
+    except Exception:
+        pass
+    try:
+        from backend.apps.analytics.analytics import APP_VERSION
+        env["app_version"] = APP_VERSION
+    except Exception:
+        pass
+    return env
+
+
+def _base_url() -> str:
+    try:
+        from backend.apps.settings.settings import load_settings
+        from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
+        s = load_settings()
+        return (getattr(s, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
+    except Exception:
+        return _DEFAULT_BASE
+
+
+async def _post(path: str, body: dict) -> bool:
+    url = f"{_base_url()}{path}"
+    try:
+        async with httpx.AsyncClient(timeout=_TIMEOUT_SECONDS) as c:
+            r = await c.post(url, json=body)
+        return 200 <= r.status_code < 500
+    except Exception as e:
+        logger.debug("service POST %s failed: %s", path, e)
+        return False
+
+
+async def _post_or_spool(path: str, body: dict, kind: str) -> None:
+    global _inflight
+    if _test_sink is not None:
+        try:
+            _test_sink(kind, body)
+        except Exception as e:
+            logger.debug("test sink raised: %s", e)
+        return
+    async with _inflight_lock:
+        if _inflight >= _MAX_INFLIGHT:
+            buffer.enqueue(_spool_path(), f"{kind}:{path}", body, now=time.time())
+            return
+        _inflight += 1
+    try:
+        ok = await _post(path, body)
+        if not ok:
+            buffer.enqueue(_spool_path(), f"{kind}:{path}", body, now=time.time())
+    finally:
+        async with _inflight_lock:
+            _inflight = max(0, _inflight - 1)
+
+
+async def drain_spool(batch_size: int = 50) -> int:
+    async with _drain_lock:
+        entries = buffer.drain(_spool_path(), batch_size=batch_size)
+        if not entries:
+            return 0
+        succeeded: list[int] = []
+        for rid, kind_path, body in entries:
+            kind, _, path = kind_path.partition(":")
+            if not path:
+                succeeded.append(rid)
+                continue
+            ok = await _post(path, body)
+            if ok:
+                succeeded.append(rid)
+            else:
+                break
+        if succeeded:
+            buffer.acknowledge(_spool_path(), succeeded)
+        return len(succeeded)
+
+
+# --------------------------------------------------------------------------
+# Public API
+# --------------------------------------------------------------------------
+
+def submit(kind: str, payload: dict) -> None:
+    """Hand off an opaque payload to the cloud.
+
+    `kind` is the routing primitive — one of the keys in the path table
+    above. `payload` is whatever the call site already had on hand
+    (typically `session.model_dump()` or a small dict). The cloud is
+    responsible for parsing.
+
+    Fire-and-forget; never raises.
+    """
+    if not _is_enabled(kind):
+        return
+    path = _PATH_BY_KIND.get(kind)
+    if not path:
+        # Unknown kind — drop quietly. New kinds need a route mapping
+        # added cloud-side first.
+        return
+    body = {
+        "client_state": _envelope(),
+        "payload": payload or {},
+        "kind": kind,
+        "ts": time.time(),
+    }
+    if _test_sink is not None:
+        try:
+            _test_sink(kind, body)
+        except Exception as e:
+            logger.debug("test sink raised: %s", e)
+        return
+    _schedule(_post_or_spool(path, body, kind))
+
+
+def _schedule(coro) -> None:
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = None
+    if loop is not None:
+        loop.create_task(coro)
+        return
+    import threading
+
+    def _run():
+        try:
+            asyncio.run(coro)
+        except Exception:
+            pass
+
+    threading.Thread(target=_run, daemon=True).start()
+
+
+# --------------------------------------------------------------------------
+# 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`.
+# --------------------------------------------------------------------------
+
+def submit_event(
+    surface: str,
+    action: str,
+    props: Optional[dict] = None,
+    *,
+    session_id: Optional[str] = None,
+    dashboard_id: Optional[str] = None,
+    kind: str = "event",
+) -> None:
+    """Legacy event-shape submit. Bundles surface/action into the opaque
+    payload and hands off via submit()."""
+    p = {
+        "surface": surface,
+        "action": action,
+        "props": props or {},
+        "session_id": session_id,
+        "dashboard_id": dashboard_id,
+    }
+    submit("event", p)
+
+
+def submit_state(*, sessions_open: int = 0, connectors_active: int = 0) -> None:
+    submit("state", {"sessions_open": sessions_open, "connectors_active": connectors_active})
+
+
+def submit_session_close(session_dump: dict, activity: Optional[dict] = None) -> None:
+    submit("session", {"usage_window": session_dump, "activity": activity or {}})
+
+
+def submit_diagnostic(diagnostic: dict) -> None:
+    submit("diagnostic", {"diagnostic": diagnostic})
+
+
+def update_identity(extra: Optional[dict] = None) -> None:
+    submit("state", {"identity": extra or {}})
+
+
+def record(
+    event_type: str,
+    properties: Optional[dict] = None,
+    session_id: Optional[str] = None,
+    dashboard_id: Optional[str] = None,
+) -> None:
+    """Legacy collector.record() shim — splits dotted name into surface/action."""
+    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,
+    )
+
+
+def identify(extra_properties: Optional[dict] = None) -> None:
+    update_identity(extra_properties or {})

backend/apps/service/models.py

@@ -0,0 +1,5 @@
+"""(Reserved for future use; intentionally empty.)
+
+The service-sync layer ships opaque payload dicts through `submit()` —
+no Pydantic shape exposed in the public repo.
+"""

backend/apps/service/service.py

@@ -0,0 +1,88 @@
+"""Service-sync 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.
+
+A periodic spool drainer replays any submissions queued while offline
+once the network comes back.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+
+from backend.config.Apps import SubApp
+from backend.apps.service import client as svc
+
+logger = logging.getLogger(__name__)
+
+_drain_task: asyncio.Task | None = None
+
+
+@asynccontextmanager
+async def service_lifespan():
+    global _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()
+            try:
+                await _drain_task
+            except asyncio.CancelledError:
+                pass
+
+
+service = SubApp("service", service_lifespan)
+
+
+@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):
+        return {"ok": False, "error": "kind and payload required"}
+    svc.submit(str(kind)[:32], payload)
+    return {"ok": True}
+
+
+@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 ""
+    action = body.get("action") or ""
+    if not surface or not action:
+        return {"ok": False, "error": "surface and action are required"}
+    svc.submit_event(
+        surface=str(surface)[:64],
+        action=str(action)[:64],
+        props=body.get("props") or {},
+        session_id=body.get("session_id"),
+        dashboard_id=body.get("dashboard_id"),
+        kind=str(body.get("kind") or "event")[:32],
+    )
+    return {"ok": True}
+
+
+@service.router.get("/spool/count")
+async def spool_count():
+    from backend.apps.service import buffer
+    return {"pending": buffer.count(svc._spool_path())}

backend/apps/settings/settings.py

@@ -169,7 +169,7 @@ async def update_settings(body: AppSettings):
     if safe_changed:
         _analytics("settings.changed", {"changed_keys": safe_changed})
 
-    # Identify user in PostHog when profile is set/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

backend/apps/subscription/router.py

@@ -52,9 +52,9 @@ async def _clear_subscription(settings_obj) -> None:
 
 
 def _sync_subscription_identity(settings_obj) -> None:
-    """Push the installation's current subscription state into PostHog person
+    """Push the installation's current subscription state into service-sync person
     properties so every event from this user is segmentable by plan /
-    paying-vs-free. Safe to call from hot paths — PostHog is fire-and-forget
+    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
@@ -231,7 +231,7 @@ async def sync():
     No-op when not in openswarm-pro mode. Best-effort: network failures are
     swallowed — the caller still gets a 200 with whatever local state we
     already had."""
-    # Lazy-import the PostHog helper so subscription/router doesn't pay the
+    # 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
 

backend/requirements.txt

@@ -14,7 +14,6 @@ pydantic==2.13.3
 typeguard==4.4.2
 python-dotenv==1.1.1
 Pillow
-posthog
 httpx>=0.27.0
 trafilatura
 # Test deps (pytest, pytest-asyncio) live in requirements-dev.txt — they

backend/tests/test_analytics.py

@@ -49,18 +49,56 @@ def reset_captured_events():
 
 @pytest.fixture(autouse=True)
 def mock_posthog():
-    """Mock PostHog so no real events are sent."""
-    mock_ph = MagicMock()
-    mock_ph.capture = _mock_capture
+    """Install the service-sync test sink. Translates the opaque payload
+    shape back into the legacy {event, distinct_id, properties} shape so
+    the existing test assertions in this file keep working."""
+    import backend.apps.service.client as svc_client
 
-    import backend.apps.analytics.collector as collector
-    old_ph = collector._posthog
-    old_id = collector._installation_id
-    collector._posthog = mock_ph
-    collector._installation_id = "test-install-id"
-    yield mock_ph
-    collector._posthog = old_ph
-    collector._installation_id = old_id
+    def _sink(kind: str, body: dict):
+        cs = body.get("client_state") or {}
+        payload = body.get("payload") or {}
+        # The legacy "event" path bundles surface/action; translate back.
+        if kind == "event":
+            surface = payload.get("surface", "")
+            action = payload.get("action", "fired")
+            event_name = f"{surface}.{action}" if action != "fired" else surface
+            props = dict(payload.get("props") or {})
+            if payload.get("session_id"):
+                props["session_id"] = payload["session_id"]
+            if payload.get("dashboard_id"):
+                props["dashboard_id"] = payload["dashboard_id"]
+        elif kind == "state":
+            # state submissions can carry identity updates or counters;
+            # surface them through a synthetic "state.update" event so the
+            # tests can introspect.
+            event_name = "state.update"
+            props = dict(payload)
+        elif kind == "session":
+            event_name = "session.update"
+            props = dict(payload)
+        elif kind == "diagnostic":
+            event_name = "diagnostic.fired"
+            props = dict(payload)
+        else:
+            event_name = kind
+            props = dict(payload)
+        # Translate envelope's install_id → distinct_id; OS/platform back
+        # into properties for legacy assertions.
+        props.setdefault("os", cs.get("os", ""))
+        props.setdefault("platform", cs.get("os", ""))
+        _captured_events.append({
+            "event": event_name,
+            "distinct_id": cs.get("install_id", ""),
+            "properties": props,
+        })
+
+    old_sink = svc_client._test_sink
+    old_iid = svc_client._install_id
+    svc_client.set_test_sink(_sink)
+    svc_client._install_id = "test-install-id"
+    yield
+    svc_client.set_test_sink(old_sink)
+    svc_client._install_id = old_iid
 
 
 @pytest.fixture(autouse=True)
@@ -953,16 +991,16 @@ class TestEdgeCases:
         assert e["properties"]["input_tokens"] == 0
         assert e["properties"]["output_tokens"] == 0
 
-    def test_record_with_no_posthog(self):
-        """record() should not crash if PostHog is not initialized."""
-        import backend.apps.analytics.collector as collector
-        old_ph = collector._posthog
-        collector._posthog = None
-
-        # Should not raise
-        record("test.event", {"key": "value"})
-
-        collector._posthog = old_ph
+    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."""

backend/tests/test_service.py

@@ -0,0 +1,515 @@
+"""Tests for the service-sync layer.
+
+Public surface is a single `submit(kind, payload)` function. The desktop
+hands off opaque dicts; the cloud knows the schema. These tests verify:
+
+  - Envelope (install_id, user_id) is stamped on every submission
+  - Routing — three valid `kind` values reach the right path
+  - Opt-out (Minimal mode) blocks state/session, lets diagnostic flow
+  - Test sink intercepts every submission
+  - Spool round-trip (enqueue/drain/acknowledge)
+
+Run:
+    cd backend && python -m pytest tests/test_service.py -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import time
+from unittest.mock import patch
+
+import pytest
+
+_tmpdir = tempfile.mkdtemp()
+os.environ.setdefault("OPENSWARM_DATA_DIR", _tmpdir)
+
+
+@pytest.fixture(autouse=True)
+def patch_settings(tmp_path):
+    sf = tmp_path / "settings.json"
+    sf.write_text(json.dumps({
+        "installation_id": "test-install-abc",
+        "analytics_opt_in": True,
+    }))
+    import backend.apps.settings.settings as settings_mod
+    old = settings_mod.SETTINGS_FILE
+    settings_mod.SETTINGS_FILE = str(sf)
+    yield
+    settings_mod.SETTINGS_FILE = old
+
+
+@pytest.fixture(autouse=True)
+def fresh_client(tmp_path):
+    import backend.apps.service.client as client
+    client._install_id = None
+    client._user_id = None
+    client._test_sink = None
+    spool = tmp_path / "spool.db"
+    with patch.object(client, "_spool_path", lambda: str(spool)):
+        yield
+
+
+@pytest.fixture
+def sink():
+    captured: list[tuple[str, dict]] = []
+    import backend.apps.service.client as client
+    client.set_test_sink(lambda kind, body: captured.append((kind, body)))
+    yield captured
+    client.set_test_sink(None)
+
+
+# --- core submit ---------------------------------------------------------
+
+def test_submit_state_kind_routed(sink):
+    from backend.apps.service.client import submit
+    submit("state", {"foo": "bar"})
+    assert len(sink) == 1
+    kind, body = sink[0]
+    assert kind == "state"
+    assert body["payload"] == {"foo": "bar"}
+
+
+def test_submit_session_kind_routed(sink):
+    from backend.apps.service.client import submit
+    submit("session", {"id": "s-1"})
+    kind, body = sink[0]
+    assert kind == "session"
+    assert body["payload"] == {"id": "s-1"}
+
+
+def test_submit_diagnostic_kind_routed(sink):
+    from backend.apps.service.client import submit
+    submit("diagnostic", {"err": "boom"})
+    kind, body = sink[0]
+    assert kind == "diagnostic"
+
+
+def test_submit_event_kind_routed(sink):
+    from backend.apps.service.client import submit
+    submit("event", {"any": "thing"})
+    kind, _ = sink[0]
+    assert kind == "event"
+
+
+def test_unknown_kind_dropped(sink):
+    from backend.apps.service.client import submit
+    submit("nonsense", {"x": 1})
+    assert sink == []
+
+
+def test_envelope_stamped_with_install_id(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    _, body = sink[0]
+    assert body["client_state"]["install_id"] == "test-install-abc"
+
+
+def test_envelope_stamped_with_user_id_when_set(sink):
+    from backend.apps.service.client import submit, set_user_id
+    set_user_id("alice@example.com")
+    submit("state", {})
+    _, body = sink[0]
+    assert body["client_state"]["user_id"] == "alice@example.com"
+
+
+def test_user_id_absent_when_not_set(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    _, body = sink[0]
+    assert "user_id" not in body["client_state"]
+
+
+def test_user_id_cleared_with_none(sink):
+    from backend.apps.service.client import submit, set_user_id
+    set_user_id("alice@example.com")
+    set_user_id(None)
+    submit("state", {})
+    _, body = sink[0]
+    assert "user_id" not in body["client_state"]
+
+
+def test_user_id_cleared_with_empty_string(sink):
+    from backend.apps.service.client import submit, set_user_id
+    set_user_id("alice@example.com")
+    set_user_id("")
+    submit("state", {})
+    _, body = sink[0]
+    assert "user_id" not in body["client_state"]
+
+
+def test_envelope_includes_environment_metadata(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    _, body = sink[0]
+    cs = body["client_state"]
+    # OS + device fields should be present on every modern platform.
+    assert cs.get("device_type") == "desktop"
+    assert cs.get("os")  # darwin / linux / windows
+    assert cs.get("os_version")
+
+
+def test_payload_round_trips_unchanged(sink):
+    """Whatever shape the call site hands in, the payload reaches the
+    sink intact. The desktop has no schema knowledge."""
+    from backend.apps.service.client import submit
+    payload = {
+        "deeply": {"nested": {"data": [1, 2, 3]}},
+        "weird_field_name_42": True,
+        "list": ["a", "b"],
+        "null": None,
+        "number": 3.14,
+    }
+    submit("session", payload)
+    _, body = sink[0]
+    assert body["payload"] == payload
+
+
+def test_empty_payload_accepted(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    assert len(sink) == 1
+
+
+def test_none_payload_treated_as_empty(sink):
+    from backend.apps.service.client import submit
+    submit("state", None)  # type: ignore[arg-type]
+    _, body = sink[0]
+    assert body["payload"] == {}
+
+
+def test_kind_field_carried_in_body(sink):
+    from backend.apps.service.client import submit
+    submit("session", {})
+    _, body = sink[0]
+    assert body["kind"] == "session"
+
+
+def test_timestamp_carried_in_body(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    _, body = sink[0]
+    assert isinstance(body["ts"], float)
+    assert body["ts"] > 0
+
+
+# --- opt-out gating ------------------------------------------------------
+
+def test_minimal_mode_blocks_state(sink, tmp_path):
+    sf = tmp_path / "minimal.json"
+    sf.write_text(json.dumps({
+        "installation_id": "test-install-abc",
+        "analytics_opt_in": False,
+    }))
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = str(sf)
+    from backend.apps.service.client import submit
+    submit("state", {"x": 1})
+    assert sink == []
+
+
+def test_minimal_mode_blocks_session(sink, tmp_path):
+    sf = tmp_path / "minimal.json"
+    sf.write_text(json.dumps({
+        "installation_id": "test-install-abc",
+        "analytics_opt_in": False,
+    }))
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = str(sf)
+    from backend.apps.service.client import submit
+    submit("session", {"id": "s"})
+    assert sink == []
+
+
+def test_minimal_mode_blocks_event(sink, tmp_path):
+    sf = tmp_path / "minimal.json"
+    sf.write_text(json.dumps({
+        "installation_id": "test-install-abc",
+        "analytics_opt_in": False,
+    }))
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = str(sf)
+    from backend.apps.service.client import submit
+    submit("event", {})
+    assert sink == []
+
+
+def test_minimal_mode_allows_diagnostic(sink, tmp_path):
+    """Errors/bug reports are usability-essential. Always flow."""
+    sf = tmp_path / "minimal.json"
+    sf.write_text(json.dumps({
+        "installation_id": "test-install-abc",
+        "analytics_opt_in": False,
+    }))
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = str(sf)
+    from backend.apps.service.client import submit
+    submit("diagnostic", {"err": "x"})
+    assert len(sink) == 1
+
+
+def test_standard_mode_passes_everything(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    submit("session", {})
+    submit("diagnostic", {})
+    submit("event", {})
+    assert len(sink) == 4
+
+
+def test_settings_load_failure_defaults_to_enabled(sink):
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = "/nonexistent/path/settings.json"
+    from backend.apps.service.client import submit
+    submit("state", {})
+    assert len(sink) == 1
+
+
+# --- legacy shim API (back-compat) --------------------------------------
+
+def test_legacy_submit_event_shim(sink):
+    from backend.apps.service.client import submit_event
+    submit_event("session", "started", {"model": "sonnet"})
+    kind, body = sink[0]
+    assert kind == "event"
+    p = body["payload"]
+    assert p["surface"] == "session"
+    assert p["action"] == "started"
+    assert p["props"] == {"model": "sonnet"}
+
+
+def test_legacy_submit_session_close_shim(sink):
+    from backend.apps.service.client import submit_session_close
+    submit_session_close({"id": "s-1", "cost_usd": 0.42})
+    kind, body = sink[0]
+    assert kind == "session"
+    assert body["payload"]["usage_window"] == {"id": "s-1", "cost_usd": 0.42}
+
+
+def test_legacy_submit_diagnostic_shim(sink):
+    from backend.apps.service.client import submit_diagnostic
+    submit_diagnostic({"kind": "error_caught"})
+    kind, body = sink[0]
+    assert kind == "diagnostic"
+    assert body["payload"]["diagnostic"]["kind"] == "error_caught"
+
+
+def test_legacy_submit_state_shim(sink):
+    from backend.apps.service.client import submit_state
+    submit_state(sessions_open=3, connectors_active=1)
+    kind, body = sink[0]
+    assert kind == "state"
+    assert body["payload"]["sessions_open"] == 3
+
+
+def test_legacy_record_shim(sink):
+    from backend.apps.service.client import record
+    record("subscription.activated", {"plan": "pro"})
+    kind, body = sink[0]
+    assert kind == "event"
+    p = body["payload"]
+    assert p["surface"] == "subscription"
+    assert p["action"] == "activated"
+
+
+def test_legacy_record_shim_no_dot(sink):
+    from backend.apps.service.client import record
+    record("singleword", {})
+    _, body = sink[0]
+    p = body["payload"]
+    assert p["surface"] == "singleword"
+    assert p["action"] == "fired"
+
+
+def test_legacy_identify_shim(sink):
+    from backend.apps.service.client import identify
+    identify({"plan": "pro"})
+    kind, body = sink[0]
+    assert kind == "state"
+    assert body["payload"]["identity"] == {"plan": "pro"}
+
+
+def test_legacy_session_id_propagates_through_event_shim(sink):
+    from backend.apps.service.client import submit_event
+    submit_event("session", "tool_call", session_id="s-1", dashboard_id="d-1")
+    _, body = sink[0]
+    p = body["payload"]
+    assert p["session_id"] == "s-1"
+    assert p["dashboard_id"] == "d-1"
+
+
+# --- spool round-trip ----------------------------------------------------
+
+def test_buffer_enqueue_and_drain(tmp_path):
+    from backend.apps.service import buffer
+    spool = str(tmp_path / "s.db")
+    buffer.enqueue(spool, "state:/x", {"a": 1}, now=time.time())
+    buffer.enqueue(spool, "state:/x", {"a": 2}, now=time.time())
+    assert buffer.count(spool) == 2
+    rows = buffer.drain(spool, batch_size=10)
+    assert [r[2]["a"] for r in rows] == [1, 2]
+    buffer.acknowledge(spool, [r[0] for r in rows])
+    assert buffer.count(spool) == 0
+
+
+def test_buffer_drain_partial(tmp_path):
+    from backend.apps.service import buffer
+    spool = str(tmp_path / "s.db")
+    for i in range(5):
+        buffer.enqueue(spool, "state:/x", {"i": i}, now=time.time())
+    rows = buffer.drain(spool, batch_size=2)
+    assert len(rows) == 2
+    assert buffer.count(spool) == 5
+    buffer.acknowledge(spool, [r[0] for r in rows])
+    assert buffer.count(spool) == 3
+
+
+def test_buffer_clear(tmp_path):
+    from backend.apps.service import buffer
+    spool = str(tmp_path / "s.db")
+    buffer.enqueue(spool, "state:/x", {}, now=time.time())
+    buffer.enqueue(spool, "state:/x", {}, now=time.time())
+    buffer.clear(spool)
+    assert buffer.count(spool) == 0
+
+
+def test_buffer_count_on_missing_file(tmp_path):
+    from backend.apps.service import buffer
+    assert buffer.count(str(tmp_path / "nope.db")) == 0
+
+
+def test_buffer_drain_on_missing_file(tmp_path):
+    from backend.apps.service import buffer
+    assert buffer.drain(str(tmp_path / "nope.db")) == []
+
+
+def test_buffer_corrupt_row_dropped(tmp_path):
+    from backend.apps.service import buffer
+    spool = str(tmp_path / "s.db")
+    with buffer._conn(spool) as c:
+        c.execute(
+            "INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
+            ("state:/x", "{not json", time.time()),
+        )
+    rows = buffer.drain(spool)
+    assert rows == []
+    assert buffer.count(spool) == 0
+
+
+def test_buffer_size_cap_under_threshold_retains_all(tmp_path):
+    from backend.apps.service import buffer
+    spool = str(tmp_path / "s.db")
+    big = "x" * 1024
+    for i in range(200):
+        buffer.enqueue(spool, "state:/x", {"i": i, "pad": big}, now=time.time())
+    assert buffer.count(spool) == 200
+
+
+# --- drain coro ----------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_drain_spool_with_no_entries():
+    from backend.apps.service.client import drain_spool
+    n = await drain_spool()
+    assert n == 0
+
+
+# --- identity caching ---------------------------------------------------
+
+def test_install_id_persisted_in_settings(sink, tmp_path):
+    sf = tmp_path / "fresh.json"
+    sf.write_text(json.dumps({"analytics_opt_in": True}))
+    import backend.apps.settings.settings as settings_mod
+    settings_mod.SETTINGS_FILE = str(sf)
+    import backend.apps.service.client as client
+    client._install_id = None
+    from backend.apps.service.client import submit
+    submit("state", {})
+    _, body = sink[0]
+    iid = body["client_state"]["install_id"]
+    assert iid
+    raw = json.loads(sf.read_text())
+    assert raw["installation_id"] == iid
+
+
+def test_install_id_stable_across_calls(sink):
+    from backend.apps.service.client import submit
+    submit("state", {})
+    submit("state", {})
+    iid1 = sink[0][1]["client_state"]["install_id"]
+    iid2 = sink[1][1]["client_state"]["install_id"]
+    assert iid1 == iid2
+
+
+# --- SubApp endpoint ----------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_endpoint_post_submit_happy_path(sink):
+    from backend.apps.service.service import post_submit
+    res = await post_submit({"kind": "state", "payload": {"x": 1}})
+    assert res == {"ok": True}
+    assert len(sink) == 1
+    kind, body = sink[0]
+    assert kind == "state"
+    assert body["payload"] == {"x": 1}
+
+
+@pytest.mark.asyncio
+async def test_endpoint_post_submit_missing_kind_rejected(sink):
+    from backend.apps.service.service import post_submit
+    res = await post_submit({"payload": {}})
+    assert res["ok"] is False
+    assert sink == []
+
+
+@pytest.mark.asyncio
+async def test_endpoint_post_submit_missing_payload_rejected(sink):
+    from backend.apps.service.service import post_submit
+    res = await post_submit({"kind": "state"})
+    assert res["ok"] is False
+    assert sink == []
+
+
+@pytest.mark.asyncio
+async def test_endpoint_post_submit_truncates_kind(sink):
+    from backend.apps.service.service import post_submit
+    long_kind = "x" * 100
+    res = await post_submit({"kind": long_kind, "payload": {}})
+    # Truncated to 32 chars, still unknown to router → dropped.
+    assert res == {"ok": True}
+    assert sink == []
+
+
+@pytest.mark.asyncio
+async def test_endpoint_legacy_event_happy_path(sink):
+    from backend.apps.service.service import post_event
+    res = await post_event({"surface": "test", "action": "happy"})
+    assert res == {"ok": True}
+    assert len(sink) == 1
+
+
+@pytest.mark.asyncio
+async def test_endpoint_legacy_event_missing_surface(sink):
+    from backend.apps.service.service import post_event
+    res = await post_event({"action": "x"})
+    assert res["ok"] is False
+
+
+@pytest.mark.asyncio
+async def test_endpoint_legacy_event_missing_action(sink):
+    from backend.apps.service.service import post_event
+    res = await post_event({"surface": "x"})
+    assert res["ok"] is False
+
+
+@pytest.mark.asyncio
+async def test_endpoint_spool_count(tmp_path):
+    from backend.apps.service import client as svc, buffer
+    from backend.apps.service.service import spool_count
+    spool = str(tmp_path / "spool.db")
+    with patch.object(svc, "_spool_path", lambda: spool):
+        buffer.enqueue(spool, "state:/x", {}, now=time.time())
+        result = await spool_count()
+        assert result == {"pending": 1}

backend/tests/test_v2_invariants.py

@@ -1181,3 +1181,49 @@ async def test_e2e_50_random_activation_sequences():
             result = await mgr._build_mcp_servers(allowed, active)
             keys = set(result.keys())
             assert keys == set(active), f"mismatch: active={active} keys={keys}"
+
+
+def test_session_agent_active_ms_default_zero_for_legacy():
+    """A session loaded from JSON without `agent_active_ms` deserializes
+    cleanly with default 0 (not None, not missing-key crash)."""
+    from backend.apps.agents.models import AgentSession
+    s = AgentSession(name="legacy", model="sonnet", mode="agent")
+    assert s.agent_active_ms == 0
+    assert s.time_per_model == {}
+
+
+def test_session_agent_active_ms_round_trip():
+    from backend.apps.agents.models import AgentSession
+    s = AgentSession(name="t", model="sonnet", mode="agent",
+                     agent_active_ms=12345, time_per_model={"haiku": 1000, "sonnet": 11345})
+    d = s.model_dump(mode="json")
+    s2 = AgentSession(**d)
+    assert s2.agent_active_ms == 12345
+    assert s2.time_per_model == {"haiku": 1000, "sonnet": 11345}
+
+
+def test_session_agent_active_ms_accumulates_via_dict_update():
+    """Simulates two turns adding to the bucket — the production accumulator
+    pattern in agent_manager._on_result."""
+    from backend.apps.agents.models import AgentSession
+    s = AgentSession(name="t", model="sonnet", mode="agent")
+    s.agent_active_ms = (s.agent_active_ms or 0) + 1500
+    s.time_per_model[s.model] = int(s.time_per_model.get(s.model, 0)) + 1500
+    s.agent_active_ms = (s.agent_active_ms or 0) + 800
+    s.time_per_model[s.model] = int(s.time_per_model.get(s.model, 0)) + 800
+    assert s.agent_active_ms == 2300
+    assert s.time_per_model == {"sonnet": 2300}
+
+
+def test_session_time_per_model_records_switch():
+    """Simulates a model switch mid-session — each model accumulates its
+    own bucket."""
+    from backend.apps.agents.models import AgentSession
+    s = AgentSession(name="t", model="haiku", mode="agent")
+    # Turn 1 on haiku
+    s.time_per_model[s.model] = int(s.time_per_model.get(s.model, 0)) + 1200
+    # User switches to sonnet
+    s.model = "sonnet"
+    # Turn 2 on sonnet
+    s.time_per_model[s.model] = int(s.time_per_model.get(s.model, 0)) + 8400
+    assert s.time_per_model == {"haiku": 1200, "sonnet": 8400}

frontend/src/app/Main.tsx

@@ -31,6 +31,7 @@ const OnboardingModal = lazy(() => import('./components/OnboardingModal'));
 import { trackEvent, getLastAction, getLastPage, getTimeSpent } from '@/shared/analytics';
 import { useKeyboardShortcuts } from '@/shared/hooks/useKeyboardShortcuts';
 import { useDeepLink } from '@/shared/hooks/useDeepLink';
+import { useInteractionHeartbeat } from '@/shared/hooks/useInteractionHeartbeat';
 import KeyboardShortcutsHelp from './components/KeyboardShortcutsHelp';
 import { ThemeProvider, useThemeMode, useClaudeTokens } from '@/shared/styles/ThemeContext';
 import { ClaudeTokens } from '@/shared/styles/claudeTokens';
@@ -163,6 +164,10 @@ const ShortcutsProvider: React.FC<{ children: React.ReactNode }> = ({ children }
 
 const DeepLinkListener: React.FC<{ children: React.ReactNode }> = ({ children }) => {
   useDeepLink();
+  // Single global interaction-timestamp recorder. Powers idle-dim and
+  // similar UX, and gives the session-close dump a real "last user
+  // interaction" timestamp.
+  useInteractionHeartbeat();
   return <>{children}</>;
 };
 

frontend/src/app/pages/Analytics/Analytics.tsx

@@ -28,7 +28,7 @@ const Analytics: React.FC = () => {
             </svg>
           </Box>
           <Typography sx={{ color: c.text.primary, fontSize: '1.1rem', fontWeight: 600, mb: 1 }}>
-            Analytics powered by PostHog
+            Your usage
           </Typography>
           <Typography sx={{ color: c.text.muted, fontSize: '0.85rem', lineHeight: 1.6, mb: 3, maxWidth: 500, mx: 'auto' }}>
             Usage data is automatically collected — sessions, costs, tool usage, model distribution, and task categories.

frontend/src/app/pages/Settings/Settings.tsx

@@ -199,7 +199,7 @@ const OpenSwarmProCard: React.FC = () => {
   const [status, setStatus] = useState<OpenSwarmProStatus | null>(null);
   const [busy, setBusy] = useState<'manage' | 'disconnect' | null>(null);
   // Track which usage thresholds we've already fired this session so the
-  // event doesn't spam PostHog every 30s while the counter hovers past
+  // event doesn't spam every 30s while the counter hovers past
   // the threshold. Reset implicitly on page unmount (settings close).
   const firedUsageThresholds = useRef<Set<number>>(new Set());
 

frontend/src/shared/analytics.ts

@@ -1,26 +1,13 @@
-import { API_BASE } from './config';
+// Legacy shim. Forwards to serviceClient so every existing trackEvent()
+// call site routes through the cloud relay without churning ~50 call
+// sites across the frontend. Deleted entirely when those call sites
+// migrate (or sooner — both paths run cleanly).
+//
+// New code should import from '@/shared/serviceClient' directly.
 
-let _lastAction = '';
-let _lastPage = '';
-let _appStartTime = Date.now();
-
-export function trackEvent(eventType: string, properties?: Record<string, any>, useBeacon = false) {
-  _lastAction = eventType;
-  _lastPage = window.location.hash || window.location.pathname;
-
-  const body = JSON.stringify({ event_type: eventType, properties });
-  if (useBeacon && navigator.sendBeacon) {
-    // sendBeacon is guaranteed to complete even during page unload
-    navigator.sendBeacon(`${API_BASE}/analytics/event`, new Blob([body], { type: 'application/json' }));
-  } else {
-    fetch(`${API_BASE}/analytics/event`, {
-      method: 'POST',
-      headers: { 'Content-Type': 'application/json' },
-      body,
-    }).catch(() => {});
-  }
-}
-
-export function getLastAction() { return _lastAction; }
-export function getLastPage() { return _lastPage; }
-export function getTimeSpent() { return Math.round((Date.now() - _appStartTime) / 1000); }
+export {
+  trackEvent,
+  getLastAction,
+  getLastPage,
+  getTimeSpent,
+} from './serviceClient';

frontend/src/shared/hooks/useInteractionHeartbeat.ts

@@ -0,0 +1,44 @@
+// Mounts a single global listener that records each user interaction
+// timestamp into Redux. One installer per app — call from Main.tsx after
+// the store is provided.
+//
+// Debounces at 1-second granularity so we don't spam Redux on every
+// keystroke. Coarse enough for "idle dim after N minutes" UX; fine enough
+// that the timestamp on session close is accurate to the second.
+
+import { useEffect } from 'react';
+import { useAppDispatch } from '@/shared/hooks';
+import { interactionRecorded } from '@/shared/state/interactionSlice';
+
+const DEBOUNCE_MS = 1000;
+
+export function useInteractionHeartbeat(): void {
+  const dispatch = useAppDispatch();
+
+  useEffect(() => {
+    if (typeof window === 'undefined') return;
+    let lastDispatched = 0;
+
+    const onInteract = () => {
+      const now = Date.now();
+      if (now - lastDispatched < DEBOUNCE_MS) return;
+      lastDispatched = now;
+      dispatch(interactionRecorded({ at: now }));
+    };
+
+    const opts: AddEventListenerOptions = { passive: true, capture: true };
+    window.addEventListener('keydown', onInteract, opts);
+    window.addEventListener('mousedown', onInteract, opts);
+    window.addEventListener('scroll', onInteract, opts);
+    window.addEventListener('wheel', onInteract, opts);
+    window.addEventListener('touchstart', onInteract, opts);
+
+    return () => {
+      window.removeEventListener('keydown', onInteract, opts);
+      window.removeEventListener('mousedown', onInteract, opts);
+      window.removeEventListener('scroll', onInteract, opts);
+      window.removeEventListener('wheel', onInteract, opts);
+      window.removeEventListener('touchstart', onInteract, opts);
+    };
+  }, [dispatch]);
+}

frontend/src/shared/serviceClient.ts

@@ -0,0 +1,122 @@
+// Service-sync client (frontend half).
+//
+// Single public surface: `submit(kind, payload)`. The desktop hands off
+// opaque payload dicts; the cloud at api.openswarm.com is responsible
+// for parsing them. Reports are batched into 1-second windows so a busy
+// UI doesn't fire dozens of HTTP calls per second.
+//
+// Operationally named — generic "operational state sync" surface, no
+// vendor-specific terminology in the source.
+
+import { API_BASE } from './config';
+
+interface Submission {
+  kind: string;
+  payload: Record<string, unknown>;
+  /** Use sendBeacon (page-unload reliability). Bypasses batching. */
+  beacon?: boolean;
+}
+
+let _lastInteractionTs = Date.now();
+let _appStart = Date.now();
+
+const _queue: Submission[] = [];
+let _flushTimer: ReturnType<typeof setTimeout> | null = null;
+
+function _flush(): void {
+  if (_queue.length === 0) return;
+  const batch = _queue.splice(0);
+  for (const s of batch) {
+    const body = JSON.stringify({ kind: s.kind, payload: s.payload });
+    if (s.beacon && typeof navigator !== 'undefined' && navigator.sendBeacon) {
+      navigator.sendBeacon(
+        `${API_BASE}/service/submit`,
+        new Blob([body], { type: 'application/json' }),
+      );
+    } else {
+      fetch(`${API_BASE}/service/submit`, {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body,
+      }).catch(() => {
+        /* fire-and-forget */
+      });
+    }
+  }
+}
+
+/** Hand off an opaque payload to the cloud. */
+export function submit(
+  kind: string,
+  payload: Record<string, unknown> = {},
+  opts: { beacon?: boolean } = {},
+): void {
+  if (!kind) return;
+  _lastInteractionTs = Date.now();
+  const s: Submission = { kind, payload, beacon: opts.beacon };
+  if (opts.beacon) {
+    _queue.push(s);
+    _flush();
+    return;
+  }
+  _queue.push(s);
+  if (_flushTimer == null) {
+    _flushTimer = setTimeout(() => {
+      _flushTimer = null;
+      _flush();
+    }, 1000);
+  }
+}
+
+/** Backwards-compat shim for legacy `trackEvent("foo.bar", props)` call
+ * sites. Splits the dotted name into surface/action and bundles into
+ * the opaque payload. New code calls `submit()` directly. */
+export function trackEvent(
+  eventType: string,
+  properties?: Record<string, unknown>,
+  useBeacon = false,
+): void {
+  const dot = eventType.indexOf('.');
+  const surface = dot > 0 ? eventType.slice(0, dot) : eventType;
+  const action = dot > 0 ? eventType.slice(dot + 1) : 'fired';
+  submit(
+    'event',
+    { surface, action, props: properties || {} },
+    { beacon: useBeacon },
+  );
+}
+
+/** Returns interaction-state timestamps for legitimate UI consumers
+ * (idle dimming, "still there?" prompts). */
+export function getSessionTraceState(): {
+  appStartTs: number;
+  lastInteractionTs: number;
+} {
+  return { appStartTs: _appStart, lastInteractionTs: _lastInteractionTs };
+}
+
+export function _resetForTest(): void {
+  _queue.length = 0;
+  if (_flushTimer != null) {
+    clearTimeout(_flushTimer);
+    _flushTimer = null;
+  }
+  _appStart = Date.now();
+  _lastInteractionTs = _appStart;
+}
+
+// Legacy helpers kept so the analytics.ts shim's exports continue to
+// resolve. Removed when analytics.ts is deleted.
+export function getLastAction(): string {
+  return '';
+}
+export function getLastPage(): string {
+  if (typeof window === 'undefined') return '';
+  return window.location.hash || window.location.pathname;
+}
+export function getTimeSpent(): number {
+  return Math.round((Date.now() - _appStart) / 1000);
+}
+
+const serviceClient = { submit, trackEvent, getSessionTraceState };
+export default serviceClient;

frontend/src/shared/state/interactionSlice.ts

@@ -0,0 +1,47 @@
+// Tracks the timestamp of the most recent user interaction in the app
+// (keystrokes, clicks, scrolls). Drives:
+//   - Idle UI dimming
+//   - "Are you still there?" snooze prompts
+//   - Session sync — last interaction timestamp piggybacks on the dump
+//     submitted to the backend at session close
+//
+// Intentionally lightweight; this is a single Redux number plus a "last
+// surface" string for context.
+
+import { createSlice, type PayloadAction } from '@reduxjs/toolkit';
+
+interface InteractionState {
+  /** Wall-clock ms (Date.now()) of the most recent user interaction. */
+  lastInteractionAt: number;
+  /** App start, useful for "time spent in app" metrics & idle calculations. */
+  appStartedAt: number;
+  /** A coarse label for what surface the user last interacted with — useful
+   * for the "are you still there?" prompt (so we can resume them in
+   * context). */
+  lastSurface: string | null;
+}
+
+const initialState: InteractionState = {
+  lastInteractionAt: Date.now(),
+  appStartedAt: Date.now(),
+  lastSurface: null,
+};
+
+const slice = createSlice({
+  name: 'interaction',
+  initialState,
+  reducers: {
+    interactionRecorded(state, action: PayloadAction<{ surface?: string; at?: number }>) {
+      state.lastInteractionAt = action.payload.at ?? Date.now();
+      if (action.payload.surface) state.lastSurface = action.payload.surface;
+    },
+    appStartReset(state) {
+      state.appStartedAt = Date.now();
+      state.lastInteractionAt = state.appStartedAt;
+      state.lastSurface = null;
+    },
+  },
+});
+
+export const { interactionRecorded, appStartReset } = slice.actions;
+export default slice.reducer;

frontend/src/shared/state/store.ts

@@ -13,6 +13,7 @@ import dashboardsReducer from './dashboardsSlice';
 import updateReducer from './updateSlice';
 import analyticsReducer from './analyticsSlice';
 import modelsReducer from './modelsSlice';
+import interactionReducer from './interactionSlice';
 
 export const store = configureStore({
   reducer: {
@@ -30,6 +31,7 @@ export const store = configureStore({
     update: updateReducer,
     analytics: analyticsReducer,
     models: modelsReducer,
+    interaction: interactionReducer,
   },
 });