mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-06 09:47:44 +02:00
346 lines
9.9 KiB
Python
346 lines
9.9 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()
|
|
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.service.service 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 _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 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(),
|
|
}
|
|
_log(kind, payload)
|
|
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 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 {})
|