mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-22 20:52:23 +02:00
395 lines
12 KiB
Python
395 lines
12 KiB
Python
"""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()
|
|
# Prefer the cloud-issued user_id (UUID) if the user has signed in
|
|
# via Google OAuth, magic link, or Stripe checkout; that's the
|
|
# authoritative identity. Falls back to user_email for installs
|
|
# that haven't completed sign-in yet (so existing onboarding-only
|
|
# installs don't lose their Person history during the v1.0.29
|
|
# rollout). After every install signs in, this fallback drops out.
|
|
return (
|
|
getattr(s, "user_id", None)
|
|
or 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
|
|
# Timezone: prefer the IANA zone name passed in by Electron (always
|
|
# canonical, e.g. "America/Los_Angeles") so cloud-side localTimeFields()
|
|
# can format hour-of-day correctly. Fall back to Python's local zone
|
|
# which sometimes returns abbreviations (PDT, CDT) or localized names
|
|
# ("Romance (zomertijd)") that don't round-trip through tzdata.
|
|
try:
|
|
ianatz = os.environ.get("OPENSWARM_TIMEZONE", "").strip()
|
|
if not ianatz:
|
|
try:
|
|
from tzlocal import get_localzone_name # type: ignore
|
|
ianatz = get_localzone_name() or ""
|
|
except Exception:
|
|
pass
|
|
if not ianatz:
|
|
import datetime as _dt
|
|
local_tz = _dt.datetime.now().astimezone().tzinfo
|
|
if local_tz:
|
|
ianatz = str(local_tz)
|
|
if ianatz:
|
|
env["timezone"] = ianatz
|
|
except Exception:
|
|
pass
|
|
# Locale: BCP 47 string ("en-US", "es-ES", etc.) injected by Electron via
|
|
# app.getLocale(); see electron/main.js. We don't fall back to Python's
|
|
# locale.getdefaultlocale() because that's deprecated, often empty, and
|
|
# returns inconsistent OS-specific values across macOS/Windows/Linux.
|
|
try:
|
|
loc = os.environ.get("OPENSWARM_LOCALE", "").strip()
|
|
if loc:
|
|
env["locale"] = loc
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from backend.apps.service.service import APP_VERSION
|
|
env["app_version"] = APP_VERSION
|
|
except Exception:
|
|
pass
|
|
# How this build was packaged. Set by the platform-specific build script
|
|
# (electron-builder afterPack hooks for dmg / exe / appimage / deb / rpm).
|
|
# Defaults to "dev" when running from `bash run.sh` in a checked-out repo.
|
|
env["install_method"] = os.environ.get("OPENSWARM_INSTALL_METHOD", "dev")
|
|
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 _log(kind: str, payload: dict) -> None:
|
|
"""Append to the rolling operational log for diagnostics."""
|
|
try:
|
|
from backend.apps.service.ring_buffer import record
|
|
record(kind)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def sync(data: dict | None = None) -> None:
|
|
"""Sync operational state to the cloud. Single entry point.
|
|
|
|
Accepts any dict; the cloud determines what it is from the shape.
|
|
The desktop has no knowledge of event types, schemas, or routing.
|
|
|
|
Each call carries:
|
|
- `t`: client-side timestamp at submit time (unix seconds, float).
|
|
- `submission_id`: uuid generated per call. The cloud uses
|
|
(install_id, submission_id) as an idempotency key, so a retry
|
|
from the offline spool is a no-op rather than a double-write.
|
|
|
|
Fire-and-forget; never raises.
|
|
"""
|
|
payload = data or {}
|
|
if not _is_enabled("state"):
|
|
return
|
|
body = {
|
|
"client_state": _envelope(),
|
|
"d": payload,
|
|
"t": time.time(),
|
|
"submission_id": uuid4().hex,
|
|
}
|
|
_log("s", payload)
|
|
if _test_sink is not None:
|
|
try:
|
|
_test_sink("s", body)
|
|
except Exception as e:
|
|
logger.debug("test sink raised: %s", e)
|
|
return
|
|
_schedule(_post_or_spool(_DEFAULT_SYNC_PATH, body, "s"))
|
|
|
|
|
|
# Internal routing; the cloud has one endpoint for everything.
|
|
_DEFAULT_SYNC_PATH = "/api/service/sync"
|
|
|
|
|
|
def submit(kind: str, payload: dict) -> None:
|
|
"""Legacy shim; routes through sync(). Kept for back-compat during
|
|
migration. New code should call sync() directly."""
|
|
sync(payload)
|
|
|
|
|
|
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 older import paths.
|
|
# --------------------------------------------------------------------------
|
|
|
|
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:
|
|
try:
|
|
from backend.apps.service.ring_buffer import snapshot
|
|
diagnostic["recent_log"] = snapshot()
|
|
except Exception:
|
|
pass
|
|
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 {})
|