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

[haik]: refactor naming convention and clean up dead code across backend outputs and service modules. Replace all leading-underscore private names with explicit p_/P_ prefix for module-internal symbols and drop the underscore for cross-module public APIs (load_all, save, load, walk_directory, find_free_port, etc). Rename the runtime module singleton from manager to RUNTIME_MANAGER. Remove unused legacy shims from client.py (submit_event, submit_session_close, record, set_test_sink, set_user_id, update_identity), drop clear() from ring_buffer.py and buffer.py, and remove the VIEW_BUILDER_SKILL backward-compat alias. Switch runtime.py from relative to absolute imports, rename the _d route param to data, and add public-usage comments to clarify cross-module API boundaries

Browse Source
This commit is contained in:
haikdc
2026-06-13 22:01:17 -07:00
parent fafbe45187
commit 87f09e9695
16 changed files with 501 additions and 566 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/agent_manager.py
+2 -4
View File
@@ -893,10 +893,8 @@ class AgentManager:
if file_path:
try:
await asyncio.sleep(0.4)
from backend.apps.outputs.runtime import (
manager as _outputs_runtime_manager,
)
errs = _outputs_runtime_manager.drain_errors_for_path(file_path)
from backend.apps.outputs.runtime import RUNTIME_MANAGER
errs = RUNTIME_MANAGER.drain_errors_for_path(file_path)
except Exception:
errs = []
if errs:
backend/apps/outputs/html_inject.py
+16 -16
View File
@@ -14,13 +14,13 @@ from jsonschema import validate as schema_validate, ValidationError as SchemaVal
logger = logging.getLogger(__name__)
MODEL_MAP = {
P_MODEL_MAP = {
"sonnet": "claude-sonnet-4-20250514",
"opus": "claude-opus-4-20250514",
"haiku": "claude-haiku-4-5-20251001",
}
def _get_anthropic_client(api_model: str | None = None):
def get_anthropic_client(api_model: str | None = None):
"""Create an AsyncAnthropic client using the API key from app settings.
When `api_model` is provided and carries a 9Router prefix (cc/, cx/, gc/),
@@ -39,7 +39,7 @@ def _get_anthropic_client(api_model: str | None = None):
return get_anthropic_client(settings)
def _validate_against_schema(data: dict, schema: dict) -> str | None:
def validate_against_schema(data: dict, schema: dict) -> str | None:
"""Validate *data* against *schema*. Return an error string or None."""
try:
schema_validate(instance=data, schema=schema)
@@ -49,7 +49,7 @@ def _validate_against_schema(data: dict, schema: dict) -> str | None:
return f"Schema validation failed at {path}: {exc.message}"
def _build_data_injection(input_json: str, result_json: str, backend_url_json: str = "null") -> str:
def p_build_data_injection(input_json: str, result_json: str, backend_url_json: str = "null") -> str:
"""Build a <script> tag that sets OUTPUT_INPUT / OUTPUT_BACKEND_RESULT /
OUTPUT_BACKEND_URL and listens for postMessage updates.
@@ -76,8 +76,8 @@ def _build_data_injection(input_json: str, result_json: str, backend_url_json: s
)
def _inject_data_into_html(html: str, input_json: str = "{}", result_json: str = "null", backend_url_json: str = "null") -> str:
injection = _build_data_injection(input_json, result_json, backend_url_json)
def inject_data_into_html(html: str, input_json: str = "{}", result_json: str = "null", backend_url_json: str = "null") -> str:
injection = p_build_data_injection(input_json, result_json, backend_url_json)
if "</head>" in html:
return html.replace("</head>", f"{injection}\n</head>", 1)
if "<body" in html:
@@ -85,13 +85,13 @@ def _inject_data_into_html(html: str, input_json: str = "{}", result_json: str =
return f"{injection}\n{html}"
def _backend_url_for_workspace(workspace_id: str) -> str:
def backend_url_for_workspace(workspace_id: str) -> str:
"""Return the JSON-encoded backend URL for the given workspace, or
"null" if no runtime is active. Cheap inline lookup so serve_workspace_file
doesn't have to think about it."""
try:
from backend.apps.outputs.runtime import manager as runtime_manager
rt = runtime_manager.get(workspace_id)
from backend.apps.outputs.runtime import RUNTIME_MANAGER
rt = RUNTIME_MANAGER.get(workspace_id)
if rt and rt.running and rt.port:
return json.dumps(f"http://127.0.0.1:{rt.port}")
except Exception:
@@ -103,18 +103,18 @@ def _backend_url_for_workspace(workspace_id: str) -> str:
# external (CDNs, mailto) or non-network references that the auth middleware
# never sees. Anything else is treated as a same-origin relative URL pointing
# at our /api/outputs/.../serve/ subtree, which DOES need the token.
_ABSOLUTE_URL_PREFIXES = (
P_ABSOLUTE_URL_PREFIXES = (
"http://", "https://", "//", "data:", "blob:",
"mailto:", "tel:", "javascript:", "about:", "#",
)
_HREF_SRC_ATTR_RE = re.compile(
P_HREF_SRC_ATTR_RE = re.compile(
r"""(\s(?:href|src))\s*=\s*(["'])([^"']+)\2""",
re.IGNORECASE,
)
def _inject_token_into_relative_urls(html: str, token: str) -> str:
def inject_token_into_relative_urls(html: str, token: str) -> str:
"""Append `?token=<t>` to every relative href/src in the served HTML.
Browsers strip the parent iframe URL's query string before resolving
@@ -126,10 +126,10 @@ def _inject_token_into_relative_urls(html: str, token: str) -> str:
if not token:
return html
def _patch(match: re.Match) -> str:
def patch(match: re.Match) -> str:
attr, quote, url = match.group(1), match.group(2), match.group(3)
lowered = url.lower().lstrip()
if lowered.startswith(_ABSOLUTE_URL_PREFIXES):
if lowered.startswith(P_ABSOLUTE_URL_PREFIXES):
return match.group(0)
if "token=" in url:
return match.group(0)
@@ -143,10 +143,10 @@ def _inject_token_into_relative_urls(html: str, token: str) -> str:
sep = "&" if "?" in base else "?"
return f'{attr}={quote}{base}{sep}token={token}{frag}{quote}'
return _HREF_SRC_ATTR_RE.sub(_patch, html)
return P_HREF_SRC_ATTR_RE.sub(patch, html)
def _decode_data_param(d: str) -> tuple[str, str]:
def decode_data_param(d: str) -> tuple[str, str]:
"""Decode the base64-encoded _d query param into (input_json, result_json)."""
try:
decoded = json.loads(base64.b64decode(d))
backend/apps/outputs/models.py
+4 -4
View File
@@ -28,7 +28,7 @@ class Output(BaseModel):
@model_validator(mode="before")
@classmethod
def _migrate_flat_fields(cls, data: Any) -> Any:
def p_migrate_flat_fields(cls, data: Any) -> Any:
"""Migrate legacy frontend_code/backend_code fields into the files dict."""
if not isinstance(data, dict):
return data
@@ -71,7 +71,7 @@ class OutputCreate(BaseModel):
@model_validator(mode="before")
@classmethod
def _migrate_flat_fields(cls, data: Any) -> Any:
def p_migrate_flat_fields(cls, data: Any) -> Any:
if not isinstance(data, dict):
return data
if "files" not in data or not data["files"]:
@@ -101,7 +101,7 @@ class OutputUpdate(BaseModel):
@model_validator(mode="before")
@classmethod
def _migrate_flat_fields(cls, data: Any) -> Any:
def p_migrate_flat_fields(cls, data: Any) -> Any:
if not isinstance(data, dict):
return data
if "files" not in data:
@@ -166,7 +166,7 @@ class WorkspaceSeedRequest(BaseModel):
@model_validator(mode="before")
@classmethod
def _migrate_flat_fields(cls, data: Any) -> Any:
def p_migrate_flat_fields(cls, data: Any) -> Any:
"""Accept legacy frontend_code/backend_code/schema_json fields."""
if not isinstance(data, dict):
return data
backend/apps/outputs/outputs.py
+66 -66
View File
@@ -22,18 +22,18 @@ from backend.apps.outputs.view_builder_templates import (
from backend.apps.settings.settings import load_settings
from backend.config.paths import OUTPUTS_DIR as DATA_DIR, OUTPUTS_WORKSPACE_DIR as WORKSPACE_DIR
from backend.apps.outputs.html_inject import (
_get_anthropic_client,
_validate_against_schema,
_inject_data_into_html,
_backend_url_for_workspace,
_inject_token_into_relative_urls,
_decode_data_param,
get_anthropic_client,
validate_against_schema,
inject_data_into_html,
backend_url_for_workspace,
inject_token_into_relative_urls,
decode_data_param,
)
from backend.apps.outputs.workspace_io import (
_load_all,
_save,
_load,
_walk_directory,
load_all,
save,
load,
walk_directory,
)
from backend.apps.outputs.prompts import VIBE_CODE_SYSTEM_PROMPT
@@ -52,8 +52,8 @@ async def outputs_lifespan():
# main backend dies, leaving ghost listeners on the .env-pinned
# ports that block the next OpenSwarm launch's reload preview.
try:
from backend.apps.outputs.runtime import manager as runtime_manager
killed = await runtime_manager.stop_all()
from backend.apps.outputs.runtime import RUNTIME_MANAGER
killed = await RUNTIME_MANAGER.stop_all()
if killed:
logger.info("outputs lifespan: reaped %d workspace runtimes on shutdown", killed)
except Exception:
@@ -68,7 +68,7 @@ outputs = SubApp("outputs", outputs_lifespan)
# ---------------------------------------------------------------------------
@outputs.router.get("/workspace/{workspace_id}/serve/{filepath:path}")
async def serve_workspace_file(workspace_id: str, filepath: str, _d: str = ""):
async def serve_workspace_file(workspace_id: str, filepath: str, data: str = ""):
"""Serve a file from a workspace folder. For index.html, inject OUTPUT data."""
folder = os.path.join(WORKSPACE_DIR, workspace_id)
full_path = os.path.normpath(os.path.join(folder, filepath))
@@ -81,31 +81,31 @@ async def serve_workspace_file(workspace_id: str, filepath: str, _d: str = ""):
content = f.read()
if filepath == "index.html":
input_json, result_json = _decode_data_param(_d) if _d else ("{}", "null")
backend_url_json = _backend_url_for_workspace(workspace_id)
content = _inject_data_into_html(content, input_json, result_json, backend_url_json)
input_json, result_json = decode_data_param(data) if data else ("{}", "null")
backend_url_json = backend_url_for_workspace(workspace_id)
content = inject_data_into_html(content, input_json, result_json, backend_url_json)
# Iframe sub-resource fetches (<link>, <script src>, <img>) drop the
# parent's ?token= query string, so rewrite the HTML to put the token
# back on every relative URL; otherwise sub-resources 401.
content = _inject_token_into_relative_urls(content, get_auth_token())
content = inject_token_into_relative_urls(content, get_auth_token())
mime, _ = mimetypes.guess_type(filepath)
return Response(content=content, media_type=mime or "text/plain")
@outputs.router.get("/{output_id}/serve/{filepath:path}")
async def serve_output_file(output_id: str, filepath: str, _d: str = ""):
async def serve_output_file(output_id: str, filepath: str, data: str = ""):
"""Serve a file from a saved output's files dict. For index.html, inject OUTPUT data."""
output = _load(output_id)
output = load(output_id)
content = output.files.get(filepath)
if content is None:
raise HTTPException(status_code=404, detail="File not found in output")
if filepath == "index.html":
input_json, result_json = _decode_data_param(_d) if _d else ("{}", "null")
backend_url_json = _backend_url_for_workspace(output.workspace_id) if output.workspace_id else "null"
content = _inject_data_into_html(content, input_json, result_json, backend_url_json)
content = _inject_token_into_relative_urls(content, get_auth_token())
input_json, result_json = decode_data_param(data) if data else ("{}", "null")
backend_url_json = backend_url_for_workspace(output.workspace_id) if output.workspace_id else "null"
content = inject_data_into_html(content, input_json, result_json, backend_url_json)
content = inject_token_into_relative_urls(content, get_auth_token())
mime, _ = mimetypes.guess_type(filepath)
return Response(content=content, media_type=mime or "text/plain")
@@ -117,7 +117,7 @@ async def serve_output_file(output_id: str, filepath: str, _d: str = ""):
@outputs.router.get("/list")
async def list_outputs():
return {"outputs": [o.model_dump() for o in _load_all()]}
return {"outputs": [o.model_dump() for o in load_all()]}
@outputs.router.get("/workspace/{workspace_id}")
@@ -127,7 +127,7 @@ async def read_workspace(workspace_id: str):
if not os.path.isdir(folder):
raise HTTPException(status_code=404, detail="Workspace not found")
files = _walk_directory(folder)
files = walk_directory(folder)
meta = None
if "meta.json" in files:
@@ -171,7 +171,7 @@ def sync_output_from_meta_json(workspace_id: str) -> bool:
description = str(meta.get("description") or "").strip()
if not name and not description:
return False
matching = [o for o in _load_all() if o.workspace_id == workspace_id]
matching = [o for o in load_all() if o.workspace_id == workspace_id]
if not matching:
return False
output = matching[0]
@@ -187,7 +187,7 @@ def sync_output_from_meta_json(workspace_id: str) -> bool:
changed = True
if changed:
output.updated_at = datetime.now().isoformat()
_save(output)
save(output)
return changed
except (OSError, json.JSONDecodeError, ValueError):
return False
@@ -223,18 +223,18 @@ def ensure_webapp_workspace_seeded_and_registered(
os.makedirs(folder, exist_ok=True)
already_seeded = os.path.exists(os.path.join(folder, "run.sh"))
if not already_seeded:
from backend.apps.outputs.runtime import _find_free_port
frontend_port = _find_free_port()
from backend.apps.outputs.runtime_proc import find_free_port
frontend_port = find_free_port()
seed_webapp_template_workspace(folder, frontend_port)
with open(os.path.join(folder, "SKILL.md"), "w", encoding="utf-8") as f:
f.write(load_app_builder_skill())
existing = [o for o in _load_all() if o.workspace_id == workspace_id]
existing = [o for o in load_all() if o.workspace_id == workspace_id]
if existing:
output = existing[0]
if session_id and output.session_id != session_id:
output.session_id = session_id
output.updated_at = datetime.now().isoformat()
_save(output)
save(output)
return output.id
now = datetime.now().isoformat()
output = Output(
@@ -247,7 +247,7 @@ def ensure_webapp_workspace_seeded_and_registered(
created_at=now,
updated_at=now,
)
_save(output)
save(output)
return output.id
except Exception:
logger.exception("ensure_webapp_workspace_seeded_and_registered failed for %s", workspace_id)
@@ -293,16 +293,16 @@ async def seed_workspace(body: WorkspaceSeedRequest):
# dirs_exist_ok=True + copytree). If `run.sh` already exists,
# the workspace was seeded on a previous visit; skip the file
# copy and only re-derive the frontend port from .env.
from backend.apps.outputs.runtime import _find_free_port, _read_env_value
from backend.apps.outputs.runtime_proc import find_free_port, read_env_value
already_seeded = os.path.exists(os.path.join(folder, "run.sh"))
if already_seeded:
fp_raw = _read_env_value(os.path.join(folder, ".env"), "FRONTEND_PORT")
fp_raw = read_env_value(os.path.join(folder, ".env"), "FRONTEND_PORT")
try:
frontend_port = int(fp_raw) if fp_raw else _find_free_port()
frontend_port = int(fp_raw) if fp_raw else find_free_port()
except (TypeError, ValueError):
frontend_port = _find_free_port()
frontend_port = find_free_port()
else:
frontend_port = _find_free_port()
frontend_port = find_free_port()
seed_webapp_template_workspace(folder, frontend_port)
# SKILL.md still goes in workspace root; agent reads it for
# context. Live content (user-editable via Skills page) is
@@ -323,7 +323,7 @@ async def seed_workspace(body: WorkspaceSeedRequest):
# remains the source of truth for the code.
output_id: Optional[str] = None
try:
existing = [o for o in _load_all() if o.workspace_id == body.workspace_id]
existing = [o for o in load_all() if o.workspace_id == body.workspace_id]
if existing:
output_id = existing[0].id
else:
@@ -337,7 +337,7 @@ async def seed_workspace(body: WorkspaceSeedRequest):
created_at=now,
updated_at=now,
)
_save(output)
save(output)
output_id = output.id
except Exception:
logger.exception("seed-time Output create failed for %s", body.workspace_id)
@@ -387,10 +387,10 @@ async def seed_workspace(body: WorkspaceSeedRequest):
# ---------------------------------------------------------------------------
def _runtime_status_payload(workspace_id: str) -> dict:
from backend.apps.outputs.runtime import manager as runtime_manager
from backend.apps.outputs.runtime import _is_new_mode
rt = runtime_manager.get(workspace_id)
def runtime_status_payload(workspace_id: str) -> dict:
from backend.apps.outputs.runtime import RUNTIME_MANAGER
from backend.apps.outputs.runtime_proc import is_new_mode
rt = RUNTIME_MANAGER.get(workspace_id)
if not rt:
# Even without a live runtime, the editor needs is_new_mode to
# decide whether the preview pane should fall back to the legacy
@@ -399,7 +399,7 @@ def _runtime_status_payload(workspace_id: str) -> dict:
# Compute from disk so a failed runtime/start still gives the
# client the right hint instead of dumping it onto a 404.
folder = os.path.join(WORKSPACE_DIR, workspace_id)
is_new = _is_new_mode(folder) if os.path.isdir(folder) else False
is_new = is_new_mode(folder) if os.path.isdir(folder) else False
return {
"running": False,
"port": None,
@@ -431,16 +431,16 @@ async def runtime_start(workspace_id: str):
folder = os.path.join(WORKSPACE_DIR, workspace_id)
if not os.path.isdir(folder):
raise HTTPException(status_code=404, detail="Workspace not found")
from backend.apps.outputs.runtime import manager as runtime_manager
await runtime_manager.attach(workspace_id, os.path.abspath(folder))
return _runtime_status_payload(workspace_id)
from backend.apps.outputs.runtime import RUNTIME_MANAGER
await RUNTIME_MANAGER.attach(workspace_id, os.path.abspath(folder))
return runtime_status_payload(workspace_id)
@outputs.router.post("/workspace/{workspace_id}/runtime/stop")
async def runtime_stop(workspace_id: str):
from backend.apps.outputs.runtime import manager as runtime_manager
await runtime_manager.detach(workspace_id)
return _runtime_status_payload(workspace_id)
from backend.apps.outputs.runtime import RUNTIME_MANAGER
await RUNTIME_MANAGER.detach(workspace_id)
return runtime_status_payload(workspace_id)
@outputs.router.post("/workspace/{workspace_id}/runtime/restart")
@@ -448,19 +448,19 @@ async def runtime_restart(workspace_id: str):
folder = os.path.join(WORKSPACE_DIR, workspace_id)
if not os.path.isdir(folder):
raise HTTPException(status_code=404, detail="Workspace not found")
from backend.apps.outputs.runtime import manager as runtime_manager
from backend.apps.outputs.runtime import RUNTIME_MANAGER
# Restart only if something's attached; otherwise this is a no-op
# silently (a hard-reload click while the runtime was already torn
# down; we'd rather not silently respawn an orphan).
rt = runtime_manager.get(workspace_id)
rt = RUNTIME_MANAGER.get(workspace_id)
if rt:
await runtime_manager.restart(workspace_id, os.path.abspath(folder))
return _runtime_status_payload(workspace_id)
await RUNTIME_MANAGER.restart(workspace_id, os.path.abspath(folder))
return runtime_status_payload(workspace_id)
@outputs.router.get("/workspace/{workspace_id}/runtime/status")
async def runtime_get_status(workspace_id: str):
return _runtime_status_payload(workspace_id)
return runtime_status_payload(workspace_id)
@outputs.router.post("/shutdown-all")
@@ -469,8 +469,8 @@ async def runtime_shutdown_all():
will-quit so app subprocesses die BEFORE the main backend gets
SIGTERM'd; without it `bash run.sh` + its vite/uvicorn descendants
reparent to PID 1 and squat on .env-pinned ports forever."""
from backend.apps.outputs.runtime import manager as runtime_manager
killed = await runtime_manager.stop_all()
from backend.apps.outputs.runtime import RUNTIME_MANAGER
killed = await RUNTIME_MANAGER.stop_all()
return {"ok": True, "killed": killed}
@@ -519,7 +519,7 @@ async def delete_workspace_file(workspace_id: str, filepath: str):
@outputs.router.get("/{output_id}")
async def get_output(output_id: str):
return _load(output_id).model_dump()
return load(output_id).model_dump()
@outputs.router.post("/create")
@@ -535,14 +535,14 @@ async def create_output(body: OutputCreate):
created_at=now,
updated_at=now,
)
_save(output)
save(output)
pass
return {"ok": True, "output": output.model_dump()}
@outputs.router.put("/{output_id}")
async def update_output(output_id: str, body: OutputUpdate):
output = _load(output_id)
output = load(output_id)
# exclude_unset, NOT exclude_none: a PUT that explicitly sends session_id=null
# (the Apps stale-link self-heal) must clear the field. exclude_none silently
# dropped that null, so the dead pointer never cleared and the app 404'd on
@@ -554,13 +554,13 @@ async def update_output(output_id: str, body: OutputUpdate):
# Only a real screenshot write moves the sort key; files/linkage saves don't reorder.
if body.thumbnail is not None:
output.preview_updated_at = now
_save(output)
save(output)
return {"ok": True, "output": output.model_dump()}
@outputs.router.delete("/{output_id}")
async def delete_output(output_id: str):
_load(output_id)
load(output_id)
path = os.path.join(DATA_DIR, f"{output_id}.json")
if os.path.exists(path):
os.remove(path)
@@ -589,7 +589,7 @@ async def vibe_code(body: VibeCodeRequest):
from backend.apps.agents.providers.registry import resolve_aux_model
try:
aux_model, _aux_base = await resolve_aux_model(load_settings(), preferred_tier="sonnet")
aux_model, _ = await resolve_aux_model(load_settings(), preferred_tier="sonnet")
except ValueError as e:
return {
"message": f"Error: {str(e)}",
@@ -597,7 +597,7 @@ async def vibe_code(body: VibeCodeRequest):
"backend_code": body.current_backend_code,
"input_schema": body.current_schema,
}
client = _get_anthropic_client(aux_model)
client = get_anthropic_client(aux_model)
try:
resp = await client.messages.create(
model=aux_model,
@@ -648,9 +648,9 @@ async def vibe_code(body: VibeCodeRequest):
@outputs.router.post("/execute")
async def execute_output(body: OutputExecute):
output = _load(body.output_id)
output = load(body.output_id)
validation_err = _validate_against_schema(body.input_data, output.input_schema)
validation_err = validate_against_schema(body.input_data, output.input_schema)
if validation_err:
return OutputExecuteResult(
output_id=output.id,
backend/apps/outputs/runtime.py
+158 -151
View File
@@ -10,7 +10,7 @@ from dataclasses import dataclass
from typing import Callable, Optional
def _resolve_bash() -> str:
def p_resolve_bash() -> str:
# Windows: Python's subprocess uses Windows-style PATH resolution and doesn't follow Git Bash's Unix-style entries like /mingw64/bin/..., so a bare "bash" call hits [WinError 2]. shutil.which goes through Windows PATHEXT lookup; fall back to the conventional Git for Windows install path so users without bash in their Windows PATH still work. POSIX: just return "bash" since the kernel finds it via PATH like any other exec.
found = shutil.which("bash")
if found:
@@ -25,29 +25,29 @@ def _resolve_bash() -> str:
return candidate
return "bash"
from .runtime_proc import (
_ERROR_PATTERNS,
_FRONTEND_BIND_POLL_INTERVAL,
_FRONTEND_BIND_TIMEOUT_SECONDS,
_LOG_BUFFER_LINES,
_MAX_IDLE_RUNTIMES,
_RECENT_ERRORS_MAX,
_TERMINATE_GRACE_SECONDS,
_background_priority_kwargs,
_find_free_port,
_is_new_mode,
_is_port_free,
_kill_descendant_tree,
_read_env_value,
_resume_process_tree,
_suspend_process_tree,
_write_env_value,
from backend.apps.outputs.runtime_proc import (
ERROR_PATTERNS,
FRONTEND_BIND_POLL_INTERVAL,
FRONTEND_BIND_TIMEOUT_SECONDS,
LOG_BUFFER_LINES,
MAX_IDLE_RUNTIMES,
RECENT_ERRORS_MAX,
TERMINATE_GRACE_SECONDS,
background_priority_kwargs,
find_free_port,
is_new_mode,
is_port_free,
kill_descendant_tree,
read_env_value,
resume_process_tree,
suspend_process_tree,
write_env_value,
)
logger = logging.getLogger(__name__)
# Module-level lock so only ONE vite optimizeDeps runs at a time; must be acquired before manager._lock to avoid deadlock with manager.attach.
_vite_boot_lock = asyncio.Lock()
# Module-level lock so only ONE vite optimizeDeps runs at a time; must be acquired before manager.p_lock to avoid deadlock with manager.attach.
P_VITE_BOOT_LOCK = asyncio.Lock()
@dataclass
@@ -82,28 +82,28 @@ class AppRuntime:
self.frontend_port: Optional[int] = None
# New-mode only: flips True once something is actually listening
# on frontend_port (we kick off a background poll task in
# _start_new_mode). frontend_url returns null until this flips,
# p_start_new_mode). frontend_url returns null until this flips,
# so the preview pane doesn't try to navigate to an unbound port
# and show a "Site can't be reached" error mid-npm-install.
self._frontend_ready: bool = False
self.p_frontend_ready: bool = False
# True while the process tree is SIGSTOP'd in the idle pool. A frozen
# vite still holds its port but can't answer it, so frontend_url must
# stay null while suspended (else the webview loads a dead port = the
# ERR_FAILED on fast app-switching).
self._suspended: bool = False
self.p_suspended: bool = False
self.process: Optional[asyncio.subprocess.Process] = None
self.log_buffer: deque[LogLine] = deque(maxlen=_LOG_BUFFER_LINES)
self._subscribers: set[LogSubscriber] = set()
self.log_buffer: deque[LogLine] = deque(maxlen=LOG_BUFFER_LINES)
self.p_subscribers: set[LogSubscriber] = set()
# Recent build/runtime errors scraped from stderr; drained by
# the agent's post-tool hook after Write/Edit so the agent sees
# vite/babel/uvicorn errors in its next turn and can self-fix
# instead of leaving the user with a red iframe overlay.
self.recent_errors: deque[str] = deque(maxlen=_RECENT_ERRORS_MAX)
self._stdout_task: Optional[asyncio.Task] = None
self._stderr_task: Optional[asyncio.Task] = None
self._wait_task: Optional[asyncio.Task] = None
self._frontend_ready_task: Optional[asyncio.Task] = None
self._lock = asyncio.Lock()
self.recent_errors: deque[str] = deque(maxlen=RECENT_ERRORS_MAX)
self.p_stdout_task: Optional[asyncio.Task] = None
self.p_stderr_task: Optional[asyncio.Task] = None
self.p_wait_task: Optional[asyncio.Task] = None
self.p_frontend_ready_task: Optional[asyncio.Task] = None
self.p_lock = asyncio.Lock()
def drain_errors(self) -> list[str]:
"""Pop and return all accumulated error lines. Used by the
@@ -123,21 +123,21 @@ class AppRuntime:
@property
def is_new_mode(self) -> bool:
return _is_new_mode(self.workspace_path)
return is_new_mode(self.workspace_path)
@property
def frontend_url(self) -> Optional[str]:
# Gated on `_frontend_ready` (set by the background bind-poll
# task in _start_new_mode) so the preview pane only switches
# Gated on `p_frontend_ready` (set by the background bind-poll
# task in p_start_new_mode) so the preview pane only switches
# over once Vite is actually accepting connections. Without
# this, the editor flashes a "Site can't be reached" error
# while `npm install` is running.
# Also gated on `running`: a vite that crashed or got orphaned still
# has _frontend_ready=True, and handing the webview that dead port is
# has p_frontend_ready=True, and handing the webview that dead port is
# the ERR_FAILED you see on reopen. No live process, no URL.
# And gated on `not _suspended`: a SIGSTOP'd idle runtime is "running"
# And gated on `not p_suspended`: a SIGSTOP'd idle runtime is "running"
# (returncode is None) but frozen, so its port won't answer.
if self.frontend_port and self._frontend_ready and self.running and not self._suspended:
if self.frontend_port and self.p_frontend_ready and self.running and not self.p_suspended:
return f"http://127.0.0.1:{self.frontend_port}/"
return None
@@ -160,11 +160,11 @@ class AppRuntime:
exists so the Terminal pane can host `[FRONTEND]` lines.
New-mode spawns are serialized through the module-level
`_vite_boot_lock` (see comment at the lock declaration) so a
`P_VITE_BOOT_LOCK` (see comment at the lock declaration) so a
burst of "create 3 apps in 5 seconds" doesn't trigger 3 parallel
MUI pre-bundle runs each pegging a core.
"""
async with self._lock:
async with self.p_lock:
if self.running:
return True
@@ -175,45 +175,45 @@ class AppRuntime:
# vite emits "frontend ready" (or its 180s timeout
# fires), which is the moment the next workspace can
# start its own vite without competing for the same
# CPU. See `_await_frontend_bind` for the release.
await _vite_boot_lock.acquire()
# CPU. See `p_await_frontend_bind` for the release.
await P_VITE_BOOT_LOCK.acquire()
try:
ok = await self._start_new_mode()
ok = await self.p_start_new_mode()
if not ok:
# Spawn failed before the bind-poll task was
# created; release synchronously so we don't
# wedge the next workspace.
_vite_boot_lock.release()
P_VITE_BOOT_LOCK.release()
return ok
except Exception:
_vite_boot_lock.release()
P_VITE_BOOT_LOCK.release()
raise
return await self._start_old_mode()
return await self.p_start_old_mode()
async def _start_new_mode(self) -> bool:
async def p_start_new_mode(self) -> bool:
env_path = os.path.join(self.workspace_path, ".env")
fp_raw = _read_env_value(env_path, "FRONTEND_PORT")
bp_raw = _read_env_value(env_path, "BACKEND_PORT")
fp_raw = read_env_value(env_path, "FRONTEND_PORT")
bp_raw = read_env_value(env_path, "BACKEND_PORT")
# FRONTEND_PORT is allocated by seed_workspace; should always be
# a number. If missing, fall back to a fresh allocation (rare
# edge case: workspace seeded by an older OpenSwarm).
try:
self.frontend_port = int(fp_raw) if fp_raw else _find_free_port()
self.frontend_port = int(fp_raw) if fp_raw else find_free_port()
except ValueError:
self.frontend_port = _find_free_port()
self.frontend_port = find_free_port()
# Port-collision safety net: if a ghost subprocess from a prior
# OpenSwarm run is still bound to the persisted port (force-quit,
# crash, OS killed the parent before stop_all could reap), Vite
# would EADDRINUSE silently. Re-probe and reallocate, then rewrite
# .env so the bash run.sh subprocess reads the new port.
if self.frontend_port and not _is_port_free(self.frontend_port):
new_port = _find_free_port()
self._broadcast(LogLine(
if self.frontend_port and not is_port_free(self.frontend_port):
new_port = find_free_port()
self.p_broadcast(LogLine(
"runtime",
f"[runtime] persisted FRONTEND_PORT {self.frontend_port} is in use; reallocating to {new_port}",
))
self.frontend_port = new_port
_write_env_value(env_path, "FRONTEND_PORT", str(new_port))
write_env_value(env_path, "FRONTEND_PORT", str(new_port))
# BACKEND_PORT may be the literal string "NONE" (frontend-only
# app; the common case) or a number once `backend_init.sh` has
# run. Only populate self.port when there's a real backend.
@@ -224,18 +224,18 @@ class AppRuntime:
self.port = None
# Same collision check for the backend port; a leaked uvicorn
# from a prior session would otherwise block the new spawn.
if self.port and not _is_port_free(self.port):
new_port = _find_free_port()
self._broadcast(LogLine(
if self.port and not is_port_free(self.port):
new_port = find_free_port()
self.p_broadcast(LogLine(
"runtime",
f"[runtime] persisted BACKEND_PORT {self.port} is in use; reallocating to {new_port}",
))
self.port = new_port
_write_env_value(env_path, "BACKEND_PORT", str(new_port))
write_env_value(env_path, "BACKEND_PORT", str(new_port))
else:
self.port = None
env = self._spawn_env_base()
env = self.p_spawn_env_base()
# bash run.sh reads .env itself; we don't need to set
# FRONTEND_PORT / BACKEND_PORT here. We DO export the install
# paths so the template's `backend/run.sh` can find our
@@ -244,47 +244,47 @@ class AppRuntime:
# the more reliable read site for subshells.)
# NOTE: keep these in sync with seed_webapp_template_workspace.
from backend.apps.outputs.view_builder_templates import (
_DEBUGGER_PATH,
_TEMPLATE_BACKEND_PATH,
DEBUGGER_PATH,
TEMPLATE_BACKEND_PATH,
)
env["OPENSWARM_DEBUGGER_PATH"] = _DEBUGGER_PATH
env["OPENSWARM_TEMPLATE_BACKEND_PATH"] = _TEMPLATE_BACKEND_PATH
env["OPENSWARM_DEBUGGER_PATH"] = DEBUGGER_PATH
env["OPENSWARM_TEMPLATE_BACKEND_PATH"] = TEMPLATE_BACKEND_PATH
try:
self.process = await asyncio.create_subprocess_exec(
_resolve_bash(), "run.sh",
p_resolve_bash(), "run.sh",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=self.workspace_path,
env=env,
**_background_priority_kwargs(),
**background_priority_kwargs(),
)
except Exception as e:
logger.exception("failed to start new-mode runtime for %s", self.workspace_id)
self._broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
self.p_broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
self.frontend_port = None
self.port = None
self.process = None
return False
backend_note = f" + backend on {self.port}" if self.port else ""
self._broadcast(LogLine("runtime", f"[runtime] bash run.sh started; frontend on {self.frontend_port}{backend_note} (pid {self.process.pid})"))
self._stdout_task = asyncio.create_task(self._pipe_stream(self.process.stdout, "stdout"))
self._stderr_task = asyncio.create_task(self._pipe_stream(self.process.stderr, "stderr"))
self._wait_task = asyncio.create_task(self._await_exit())
self.p_broadcast(LogLine("runtime", f"[runtime] bash run.sh started; frontend on {self.frontend_port}{backend_note} (pid {self.process.pid})"))
self.p_stdout_task = asyncio.create_task(self.p_pipe_stream(self.process.stdout, "stdout"))
self.p = asyncio.create_task(self.p_pipe_stream(self.process.stderr, "stderr"))
self.p_wait_task = asyncio.create_task(self.p_await_exit())
# Kick off the port-bind poller so frontend_url flips on once
# Vite is actually accepting connections.
self._frontend_ready = False
self._frontend_ready_task = asyncio.create_task(self._await_frontend_bind())
self.p_frontend_ready = False
self.p_frontend_ready_task = asyncio.create_task(self.p_await_frontend_bind())
return True
async def _await_frontend_bind(self) -> None:
"""Poll `frontend_port` every _FRONTEND_BIND_POLL_INTERVAL until
async def p_await_frontend_bind(self) -> None:
"""Poll `frontend_port` every FRONTEND_BIND_POLL_INTERVAL until
something binds (Vite dev server) or we hit the timeout. Emits a
`[runtime]` log line on success/failure so the Terminal pane
shows the transition; flips `_frontend_ready` which the
shows the transition; flips `p_frontend_ready` which the
`frontend_url` property reads.
Also responsible for releasing the module-level `_vite_boot_lock`
Also responsible for releasing the module-level `P_VITE_BOOT_LOCK`
; every exit path (success, process death, hard timeout) MUST
release exactly once so the next queued workspace can start its
own vite spawn. A try/finally on the lock guarantees that even
@@ -293,13 +293,13 @@ class AppRuntime:
# end doesn't double-release if a success path beat it.
lock_released = False
def _release_boot_lock() -> None:
def release_boot_lock() -> None:
nonlocal lock_released
if lock_released:
return
lock_released = True
try:
_vite_boot_lock.release()
P_VITE_BOOT_LOCK.release()
except RuntimeError:
# Lock already released (e.g. start() failure path
# released synchronously before spawning the poll task).
@@ -309,7 +309,7 @@ class AppRuntime:
if not self.frontend_port:
return
port = self.frontend_port
deadline = asyncio.get_event_loop().time() + _FRONTEND_BIND_TIMEOUT_SECONDS
deadline = asyncio.get_event_loop().time() + FRONTEND_BIND_TIMEOUT_SECONDS
while asyncio.get_event_loop().time() < deadline:
# Stop polling if the process died; pointless to keep
# checking a port nothing will bind.
@@ -326,8 +326,8 @@ class AppRuntime:
await writer.wait_closed()
except Exception:
pass
self._frontend_ready = True
self._broadcast(LogLine(
self.p_frontend_ready = True
self.p_broadcast(LogLine(
"runtime",
f"[runtime] frontend ready at http://127.0.0.1:{port}/",
))
@@ -335,32 +335,32 @@ class AppRuntime:
# ready; the next queued workspace can start its
# own bundle now even though we'll keep streaming
# logs for this one.
_release_boot_lock()
release_boot_lock()
return
except (OSError, asyncio.TimeoutError):
pass
await asyncio.sleep(_FRONTEND_BIND_POLL_INTERVAL)
await asyncio.sleep(FRONTEND_BIND_POLL_INTERVAL)
# Timed out; keep the runtime up (Terminal might show useful
# errors) but surface why the preview never appeared.
self._broadcast(LogLine(
self.p_broadcast(LogLine(
"runtime",
f"[runtime] frontend did NOT bind on port {port} after "
f"{_FRONTEND_BIND_TIMEOUT_SECONDS}s; check the Terminal "
f"{FRONTEND_BIND_TIMEOUT_SECONDS}s; check the Terminal "
f"for npm/vite errors.",
))
finally:
# Catches process-death return, timeout fall-through, and
# any exception in the poll body. _release_boot_lock is
# any exception in the poll body. release_boot_lock is
# idempotent so this is safe even after the success path
# already released.
_release_boot_lock()
release_boot_lock()
async def _start_old_mode(self) -> bool:
async def p_start_old_mode(self) -> bool:
if not self.has_backend_file:
self.port = None
return False
self.port = _find_free_port()
env = self._spawn_env_base()
self.port = find_free_port()
env = self.p_spawn_env_base()
env["PORT"] = str(self.port)
env["BACKEND_PORT"] = str(self.port) # alias; both common names work
try:
@@ -373,21 +373,21 @@ class AppRuntime:
stderr=asyncio.subprocess.PIPE,
cwd=self.workspace_path,
env=env,
**_background_priority_kwargs(),
**background_priority_kwargs(),
)
except Exception as e:
logger.exception("failed to start backend for %s", self.workspace_id)
self._broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
self.p_broadcast(LogLine("runtime", f"[runtime] failed to start: {e}"))
self.port = None
self.process = None
return False
self._broadcast(LogLine("runtime", f"[runtime] backend started on port {self.port} (pid {self.process.pid})"))
self._stdout_task = asyncio.create_task(self._pipe_stream(self.process.stdout, "stdout"))
self._stderr_task = asyncio.create_task(self._pipe_stream(self.process.stderr, "stderr"))
self._wait_task = asyncio.create_task(self._await_exit())
self.p_broadcast(LogLine("runtime", f"[runtime] backend started on port {self.port} (pid {self.process.pid})"))
self.p_stdout_task = asyncio.create_task(self.p_pipe_stream(self.process.stdout, "stdout"))
self.p = asyncio.create_task(self.p_pipe_stream(self.process.stderr, "stderr"))
self.p_wait_task = asyncio.create_task(self.p_await_exit())
return True
def _spawn_env_base(self) -> dict[str, str]:
def p_spawn_env_base(self) -> dict[str, str]:
"""Inherited env minus the install token. Backend.py can hit our
REST API back via its own creds if it really needs to, but it
shouldn't inherit the host process's token by default."""
@@ -402,33 +402,33 @@ class AppRuntime:
return env
async def stop(self) -> None:
async with self._lock:
async with self.p_lock:
if not self.process or self.process.returncode is not None:
# Still cancel the bind poller in case stop() races a
# never-launched runtime; defensive no-op otherwise.
if self._frontend_ready_task and not self._frontend_ready_task.done():
self._frontend_ready_task.cancel()
if self.p_frontend_ready_task and not self.p_frontend_ready_task.done():
self.p_frontend_ready_task.cancel()
return
try:
# Walk the descendant tree first so vite/uvicorn grandchildren
# die before bash exits and orphans them to PID 1. The webapp
# template's run.sh only traps EXIT, not TERM, so a flat
# SIGTERM to bash kills bash silently and leaves vite alive.
_kill_descendant_tree(self.process.pid, "TERM")
kill_descendant_tree(self.process.pid, "TERM")
self.process.terminate()
try:
await asyncio.wait_for(self.process.wait(), timeout=_TERMINATE_GRACE_SECONDS)
await asyncio.wait_for(self.process.wait(), timeout=TERMINATE_GRACE_SECONDS)
except asyncio.TimeoutError:
_kill_descendant_tree(self.process.pid, "KILL")
kill_descendant_tree(self.process.pid, "KILL")
self.process.kill()
await self.process.wait()
except ProcessLookupError:
pass
# Cancel the bind poller so it stops scanning a port that's
# gone away, and reset the readiness flag.
if self._frontend_ready_task and not self._frontend_ready_task.done():
self._frontend_ready_task.cancel()
self._frontend_ready = False
if self.p_frontend_ready_task and not self.p_frontend_ready_task.done():
self.p_frontend_ready_task.cancel()
self.p_frontend_ready = False
async def restart(self) -> bool:
await self.stop()
@@ -438,36 +438,36 @@ class AppRuntime:
"""Register a log subscriber. Immediately replays the ring buffer
so a Terminal pane that opens mid-session shows context. Returns
an unsubscribe function."""
self._subscribers.add(cb)
self.p_subscribers.add(cb)
for line in list(self.log_buffer):
try:
cb(line)
except Exception:
pass
def _unsub() -> None:
self._subscribers.discard(cb)
def unsub() -> None:
self.p_subscribers.discard(cb)
return _unsub
return unsub
def _broadcast(self, line: LogLine) -> None:
def p_broadcast(self, line: LogLine) -> None:
self.log_buffer.append(line)
# Snapshot subscribers; they can self-remove during dispatch.
for cb in list(self._subscribers):
for cb in list(self.p_subscribers):
try:
cb(line)
except Exception:
pass
def _maybe_capture_error(self, text: str) -> None:
def p_maybe_capture_error(self, text: str) -> None:
"""If a stderr/stdout line matches a known build-error pattern,
record it for the next agent-tool drain. Tests every line ,
cheap (single regex search) and only the matching ones land in
the buffer."""
if _ERROR_PATTERNS.search(text):
if ERROR_PATTERNS.search(text):
self.recent_errors.append(text.rstrip())
async def _pipe_stream(self, stream: Optional[asyncio.StreamReader], name: str) -> None:
async def p_pipe_stream(self, stream: Optional[asyncio.StreamReader], name: str) -> None:
if stream is None:
return
try:
@@ -477,21 +477,21 @@ class AppRuntime:
break
text = raw.decode(errors="replace").rstrip("\r\n")
if text:
self._broadcast(LogLine(name, text))
self.p_broadcast(LogLine(name, text))
if name == "stderr" or name == "stdout":
self._maybe_capture_error(text)
self.p_maybe_capture_error(text)
except Exception:
logger.exception("log pipe error (%s) for %s", name, self.workspace_id)
async def _await_exit(self) -> None:
async def p_await_exit(self) -> None:
if not self.process:
return
rc = await self.process.wait()
# Unclean death (vite crash, OOM, orphaned parent) must drop readiness;
# otherwise frontend_url keeps advertising a dead port and the preview
# navigates into ERR_FAILED. stop() already does this for clean stops.
self._frontend_ready = False
self._broadcast(LogLine("runtime", f"[runtime] backend exited with code {rc}"))
self.p_frontend_ready = False
self.p_broadcast(LogLine("runtime", f"[runtime] backend exited with code {rc}"))
class AppRuntimeManager:
@@ -502,30 +502,32 @@ class AppRuntimeManager:
spawns; final detach moves the runtime into an LRU idle pool
instead of stopping it immediately; so re-clicking a recent App
is instant. The oldest runtime gets reaped once the pool exceeds
_MAX_IDLE_RUNTIMES."""
MAX_IDLE_RUNTIMES."""
def __init__(self) -> None:
# NOTE: None of the attributes ever got modified externally, so they're all private.
# workspace_id → AppRuntime, currently has >=1 subscriber.
self.runtimes: dict[str, AppRuntime] = {}
self._attached: dict[str, int] = {}
self.p_attached: dict[str, int] = {}
# workspace_id → AppRuntime with no subscribers but still
# alive. OrderedDict gives O(1) move_to_end + popitem(last=False)
# for LRU semantics.
self._idle_lru: "OrderedDict[str, AppRuntime]" = OrderedDict()
self._lock = asyncio.Lock()
self.p_idle_lru: "OrderedDict[str, AppRuntime]" = OrderedDict()
self.p_lock = asyncio.Lock()
# Public - used by outputs.py
async def attach(self, workspace_id: str, workspace_path: str) -> AppRuntime:
revived = False
# Defined here so every code path below leaves it bound; the
# revive-idle branch used to skip the assignment, leaving the
# post-lock `if dead is not None:` check throwing UnboundLocalError.
dead: Optional[AppRuntime] = None
async with self._lock:
async with self.p_lock:
rt = self.runtimes.get(workspace_id)
if rt is None:
# Maybe the runtime is sitting idle in the LRU; revive
# it without paying the spawn cost again.
idle_rt = self._idle_lru.pop(workspace_id, None)
idle_rt = self.p_idle_lru.pop(workspace_id, None)
if idle_rt is not None and idle_rt.running:
rt = idle_rt
rt.workspace_path = workspace_path
@@ -533,8 +535,8 @@ class AppRuntimeManager:
revived = True
# SIGCONT the process tree if A2 had it paused while
# idle. Pair with the SIGSTOP in detach() below.
_resume_process_tree(rt.process)
rt._suspended = False
resume_process_tree(rt.process)
rt.p_suspended = False
else:
if idle_rt is not None:
# Stale idle entry; process died while idling.
@@ -549,7 +551,7 @@ class AppRuntimeManager:
# folder), trust the latest caller; they have the
# current truth.
rt.workspace_path = workspace_path
self._attached[workspace_id] = self._attached.get(workspace_id, 0) + 1
self.p_attached[workspace_id] = self.p_attached.get(workspace_id, 0) + 1
if not revived and not rt.running:
await rt.start()
# Stop any dead idle runtime outside the lock to avoid blocking.
@@ -560,15 +562,16 @@ class AppRuntimeManager:
logger.exception("failed to reap dead idle runtime %s", workspace_id)
return rt
# Public - used by outputs.py
async def detach(self, workspace_id: str) -> None:
to_idle: Optional[AppRuntime] = None
to_reap: list[AppRuntime] = []
async with self._lock:
count = self._attached.get(workspace_id, 0) - 1
async with self.p_lock:
count = self.p_attached.get(workspace_id, 0) - 1
if count > 0:
self._attached[workspace_id] = count
self.p_attached[workspace_id] = count
return
self._attached.pop(workspace_id, None)
self.p_attached.pop(workspace_id, None)
rt = self.runtimes.pop(workspace_id, None)
if rt is None:
return
@@ -579,21 +582,21 @@ class AppRuntimeManager:
if not rt.running:
to_reap.append(rt)
else:
self._idle_lru[workspace_id] = rt
self._idle_lru.move_to_end(workspace_id)
_suspend_process_tree(rt.process)
rt._suspended = True
while len(self._idle_lru) > _MAX_IDLE_RUNTIMES:
_, old_rt = self._idle_lru.popitem(last=False)
self.p_idle_lru[workspace_id] = rt
self.p_idle_lru.move_to_end(workspace_id)
suspend_process_tree(rt.process)
rt.p_suspended = True
while len(self.p_idle_lru) > MAX_IDLE_RUNTIMES:
_, old_rt = self.p_idle_lru.popitem(last=False)
# Reaping a stopped process: SIGCONT first so the
# SIGTERM in stop() can be delivered cleanly (a
# SIGSTOP'd process can't run its own shutdown).
_resume_process_tree(old_rt.process)
resume_process_tree(old_rt.process)
to_reap.append(old_rt)
to_idle = rt if rt.running else None
# Stop any reaped runtimes OUTSIDE the lock. stop() is async and
# can take up to _TERMINATE_GRACE_SECONDS; holding the lock for
# can take up to TERMINATE_GRACE_SECONDS; holding the lock for
# it would block every other attach/detach.
for old in to_reap:
try:
@@ -601,8 +604,9 @@ class AppRuntimeManager:
except Exception:
logger.exception("failed to reap idle runtime %s", workspace_id)
if to_idle is not None:
logger.debug("workspace %s idled (LRU size now %d)", workspace_id, len(self._idle_lru))
logger.debug("workspace %s idled (LRU size now %d)", workspace_id, len(self.p_idle_lru))
# Public - used by html_inject.py, main.py, outputs.py
def get(self, workspace_id: str) -> Optional[AppRuntime]:
# Active subscribers see the live runtime; idle-pool members
# are also accessible so a status probe between detach and
@@ -610,8 +614,9 @@ class AppRuntimeManager:
rt = self.runtimes.get(workspace_id)
if rt is not None:
return rt
return self._idle_lru.get(workspace_id)
return self.p_idle_lru.get(workspace_id)
# Public - used by agent_manager.py
def drain_errors_for_path(self, file_path: str) -> list[str]:
"""If `file_path` falls under one of the live workspace
runtimes' workspace_path, drain that workspace's recent
@@ -629,7 +634,7 @@ class AppRuntimeManager:
# navigated away from the workspace mid-build, but the agent
# could still be editing files; the LRU keeps the runtime alive
# for ~3 idle slots.
for rt in (*self.runtimes.values(), *self._idle_lru.values()):
for rt in (*self.runtimes.values(), *self.p_idle_lru.values()):
try:
ws_root = os.path.abspath(rt.workspace_path)
except Exception:
@@ -638,8 +643,9 @@ class AppRuntimeManager:
return rt.drain_errors()
return []
# Public - used by outputs.py
async def restart(self, workspace_id: str, workspace_path: Optional[str] = None) -> Optional[AppRuntime]:
rt = self.runtimes.get(workspace_id) or self._idle_lru.get(workspace_id)
rt = self.runtimes.get(workspace_id) or self.p_idle_lru.get(workspace_id)
if rt is None:
return None
if workspace_path:
@@ -647,6 +653,7 @@ class AppRuntimeManager:
await rt.restart()
return rt
# Public - used by outputs.py
async def stop_all(self) -> int:
"""Terminate every active + idle workspace subprocess. Called on
FastAPI lifespan shutdown AND from Electron's pre-quit POST. Without
@@ -657,16 +664,16 @@ class AppRuntimeManager:
so they can run their own shutdown. Parallel via gather; with the
per-runtime 3s SIGTERM grace, worst case is one ~3s wait rather than
N*3s. Idempotent; safe to invoke from multiple shutdown paths."""
async with self._lock:
async with self.p_lock:
victims: list[AppRuntime] = []
for rt in list(self.runtimes.values()):
victims.append(rt)
for rt in list(self._idle_lru.values()):
_resume_process_tree(rt.process)
for rt in list(self.p_idle_lru.values()):
resume_process_tree(rt.process)
victims.append(rt)
self.runtimes.clear()
self._idle_lru.clear()
self._attached.clear()
self.p_idle_lru.clear()
self.p_attached.clear()
if not victims:
return 0
await asyncio.gather(
@@ -676,4 +683,4 @@ class AppRuntimeManager:
return len(victims)
manager = AppRuntimeManager()
RUNTIME_MANAGER = AppRuntimeManager()
backend/apps/outputs/runtime_proc.py
+17 -17
View File
@@ -14,24 +14,24 @@ from typing import Optional
logger = logging.getLogger(__name__)
# SIGTERM grace; well-behaved servers shut down under a second so 3s is enough.
_TERMINATE_GRACE_SECONDS = 3
TERMINATE_GRACE_SECONDS = 3
# 180s covers npm install (60-90s on typical hardware) plus the Vite bind.
_FRONTEND_BIND_TIMEOUT_SECONDS = 180
FRONTEND_BIND_TIMEOUT_SECONDS = 180
# 80ms probe: dropping from 500ms was pure user-visible preview latency win; cheap on localhost.
_FRONTEND_BIND_POLL_INTERVAL = 0.08
FRONTEND_BIND_POLL_INTERVAL = 0.08
# 2000 lines per runtime; lets a Terminal tab opened mid-session replay context. ~few hundred KB at worst.
_LOG_BUFFER_LINES = 2000
LOG_BUFFER_LINES = 2000
# Idle runtimes kept in LRU; trades memory for instant switch-back, beyond 1 because typical users ping-pong 2-3 apps.
_MAX_IDLE_RUNTIMES = 3
MAX_IDLE_RUNTIMES = 3
# Cap on recent error lines the agent gets; 50 is enough for babel error + stack + a few warnings.
_RECENT_ERRORS_MAX = 50
RECENT_ERRORS_MAX = 50
# Narrow regex for build errors (vite, babel, tsc, uvicorn); keeps routine logs out of agent context.
_ERROR_PATTERNS = re.compile(
ERROR_PATTERNS = re.compile(
r"(?:"
r"\[plugin:[^\]]+\]|" # vite plugin errors
r"SyntaxError|" # node / babel
@@ -49,7 +49,7 @@ _ERROR_PATTERNS = re.compile(
)
def _suspend_process_tree(proc) -> None:
def suspend_process_tree(proc) -> None:
"""Send SIGSTOP to a workspace's subprocess so it consumes 0% CPU
while sitting in the LRU idle pool. The signal is delivered to the
PROCESS GROUP (negative PID) when the child is a session leader,
@@ -71,7 +71,7 @@ def _suspend_process_tree(proc) -> None:
pass
def _resume_process_tree(proc) -> None:
def resume_process_tree(proc) -> None:
"""SIGCONT a previously-suspended workspace process. Pair with
_suspend_process_tree. Microsecond cost; idempotent if the process
was never paused."""
@@ -85,7 +85,7 @@ def _resume_process_tree(proc) -> None:
pass
def _background_priority_kwargs() -> dict:
def background_priority_kwargs() -> dict:
"""Return the kwargs that lower the spawned subprocess's OS priority
to a "background" level. On POSIX this is `preexec_fn=os.nice(10)`,
which sets the child's nice to +10 BEFORE exec (so the renice covers
@@ -112,7 +112,7 @@ def _background_priority_kwargs() -> dict:
return {"preexec_fn": lambda: os.nice(10)}
def _find_free_port() -> int:
def find_free_port() -> int:
"""Ask the kernel for an unused localhost port. There's a tiny race
between this socket closing and the backend re-binding, but we hand
each port to exactly one runtime so no caller competes for it, and
@@ -122,7 +122,7 @@ def _find_free_port() -> int:
return s.getsockname()[1]
def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
def kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
"""Recursively signal every descendant of `pid`, leaves-first. The
webapp template's run.sh installs `trap cleanup EXIT` (no TERM), so a
plain SIGTERM to the bash wrapper exits bash silently and leaves
@@ -154,7 +154,7 @@ def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
except Exception:
children = []
for child in children:
_kill_descendant_tree(child, sig_name)
kill_descendant_tree(child, sig_name)
sig = getattr(signal, f"SIG{sig_name}", signal.SIGTERM)
for child in children:
try:
@@ -163,7 +163,7 @@ def _kill_descendant_tree(pid: int, sig_name: str = "TERM") -> None:
pass
def _is_port_free(port: int) -> bool:
def is_port_free(port: int) -> bool:
"""True if nothing currently holds a TCP listener on 127.0.0.1:port.
Cheap kernel-probe; resolves on bind success. Used as the cross-session
safety net: if a prior OpenSwarm run left a ghost subprocess holding
@@ -177,7 +177,7 @@ def _is_port_free(port: int) -> bool:
return False
def _write_env_value(env_path: str, key: str, value: str) -> None:
def write_env_value(env_path: str, key: str, value: str) -> None:
"""Update KEY=VALUE in an existing `.env`, preserving every other
line. Creates the file if missing. Used when a persisted port collides
with a ghost from a prior session and we have to reallocate before
@@ -210,7 +210,7 @@ def _write_env_value(env_path: str, key: str, value: str) -> None:
logger.exception("failed writing %s=%s to %s", key, value, env_path)
def _is_new_mode(workspace_path: str) -> bool:
def is_new_mode(workspace_path: str) -> bool:
"""A workspace is "new-mode" (webapp-template scaffold) if it has a
`run.sh` at its root. Old-mode workspaces are flat `index.html`-only
apps that pre-date the template swap; they're served by OpenSwarm's
@@ -222,7 +222,7 @@ def _is_new_mode(workspace_path: str) -> bool:
return os.path.isfile(os.path.join(workspace_path, "run.sh"))
def _read_env_value(env_path: str, key: str) -> Optional[str]:
def read_env_value(env_path: str, key: str) -> Optional[str]:
"""Parse one value out of a workspace's `.env` without the cost of a
full subprocess-source. Strips quotes + trailing comments. Returns
None if the file or key is missing."""
backend/apps/outputs/view_builder_templates.py
+54 -60
View File
@@ -13,7 +13,7 @@ import threading
logger = logging.getLogger(__name__)
def _resolve_npm() -> list[str] | None:
def p_resolve_npm() -> list[str] | None:
"""Resolve an invokable npm command. Windows ships npm as npm.cmd (a
batch shim), which Python's subprocess won't find via a bare "npm";
and the packaged Electron build bundles only node.exe (no npm) but
@@ -42,7 +42,7 @@ def _resolve_npm() -> list[str] | None:
return None
def _resolve_python() -> str:
def p_resolve_python() -> str:
"""The interpreter to build warm/workspace venvs with. sys.executable
is the running backend's python (bundled standalone in the packaged
build, system python in dev) and is always valid, sidestepping the
@@ -69,8 +69,8 @@ WEBAPP_TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "webapp_template")
# Bundled default; used as the read-once fallback if the user-editable
# copy at ~/.claude/skills/app_builder_skill.md has been removed despite
# the built-in flag (defensive; shouldn't happen in normal use).
with open(APP_BUILDER_SKILL_SOURCE_PATH, encoding="utf-8") as _f:
APP_BUILDER_SKILL_DEFAULT = _f.read()
with open(APP_BUILDER_SKILL_SOURCE_PATH, encoding="utf-8") as f:
APP_BUILDER_SKILL_DEFAULT = f.read()
def load_app_builder_skill() -> str:
@@ -88,13 +88,7 @@ def load_app_builder_skill() -> str:
pass
return APP_BUILDER_SKILL_DEFAULT
# Backward-compat alias. Older callers import VIEW_BUILDER_SKILL directly ,
# point them at the same content as the user-editable version so a "frozen
# at import" stale copy can't drift from what the skills page shows.
VIEW_BUILDER_SKILL = APP_BUILDER_SKILL_DEFAULT
VIEW_TEMPLATE_INDEX = """\
P_VIEW_TEMPLATE_INDEX = """\
<!DOCTYPE html>
<html lang="en">
<head>
@@ -139,7 +133,7 @@ VIEW_TEMPLATE_INDEX = """\
</html>
"""
VIEW_TEMPLATE_SCHEMA = """\
P_VIEW_TEMPLATE_SCHEMA = """\
{
"type": "object",
"properties": {},
@@ -147,7 +141,7 @@ VIEW_TEMPLATE_SCHEMA = """\
}
"""
VIEW_TEMPLATE_META = """\
P_VIEW_TEMPLATE_META = """\
{
"name": "",
"description": ""
@@ -155,9 +149,9 @@ VIEW_TEMPLATE_META = """\
"""
VIEW_TEMPLATE_FILES = {
"index.html": VIEW_TEMPLATE_INDEX,
"schema.json": VIEW_TEMPLATE_SCHEMA,
"meta.json": VIEW_TEMPLATE_META,
"index.html": P_VIEW_TEMPLATE_INDEX,
"schema.json": P_VIEW_TEMPLATE_SCHEMA,
"meta.json": P_VIEW_TEMPLATE_META,
}
@@ -165,7 +159,7 @@ VIEW_TEMPLATE_FILES = {
# webapp_template (new-mode) seed helpers
# ---------------------------------------------------------------------------
def _ignore_backend(src: str, names: list[str]) -> list[str]:
def p_ignore_backend(src: str, names: list[str]) -> list[str]:
"""copytree filter; when copying the template root, drop only the
top-level `backend/` directory. Subdirectories named `backend` deeper
in the tree (none today, but defensively scoped) are unaffected."""
@@ -174,10 +168,10 @@ def _ignore_backend(src: str, names: list[str]) -> list[str]:
return []
_DEBUGGER_PATH = os.path.abspath(
DEBUGGER_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "..", "..", "debugger")
)
_TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "backend"))
TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "backend"))
# ---------------------------------------------------------------------------
@@ -191,8 +185,8 @@ _TEMPLATE_BACKEND_PATH = os.path.abspath(os.path.join(WEBAPP_TEMPLATE_DIR, "back
# until the user clears ~/.openswarm/cache.
# ---------------------------------------------------------------------------
_warm_cache_lock = threading.Lock()
_warm_cache_thread: threading.Thread | None = None
P_WARM_CACHE_LOCK = threading.Lock()
P_WARM_CACHE_THREAD: threading.Thread | None = None
# Pre-built node_modules archive bundled with packaged releases. Generated
@@ -202,25 +196,25 @@ _warm_cache_thread: threading.Thread | None = None
# ~3 s vs ~22 s for the live install. Stale archives (package.json bumped
# but archive not rebuilt) are silently ignored, so the live-install
# fallback always wins on correctness.
_BUNDLED_ARCHIVE_DIR = os.path.join(
P_BUNDLED_ARCHIVE_DIR = os.path.join(
os.path.dirname(__file__), "webapp_template_cache"
)
def _bundled_archive_path_for(digest: str) -> str:
def p_bundled_archive_path_for(digest: str) -> str:
"""Sha-tagged archive path so a stale archive from a prior template
version is automatically skipped instead of overwriting the cache with
out-of-date modules."""
return os.path.join(_BUNDLED_ARCHIVE_DIR, f"node_modules.{digest}.tar.gz")
return os.path.join(P_BUNDLED_ARCHIVE_DIR, f"node_modules.{digest}.tar.gz")
def _try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
def p_try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
"""Unpack the sha-tagged bundled archive into `cache_dir` if one
exists for the current template digest. Returns True on success,
False to signal the caller should fall back to a live `npm install`.
The archive is built from the same package.json + package-lock.json
sha so the extracted tree is byte-equivalent to `npm ci`."""
archive_path = _bundled_archive_path_for(digest)
archive_path = p_bundled_archive_path_for(digest)
if not os.path.exists(archive_path):
return False
try:
@@ -252,7 +246,7 @@ def _try_extract_bundled_archive(cache_dir: str, digest: str) -> bool:
return False
def _warm_cache_digest() -> str:
def p_warm_cache_digest() -> str:
"""Sha of the template's frontend/package.json; used as the cache
key + the bundled-archive filename suffix so a package.json bump
invalidates both at once."""
@@ -264,34 +258,34 @@ def _warm_cache_digest() -> str:
return "fallback"
def _warm_cache_dir() -> str:
def p_warm_cache_dir() -> str:
"""Path the warm node_modules lives under. Hashed by package.json so
upgrades automatically force a re-populate."""
base = os.environ.get("OPENSWARM_WEBAPP_CACHE_DIR") or os.path.expanduser(
"~/.openswarm/cache/webapp_template_node_modules"
)
return os.path.join(base, _warm_cache_digest())
return os.path.join(base, p_warm_cache_digest())
def _ensure_warm_cache() -> str | None:
def p_ensure_warm_cache() -> str | None:
"""Populate the warm-cache node_modules if missing. Returns the
absolute path to the populated `node_modules` directory, or None on
failure. Thread-safe; concurrent callers block on a single install
instead of racing. Idempotent and fast after the first call."""
cache_dir = _warm_cache_dir()
cache_dir = p_warm_cache_dir()
cache_modules = os.path.join(cache_dir, "node_modules")
if os.path.isdir(cache_modules):
return cache_modules
with _warm_cache_lock:
with P_WARM_CACHE_LOCK:
if os.path.isdir(cache_modules):
return cache_modules
# Fast path: pre-built archive shipped inside the release. The
# build script generates this so users hitting OpenSwarm for the
# first time skip the ~22 s live `npm install`. Falls through on
# any failure so dev installs (no archive) keep working.
if _try_extract_bundled_archive(cache_dir, _warm_cache_digest()):
if p_try_extract_bundled_archive(cache_dir, p_warm_cache_digest()):
logger.info("webapp-template: warm cache ready from bundled archive")
return cache_modules
try:
@@ -304,7 +298,7 @@ def _ensure_warm_cache() -> str | None:
tmpl_lock = os.path.join(WEBAPP_TEMPLATE_DIR, "frontend", "package-lock.json")
shutil.copyfile(tmpl_pkg, os.path.join(cache_dir, "package.json"))
base_flags = ["--prefer-offline", "--no-audit", "--no-fund", "--loglevel=error"]
npm = _resolve_npm()
npm = p_resolve_npm()
if npm is None:
logger.info("webapp-template: no npm available; skipping warm cache (workspace will install on first run)")
return None
@@ -344,11 +338,11 @@ def _ensure_warm_cache() -> str | None:
return None
def _link_node_modules(workspace_dir: str) -> None:
def p_link_node_modules(workspace_dir: str) -> None:
"""After copytree, point the workspace's frontend/node_modules at
the warm-cache directory. Safe fallback; if the cache isn't ready,
the workspace's run.sh will fall through to its own install path."""
cache_modules = _ensure_warm_cache()
cache_modules = p_ensure_warm_cache()
if not cache_modules:
return
target = os.path.join(workspace_dir, "frontend", "node_modules")
@@ -392,10 +386,10 @@ def _link_node_modules(workspace_dir: str) -> None:
# otherwise pays per workspace.
# ---------------------------------------------------------------------------
_warm_venv_lock = threading.Lock()
P_WARM_VENV_LOCK = threading.Lock()
def _warm_venv_dir() -> str:
def p_warm_venv_dir() -> str:
"""Cache root for the shared backend venv, keyed by a sha of the
template backend's pyproject.toml so a dep bump auto-invalidates."""
pyproject = os.path.join(WEBAPP_TEMPLATE_DIR, "backend", "pyproject.toml")
@@ -410,18 +404,18 @@ def _warm_venv_dir() -> str:
return os.path.join(base, digest)
def _ensure_warm_python_venv() -> str | None:
def p_ensure_warm_python_venv() -> str | None:
"""Populate the warm-cache backend venv if missing. Returns the
absolute path to the populated `.venv` directory, or None on
failure. Thread-safe and idempotent; fast return after first call."""
cache_dir = _warm_venv_dir()
cache_dir = p_warm_venv_dir()
venv_dir = os.path.join(cache_dir, ".venv")
sentinel = os.path.join(cache_dir, ".populated")
if os.path.isfile(sentinel) and os.path.isdir(venv_dir):
return venv_dir
with _warm_venv_lock:
with P_WARM_VENV_LOCK:
if os.path.isfile(sentinel) and os.path.isdir(venv_dir):
return venv_dir
try:
@@ -433,7 +427,7 @@ def _ensure_warm_python_venv() -> str | None:
# `python.exe`. On macOS/Linux the versioned candidates
# match first so we don't accidentally pick a system
# Python 2.x via the bare name.
py = _resolve_python()
py = p_resolve_python()
# Wipe any half-populated venv from a previous crashed run.
if os.path.isdir(venv_dir):
@@ -481,28 +475,28 @@ def warm_cache_in_background() -> None:
node_modules cache and the backend-venv cache so the user's FIRST
webapp-template seed doesn't pay the install costs. No-op (fast
return) if both caches are already there or a thread is in flight."""
global _warm_cache_thread
if _warm_cache_thread is not None and _warm_cache_thread.is_alive():
global P_WARM_CACHE_THREAD
if P_WARM_CACHE_THREAD is not None and P_WARM_CACHE_THREAD.is_alive():
return
node_done = os.path.isdir(os.path.join(_warm_cache_dir(), "node_modules"))
venv_done = os.path.isfile(os.path.join(_warm_venv_dir(), ".populated"))
node_done = os.path.isdir(os.path.join(p_warm_cache_dir(), "node_modules"))
venv_done = os.path.isfile(os.path.join(p_warm_venv_dir(), ".populated"))
if node_done and venv_done:
return
def _runner() -> None:
def runner() -> None:
try:
_ensure_warm_cache()
p_ensure_warm_cache()
except Exception:
logger.exception("background warm node_modules crashed")
try:
_ensure_warm_python_venv()
p_ensure_warm_python_venv()
except Exception:
logger.exception("background warm python venv crashed")
_warm_cache_thread = threading.Thread(
target=_runner, daemon=True, name="webapp-template-warm-cache"
P_WARM_CACHE_THREAD = threading.Thread(
target=runner, daemon=True, name="webapp-template-warm-cache"
)
_warm_cache_thread.start()
P_WARM_CACHE_THREAD.start()
# Trigger pre-warm on module import; backend startup hits this and the
@@ -512,7 +506,7 @@ def warm_cache_in_background() -> None:
warm_cache_in_background()
def _patch_env_port(env_path: str, key: str, value: str) -> None:
def p_patch_env_port(env_path: str, key: str, value: str) -> None:
"""Idempotent in-place rewrite: `KEY=...` → `KEY=value`. Appends if
the key isn't present. Preserves surrounding lines untouched."""
if not os.path.exists(env_path):
@@ -557,12 +551,12 @@ def seed_webapp_template_workspace(workspace_dir: str, frontend_port: int) -> No
shutil.copytree(
WEBAPP_TEMPLATE_DIR,
workspace_dir,
ignore=_ignore_backend,
ignore=p_ignore_backend,
dirs_exist_ok=True,
)
# Symlink the workspace's frontend/node_modules at the warm cache so
# `npm install` can be skipped entirely by the workspace run.sh.
_link_node_modules(workspace_dir)
p_link_node_modules(workspace_dir)
env_path = os.path.join(workspace_dir, ".env")
env_example_path = os.path.join(workspace_dir, ".env.example")
src_example = os.path.join(WEBAPP_TEMPLATE_DIR, ".env.example")
@@ -579,17 +573,17 @@ def seed_webapp_template_workspace(workspace_dir: str, frontend_port: int) -> No
with open(env_path, "w", encoding="utf-8") as f:
f.write("BACKEND_PORT=NONE\nFRONTEND_PORT=4949\n")
_patch_env_port(env_path, "FRONTEND_PORT", str(frontend_port))
_patch_env_port(env_example_path, "FRONTEND_PORT", str(frontend_port))
p_patch_env_port(env_path, "FRONTEND_PORT", str(frontend_port))
p_patch_env_port(env_example_path, "FRONTEND_PORT", str(frontend_port))
# Install-specific paths; .env only.
_patch_env_port(env_path, "OPENSWARM_TEMPLATE_BACKEND_PATH", _TEMPLATE_BACKEND_PATH)
_patch_env_port(env_path, "OPENSWARM_DEBUGGER_PATH", _DEBUGGER_PATH)
p_patch_env_port(env_path, "OPENSWARM_TEMPLATE_BACKEND_PATH", TEMPLATE_BACKEND_PATH)
p_patch_env_port(env_path, "OPENSWARM_DEBUGGER_PATH", DEBUGGER_PATH)
# Backend-venv warm-cache path; backend_init.sh checks this for a
# pre-populated `.venv/` to cp -aR into the workspace instead of
# paying the ~25s venv-create + pip-install cost. Written even if
# the cache isn't ready yet; backend_init.sh re-checks at run time.
_patch_env_port(env_path, "OPENSWARM_BACKEND_VENV_CACHE", _warm_venv_dir())
p_patch_env_port(env_path, "OPENSWARM_BACKEND_VENV_CACHE", p_warm_venv_dir())
# Make the shipped scripts executable. tarball/git extracts may strip
# the +x bit depending on how the snapshot was vendored.
backend/apps/outputs/workspace_io.py
+9 -9
View File
@@ -13,7 +13,7 @@ from backend.config.json_store import read_json_or_none, atomic_write_json
logger = logging.getLogger(__name__)
def _load_all() -> list[Output]:
def load_all() -> list[Output]:
result = []
if not os.path.exists(DATA_DIR):
return result
@@ -29,11 +29,11 @@ def _load_all() -> list[Output]:
return result
def _save(output: Output):
def save(output: Output):
atomic_write_json(os.path.join(DATA_DIR, f"{output.id}.json"), output.model_dump())
def _load(output_id: str) -> Output:
def load(output_id: str) -> Output:
data = read_json_or_none(os.path.join(DATA_DIR, f"{output_id}.json"))
if data is None:
raise HTTPException(status_code=404, detail="Output not found")
@@ -54,7 +54,7 @@ def load_output(output_id: str) -> Output | None:
# agent is active. Result: backend CPU pegged on JSON-serializing
# auto-generated chunks the frontend will then throw away. The frontend
# already filters these for display; this skip is the real fix.
_WALK_SKIP_DIRS = frozenset({
P_WALK_SKIP_DIRS = frozenset({
"node_modules",
".vite",
".vite-cache",
@@ -75,10 +75,10 @@ _WALK_SKIP_DIRS = frozenset({
# they're not what the user/agent is editing. Anything over the cap
# returns a truncated stub the frontend treats as "open the file
# directly to see full contents."
_WALK_MAX_FILE_BYTES = 256 * 1024
P_WALK_MAX_FILE_BYTES = 256 * 1024
def _walk_directory(folder: str) -> dict[str, str]:
def walk_directory(folder: str) -> dict[str, str]:
"""Walk a directory tree and return {relative_path: content} for all
text files the user is actually authoring. Skips build/install
directories AND truncates oversize files; both critical for the
@@ -92,7 +92,7 @@ def _walk_directory(folder: str) -> dict[str, str]:
# Doing it here means we never even stat the children, so a
# 10k-file `.venv/` costs ~one stat (on the dir itself) instead
# of 10k.
dirs[:] = [d for d in dirs if d not in _WALK_SKIP_DIRS]
dirs[:] = [d for d in dirs if d not in P_WALK_SKIP_DIRS]
for fname in filenames:
full_path = os.path.join(root, fname)
# Normalize to forward-slash keys so the frontend's
@@ -106,10 +106,10 @@ def _walk_directory(folder: str) -> dict[str, str]:
# Stat first; cheap, lets us skip giant files without
# opening + reading them.
size = os.path.getsize(full_path)
if size > _WALK_MAX_FILE_BYTES:
if size > P_WALK_MAX_FILE_BYTES:
files[rel_path] = (
f"// [openswarm] file truncated ({size} bytes > "
f"{_WALK_MAX_FILE_BYTES} byte cap). Open directly "
f"{P_WALK_MAX_FILE_BYTES} byte cap). Open directly "
f"to view full contents."
)
continue
backend/apps/service/buffer.py
+19 -21
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 p_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:
@@ -54,11 +54,12 @@ def _conn(spool_path: str) -> Iterator[sqlite3.Connection]:
c.close()
# Public - Used by client.py
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, p_conn(spool_path) as c:
c.execute(
"INSERT INTO spool (kind, payload, created_at) VALUES (?, ?, ?)",
(kind, body, now),
@@ -68,8 +69,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,23 +87,24 @@ 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
# Public - Used by client.py
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:
with P_LOCK, p_conn(spool_path) as c:
rows = c.execute(
"SELECT id, kind, payload FROM spool ORDER BY id ASC LIMIT ?",
(batch_size,),
@@ -113,30 +115,26 @@ 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, p_conn(spool_path) as c:
c.execute("DELETE FROM spool WHERE id = ?", (rid,))
logger.warning("Dropped corrupt spool row id=%s", rid)
return out
# Public - Used by client.py
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, p_conn(spool_path) as c:
c.executemany("DELETE FROM spool WHERE id = ?", [(i,) for i in ids])
# Public - Used by service.py
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, p_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")
return int(row[0]) if row else 0
backend/apps/service/client.py
+80 -130
View File
@@ -1,6 +1,6 @@
"""Operational state forwarder.
Single public surface: `submit(kind, payload)`. The desktop hands off
Single public surface: `p_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.
@@ -34,26 +34,27 @@ 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()
P_TEST_SINK: Optional[Any] = None
P_INSTALL_ID: Optional[str] = None
P_USER_ID: Optional[str] = None
P_INFLIGHT: int = 0
P_INFLIGHT_LOCK = asyncio.Lock()
P_DRAIN_LOCK = asyncio.Lock()
def _spool_path() -> str:
# Public - Used by service.py
def spool_path() -> str:
try:
from backend.config.paths import SETTINGS_DIR
return os.path.join(SETTINGS_DIR, "service_spool.db")
@@ -61,34 +62,28 @@ def _spool_path() -> str:
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
def p_get_install_id() -> str:
global P_INSTALL_ID
if P_INSTALL_ID:
return P_INSTALL_ID
try:
from backend.apps.settings.store import load_settings, _save_settings
from backend.apps.settings.store 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
save_settings(s)
P_INSTALL_ID = iid
except Exception:
_install_id = uuid4().hex
return _install_id
P_INSTALL_ID = uuid4().hex
return P_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()
@@ -107,12 +102,8 @@ def _get_user_id() -> Optional[str]:
return None
def set_user_id(uid: Optional[str]) -> None:
global _user_id
_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 +121,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:
@@ -156,8 +147,8 @@ def _envelope() -> dict:
except Exception:
pass
if not ianatz:
import datetime as _dt
local_tz = _dt.datetime.now().astimezone().tzinfo
import datetime as dt
local_tz = dt.datetime.now().astimezone().tzinfo
if local_tz:
ianatz = str(local_tz)
if ianatz:
@@ -182,20 +173,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 +194,43 @@ 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 P_TEST_SINK is not None:
try:
_test_sink(kind, body)
P_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)
# Public - Used by service.py
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 +239,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,15 +256,16 @@ async def drain_spool(batch_size: int = 50) -> int:
# Public API
# --------------------------------------------------------------------------
def _log(kind: str) -> None:
def p_log(kind: str) -> None:
"""Append to the rolling operational log for diagnostics."""
try:
from backend.apps.service.ring_buffer import record
record(kind)
from backend.apps.service.ring_buffer import record as ring_record
ring_record(kind)
except Exception:
pass
# Public - Used by agents.py, settings.py, cloud_sync.py, subscription.router.py, service.py
def sync(data: dict | None = None) -> None:
"""Sync operational state to the cloud. Single entry point.
@@ -288,29 +281,29 @@ 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")
if _test_sink is not None:
p_log("s")
if P_TEST_SINK is not None:
try:
_test_sink("s", body)
P_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(payload: dict) -> None:
def p_submit(payload: dict) -> None:
"""Routes through sync(). The cloud demuxes by payload shape (state /
sync / diagnostic / event), so kind here is informational; the routing
happens server-side in openswarm-cloud/src/routes/service/ingest.ts.
@@ -318,7 +311,7 @@ def submit(payload: dict) -> None:
sync(payload)
def _schedule(coro) -> None:
def p_schedule(coro) -> None:
try:
loop = asyncio.get_running_loop()
except RuntimeError:
@@ -328,75 +321,32 @@ def _schedule(coro) -> None:
return
import threading
def _run():
def run_coro():
try:
asyncio.run(coro)
except Exception:
pass
threading.Thread(target=_run, daemon=True).start()
threading.Thread(target=run_coro, daemon=True).start()
# --------------------------------------------------------------------------
# Backwards-compat shims for legacy call sites. New code calls submit()
# Backwards-compat shims for legacy call sites. New code calls p_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,
) -> 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_session_close(session_dump: dict, activity: Optional[dict] = None) -> None:
submit("session", {"usage_window": session_dump, "activity": activity or {}})
# Public - Used by agent_manager.py
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,
)
p_submit("diagnostic", {"diagnostic": diagnostic})
# Public - Used by settings.py, auth.router.py, subscription.router.py
def identify(extra_properties: Optional[dict] = None) -> None:
update_identity(extra_properties or {})
p_submit("state", {"identity": extra_properties or {}})
backend/apps/service/ring_buffer.py
+7 -12
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,5 @@ 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)
def clear() -> None:
with _lock:
_buffer.clear()
with P_LOCK:
return list(P_BUFFER)
backend/apps/service/service.py
+55 -61
View File
@@ -26,22 +26,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.client import sync, drain_spool, spool_path
from backend.apps.service.version import 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
_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:
@@ -51,25 +51,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)
import datetime as dt
P_PULSE_HOURS.add(dt.datetime.now().hour)
except Exception:
pass
@@ -80,40 +80,34 @@ async def _pulse_loop():
stats = await get_usage_stats()
if stats:
cur_cost = stats.get("totalCost", 0) or 0
cur_prompt = stats.get("totalPromptTokens", 0) or 0
cur_completion = stats.get("totalCompletionTokens", 0) or 0
cur_requests = stats.get("totalRequests", 0) or 0
cost_delta, _last_9r_cost = _compute_delta(cur_cost, _last_9r_cost)
prompt_delta, _last_9r_prompt_tokens = _compute_delta(cur_prompt, _last_9r_prompt_tokens, threshold=1000)
completion_delta, _last_9r_completion_tokens = _compute_delta(cur_completion, _last_9r_completion_tokens, threshold=1000)
requests_delta, _last_9r_requests = _compute_delta(cur_requests, _last_9r_requests, threshold=10)
_pulse_delta_cost_total += cost_delta
cost_delta, P_LAST_9R_COST = p_compute_delta(cur_cost, P_LAST_9R_COST)
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({
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()
await drain_spool()
except Exception:
pass
await asyncio.sleep(60)
@@ -121,7 +115,7 @@ async def _drain_loop():
@asynccontextmanager
async def service_lifespan():
global _pulse_task, _drain_task
global P_PULSE_TASK, P_DRAIN_TASK
try:
from backend.apps.settings.settings import load_settings
@@ -153,7 +147,7 @@ async def service_lifespan():
for cp in getattr(settings, "custom_providers", []):
providers.append(cp.name)
svc.sync({
sync({
"os": platform.system(),
"platform": platform.platform(),
"provider_count": len(providers),
@@ -188,7 +182,7 @@ async def service_lifespan():
if is_paying and getattr(settings, "openswarm_subscription_expires", None):
id_props["subscription_expires"] = settings.openswarm_subscription_expires
svc.sync({"identity": id_props})
sync({"identity": id_props})
except Exception as e:
logger.debug(f"Service startup event failed (non-critical): {e}")
@@ -198,26 +192,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
try:
from backend.apps.nine_router.process import stop
@@ -235,7 +229,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
@@ -253,11 +247,11 @@ 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"))
def _is_real(sess: dict) -> bool:
def is_real(sess: dict) -> bool:
# "Real" = actually ran. Empty draft/abandoned sessions (no assistant turn, no tokens,
# no active time) otherwise inflate the count and drag every average toward zero.
if (sess.get("agent_active_ms") or 0) > 0 or (sess.get("cost_usd") or 0) > 0:
@@ -267,7 +261,7 @@ async def usage_summary():
return True
return any(m.get("role") == "assistant" for m in sess.get("messages", []))
sessions = [s for s in sessions if _is_real(s)]
sessions = [s for s in sessions if is_real(s)]
total_sessions = len(sessions)
total_cost = sum(s.get("cost_usd", 0) for s in sessions)
@@ -438,25 +432,25 @@ async def post_submit(body=Body(...)):
for item in body:
if isinstance(item, dict):
if any(k in item for k in ("s", "a", "p")):
svc.sync(item)
sync(item)
continue
kind = item.get("kind") or ""
payload = item.get("payload") or {}
if isinstance(payload, dict):
payload.setdefault("kind", kind)
svc.sync(payload)
sync(payload)
return {"ok": True}
if not isinstance(body, dict):
return {"ok": False, "error": "JSON object or array required"}
# Shape 1: frontend `report()`; flat {s, a, p, ...}
if any(k in body for k in ("s", "a", "p")):
svc.sync(body)
sync(body)
return {"ok": True}
# Shape 2: legacy {kind, payload}
kind = body.get("kind") or ""
payload = body.get("payload")
if kind and isinstance(payload, dict):
svc.sync(payload)
sync(payload)
return {"ok": True}
return {"ok": False, "error": "expected {s,a,p,...} or {kind,payload}"}
@@ -474,7 +468,7 @@ async def post_event(body: dict):
if not action:
action = "fired"
svc.sync({
sync({
"s": str(surface)[:64],
"a": str(action)[:64],
"p": body.get("props") or body.get("properties") or {},
@@ -485,4 +479,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(spool_path())}
backend/apps/service/version.py
+7 -8
View File
@@ -7,8 +7,7 @@ path to electron/package.json hops the same three dirnames up to the repo root.
import json
import os
def _read_app_version() -> str:
def p_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
@@ -23,13 +22,13 @@ def _read_app_version() -> str:
# app_version="unknown" pre-fix. Kept for backward compatibility with
# dev runs and as a safety net if the env var is ever unset.
try:
_here = os.path.dirname(os.path.abspath(__file__))
_repo = os.path.dirname(os.path.dirname(os.path.dirname(_here)))
_pkg = os.path.join(_repo, "electron", "package.json")
with open(_pkg, encoding="utf-8") as _f:
return json.load(_f).get("version", "unknown")
here = os.path.dirname(os.path.abspath(__file__))
repo = os.path.dirname(os.path.dirname(os.path.dirname(here)))
pkg = os.path.join(repo, "electron", "package.json")
with open(pkg, encoding="utf-8") as f:
return json.load(f).get("version", "unknown")
except (OSError, ValueError, KeyError):
return "unknown"
APP_VERSION = _read_app_version()
APP_VERSION = p_read_app_version()
backend/apps/settings/credentials.py
+2 -2
View File
@@ -31,7 +31,7 @@ def proxy_auth(settings: AppSettings) -> tuple[str | None, str | None]:
return (None, None)
def _check_9router() -> bool:
def p_check_9router() -> bool:
"""Check if 9Router is running locally."""
try:
import httpx
@@ -56,7 +56,7 @@ def get_anthropic_client(settings: AppSettings) -> anthropic.AsyncAnthropic:
return anthropic.AsyncAnthropic(api_key=settings.anthropic_api_key)
# Fall back to 9Router (free for users with Claude/ChatGPT/Gemini subscriptions).
if _check_9router():
if p_check_9router():
return anthropic.AsyncAnthropic(
api_key="9router",
base_url="http://localhost:20128",
backend/main.py
+4 -4
View File
@@ -288,8 +288,8 @@ async def websocket_runtime_logs(websocket: WebSocket, workspace_id: str):
if not _ws_auth_ok(websocket):
return
await websocket.accept()
from backend.apps.outputs.runtime import manager as runtime_manager
rt = runtime_manager.get(workspace_id)
from backend.apps.outputs.runtime import RUNTIME_MANAGER
rt = RUNTIME_MANAGER.get(workspace_id)
if rt is None:
# No active runtime, surface that to the client and close. The
# frontend will call /runtime/start and reconnect. Also emit a
@@ -298,8 +298,8 @@ async def websocket_runtime_logs(websocket: WebSocket, workspace_id: str):
# webapp_template workspaces instead of falling back to the
# legacy /serve/index.html URL (which 404s in new-mode).
try:
from backend.apps.outputs.outputs import _runtime_status_payload
status = _runtime_status_payload(workspace_id)
from backend.apps.outputs.outputs import runtime_status_payload
status = runtime_status_payload(workspace_id)
await websocket.send_text(json.dumps({
"event": "runtime:status",
"workspace_id": workspace_id,
backend/tests/test_service.py
+1 -1
View File
@@ -244,7 +244,7 @@ def test_buffer_clear(tmp_path):
from backend.apps.service import buffer
spool = str(tmp_path / "s.db")
buffer.enqueue(spool, "s:/x", {}, now=time.time())
buffer.clear(spool)
buffer.p_clear(spool) # p-private-ignore
assert buffer.count(spool) == 0