WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-12 20:57:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] service+subscription+auth+shims: leading-_ -> p_/public; promote cross-file public (spool_path/conn/install_id/get_user_id/sync_pro_routing/etc.), fix qualified+string test refs

Browse Source
This commit is contained in:
ciregenz
2026-06-23 20:53:18 -07:00
parent e1bd57a629
commit 1750363f11
15 changed files with 269 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agents.py
+11 -11
View File
@@ -13,7 +13,7 @@ import logging
logger = logging.getLogger(__name__)
# Dedup concurrent generate-group-meta calls; collapses the 429 thundering herd by sharing one upstream Future per (session, group).
_group_meta_inflight: dict[tuple[str, str], asyncio.Future] = {}
p_group_meta_inflight: dict[tuple[str, str], asyncio.Future] = {}
@asynccontextmanager
async def agents_lifespan():
@@ -164,7 +164,7 @@ async def generate_group_meta(session_id: str, body: dict):
is_refinement = body.get("is_refinement", False)
key = (session_id, group_id)
if not is_refinement:
existing = _group_meta_inflight.get(key)
existing = p_group_meta_inflight.get(key)
if existing is not None and not existing.done():
try:
return await existing
@@ -174,7 +174,7 @@ async def generate_group_meta(session_id: str, body: dict):
future: asyncio.Future = asyncio.get_event_loop().create_future()
if not is_refinement:
_group_meta_inflight[key] = future
p_group_meta_inflight[key] = future
try:
result = await agent_manager.generate_group_meta(
session_id,
@@ -191,8 +191,8 @@ async def generate_group_meta(session_id: str, body: dict):
future.set_exception(e)
raise
finally:
if not is_refinement and _group_meta_inflight.get(key) is future:
_group_meta_inflight.pop(key, None)
if not is_refinement and p_group_meta_inflight.get(key) is future:
p_group_meta_inflight.pop(key, None)
@agents.router.patch("/sessions/{session_id}")
async def update_session(session_id: str, body: dict):
@@ -383,10 +383,10 @@ async def subscriptions_connect(body: dict):
raise HTTPException(status_code=503, detail="9Router not available. Please install Node.js.")
# Reconnecting gemini-cli must wipe antigravity; registry prefers AG and a stale AG token would 400 after gemini-cli refreshes.
cascade = _PROVIDER_CASCADE_REMOVES.get(provider, [])
cascade = P_PROVIDER_CASCADE_REMOVES.get(provider, [])
if cascade:
try:
await _delete_provider_connections(cascade)
await p_delete_provider_connections(cascade)
except Exception:
pass
@@ -809,12 +809,12 @@ async def list_models():
# gemini-cli and antigravity are two Google OAuth lanes; registry prefers AG, so we cascade-wipe AG when reconnecting gemini-cli to avoid stale-AG 400s. One-directional: AG operations MUST NOT cascade back.
_PROVIDER_CASCADE_REMOVES: dict[str, list[str]] = {
P_PROVIDER_CASCADE_REMOVES: dict[str, list[str]] = {
"gemini-cli": ["antigravity"],
}
async def _delete_provider_connections(providers: list[str]) -> int:
async def p_delete_provider_connections(providers: list[str]) -> int:
"""Delete 9Router connections in `providers`; returns count removed, silent on 9Router unreachable."""
import httpx
from backend.apps.nine_router import NINE_ROUTER_API, get_providers
@@ -842,8 +842,8 @@ async def subscriptions_disconnect(body: dict):
raise HTTPException(status_code=400, detail="provider required")
try:
to_remove = [provider, *_PROVIDER_CASCADE_REMOVES.get(provider, [])]
removed = await _delete_provider_connections(to_remove)
to_remove = [provider, *P_PROVIDER_CASCADE_REMOVES.get(provider, [])]
removed = await p_delete_provider_connections(to_remove)
if removed:
from backend.apps.service.client import sync as _sync
from backend.apps.settings.settings import load_settings
backend/apps/agents/core/error_classify.py
+4 -4
View File
@@ -7,7 +7,7 @@ from typeguard import typechecked
# error string to telemetry. own_key mode means the subprocess stderr can echo
# the user's OWN provider key, so this scrub is the wall between a diagnostic
# and a key leak; over-redacting is fine, leaking is not.
_TELEMETRY_SECRET_PATTERNS = (
P_TELEMETRY_SECRET_PATTERNS = (
re.compile(r"sk-ant-[A-Za-z0-9_\-]{12,}"),
re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),
@@ -23,7 +23,7 @@ def redact_for_telemetry(text: str, *, limit: int = 2000) -> str:
error/stderr string goes through here before it leaves the machine."""
if not text:
return ""
for pat in _TELEMETRY_SECRET_PATTERNS:
for pat in P_TELEMETRY_SECRET_PATTERNS:
text = pat.sub("[redacted]", text)
return text[-limit:]
@@ -51,7 +51,7 @@ TRANSIENT_CAPACITY_PATTERNS = re.compile(
# zero tokens. That is NOT auth, reconnecting won't help, the request shape is
# wrong, so we classify it apart and stop the catch-all from showing a
# "reconnect your subscription" card for a tool-schema 400.
_TRANSLATION_ERROR_PATTERNS = re.compile(
P_TRANSLATION_ERROR_PATTERNS = re.compile(
r"(?:function_declarations"
r"|invalid_argument"
r"|invalid\s+json\s+payload"
@@ -128,7 +128,7 @@ def is_translation_error(exc: BaseException, extra_text: str = "") -> bool:
combined = f"{exc!s}\n{extra_text}".strip()
if not combined:
return False
return bool(_TRANSLATION_ERROR_PATTERNS.search(combined))
return bool(P_TRANSLATION_ERROR_PATTERNS.search(combined))
@typechecked
backend/apps/auth/router.py
+9 -9
View File
@@ -40,14 +40,14 @@ async def auth_lifespan():
auth = SubApp("auth", auth_lifespan)
def _proxy_url() -> str:
def p_proxy_url() -> str:
settings_obj = load_settings()
url = (getattr(settings_obj, "openswarm_proxy_url", None)
or OPENSWARM_DEFAULT_PROXY_URL)
return url.rstrip("/")
async def _sync_pro_routing(settings_obj) -> None:
async def p_sync_pro_routing(settings_obj) -> None:
"""Mirror connection state into 9Router's Claude lane; sign-in can flip a
paying user into pro mode and sign-out must tear the lane down so a
revoked bearer doesn't linger in the router."""
@@ -58,7 +58,7 @@ async def _sync_pro_routing(settings_obj) -> None:
logger.debug("pro routing sync skipped: %s", e)
def _sync_identity_to_service(settings_obj) -> None:
def p_sync_identity_to_service(settings_obj) -> None:
"""Push user_id + email + signin_method into the service-sync identify
pipeline so every event from this user has the right Person properties."""
try:
@@ -101,7 +101,7 @@ async def signin_activate(body: SigninActivateRequest):
if not body.token or len(body.token) < 16:
raise HTTPException(status_code=400, detail="Invalid token")
proxy = _proxy_url()
proxy = p_proxy_url()
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.post(
@@ -157,8 +157,8 @@ async def signin_activate(body: SigninActivateRequest):
settings_obj.openswarm_proxy_url = proxy
await save_settings_async(settings_obj)
_sync_identity_to_service(settings_obj)
await _sync_pro_routing(settings_obj)
p_sync_identity_to_service(settings_obj)
await p_sync_pro_routing(settings_obj)
return {
"ok": True,
@@ -185,7 +185,7 @@ async def signout():
"""
settings_obj = load_settings()
bearer = getattr(settings_obj, "openswarm_bearer_token", None)
proxy = _proxy_url()
proxy = p_proxy_url()
if bearer:
try:
async with httpx.AsyncClient(timeout=5.0) as client:
@@ -245,6 +245,6 @@ async def signout():
settings_obj.openswarm_subscription_expires = None
settings_obj.openswarm_usage_cached = None
await save_settings_async(settings_obj)
_sync_identity_to_service(settings_obj)
await _sync_pro_routing(settings_obj)
p_sync_identity_to_service(settings_obj)
await p_sync_pro_routing(settings_obj)
return {"ok": True}
backend/apps/discord_mcp_shim/server.py
+49 -49
View File
@@ -214,7 +214,7 @@ TOOLS = [
# -- HTTP plumbing ---------------------------------------------------------
def _call(
def p_call(
method: str,
path: str,
*,
@@ -267,17 +267,17 @@ def _call(
return 0, f"Request failed: {e!r}"
def _err(text: str) -> dict:
def p_err(text: str) -> dict:
return {"content": [{"type": "text", "text": f"Error: {text}"}], "isError": True}
def _ok(payload) -> dict:
def p_ok(payload) -> dict:
if isinstance(payload, str):
return {"content": [{"type": "text", "text": payload}]}
return {"content": [{"type": "text", "text": json.dumps(payload, indent=2, default=str)}]}
def _check_guild(guild_id: str) -> str | None:
def p_check_guild(guild_id: str) -> str | None:
"""Return an error string if guild_id is outside the user-authorized set, else None.
The set is sourced from OPENSWARM_DISCORD_GUILD_IDS env var (CSV) which
@@ -299,69 +299,69 @@ def _check_guild(guild_id: str) -> str | None:
def handle_tool_call(name: str, args: dict) -> dict:
if name == "discord_login":
status, body = _call("GET", "/users/@me/guilds")
status, body = p_call("GET", "/users/@me/guilds")
if status != 200:
return _err(f"Discord proxy unreachable (HTTP {status}): {body}")
return _ok({"connected": True, "guilds": body})
return p_err(f"Discord proxy unreachable (HTTP {status}): {body}")
return p_ok({"connected": True, "guilds": body})
if name == "discord_get_server_info":
gid = str(args.get("guild_id", ""))
if (e := _check_guild(gid)): return _err(e)
status, body = _call("GET", f"/guilds/{gid}")
return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
if (e := p_check_guild(gid)): return p_err(e)
status, body = p_call("GET", f"/guilds/{gid}")
return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
if name == "discord_list_channels":
gid = str(args.get("guild_id", ""))
if (e := _check_guild(gid)): return _err(e)
status, body = _call("GET", f"/guilds/{gid}/channels")
return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
if (e := p_check_guild(gid)): return p_err(e)
status, body = p_call("GET", f"/guilds/{gid}/channels")
return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
if name == "discord_create_text_channel":
gid = str(args.get("guild_id", ""))
if (e := _check_guild(gid)): return _err(e)
if (e := p_check_guild(gid)): return p_err(e)
payload: dict = {"name": args.get("name", ""), "type": 0}
if args.get("parent_id"): payload["parent_id"] = args["parent_id"]
if args.get("topic"): payload["topic"] = args["topic"]
status, body = _call("POST", f"/guilds/{gid}/channels", body=payload)
return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
status, body = p_call("POST", f"/guilds/{gid}/channels", body=payload)
return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
if name == "discord_create_category":
gid = str(args.get("guild_id", ""))
if (e := _check_guild(gid)): return _err(e)
status, body = _call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4})
return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
if (e := p_check_guild(gid)): return p_err(e)
status, body = p_call("POST", f"/guilds/{gid}/channels", body={"name": args.get("name", ""), "type": 4})
return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
if name == "discord_edit_category":
cid = str(args.get("channel_id", ""))
payload: dict = {}
if args.get("name"): payload["name"] = args["name"]
status, body = _call("PATCH", f"/channels/{cid}", body=payload)
return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
status, body = p_call("PATCH", f"/channels/{cid}", body=payload)
return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
if name == "discord_delete_category" or name == "discord_delete_channel":
cid = str(args.get("channel_id", ""))
status, body = _call("DELETE", f"/channels/{cid}")
return _ok({"deleted": True}) if status in (200, 204) else _err(f"HTTP {status}: {body}")
status, body = p_call("DELETE", f"/channels/{cid}")
return p_ok({"deleted": True}) if status in (200, 204) else p_err(f"HTTP {status}: {body}")
if name == "discord_send":
cid = str(args.get("channel_id", ""))
content = str(args.get("content", ""))
status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content})
return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content})
return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
if name == "discord_read_messages":
cid = str(args.get("channel_id", ""))
limit = max(1, min(int(args.get("limit", 50) or 50), 100))
status, body = _call("GET", f"/channels/{cid}/messages", query={"limit": limit})
return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
status, body = p_call("GET", f"/channels/{cid}/messages", query={"limit": limit})
return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
if name == "discord_add_reaction":
cid = str(args.get("channel_id", ""))
mid = str(args.get("message_id", ""))
emoji = str(args.get("emoji", ""))
# Discord's URL needs the emoji urlencoded; passes through.
status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me")
return _ok({"added": emoji}) if status in (200, 204) else _err(f"HTTP {status}: {body}")
status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(emoji, safe='')}/@me")
return p_ok({"added": emoji}) if status in (200, 204) else p_err(f"HTTP {status}: {body}")
if name == "discord_add_multiple_reactions":
cid = str(args.get("channel_id", ""))
@@ -369,46 +369,46 @@ def handle_tool_call(name: str, args: dict) -> dict:
emojis = args.get("emojis", []) or []
results = []
for e in emojis:
status, body = _call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me")
status, body = p_call("PUT", f"/channels/{cid}/messages/{mid}/reactions/{urllib.parse.quote(str(e), safe='')}/@me")
results.append({"emoji": e, "ok": status in (200, 204), "status": status})
return _ok({"reactions": results})
return p_ok({"reactions": results})
if name == "discord_get_forum_channels":
gid = str(args.get("guild_id", ""))
if (e := _check_guild(gid)): return _err(e)
status, body = _call("GET", f"/guilds/{gid}/channels")
if status != 200: return _err(f"HTTP {status}: {body}")
if (e := p_check_guild(gid)): return p_err(e)
status, body = p_call("GET", f"/guilds/{gid}/channels")
if status != 200: return p_err(f"HTTP {status}: {body}")
# Filter to type 15 (forum). Discord channel types reference:
# GUILD_FORUM = 15
forums = [ch for ch in (body or []) if isinstance(ch, dict) and ch.get("type") == 15]
return _ok(forums)
return p_ok(forums)
if name == "discord_create_forum_post":
fid = str(args.get("forum_id", ""))
status, body = _call("POST", f"/channels/{fid}/threads", body={
status, body = p_call("POST", f"/channels/{fid}/threads", body={
"name": args.get("name", ""),
"message": {"content": args.get("content", "")},
})
return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
if name == "discord_get_forum_post":
cid = str(args.get("channel_id", ""))
mid = str(args.get("message_id", ""))
status, body = _call("GET", f"/channels/{cid}/messages/{mid}")
return _ok(body) if status == 200 else _err(f"HTTP {status}: {body}")
status, body = p_call("GET", f"/channels/{cid}/messages/{mid}")
return p_ok(body) if status == 200 else p_err(f"HTTP {status}: {body}")
if name == "discord_reply_to_forum":
cid = str(args.get("channel_id", ""))
content = str(args.get("content", ""))
status, body = _call("POST", f"/channels/{cid}/messages", body={"content": content})
return _ok(body) if status in (200, 201) else _err(f"HTTP {status}: {body}")
status, body = p_call("POST", f"/channels/{cid}/messages", body={"content": content})
return p_ok(body) if status in (200, 201) else p_err(f"HTTP {status}: {body}")
return _err(f"Unknown tool: {name}")
return p_err(f"Unknown tool: {name}")
# -- JSON-RPC stdio loop ---------------------------------------------------
def _send(id_, result=None, error=None):
def p_send(id_, result=None, error=None):
msg = {"jsonrpc": "2.0", "id": id_}
if error is not None:
msg["error"] = error
@@ -433,7 +433,7 @@ def main():
params = msg.get("params", {}) or {}
if method == "initialize":
_send(id_, {
p_send(id_, {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "openswarm-discord", "version": "1.0.0"},
@@ -441,18 +441,18 @@ def main():
elif method == "notifications/initialized":
pass
elif method == "tools/list":
_send(id_, {"tools": TOOLS})
p_send(id_, {"tools": TOOLS})
elif method == "tools/call":
name = params.get("name", "")
args = params.get("arguments", {}) or {}
try:
_send(id_, handle_tool_call(name, args))
p_send(id_, handle_tool_call(name, args))
except Exception as e:
_send(id_, _err(f"shim crashed: {e!r}"))
p_send(id_, p_err(f"shim crashed: {e!r}"))
elif method == "ping":
_send(id_, {})
p_send(id_, {})
elif id_ is not None:
_send(id_, error={"code": -32601, "message": f"Method not found: {method}"})
p_send(id_, error={"code": -32601, "message": f"Method not found: {method}"})
if __name__ == "__main__":
backend/apps/google_workspace_mcp_shim/run.py
+2 -2
View File
@@ -24,7 +24,7 @@ from google.oauth2.credentials import Credentials
@functools.lru_cache(maxsize=1)
def _patched_get_credentials():
def p_patched_get_credentials():
refresh_token = os.environ.get("GOOGLE_WORKSPACE_REFRESH_TOKEN")
if not refresh_token:
raise ValueError("GOOGLE_WORKSPACE_REFRESH_TOKEN env var is required")
@@ -40,7 +40,7 @@ def _patched_get_credentials():
)
gauth.get_credentials = _patched_get_credentials
gauth.get_credentials = p_patched_get_credentials
from google_workspace_mcp import __main__ as _gw_main # noqa: E402,F401
backend/apps/service/buffer.py
+15 -15
View File
@@ -26,18 +26,18 @@ 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
P_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
P_TRIM_TARGET_FRACTION = 0.75
_lock = threading.Lock()
p_lock = threading.Lock()
@contextmanager
def _conn(spool_path: str) -> Iterator[sqlite3.Connection]:
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."""
Caller holds `p_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:
@@ -58,7 +58,7 @@ 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:
with p_lock, conn(spool_path) as c:
c.execute(
"INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
(kind, body, now),
@@ -68,8 +68,8 @@ def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
size = os.path.getsize(spool_path)
except OSError:
size = 0
if size > _MAX_BYTES:
target = int(_MAX_BYTES * _TRIM_TARGET_FRACTION)
if size > P_MAX_BYTES:
target = int(P_MAX_BYTES * P_TRIM_TARGET_FRACTION)
# Delete oldest rows until we're back under target. Use a
# reasonable batch size so we don't block forever.
dropped = 0
@@ -86,11 +86,11 @@ def enqueue(spool_path: str, kind: str, payload: dict, *, now: float) -> None:
if new_size <= target:
break
if dropped:
logger.warning("Spool over %d MB cap; dropped %d oldest entries", _MAX_BYTES // (1024 * 1024), dropped)
logger.warning("Spool over %d MB cap; dropped %d oldest entries", P_MAX_BYTES // (1024 * 1024), dropped)
# 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:
if os.path.getsize(spool_path) > P_MAX_BYTES:
c.execute("VACUUM")
except (OSError, sqlite3.DatabaseError):
pass
@@ -102,7 +102,7 @@ def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
cloud accepts them."""
if not os.path.exists(spool_path):
return []
with _lock, _conn(spool_path) as c:
with p_lock, conn(spool_path) as c:
rows = c.execute(
"SELECT id, kind, payload FROM spool ORDER BY id ASC LIMIT ?",
(batch_size,),
@@ -113,7 +113,7 @@ def drain(spool_path: str, batch_size: int = 50) -> list[tuple[int, str, dict]]:
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:
with p_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
@@ -123,7 +123,7 @@ 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:
with p_lock, conn(spool_path) as c:
c.executemany("DELETE FROM spool WHERE id = ?", [(i,) for i in ids])
@@ -131,12 +131,12 @@ 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:
with p_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:
with p_lock, conn(spool_path) as c:
c.execute("DELETE FROM spool")
backend/apps/service/client.py
+66 -66
View File
@@ -34,26 +34,26 @@ from backend.apps.service.version import APP_VERSION
logger = logging.getLogger(__name__)
_DEFAULT_BASE = "https://api.openswarm.com"
_PATH_BY_KIND = {
P_DEFAULT_BASE = "https://api.openswarm.com"
P_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
P_TIMEOUT_SECONDS = 5.0
P_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()
test_sink: Optional[Any] = None
install_id: Optional[str] = None
p_user_id: Optional[str] = None
p_inflight = 0
p_inflight_lock = asyncio.Lock()
p_drain_lock = asyncio.Lock()
def _spool_path() -> str:
def spool_path() -> str:
try:
from backend.config.paths import SETTINGS_DIR
return os.path.join(SETTINGS_DIR, "service_spool.db")
@@ -63,14 +63,14 @@ def _spool_path() -> str:
def set_test_sink(fn: Optional[Any]) -> None:
"""Test seam; receives every submission instead of the network."""
global _test_sink
_test_sink = fn
global test_sink
test_sink = fn
def _get_install_id() -> str:
global _install_id
if _install_id:
return _install_id
def p_get_install_id() -> str:
global install_id
if install_id:
return install_id
try:
from backend.apps.settings.store import load_settings, save_settings
s = load_settings()
@@ -79,16 +79,16 @@ def _get_install_id() -> str:
iid = uuid4().hex
s.installation_id = iid
save_settings(s)
_install_id = iid
install_id = iid
except Exception:
_install_id = uuid4().hex
return _install_id
install_id = uuid4().hex
return install_id
def _get_user_id() -> Optional[str]:
global _user_id
if _user_id:
return _user_id
def p_get_user_id() -> Optional[str]:
global p_user_id
if p_user_id:
return p_user_id
try:
from backend.apps.settings.store import load_settings
s = load_settings()
@@ -108,11 +108,11 @@ def _get_user_id() -> Optional[str]:
def set_user_id(uid: Optional[str]) -> None:
global _user_id
_user_id = uid or None
global p_user_id
p_user_id = uid or None
def _is_enabled(kind: str) -> bool:
def p_is_enabled(kind: str) -> bool:
"""Honour user opt-out. Diagnostic always flows (errors block usability);
state + session honour the toggle."""
if kind == "diagnostic":
@@ -130,10 +130,10 @@ def _is_enabled(kind: str) -> bool:
return True
def _envelope() -> dict:
def p_envelope() -> dict:
"""Identity + environment metadata stamped on every submission."""
env: dict[str, Any] = {"install_id": _get_install_id()}
uid = _get_user_id()
env: dict[str, Any] = {"install_id": p_get_install_id()}
uid = p_get_user_id()
if uid:
env["user_id"] = uid
try:
@@ -182,20 +182,20 @@ def _envelope() -> dict:
return env
def _base_url() -> str:
def p_base_url() -> str:
try:
from backend.apps.settings.store 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
return P_DEFAULT_BASE
async def _post(path: str, body: dict) -> int | None:
url = f"{_base_url()}{path}"
async def p_post(path: str, body: dict) -> int | None:
url = f"{p_base_url()}{path}"
try:
async with httpx.AsyncClient(timeout=_TIMEOUT_SECONDS) as c:
async with httpx.AsyncClient(timeout=P_TIMEOUT_SECONDS) as c:
r = await c.post(url, json=body)
return r.status_code
except Exception as e:
@@ -203,42 +203,42 @@ async def _post(path: str, body: dict) -> int | None:
return None
def _delivered(status: int | None) -> bool:
def p_delivered(status: int | None) -> bool:
return status is not None and 200 <= status < 300
# 429/timeouts/5xx/network are worth retrying; other 4xx means the payload itself is rejected and retrying forever would just poison the spool.
def _retryable(status: int | None) -> bool:
def p_retryable(status: int | None) -> bool:
return status is None or status >= 500 or status in (408, 429)
async def _post_or_spool(path: str, body: dict, kind: str) -> None:
global _inflight
if _test_sink is not None:
async def p_post_or_spool(path: str, body: dict, kind: str) -> None:
global p_inflight
if test_sink is not None:
try:
_test_sink(kind, body)
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())
async with p_inflight_lock:
if p_inflight >= P_MAX_INFLIGHT:
buffer.enqueue(spool_path(), f"{kind}:{path}", body, now=time.time())
return
_inflight += 1
p_inflight += 1
try:
status = await _post(path, body)
if _retryable(status):
buffer.enqueue(_spool_path(), f"{kind}:{path}", body, now=time.time())
elif not _delivered(status):
status = await p_post(path, body)
if p_retryable(status):
buffer.enqueue(spool_path(), f"{kind}:{path}", body, now=time.time())
elif not p_delivered(status):
logger.warning("service POST %s rejected with HTTP %s; payload dropped", path, status)
finally:
async with _inflight_lock:
_inflight = max(0, _inflight - 1)
async with p_inflight_lock:
p_inflight = max(0, p_inflight - 1)
async def drain_spool(batch_size: int = 50) -> int:
async with _drain_lock:
entries = buffer.drain(_spool_path(), batch_size=batch_size)
async with p_drain_lock:
entries = buffer.drain(spool_path(), batch_size=batch_size)
if not entries:
return 0
succeeded: list[int] = []
@@ -247,16 +247,16 @@ async def drain_spool(batch_size: int = 50) -> int:
if not path:
succeeded.append(rid)
continue
status = await _post(path, body)
if _delivered(status):
status = await p_post(path, body)
if p_delivered(status):
succeeded.append(rid)
elif _retryable(status):
elif p_retryable(status):
break
else:
logger.warning("service replay %s rejected with HTTP %s; dropping spooled row", path, status)
succeeded.append(rid)
if succeeded:
buffer.acknowledge(_spool_path(), succeeded)
buffer.acknowledge(spool_path(), succeeded)
return len(succeeded)
@@ -264,7 +264,7 @@ async def drain_spool(batch_size: int = 50) -> int:
# Public API
# --------------------------------------------------------------------------
def _log(kind: str, payload: dict) -> None:
def p_log(kind: str, payload: dict) -> None:
"""Append to the rolling operational log for diagnostics."""
try:
from backend.apps.service.ring_buffer import record
@@ -288,26 +288,26 @@ def sync(data: dict | None = None) -> None:
Fire-and-forget; never raises.
"""
payload = data or {}
if not _is_enabled("state"):
if not p_is_enabled("state"):
return
body = {
"client_state": _envelope(),
"client_state": p_envelope(),
"d": payload,
"t": time.time(),
"submission_id": uuid4().hex,
}
_log("s", payload)
if _test_sink is not None:
p_log("s", payload)
if test_sink is not None:
try:
_test_sink("s", body)
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"))
p_schedule(p_post_or_spool(P_DEFAULT_SYNC_PATH, body, "s"))
# Internal routing; the cloud has one endpoint for everything.
_DEFAULT_SYNC_PATH = "/api/service/sync"
P_DEFAULT_SYNC_PATH = "/api/service/sync"
def submit(kind: str, payload: dict) -> None:
@@ -318,7 +318,7 @@ def submit(kind: str, payload: dict) -> None:
sync(payload)
def _schedule(coro) -> None:
def p_schedule(coro) -> None:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
backend/apps/service/ring_buffer.py
+9 -9
View File
@@ -6,15 +6,15 @@ import threading
import time
from collections import deque
_MAX_SIZE = 50
_lock = threading.Lock()
_buffer: deque[dict] = deque(maxlen=_MAX_SIZE)
P_MAX_SIZE = 50
p_lock = threading.Lock()
p_buffer: deque[dict] = deque(maxlen=P_MAX_SIZE)
def record(label: str, **meta: str | int | float | None) -> None:
"""Append an entry. Oldest drops when full."""
with _lock:
_buffer.append({
with p_lock:
p_buffer.append({
"l": label,
"t": time.time(),
**{k: v for k, v in meta.items() if v is not None},
@@ -23,10 +23,10 @@ def record(label: str, **meta: str | int | float | None) -> None:
def snapshot() -> list[dict]:
"""Return a copy of the current buffer, oldest first."""
with _lock:
return list(_buffer)
with p_lock:
return list(p_buffer)
def clear() -> None:
with _lock:
_buffer.clear()
with p_lock:
p_buffer.clear()
backend/apps/service/service.py
+46 -46
View File
@@ -27,22 +27,22 @@ from fastapi import Body
from backend.config.Apps import SubApp
from backend.config.paths import SESSIONS_DIR
from backend.apps.service import client as svc
from backend.apps.service.version import APP_VERSION, _read_app_version
from backend.apps.service.version import APP_VERSION, read_app_version
logger = logging.getLogger(__name__)
_pulse_task: asyncio.Task | None = None
_drain_task: asyncio.Task | None = None
p_pulse_task: asyncio.Task | None = None
p_drain_task: asyncio.Task | None = None
_9r_start_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
p_last_9r_cost: float | None = None
p_last_9r_prompt_tokens: int | None = None
p_last_9r_completion_tokens: int | None = None
p_last_9r_requests: int | None = None
P_RESTART_THRESHOLD = 1.0
def _compute_delta(current: float, last: float | None, threshold: float = _RESTART_THRESHOLD) -> tuple[float, float]:
def p_compute_delta(current: float, last: float | None, threshold: float = P_RESTART_THRESHOLD) -> tuple[float, float]:
if last is None:
return 0.0, current
if current < last - threshold:
@@ -52,25 +52,25 @@ def _compute_delta(current: float, last: float | None, threshold: float = _RESTA
return current - last, current
_pulse_count = 0
_pulse_hours: set = set()
_pulse_delta_cost_total = 0.0
_pulse_batch_size = 10
p_pulse_count = 0
p_pulse_hours: set = set()
p_pulse_delta_cost_total = 0.0
p_pulse_batch_size = 10
async def _pulse_loop():
async def p_pulse_loop():
"""Periodic state-pulse loop. Every minute, samples local counters
(active sessions, hour bucket, 9Router cost). Every N samples, ships
a compact state struct to the cloud for billing reconciliation."""
global _last_9r_cost, _last_9r_prompt_tokens, _last_9r_completion_tokens, _last_9r_requests
global _pulse_count, _pulse_hours, _pulse_delta_cost_total
global p_last_9r_cost, p_last_9r_prompt_tokens, p_last_9r_completion_tokens, p_last_9r_requests
global p_pulse_count, p_pulse_hours, p_pulse_delta_cost_total
while True:
await asyncio.sleep(60)
_pulse_count += 1
p_pulse_count += 1
try:
import datetime as _dt
_pulse_hours.add(_dt.datetime.now().hour)
p_pulse_hours.add(_dt.datetime.now().hour)
except Exception:
pass
@@ -84,34 +84,34 @@ async def _pulse_loop():
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)
_pulse_delta_cost_total += cost_delta
cost_delta, p_last_9r_cost = p_compute_delta(cur_cost, p_last_9r_cost)
prompt_delta, p_last_9r_prompt_tokens = p_compute_delta(cur_prompt, p_last_9r_prompt_tokens, threshold=1000)
completion_delta, p_last_9r_completion_tokens = p_compute_delta(cur_completion, p_last_9r_completion_tokens, threshold=1000)
requests_delta, p_last_9r_requests = p_compute_delta(cur_requests, p_last_9r_requests, threshold=10)
p_pulse_delta_cost_total += cost_delta
except Exception:
pass
if _pulse_count >= _pulse_batch_size:
if p_pulse_count >= p_pulse_batch_size:
try:
from backend.apps.agents.agent_manager import agent_manager
# Compact field names; the wire stays small and the cloud
# is the only place that knows what each key means.
svc.sync({
"a": len(agent_manager.sessions), # active sessions
"h": sorted(_pulse_hours), # hour bucket set
"n": _pulse_count, # samples in batch
"c": _last_9r_cost or 0, # cumulative cost
"d1": _pulse_delta_cost_total, # cost delta since last batch
"h": sorted(p_pulse_hours), # hour bucket set
"n": p_pulse_count, # samples in batch
"c": p_last_9r_cost or 0, # cumulative cost
"d1": p_pulse_delta_cost_total, # cost delta since last batch
})
except Exception:
pass
_pulse_count = 0
_pulse_hours = set()
_pulse_delta_cost_total = 0.0
p_pulse_count = 0
p_pulse_hours = set()
p_pulse_delta_cost_total = 0.0
async def _drain_loop():
async def p_drain_loop():
while True:
try:
await svc.drain_spool()
@@ -122,7 +122,7 @@ async def _drain_loop():
@asynccontextmanager
async def service_lifespan():
global _pulse_task, _drain_task, _9r_start_task
global p_pulse_task, p_drain_task, _9r_start_task
try:
from backend.apps.settings.settings import load_settings, save_settings
@@ -204,26 +204,26 @@ async def service_lifespan():
except Exception as e:
logger.debug(f"9Router auto-start skipped: {e}")
_pulse_task = asyncio.create_task(_pulse_loop())
_drain_task = asyncio.create_task(_drain_loop())
p_pulse_task = asyncio.create_task(p_pulse_loop())
p_drain_task = asyncio.create_task(p_drain_loop())
yield
if _pulse_task:
_pulse_task.cancel()
if p_pulse_task:
p_pulse_task.cancel()
try:
await _pulse_task
await p_pulse_task
except asyncio.CancelledError:
pass
_pulse_task = None
p_pulse_task = None
if _drain_task:
_drain_task.cancel()
if p_drain_task:
p_drain_task.cancel()
try:
await _drain_task
await p_drain_task
except asyncio.CancelledError:
pass
_drain_task = None
p_drain_task = None
if _9r_start_task and not _9r_start_task.done():
_9r_start_task.cancel()
@@ -249,7 +249,7 @@ service = SubApp("service", service_lifespan)
# Usage endpoints (user-facing, read by the Settings / Usage page)
# ---------------------------------------------------------------------------
def _load_all_sessions() -> list[dict]:
def p_load_all_sessions() -> list[dict]:
results = []
if not os.path.exists(SESSIONS_DIR):
return results
@@ -267,7 +267,7 @@ def _load_all_sessions() -> list[dict]:
async def usage_summary():
from backend.apps.agents.agent_manager import agent_manager
sessions = _load_all_sessions()
sessions = p_load_all_sessions()
for s in agent_manager.get_all_sessions():
sessions.append(s.model_dump(mode="json"))
@@ -499,4 +499,4 @@ async def post_event(body: dict):
@service.router.get("/spool/count")
async def spool_count():
from backend.apps.service import buffer
return {"pending": buffer.count(svc._spool_path())}
return {"pending": buffer.count(svc.spool_path())}
backend/apps/service/version.py
+2 -2
View File
@@ -8,7 +8,7 @@ import json
import os
def _read_app_version() -> str:
def read_app_version() -> str:
# Preferred: Electron's main process injects this when spawning the
# backend (see electron/main.js; OPENSWARM_APP_VERSION). Always reliable
# in packaged builds because it comes from app.getVersion() rather than
@@ -32,4 +32,4 @@ def _read_app_version() -> str:
return "unknown"
APP_VERSION = _read_app_version()
APP_VERSION = read_app_version()
backend/apps/settings/settings.py
+2 -2
View File
@@ -230,8 +230,8 @@ async def apply_settings_update(body: AppSettings, protect_fields: set[str] | No
# owned, so the loop above just restored it to "free-trial") would keep them
# pinned to the forced Haiku lane even though they pasted a real key.
if getattr(old, "connection_mode", "own_key") == "free-trial":
from backend.apps.subscription.free_trial import _has_own_model
if _has_own_model(body):
from backend.apps.subscription.free_trial import has_own_model
if has_own_model(body):
body.connection_mode = "own_key"
body.free_trial_token = None
body.free_trial_remaining = None
backend/apps/subscription/free_trial.py
+19 -19
View File
@@ -28,10 +28,10 @@ logger = logging.getLogger(__name__)
# Namespaces the hash so a raw hardware UUID never leaves the device. Public on
# purpose (open-source): it only prevents transmitting the raw id, not a secret.
_FP_SALT = "openswarm-free-trial-v1"
P_FP_SALT = "openswarm-free-trial-v1"
def _enabled() -> bool:
def p_enabled() -> bool:
# Default ON as of 1.2.80: the cloud free-trial proxy is live on prod
# (api.openswarm.com) and arming + metered Haiku were verified end to end.
# Set OPENSWARM_FREE_TRIAL_ENABLED=0 to force it off. The pool-shed gate +
@@ -40,7 +40,7 @@ def _enabled() -> bool:
return os.environ.get("OPENSWARM_FREE_TRIAL_ENABLED", "1") == "1"
def _raw_hardware_id() -> str | None:
def p_raw_hardware_id() -> str | None:
"""A stable per-machine id that survives app reinstall / data wipe."""
system = platform.system()
try:
@@ -69,18 +69,18 @@ def _raw_hardware_id() -> str | None:
return None
def _fingerprint(settings_obj) -> str | None:
raw = _raw_hardware_id()
def p_fingerprint(settings_obj) -> str | None:
raw = p_raw_hardware_id()
if not raw:
# Fail-soft: installation_id is less durable (regenerates on wipe) but
# better than nothing on a machine where the hardware id can't be read.
raw = getattr(settings_obj, "installation_id", None)
if not raw:
return None
return hashlib.sha256((_FP_SALT + raw).encode("utf-8")).hexdigest()
return hashlib.sha256((P_FP_SALT + raw).encode("utf-8")).hexdigest()
def _has_own_model(s) -> bool:
def has_own_model(s) -> bool:
"""True if the user already has any real model path in settings; never shadow it."""
if any(getattr(s, k, None) for k in (
"anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key",
@@ -97,7 +97,7 @@ def _has_own_model(s) -> bool:
return False
async def _has_connected_subscription() -> bool:
async def p_has_connected_subscription() -> bool:
"""True if 9Router holds a live Claude/ChatGPT/Gemini subscription. Those
connections live in 9Router, not settings, so the sync check above misses
them; this catches a sub connected while the trial was armed."""
@@ -123,11 +123,11 @@ async def _has_connected_subscription() -> bool:
return False
def _proxy_base(settings_obj) -> str:
def p_proxy_base(settings_obj) -> str:
return (getattr(settings_obj, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
async def _sync_routing(settings_obj) -> None:
async def p_sync_routing(settings_obj) -> None:
try:
from backend.apps.nine_router import sync_pro_routing
await sync_pro_routing(settings_obj)
@@ -149,23 +149,23 @@ async def clear_free_trial(settings_obj) -> None:
settings_obj.default_model = "sonnet"
settings_obj.free_trial_token = None
await save_settings_async(settings_obj)
await _sync_routing(settings_obj)
await p_sync_routing(settings_obj)
async def arm_free_trial(settings_obj) -> dict:
"""Mint (or re-fetch) the machine's grant and, if runs remain, flip into
free-trial mode. Guarded: never arms over a real key/subscription."""
if not _enabled():
if not p_enabled():
return {"armed": False, "reason": "disabled"}
mode = getattr(settings_obj, "connection_mode", "own_key")
if mode not in ("own_key", "free-trial"):
return {"armed": False, "reason": "other_mode"}
own = _has_own_model(settings_obj)
own = has_own_model(settings_obj)
has_sub = False
if not own:
# A subscription lives in 9Router, not settings, and 9Router now starts in
# the BACKGROUND (non-blocking boot), so at first-launch mint time it isn't
# up yet. Without this wait _has_connected_subscription() reads False and
# up yet. Without this wait p_has_connected_subscription() reads False and
# we'd arm the free trial OVER a real Claude/ChatGPT/Gemini sub, pinning the
# user to Haiku until they manually reload. Bring 9Router up so the sub is
# actually visible before we decide. Bounded + idempotent (shares the start
@@ -183,7 +183,7 @@ async def arm_free_trial(settings_obj) -> dict:
# sub-less user exhausts these in ~1.2s and falls through to arm, so this
# never waits on a subscription that doesn't exist.
for _i in range(5):
if await _has_connected_subscription():
if await p_has_connected_subscription():
has_sub = True
break
if _i < 4:
@@ -195,11 +195,11 @@ async def arm_free_trial(settings_obj) -> dict:
await clear_free_trial(settings_obj)
return {"armed": False, "reason": "has_model"}
fp = _fingerprint(settings_obj)
fp = p_fingerprint(settings_obj)
if not fp:
return {"armed": False, "reason": "no_fingerprint"}
base = _proxy_base(settings_obj)
base = p_proxy_base(settings_obj)
payload: dict = {"fingerprint_hash": fp}
if getattr(settings_obj, "installation_id", None):
payload["install_id"] = settings_obj.installation_id
@@ -230,7 +230,7 @@ async def arm_free_trial(settings_obj) -> dict:
# Using Haiku end to end means the CLI never adds it, so the run just works.
settings_obj.default_model = "haiku"
await save_settings_async(settings_obj)
await _sync_routing(settings_obj)
await p_sync_routing(settings_obj)
return {"armed": True, "runs_remaining": remaining, "runs_limit": settings_obj.free_trial_runs_limit}
# Already spent on this machine: record it but don't arm.
@@ -245,7 +245,7 @@ async def refresh_free_trial(settings_obj) -> dict:
if getattr(settings_obj, "connection_mode", "own_key") != "free-trial" or not token:
return {"connected": False, "runs_remaining": getattr(settings_obj, "free_trial_remaining", None)}
base = _proxy_base(settings_obj)
base = p_proxy_base(settings_obj)
try:
async with httpx.AsyncClient(timeout=5.0) as client:
r = await client.post(
backend/apps/subscription/router.py
+16 -16
View File
@@ -27,7 +27,7 @@ async def subscription_lifespan():
subscription = SubApp("subscription", subscription_lifespan)
def _proxy_url() -> str:
def p_proxy_url() -> str:
"""Cloud router base URL. Overridable per-user via settings, falling back
to the module-default. No trailing slash."""
settings_obj = load_settings()
@@ -36,7 +36,7 @@ def _proxy_url() -> str:
return url.rstrip("/")
async def _sync_pro_routing(settings_obj) -> None:
async def p_sync_pro_routing(settings_obj) -> None:
"""Mirror connection state into 9Router's Claude lane (WebSearch on
non-Claude primaries). PUT /api/settings no longer carries these fields,
so the state-change endpoints here are the only trigger left."""
@@ -47,7 +47,7 @@ async def _sync_pro_routing(settings_obj) -> None:
logger.debug("pro routing sync skipped: %s", e)
async def _clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None:
async def p_clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None:
"""Revert to own_key mode and drop OpenSwarm Pro routing state.
`drop_bearer=True` (the default) is the original behavior, used when the
@@ -67,11 +67,11 @@ async def _clear_subscription(settings_obj, *, drop_bearer: bool = True) -> None
settings_obj.openswarm_subscription_expires = None
settings_obj.openswarm_usage_cached = None
await save_settings_async(settings_obj)
_sync_subscription_identity(settings_obj)
await _sync_pro_routing(settings_obj)
p_sync_subscription_identity(settings_obj)
await p_sync_pro_routing(settings_obj)
def _sync_subscription_identity(settings_obj) -> None:
def p_sync_subscription_identity(settings_obj) -> None:
"""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; service-sync is fire-and-forget
@@ -119,7 +119,7 @@ async def activate(body: ActivateRequest):
if not body.token or len(body.token) < 16:
raise HTTPException(status_code=400, detail="Invalid token")
proxy = _proxy_url()
proxy = p_proxy_url()
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.get(
@@ -166,8 +166,8 @@ async def activate(body: ActivateRequest):
settings_obj.openswarm_usage_cached = usage
await save_settings_async(settings_obj)
_sync_subscription_identity(settings_obj)
await _sync_pro_routing(settings_obj)
p_sync_subscription_identity(settings_obj)
await p_sync_pro_routing(settings_obj)
return {"ok": True, "plan": settings_obj.openswarm_subscription_plan}
@@ -200,7 +200,7 @@ async def status():
try:
async with httpx.AsyncClient(timeout=5.0) as client:
r = await client.get(
f"{_proxy_url()}/api/me",
f"{p_proxy_url()}/api/me",
headers={"Authorization": f"Bearer {bearer}"},
)
upstream_code = r.status_code
@@ -220,7 +220,7 @@ async def status():
# through a dead subscription. Settings UI sees connected=False and
# falls back to the Subscribe CTA; chat reverts to own_key routing.
if upstream_code in (401, 402):
await _clear_subscription(settings_obj)
await p_clear_subscription(settings_obj)
return {
"connected": False,
"connection_mode": "own_key",
@@ -267,7 +267,7 @@ async def sync():
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.post(
f"{_proxy_url()}/api/subscription/sync",
f"{p_proxy_url()}/api/subscription/sync",
headers={"Authorization": f"Bearer {bearer}"},
)
except httpx.HTTPError as e:
@@ -279,7 +279,7 @@ async def sync():
# the bearer is dead or the sub expired, clear local state so the app
# reverts to own_key instead of hammering a useless token.
if r.status_code in (401, 402):
await _clear_subscription(settings_obj)
await p_clear_subscription(settings_obj)
reason = "revoked" if r.status_code == 401 else "expired"
_sync(settings_obj.model_dump())
return {
@@ -308,7 +308,7 @@ async def sync():
datetime.fromtimestamp(period_end_ms / 1000, tz=timezone.utc).isoformat()
)
await save_settings_async(settings_obj)
_sync_subscription_identity(settings_obj)
p_sync_subscription_identity(settings_obj)
_sync(settings_obj.model_dump())
return {
"ok": True,
@@ -334,7 +334,7 @@ async def portal():
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.post(
f"{_proxy_url()}/api/billing/portal",
f"{p_proxy_url()}/api/billing/portal",
headers={"Authorization": f"Bearer {bearer}"},
)
if r.status_code >= 400:
@@ -374,5 +374,5 @@ async def disconnect():
does NOT sign the user out of OpenSwarm (use /api/auth/signout for that).
Useful when a user wants to temporarily route through their own API key
without losing their account state."""
await _clear_subscription(load_settings(), drop_bearer=False)
await p_clear_subscription(load_settings(), drop_bearer=False)
return {"ok": True}
backend/tests/test_free_trial.py
+12 -12
View File
@@ -12,7 +12,7 @@ from backend.apps.agents.core.error_classify import (
)
from backend.apps.agents.providers.registry import resolve_model_id_for_sdk
from backend.apps.subscription import free_trial as ft
from backend.apps.subscription.free_trial import _has_own_model, arm_free_trial, clear_free_trial
from backend.apps.subscription.free_trial import has_own_model, arm_free_trial, clear_free_trial
def test_proxy_auth_for_each_mode():
@@ -52,10 +52,10 @@ def test_exhaustion_is_classified_and_not_retried():
def test_has_own_model_never_shadows_a_real_provider():
assert not _has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
assert not _has_own_model(AppSettings())
assert _has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
assert _has_own_model(
assert not has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
assert not has_own_model(AppSettings())
assert has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
assert has_own_model(
AppSettings(connection_mode="openswarm-pro", openswarm_bearer_token="b")
)
@@ -67,7 +67,7 @@ async def test_arm_waits_for_9router_before_shadowing_a_background_started_sub(m
visible) BEFORE deciding, instead of arming the free trial over it."""
saved: list = []
monkeypatch.setattr(ft, "save_settings_async", _record(saved))
monkeypatch.setattr(ft, "_sync_routing", _noop)
monkeypatch.setattr(ft, "p_sync_routing", _noop)
started = {"called": False}
@@ -80,7 +80,7 @@ async def test_arm_waits_for_9router_before_shadowing_a_background_started_sub(m
import backend.apps.nine_router as nr
monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
monkeypatch.setattr(ft, "_has_connected_subscription", sub_visible_after_start)
monkeypatch.setattr(ft, "p_has_connected_subscription", sub_visible_after_start)
s = AppSettings() # no key, own_key mode: a subscription-only user
out = await arm_free_trial(s)
@@ -96,7 +96,7 @@ async def test_arm_tolerates_provider_load_lag(monkeypatch):
"""9Router's /api/providers can lag is_running on a cold start. arm must re-check
a few times so a sub that loads a beat late is still caught, not shadowed."""
monkeypatch.setattr(ft, "save_settings_async", _noop)
monkeypatch.setattr(ft, "_sync_routing", _noop)
monkeypatch.setattr(ft, "p_sync_routing", _noop)
async def fake_ensure_running():
return None
@@ -108,7 +108,7 @@ async def test_arm_tolerates_provider_load_lag(monkeypatch):
import backend.apps.nine_router as nr
monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
monkeypatch.setattr(ft, "_has_connected_subscription", lagging_sub)
monkeypatch.setattr(ft, "p_has_connected_subscription", lagging_sub)
s = AppSettings()
res = await ft.arm_free_trial(s)
@@ -129,10 +129,10 @@ async def test_arm_with_no_sub_is_bounded_and_falls_through_to_arm(monkeypatch):
import time
import backend.apps.nine_router as nr
monkeypatch.setattr(nr, "ensure_running", fake_ensure_running)
monkeypatch.setattr(ft, "_has_connected_subscription", never_sub)
monkeypatch.setattr(ft, "p_has_connected_subscription", never_sub)
# Short-circuit before the cloud mint so the test stays offline + deterministic;
# reaching this branch proves arm did NOT falsely conclude has_model.
monkeypatch.setattr(ft, "_fingerprint", lambda _s: None)
monkeypatch.setattr(ft, "p_fingerprint", lambda _s: None)
s = AppSettings()
t = time.monotonic()
@@ -145,7 +145,7 @@ async def test_arm_with_no_sub_is_bounded_and_falls_through_to_arm(monkeypatch):
@pytest.mark.asyncio
async def test_clear_reverts_forced_haiku_so_it_doesnt_outlive_the_trial(monkeypatch):
monkeypatch.setattr(ft, "save_settings_async", _noop)
monkeypatch.setattr(ft, "_sync_routing", _noop)
monkeypatch.setattr(ft, "p_sync_routing", _noop)
s = AppSettings(connection_mode="free-trial", free_trial_token="ftk", default_model="haiku")
await clear_free_trial(s)
backend/tests/test_service.py
+7 -7
View File
@@ -44,11 +44,11 @@ def patch_settings(tmp_path):
@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
client.install_id = None
client.p_user_id = None
client.test_sink = None
spool = tmp_path / "spool.db"
with patch.object(client, "_spool_path", lambda: str(spool)):
with patch.object(client, "spool_path", lambda: str(spool)):
yield
@@ -257,7 +257,7 @@ def test_buffer_missing_file(tmp_path):
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:
with buffer.conn(spool) as c:
c.execute(
"INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
("s:/x", "{not json", time.time()),
@@ -291,7 +291,7 @@ def test_install_id_persisted(sink, tmp_path):
import backend.apps.settings.store as settings_mod
settings_mod.SETTINGS_FILE = str(sf)
import backend.apps.service.client as client
client._install_id = None
client.install_id = None
from backend.apps.service.client import sync
sync({})
_, body = sink[0]
@@ -347,7 +347,7 @@ 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):
with patch.object(svc, "spool_path", lambda: spool):
buffer.enqueue(spool, "s:/x", {}, now=time.time())
result = await spool_count()
assert result == {"pending": 1}