diff --git a/.github/workflows/backend-tests.yml b/.github/workflows/backend-tests.yml new file mode 100644 index 000000000..39972626f --- /dev/null +++ b/.github/workflows/backend-tests.yml @@ -0,0 +1,71 @@ +name: backend-tests + +# The backend pytest suite, run on every pull request and on pushes to the mainline branches. Until +# now nothing ran it in CI, so a regression only surfaced when someone ran it by hand. +on: + pull_request: + paths: + - 'backend/**' + - '.github/workflows/backend-tests.yml' + push: + branches: [main, dev] + paths: + - 'backend/**' + - '.github/workflows/backend-tests.yml' + workflow_dispatch: + +# Runs checked-out project code on pull_request: the token stays read-only. +permissions: + contents: read + +concurrency: + group: backend-tests-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + pytest: + name: pytest (ubuntu) + runs-on: ubuntu-latest + timeout-minutes: 25 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + cache: pip + cache-dependency-path: | + backend/requirements.lock + backend/requirements-dev.txt + - name: Install backend deps (locked runtime + dev) + run: | + python -m pip install --require-hashes --only-binary=:all: -r backend/requirements.lock + python -m pip install -r backend/requirements-dev.txt + # Two numbers that must agree. A test process that dies mid-run can exit 0 with no summary + # (a hard-exit shutdown path did exactly that once and silently skipped ~42% of the suite), so + # the job also asserts that every collected test was actually run: junit testcase count == + # collect-only count. Green then means green, not "green as far as it got". + - name: Count the suite + run: | + python -m pytest backend/tests --ignore=backend/tests/formal --collect-only -q -p no:cacheprovider \ + | tail -1 | tee collected.txt + # --timeout: a test that blocks forever (a bare ws.receive_json() waiting for an event that never + # comes, say) otherwise stalls the run at 99% until timeout-minutes with no summary and no junit. + # With the cap it fails by name, the rest of the suite runs, and the assertion below still holds. + - name: Run the backend suite + run: | + python -m pytest backend/tests --ignore=backend/tests/formal -q -p no:cacheprovider \ + --timeout=300 \ + --junitxml "${RUNNER_TEMP}/pytest.xml" + - name: Every collected test ran + # Runs after a red suite too, so a failure report also says whether the run was complete. + if: ${{ !cancelled() }} + run: | + python - "${RUNNER_TEMP}/pytest.xml" collected.txt <<'PY' + import re, sys, xml.etree.ElementTree as ET + ran = sum(1 for _ in ET.parse(sys.argv[1]).getroot().iter('testcase')) + m = re.search(r'(\d+) tests? collected', open(sys.argv[2]).read()) + collected = int(m.group(1)) if m else -1 + print(f'collected={collected} ran={ran}') + if collected < 1 or ran != collected: + sys.exit(f'FAIL: {ran} of {collected} collected tests reached the report; the run was truncated') + PY diff --git a/.github/workflows/edge-tests.yml b/.github/workflows/edge-tests.yml new file mode 100644 index 000000000..82ef10d9d --- /dev/null +++ b/.github/workflows/edge-tests.yml @@ -0,0 +1,44 @@ +name: edge-tests + +# The openswarm-edge pytest suite, on every pull request that touches it and on pushes to the +# mainline branches. +on: + pull_request: + paths: + - 'openswarm-edge/**' + - '.github/workflows/edge-tests.yml' + push: + branches: [main, dev] + paths: + - 'openswarm-edge/**' + - '.github/workflows/edge-tests.yml' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: edge-tests-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + pytest: + name: pytest (openswarm-edge) + runs-on: ubuntu-latest + timeout-minutes: 10 + defaults: + run: + working-directory: openswarm-edge + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + cache: pip + cache-dependency-path: openswarm-edge/requirements.txt + - name: Install edge deps + run: | + python -m pip install -r requirements.txt + python -m pip install pytest pytest-asyncio + - name: Run the edge suite + run: python -m pytest tests -q -p no:cacheprovider diff --git a/.github/workflows/frontend-tests.yml b/.github/workflows/frontend-tests.yml new file mode 100644 index 000000000..00578bb8f --- /dev/null +++ b/.github/workflows/frontend-tests.yml @@ -0,0 +1,43 @@ +name: frontend-tests + +# Typecheck plus the renderer's node:test suite, on every pull request and on pushes to the mainline +# branches. Until now neither ran in CI; the tests were run by hand, one file at a time. +on: + pull_request: + paths: + - 'frontend/**' + - '.github/workflows/frontend-tests.yml' + push: + branches: [main, dev] + paths: + - 'frontend/**' + - '.github/workflows/frontend-tests.yml' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: frontend-tests-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + typecheck-and-tests: + name: tsc + node:test + runs-on: ubuntu-latest + timeout-minutes: 15 + defaults: + run: + working-directory: frontend + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: '20.18.1' + cache: npm + cache-dependency-path: frontend/package-lock.json + - run: npm ci + - name: Typecheck + run: npx tsc --noEmit -p tsconfig.json + - name: Unit tests (node:test via tsx) + run: node scripts/run-tests.mjs diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index cd6f5e916..fcf6840e8 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -2,13 +2,10 @@ import logging import time import os -from contextlib import asynccontextmanager -from typing import AsyncIterator, Dict, List, Optional +from typing import Any, Dict, List, Optional from typeguard import typechecked -from backend.apps.agents.core.models import ( - AgentSession, Message, -) +from backend.apps.agents.core.models import Message from backend.apps.agents.core.ws_manager import ws_manager from backend.apps.settings.settings import load_settings from backend.apps.tools_lib.tools_lib import load_builtin_permissions @@ -18,8 +15,8 @@ save_session, load_session_data as load_session_data, ) +from backend.apps.agents.manager.session.SessionStore import SessionStore from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState -from backend.apps.agents.manager.streaming.PartialReply import PartialReply from backend.apps.agents.manager.session.SessionLifecycle import SessionLifecycle from backend.apps.agents.manager.SpawnAgentRun import SpawnAgentRun from backend.apps.agents.manager.session.SessionPersistence import SessionPersistence @@ -30,76 +27,131 @@ from backend.apps.agents.manager.RunSupport import RunSupport from backend.apps.agents.manager.run.handle_run_error import handle_run_error from backend.apps.agents.manager.run.TurnRunner import TurnRunner -from backend.apps.agents.manager.run.client_pool import ClientHandle -from backend.apps.agents.manager.streaming.HookContext import HookContext +from backend.apps.agents.manager.run.TurnAdmission import TurnAdmission from backend.apps.agents.manager.run.RunOptions import RunOptions +from backend.apps.agents.events.AgentEventSink import ( + AgentEventSink, + BoundedAgentEventSink, + NullAgentEventSink, +) +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter logger = logging.getLogger(__name__) os.environ.setdefault("CLAUDE_CODE_STREAM_CLOSE_TIMEOUT", "3600000") -# Cap concurrent ROOT agent turns so firing 30 agents at once doesn't spawn 30 CLIs in the same instant; the overflow queues (agents are model/IO-bound, so they're waiting anyway). Env-tunable, 0/blank disables the gate. -MAX_CONCURRENT_TURNS = int(os.environ.get("OSW_MAX_CONCURRENT_TURNS", "8") or "0") - - -class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport): +class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnAdmission, TurnRunner, RunOptions, RunSupport): @typechecked - def __init__(self): - self.sessions: Dict[str, AgentSession] = {} + def __init__(self, store: Optional[SessionStore] = None, event_sink: Optional[AgentEventSink] = None): + self.store = store or SessionStore() + self.event_sink = event_sink or NullAgentEventSink() + # Messages queued behind an in-flight turn (upstream admission/queue); purged per session in SessionLifecycle.purge_session_memory. + self.pending_messages: Dict[str, List[QueuedMessage]] = {} from backend.apps.agents.core.flight_recorder import set_sessions_provider set_sessions_provider(lambda: self.sessions) - self.tasks: Dict[str, asyncio.Task] = {} - # Live mirror of the in-flight streamed assistant text per session, so a stop can persist the partial reply instantly instead of waiting out the multi-second SDK teardown the cancel handler sits behind. - self.live_partial: Dict[str, PartialReply] = {} - # Per-session cancel signal: the loop stashes its asyncio.Event here so a stop/close can set it. Lives on the manager, not the AgentSession model, so it stays out of serialization (an Event can't be model_dump'd). - self.cancel_events: Dict[str, asyncio.Event] = {} - # Persistent-client pool (lever A, flag-gated): one live CLI per session, reused across turns. - self.client_pool: Dict[str, ClientHandle] = {} - # Per-SESSION hook context + stderr buffer, updated in place each turn: a persistent client's hooks/stderr callback were bound at connect, so they must read stable objects, not per-turn rebuilds. - self.hook_ctxs: Dict[str, HookContext] = {} - self.stderr_buffers: Dict[str, List[str]] = {} - # Messages typed while a turn was live, replayed in order when it ends (see Messaging). - self.pending_messages: Dict[str, List[QueuedMessage]] = {} # Admission gate: one shared semaphore caps concurrent ROOT turns (children bypass). (Re)created per running loop by get_turn_admission so it never binds to a dead loop across a uvicorn reload or a test's asyncio.run. self.p_turn_admission_sema: Optional[asyncio.Semaphore] = None self.p_turn_admission_loop: Optional[asyncio.AbstractEventLoop] = None + @property + @typechecked + def sessions(self) -> Dict[str, Any]: + return self.store.sessions + + @sessions.setter + @typechecked + def sessions(self, value: Dict[str, Any]) -> None: + object.__setattr__(self.store, "sessions", value) + + @property + @typechecked + def tasks(self) -> Dict[str, Any]: + return self.store.tasks + + @tasks.setter + @typechecked + def tasks(self, value: Dict[str, Any]) -> None: + object.__setattr__(self.store, "tasks", value) + + @property + @typechecked + def live_partial(self) -> Dict[str, Any]: + return self.store.live_partial + + @live_partial.setter + @typechecked + def live_partial(self, value: Dict[str, Any]) -> None: + object.__setattr__(self.store, "live_partial", value) + + @property + @typechecked + def cancel_events(self) -> Dict[str, asyncio.Event]: + return self.store.cancel_events + + @cancel_events.setter + @typechecked + def cancel_events(self, value: Dict[str, asyncio.Event]) -> None: + object.__setattr__(self.store, "cancel_events", value) + + @property + @typechecked + def client_pool(self) -> Dict[str, Any]: + return self.store.client_pool + + @client_pool.setter + @typechecked + def client_pool(self, value: Dict[str, Any]) -> None: + object.__setattr__(self.store, "client_pool", value) + + @property + @typechecked + def hook_ctxs(self) -> Dict[str, Any]: + return self.store.hook_ctxs + + @hook_ctxs.setter + @typechecked + def hook_ctxs(self, value: Dict[str, Any]) -> None: + object.__setattr__(self.store, "hook_ctxs", value) + + @property + @typechecked + def stderr_buffers(self) -> Dict[str, List[str]]: + return self.store.stderr_buffers + + @stderr_buffers.setter + @typechecked + def stderr_buffers(self, value: Dict[str, List[str]]) -> None: + object.__setattr__(self.store, "stderr_buffers", value) @typechecked - def get_turn_admission(self) -> asyncio.Semaphore: - """The shared admission semaphore for the CURRENT loop; rebuilt if the loop changed so a - reload/test-run can never await a semaphore bound to a dead loop.""" - loop = asyncio.get_running_loop() - if self.p_turn_admission_sema is None or self.p_turn_admission_loop is not loop: - self.p_turn_admission_sema = asyncio.Semaphore(MAX_CONCURRENT_TURNS) - self.p_turn_admission_loop = loop - return self.p_turn_admission_sema - - @asynccontextmanager - async def turn_admission_slot(self, session: AgentSession, session_id: str) -> AsyncIterator[None]: - """Hold one concurrency slot for the duration of a ROOT turn. Overflow turns queue on the - semaphore (emitting agent:queued, then agent:admitted when they start). Two bypasses, both - load-bearing: (1) MAX_CONCURRENT_TURNS<=0 disables the gate entirely (kill switch); (2) a - CHILD turn (parent_session_id set) is NEVER gated, because a parent holds its own slot while - awaiting a delegated child, so gating children would deadlock the pool. `async with` release - is cancellation-safe: a stop while queued never acquired, so it can't over-release.""" - if MAX_CONCURRENT_TURNS <= 0 or session.parent_session_id is not None: - yield + async def ensure_keyed_model_route_synced(self, settings, short_name: str) -> None: + """Ensure 9Router has the provider node required by a pinned API-key model.""" + from backend.apps.agents.providers.registry import find_builtin_model + + entry = find_builtin_model(short_name) or {} + if entry.get("route") != "api": return - sema = self.get_turn_admission() - was_queued = sema.locked() - if was_queued: - try: - await ws_manager.send_to_session(session_id, "agent:queued", {"session_id": session_id}) - except Exception: - pass - async with sema: - if was_queued: - try: - await ws_manager.send_to_session(session_id, "agent:admitted", {"session_id": session_id}) - except Exception: - pass - yield + + provider = entry.get("api") + if provider == "openai" and getattr(settings, "openai_api_key", None): + from backend.apps import nine_router + if not nine_router.is_running(): + await nine_router.ensure_running() + if nine_router.is_running(): + await nine_router.sync_openai_api_key(settings.openai_api_key) + elif provider == "gemini" and getattr(settings, "google_api_key", None): + from backend.apps import nine_router + if not nine_router.is_running(): + await nine_router.ensure_running() + if nine_router.is_running(): + await nine_router.sync_gemini_api_key(settings.google_api_key) + elif provider == "custom" and getattr(settings, "custom_providers", None): + from backend.apps import nine_router + if not nine_router.is_running(): + await nine_router.ensure_running() + if nine_router.is_running(): + await nine_router.sync_custom_providers(settings.custom_providers or []) + @typechecked async def prewarm_client(self, session_id: str) -> None: @@ -156,6 +208,8 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li if not session: return + self.ensure_session_workspace_ready(session) + from backend.apps.agents.providers.registry import get_api_type as p_get_api_type p_api = p_get_api_type(session.model) prompt_content = self.build_prompt_content( @@ -184,10 +238,17 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li builtin_perms = load_builtin_permissions() # Builtins default to always_allow (frictionless); path_gate still force-prompts on catastrophic patterns (rm -rf), OS-scheduling, and sensitive paths, so poisoned-email -> destructive-command is still caught. Flip Bash to "ask" in the UI for a prompt on every command. Bind turn + stderr first: build_agent_options can raise early (no provider) and the except hands both to handle_run_error. - turn = TurnState() p_stderr_buffer: List[str] = [] # Read BEFORE build_agent_options consumes these flags: a fresh-session/fork request must force the persistent client to respawn (same branch id would otherwise fingerprint-match a client still holding the old transcript). p_force_respawn = bool(session.needs_fresh_session or session.needs_fork or fork_session) + p_event_emitter = AgentTurnEventEmitter( + sink=self.event_sink, + session_id=session_id, + provider=p_api_type_for_session, + model=p_router_model_id, + ) + turn = TurnState(event_emitter=p_event_emitter) + p_event_emitter.emit_started() try: logger.info(f"[SPAWN-PHASE] run-loop start session={session_id[:8]} t={time.monotonic():.3f}") (options, options_kwargs, prompt_content, p_stderr_buffer, @@ -195,20 +256,25 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li session, session_id, prompt, prompt_content, builtin_perms, selected_browser_ids, selected_app_output_ids, selected_setting_ids, fork_session, p_router_model_id, p_api_type_for_session) + p_hook_ctx = self.hook_ctxs.get(session_id) + if p_hook_ctx is not None: + p_hook_ctx.event_emitter = p_event_emitter resolved_model = p_router_model_id api_type = p_api_type_for_session thinking = ThinkingState() # Gate the CLI turn (spawn + stream) behind the admission slot so a burst can't run every turn at once; the slot is held ONLY for run_turn_with_retry, so the context-valve retry below re-acquires cleanly instead of nesting. - logger.info(f"[SPAWN-PHASE] admission-wait session={session_id[:8]} t={time.monotonic():.3f}") async with self.turn_admission_slot(session, session_id): - logger.info(f"[SPAWN-PHASE] admitted session={session_id[:8]} t={time.monotonic():.3f}") await self.run_turn_with_retry( session, session_id, prompt_content, options, options_kwargs, turn, thinking, p_stderr_buffer, resolved_model, api_type, global_settings, force_respawn=p_force_respawn, ) session.status = "completed" + p_event_emitter.emit_completed( + input_tokens=int(session.tokens.get("input_fresh", 0) or 0), + output_tokens=int(session.tokens.get("output", 0) or 0), + ) # Silent-quit seal: a turn that ran tools and ended with no visible answer gets ONE hidden continue nudge (dispatched by the auto-continuation block below); a second silent quit in the same ask surfaces as-is rather than looping. try: @@ -232,6 +298,7 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li except Exception: logger.exception("auto-continuation dispatch failed") except asyncio.CancelledError: + p_event_emitter.emit_failed("cancelled") # Only act if we're still the session's live task. A user stop pops this task (stop_agent already finalized status + partial), and a follow-up message may have started a newer turn; either way this dying task must NOT clobber the live status or pop the new turn's in-flight partial mirror. if self.tasks.get(session_id) is asyncio.current_task(): session.status = "stopped" @@ -287,6 +354,7 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li }) except Exception: logger.debug("submit_diagnostic context_pressure_valve failed", exc_info=True) + p_event_emitter.emit_failed("context_pressure_retry", retryable=True) await self.run_agent_loop( session_id, prompt, images, context_paths, forced_tools, attached_skills, fork_session, selected_browser_ids, @@ -294,23 +362,25 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li context_valve_retry=True, ) return + p_event_emitter.emit_failed(type(e).__name__) await handle_run_error(e, session, session_id, turn, p_stderr_buffer) except BaseException as e: # Catch BaseExceptionGroup from anyio task groups (e.g. concurrent CLI crash + pending approval cancellation) so it doesn't escape and kill the uvicorn process. logger.exception(f"Agent {session_id} fatal error: {e}") - # A group's str() names the group, not the cause; unwrap to the real member so a wrapped 429/auth error still gets its friendly card + retry-pill semantics instead of a raw group dump. + p_event_emitter.emit_failed(type(e).__name__) + # A group's str() names the group, not the cause; unwrap to the real member so a wrapped 429/auth error still gets its friendly card + retry-pill semantics. from backend.apps.agents.core.first_real_exception import first_real_exception p_real = first_real_exception(e) if p_real is not None: await handle_run_error(p_real, session, session_id, turn, p_stderr_buffer) - else: - session.status = "error" - error_msg = Message(role="system", content=f"Error: {str(e)}", branch_id=session.active_branch_id) - session.messages.append(error_msg) - await ws_manager.send_to_session(session_id, "agent:message", { - "session_id": session_id, - "message": error_msg.model_dump(mode="json"), - }) + return + session.status = "error" + error_msg = Message(role="system", content=f"Error: {str(e)}", branch_id=session.active_branch_id) + session.messages.append(error_msg) + await ws_manager.send_to_session(session_id, "agent:message", { + "session_id": session_id, + "message": error_msg.model_dump(mode="json"), + }) finally: # Only the session's live task finalizes. A stopped task (popped by stop_agent, which already finalized status + saved) or one superseded by a newer turn must not pop the new turn's partial mirror, broadcast a stale terminal status, or overwrite the snapshot the live turn is writing. p_is_live_task = self.tasks.get(session_id) is asyncio.current_task() @@ -345,4 +415,4 @@ async def run_agent_loop(self, session_id: str, prompt: str, images: Optional[Li logger.warning(f"Failed to snapshot session {session_id}: {e}") -agent_manager = AgentManager() +agent_manager = AgentManager(event_sink=BoundedAgentEventSink()) diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index 03717b6c0..621f8e113 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -1,10 +1,20 @@ +from backend.config.Apps import SubApp +from backend.apps.agents.agent_manager import agent_manager +from backend.apps.agents.core.ws_manager import ws_manager +from backend.apps.agents.core.models import AgentConfig, ApprovalResponse +from backend.apps.agents.manager.session.history_compaction import estimate_post_compact_input +from contextlib import asynccontextmanager +from fastapi import WebSocket, WebSocketDisconnect, HTTPException, Request +from fastapi.responses import JSONResponse import asyncio import logging import time -from contextlib import asynccontextmanager + +from backend.apps.hosting.policy import REQUEST_SCOPE, RequestScope +from backend.apps.nine_router.subscription_health import probe_subscription_health +from backend.apps.settings.redaction import redact_settings from typing import Any, Dict -from fastapi import HTTPException, Request from typeguard import typechecked from backend.apps.agents.agent_manager import agent_manager @@ -41,6 +51,20 @@ async def agents_lifespan(): agents = SubApp("agents", agents_lifespan) +async def p_session_or_404(session_id: str): + session = agent_manager.get_session(session_id) + if not session: + try: + session = await agent_manager.resume_session(session_id) + except ValueError: + raise HTTPException(status_code=404, detail="Session not found") + return session + + +async def p_require_owned_session(session_id: str, scope: RequestScope): + session = await p_session_or_404(session_id) + scope.require_owner_of(session.owner_account_id) + return session @typechecked def p_session_list_item(session: AgentSession) -> Dict[str, Any]: """Serialize dashboard metadata without retaining the full chat history.""" @@ -63,8 +87,8 @@ def p_session_list_item(session: AgentSession) -> Dict[str, Any]: @agents.router.get("/sessions") -async def list_sessions(dashboard_id: str = ""): - sessions = agent_manager.get_all_sessions(dashboard_id=dashboard_id or None) +async def list_sessions(dashboard_id: str = "", scope: RequestScope = REQUEST_SCOPE): + sessions = scope.filter_owned(agent_manager.get_all_sessions(dashboard_id=dashboard_id or None)) return {"sessions": [p_session_list_item(s) for s in sessions]} @agents.router.get("/sessions/{session_id}/followups") @@ -82,10 +106,13 @@ async def predict_followups_route(session_id: str, count: int = 3): @agents.router.get("/predict-prompts") -async def predict_prompts_route(count: int = 5): +async def predict_prompts_route(count: int = 5, scope: RequestScope = REQUEST_SCOPE): """Guess a few prompts the user might type next, in their own voice, from what they've already worked on. Drives the composer's ghost-text suggestion. Fails open to [] (no signal / no provider / error), so the composer just keeps its static placeholder.""" + # Hosted: the prediction scans every session on disk (all tenants), so it stays empty there (R0 containment). + if scope.hosted: + return {"suggestions": []} from backend.apps.agents.manager.predict_prompts import predict_prompts return {"suggestions": await predict_prompts(count=max(1, min(count, 8)))} @@ -105,7 +132,7 @@ async def agent_activity(): return {"active": active, "next_run_in_s": next_run_in_s} @agents.router.get("/sessions/{session_id}") -async def get_session(session_id: str): +async def get_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): """Returns the session by id. Falls back to a disk load when the session isn't in the in-memory @@ -117,21 +144,19 @@ async def get_session(session_id: str): session into agent_manager.sessions and the next GET short-circuits on the in-memory check. """ - session = agent_manager.get_session(session_id) - if not session: - try: - session = await agent_manager.resume_session(session_id) - except ValueError: - raise HTTPException(status_code=404, detail="Session not found") - # Seq read before the dump (no await between = atomic): the client seeds its WS resume cursor from this, so a REST hydrate isn't followed by a full from-zero replay of everything it just received. + session = await p_require_owned_session(session_id, scope) + # Seq read before the dump (no await between = atomic): the client seeds its WS resume cursor from this, so a REST hydrate isn't followed by a full from-zero replay. event_seq = seq_log.current_seq(session_id) payload = session.model_dump(mode="json") payload["event_seq"] = event_seq return payload @agents.router.post("/launch") -async def launch_agent(config: AgentConfig): - session = await agent_manager.launch_agent(config) +async def launch_agent(config: AgentConfig, scope: RequestScope = REQUEST_SCOPE): + config = scope.sanitize_launch_config(config) + session, run_first_turn = await scope.admit_launch(config, agent_manager.launch_agent) + if not run_first_turn: + return {"session_id": session.id, "session": session.model_dump(mode="json")} # A launch that carries a prompt runs it as the first turn through the same path /message uses. if config.prompt: asyncio.create_task(agent_manager.send_message(session.id, config.prompt)) @@ -140,11 +165,34 @@ async def launch_agent(config: AgentConfig): asyncio.create_task(agent_manager.prewarm_client(session.id)) return {"session_id": session.id, "session": session.model_dump(mode="json")} + @agents.router.post("/sessions/{session_id}/message") -async def send_message(session_id: str, body: dict): +async def send_message(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): prompt = body.get("prompt", "") if not prompt: raise HTTPException(status_code=400, detail="prompt is required") + session = await p_require_owned_session(session_id, scope) + side_effect_payload = { + "prompt": prompt, + "mode": body.get("mode"), + "model": body.get("model"), + "images": body.get("images"), + "context_paths": body.get("context_paths"), + "forced_tools": body.get("forced_tools"), + "attached_skills": body.get("attached_skills"), + "hidden": body.get("hidden", False), + "selected_browser_ids": body.get("selected_browser_ids"), + "selected_app_output_ids": body.get("selected_app_output_ids"), + "selected_setting_ids": body.get("selected_setting_ids"), + "client_message_id": body.get("client_message_id"), + } + if scope.admit_prompt( + session, + requested_mode=body.get("mode"), + forced_tools=body.get("forced_tools"), + side_effect_payload=side_effect_payload, + ): + return {"ok": True, "replayed": True} # Run MCP-suggestion classifier in parallel with the agent launch; fails open. try: @@ -174,25 +222,27 @@ async def p_emit_preflight(): except Exception: pass + await agent_manager.send_message( session_id, - prompt, - mode=body.get("mode"), - model=body.get("model"), - images=body.get("images"), - context_paths=body.get("context_paths"), - forced_tools=body.get("forced_tools"), - attached_skills=body.get("attached_skills"), - hidden=body.get("hidden", False), - selected_browser_ids=body.get("selected_browser_ids"), - selected_app_output_ids=body.get("selected_app_output_ids"), - selected_setting_ids=body.get("selected_setting_ids"), - client_message_id=body.get("client_message_id"), + side_effect_payload["prompt"], + mode=side_effect_payload["mode"], + model=side_effect_payload["model"], + images=side_effect_payload["images"], + context_paths=side_effect_payload["context_paths"], + forced_tools=side_effect_payload["forced_tools"], + attached_skills=side_effect_payload["attached_skills"], + hidden=side_effect_payload["hidden"], + selected_browser_ids=side_effect_payload["selected_browser_ids"], + selected_app_output_ids=side_effect_payload["selected_app_output_ids"], + selected_setting_ids=side_effect_payload["selected_setting_ids"], + client_message_id=side_effect_payload["client_message_id"], ) return {"ok": True} @agents.router.post("/sessions/{session_id}/stop") -async def stop_agent(session_id: str): +async def stop_agent(session_id: str, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) await agent_manager.stop_agent(session_id) # A stopped turn's parked AskUI waits would otherwise zombie for 600s and eat the next click (ENG-232). from backend.apps.agents.ui_request_bridge import cancel_session_waits @@ -200,27 +250,39 @@ async def stop_agent(session_id: str): return {"ok": True} @agents.router.post("/approval") -async def handle_approval(response: ApprovalResponse): - agent_manager.handle_approval(response.request_id, { +async def handle_approval(response: ApprovalResponse, scope: RequestScope = REQUEST_SCOPE): + approval_session_id = ws_manager.approval_session_id(response.request_id) + await scope.authorize_approval(approval_session_id, p_session_or_404) + decision = { "behavior": response.behavior, "message": response.message, "updated_input": response.updated_input, "trust_pattern": response.trust_pattern, "set_always_allow": response.set_always_allow, - }) + } + if not scope.resolve_approval(response.request_id, decision, approval_session_id): + agent_manager.handle_approval(response.request_id, decision) return {"ok": True} @agents.router.post("/sessions/{session_id}/edit_message") -async def edit_message(session_id: str, body: dict): +async def edit_message(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): + session = await p_require_owned_session(session_id, scope) message_id = body.get("message_id") new_content = body.get("content", "") if not message_id or not new_content: raise HTTPException(status_code=400, detail="message_id and content are required") + side_effect_payload = { + "message_id": message_id, + "content": new_content, + } + if scope.admit_prompt(session, requested_mode=None, forced_tools=None, side_effect_payload=side_effect_payload): + return {"ok": True, "replayed": True} await agent_manager.edit_message(session_id, message_id, new_content) return {"ok": True} @agents.router.post("/sessions/{session_id}/switch_branch") -async def switch_branch(session_id: str, body: dict): +async def switch_branch(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) branch_id = body.get("branch_id", "") if not branch_id: raise HTTPException(status_code=400, detail="branch_id is required") @@ -228,7 +290,8 @@ async def switch_branch(session_id: str, body: dict): return {"ok": True} @agents.router.post("/sessions/{session_id}/generate-title") -async def generate_title(session_id: str, body: dict): +async def generate_title(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) prompt = body.get("prompt", "") if not prompt: raise HTTPException(status_code=400, detail="prompt is required") @@ -236,7 +299,8 @@ async def generate_title(session_id: str, body: dict): return {"title": title} @agents.router.post("/sessions/{session_id}/generate-group-meta") -async def generate_group_meta(session_id: str, body: dict): +async def generate_group_meta(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) group_id = body.get("group_id", "") tool_calls = body.get("tool_calls", []) if not group_id or not tool_calls: @@ -277,25 +341,22 @@ async def generate_group_meta(session_id: str, body: dict): p_group_meta_inflight.pop(key, None) @agents.router.patch("/sessions/{session_id}") -async def update_session(session_id: str, body: dict): - session = agent_manager.get_session(session_id) - if not session: - raise HTTPException(status_code=404, detail="Session not found") +async def update_session(session_id: str, body: dict, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) await agent_manager.update_session(session_id, **body) return {"ok": True} @agents.router.get("/sessions/{session_id}/branches") -async def get_branches(session_id: str): - session = agent_manager.get_session(session_id) - if not session: - raise HTTPException(status_code=404, detail="Session not found") +async def get_branches(session_id: str, scope: RequestScope = REQUEST_SCOPE): + session = await p_require_owned_session(session_id, scope) return { "branches": {k: v.model_dump(mode="json") for k, v in session.branches.items()}, "active_branch_id": session.active_branch_id, } @agents.router.post("/sessions/{session_id}/duplicate") -async def duplicate_session(session_id: str, body: dict = {}): +async def duplicate_session(session_id: str, body: dict = {}, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) try: session = await agent_manager.duplicate_session( session_id, @@ -304,10 +365,12 @@ async def duplicate_session(session_id: str, body: dict = {}): ) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) + scope.stamp_owner(session) return {"session": session.model_dump(mode="json")} @agents.router.post("/sessions/{session_id}/close") -async def close_session(session_id: str): +async def close_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) try: await agent_manager.close_session(session_id) except ValueError as e: @@ -315,20 +378,23 @@ async def close_session(session_id: str): return {"ok": True} @agents.router.delete("/sessions/{session_id}") -async def delete_session(session_id: str): +async def delete_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) await agent_manager.delete_session(session_id) return {"ok": True} @agents.router.get("/history") -async def get_history(q: str = "", limit: int = 20, offset: int = 0, dashboard_id: str = "", closed_only: int = 0): +async def get_history(q: str = "", limit: int = 20, offset: int = 0, dashboard_id: str = "", closed_only: int = 0, scope: RequestScope = REQUEST_SCOPE): return agent_manager.get_history( q=q, limit=limit, offset=offset, dashboard_id=dashboard_id or None, + owner_account_id=scope.owner_id, closed_only=bool(closed_only), ) @agents.router.get("/sessions/{session_id}/browser-agents") -async def get_browser_agent_children(session_id: str): +async def get_browser_agent_children(session_id: str, scope: RequestScope = REQUEST_SCOPE): + await p_require_owned_session(session_id, scope) children = agent_manager.get_browser_agent_children(session_id) return {"sessions": children} @@ -359,16 +425,17 @@ async def forget_browser_memory(host: str): @agents.router.post("/sessions/{session_id}/resume") -async def resume_session(session_id: str): +async def resume_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): try: session = await agent_manager.resume_session(session_id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) + scope.require_owner_of(session.owner_account_id) return {"session": session.model_dump(mode="json")} @agents.router.post("/sessions/{session_id}/compact") -async def compact_session(session_id: str): +async def compact_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): """Run the summarizer over older turns to free up context. Wired to the 'Compact memory' button in the pre-send overflow banner and the @@ -378,9 +445,7 @@ async def compact_session(session_id: str): threshold now does the same (pre_send_context_guard); this button is the manual "do it now" for a user who wants the trim before the threshold. """ - session = agent_manager.sessions.get(session_id) - if not session: - raise HTTPException(status_code=404, detail="session not found") + session = await p_require_owned_session(session_id, scope) fired = agent_manager.maybe_compact(session, force=True) if fired: session.needs_fresh_session = True @@ -403,14 +468,12 @@ async def compact_session(session_id: str): @agents.router.post("/sessions/{session_id}/clear") -async def clear_session(session_id: str): +async def clear_session(session_id: str, scope: RequestScope = REQUEST_SCOPE): """Drop all messages from the session, keep MCPs/model/tools. Wired to the /clear slash command. Quickest path to recover from an overflow short of starting a fresh chat.""" - session = agent_manager.sessions.get(session_id) - if not session: - raise HTTPException(status_code=404, detail="session not found") + session = await p_require_owned_session(session_id, scope) session.messages = [] session.compacted_through_msg_id = None session.compacted_summary = None @@ -505,7 +568,7 @@ async def subscriptions_poll(body: dict): if result.get("success"): from backend.apps.service.client import sync as p_sync from backend.apps.settings.settings import load_settings - p_sync(load_settings().model_dump()) + p_sync(redact_settings(load_settings().model_dump())) from backend.apps.subscription.free_trial import clear_free_trial_on_connect await clear_free_trial_on_connect() return result @@ -540,7 +603,7 @@ async def subscriptions_exchange(body: dict): mark_completed(state) from backend.apps.service.client import sync as do_sync from backend.apps.settings.settings import load_settings - do_sync(load_settings().model_dump()) + do_sync(redact_settings(load_settings().model_dump())) # A connected subscription takes precedence over the free trial right away. from backend.apps.subscription.free_trial import clear_free_trial_on_connect await clear_free_trial_on_connect() @@ -599,6 +662,7 @@ async def probe_model(body: dict): from backend.apps.settings.settings import load_settings from backend.apps.nine_router import is_running as p_9r_running settings = load_settings() + await agent_manager.ensure_keyed_model_route_synced(settings, short_name) api_type = get_api_type(short_name) resolved = resolve_model_id_for_sdk(short_name, settings) entry = find_builtin_model(short_name) or {} diff --git a/backend/apps/agents/browser/browser_agent.py b/backend/apps/agents/browser/browser_agent.py index 1c8d248e6..e91393bd2 100644 --- a/backend/apps/agents/browser/browser_agent.py +++ b/backend/apps/agents/browser/browser_agent.py @@ -103,7 +103,7 @@ SYSTEM_PROMPT, ) from backend.apps.agents.core.models import AgentSession, ApprovalRequest, Message -from backend.apps.agents.core.ws_manager import ws_manager, await_reconnect +from backend.apps.agents.core.ws_manager import BrowserCommandOwner, ws_manager, await_reconnect from backend.apps.tools_lib.tools_lib import load_builtin_permissions logger = logging.getLogger(__name__) @@ -385,7 +385,10 @@ async def p_execute_browser_tool( async def p_eval_once() -> dict: rid = uuid4().hex - return await ws_manager.send_browser_command(rid, action, browser_id, params, tab_id=tab_id) + return await ws_manager.send_browser_command( + rid, action, browser_id, params, tab_id=tab_id, + owner=BrowserCommandOwner(origin="renderer"), + ) result = await p_eval_once() # Reads poll for the bridge to come up (app still mounting on turn 1). @@ -426,6 +429,7 @@ async def p_eval_once() -> dict: request_id = uuid4().hex result = await ws_manager.send_browser_command( request_id, action, browser_id, params, tab_id=tab_id, + owner=BrowserCommandOwner(origin="renderer"), ) if os.environ.get("OSW_DEBUG_LIST") == "1" and action == "list_interactives" and isinstance(result, dict): logger.info(f"[debug-list] {str(result.get('text') or '')[:2400]}") diff --git a/backend/apps/agents/browser/browser_metrics.py b/backend/apps/agents/browser/browser_metrics.py index f920bd15e..fa5168de6 100644 --- a/backend/apps/agents/browser/browser_metrics.py +++ b/backend/apps/agents/browser/browser_metrics.py @@ -87,6 +87,11 @@ def p_append(filename: str, obj: dict) -> None: path = os.path.join(metrics_dir(), filename) # owner-only: these lines can carry task text and error snippets fd = os.open(path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600) + if os.name == "posix": + try: + os.fchmod(fd, 0o600) + except Exception: + pass with os.fdopen(fd, "a", encoding="utf-8") as f: f.write(json.dumps(obj, default=str) + "\n") except Exception as e: diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 1ce658f1b..16b2a1661 100644 --- a/backend/apps/agents/core/models.py +++ b/backend/apps/agents/core/models.py @@ -16,6 +16,7 @@ class AgentConfig(BaseModel): max_turns: Optional[int] = None target_directory: Optional[str] = None dashboard_id: Optional[str] = None + owner_account_id: Optional[str] = None workflow_run_id: Optional[str] = None workflow_edit_id: Optional[str] = None # App cards the user picked to edit. When exactly one resolves, launch binds the chat's cwd to that app instead of seeding a new "Untitled App". @@ -119,6 +120,7 @@ class AgentSession(BaseModel): active_branch_id: str = "main" tool_group_meta: dict[str, "ToolGroupMeta"] = Field(default_factory=dict) dashboard_id: Optional[str] = None + owner_account_id: Optional[str] = None browser_id: Optional[str] = None parent_session_id: Optional[str] = None # Set when this session IS a workflow run's agent; the run renders in the Workflows monitor card, so the canvas suppresses the duplicate standalone agent card. diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 48fe69a0e..d5e4199a6 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/backend/apps/agents/core/ws_manager.py @@ -1,6 +1,10 @@ import asyncio +import hashlib import json import logging +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import datetime, timezone from typing import Optional from fastapi import WebSocket @@ -23,6 +27,73 @@ BROWSER_CMD_REBROADCAST_S = 3.0 # A CPU-starved renderer can briefly drop its WS (a missed heartbeat) and the frontend auto-reconnects a beat later; bridge that gap instead of hard-failing a live run into it. Short enough that a genuinely-closed window still fails quickly (and no LLM turns are ever burned waiting); long enough to ride out a reconnect even on a loaded machine. P_WS_RECONNECT_WAIT_S = 8.0 +HOSTED_AUTH_CLOSE_CODE = 4401 +HOSTED_AUTH_CLOSE_REASON = "hosted session revoked" +HOSTED_ACCOUNT_CLOSE_REASON = "hosted account revoked" +HOSTED_EXPIRY_CLOSE_REASON = "hosted session expired" +HOSTED_SOCKET_CLOSE_TIMEOUT_S = 1.0 + + +@dataclass(frozen=True) +class BrowserCommandOwner: + """Server-derived identity a browser command is correlated to at send time. + + A browser:result may resolve the command only when the submitting + connection's own BrowserCommandOwner (built from server-side connection + state, never from result payload bytes) equals this record exactly: + origin bridge, account, and auth session. Local desktop commands carry + (renderer/main, None, None); hosted identities can never match them. + """ + + origin: str # 'renderer' (dashboard sockets) | 'main' (Electron-main bridge) + account_id: str | None = None + auth_session_key: str | None = None + + +@dataclass(frozen=True) +class HostedConnectionIdentity: + account_id: str + auth_session_key: str + expires_at: datetime | None = None + + def __post_init__(self) -> None: + if self.expires_at is not None: + object.__setattr__(self, "expires_at", p_normalize_utc(self.expires_at)) + + def is_expired(self, now: datetime) -> bool: + return self.expires_at is not None and self.expires_at <= p_normalize_utc(now) + + +def p_normalize_utc(value: datetime) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def hosted_connection_identity( + account_id: str, + auth_session_key: str, + expires_at: datetime | None = None, +) -> HostedConnectionIdentity: + """Build normalized hosted socket authority from a verified auth session.""" + return HostedConnectionIdentity(account_id, auth_session_key, expires_at) + + +async def p_close_socket_with_deadline( + websocket: WebSocket, + code: int, + reason: str, + timeout: float, +) -> None: + await asyncio.wait_for(websocket.close(code=code, reason=reason), timeout=timeout) + + +def hosted_auth_session_key(cookie_value: str) -> str: + """Return a non-secret process-local lookup key for one hosted auth cookie.""" + if not cookie_value: + return "" + digest = hashlib.sha256(cookie_value.encode("utf-8")).hexdigest() + return f"hosted-session:v1:{digest}" def slim_status_data(event: str, data: dict) -> dict: @@ -65,28 +136,61 @@ async def await_reconnect(has_conn) -> bool: class ConnectionManager: """Manages WebSocket connections and HITL approval bridging; events flow through seq_log so reconnects can replay.""" - def __init__(self): + def __init__( + self, + *, + now: Callable[[], datetime] | None = None, + close_socket: Callable[[WebSocket, int, str, float], Awaitable[None]] | None = None, + close_timeout_s: float = HOSTED_SOCKET_CLOSE_TIMEOUT_S, + ): + self.p_now = now or (lambda: datetime.now(timezone.utc)) + self.p_close_socket = close_socket or p_close_socket_with_deadline + self.p_close_timeout_s = close_timeout_s self.connections: dict[str, list[WebSocket]] = {} self.global_connections: list[WebSocket] = [] + self.global_account_ids: dict[int, str | None] = {} + self.hosted_connection_identities: dict[int, HostedConnectionIdentity] = {} # Latched on the first renderer and never cleared: it answers "can a window reach this backend at all", which a momentary socket blip must not un-answer. Only the process dying resets it. self.renderer_ever_attached: bool = False # Which dashboard each global socket is currently showing, keyed by id(websocket). active_dashboard_id is the last one activated (the window the user is looking at most recently); a scheduled run targets it so its browser card spawns where the renderer can render it. self.global_dashboard_ids: dict[int, str] = {} self.active_dashboard_id: Optional[str] = None self.pending_futures: dict[str, asyncio.Future] = {} + self.pending_approval_sessions: dict[str, str] = {} self.browser_futures: dict[str, asyncio.Future] = {} + # Owner record per pending browser command; results resolve only for the exact recorded owner (resolve_browser_command). + self.browser_command_owners: dict[str, BrowserCommandOwner] = {} # The Electron MAIN process (not the renderer) holds a single WS here. Cookie reads route to it so they don't ride the renderer, which macOS throttles when the window is backgrounded (the source of the session-borrow bridge's intermittent timeouts). self.main_connection: Optional[WebSocket] = None - async def connect_session(self, session_id: str, websocket: WebSocket): + async def connect_session( + self, + session_id: str, + websocket: WebSocket, + identity: HostedConnectionIdentity | None = None, + ): await websocket.accept() if session_id not in self.connections: self.connections[session_id] = [] self.connections[session_id].append(websocket) - - async def connect_global(self, websocket: WebSocket): + if identity is not None: + self.hosted_connection_identities[id(websocket)] = identity + else: + self.hosted_connection_identities.pop(id(websocket), None) + + async def connect_global( + self, + websocket: WebSocket, + account_id: str | None = None, + identity: HostedConnectionIdentity | None = None, + ): await websocket.accept() self.global_connections.append(websocket) + self.global_account_ids[id(websocket)] = account_id + if identity is not None: + self.hosted_connection_identities[id(websocket)] = identity + else: + self.hosted_connection_identities.pop(id(websocket), None) self.renderer_ever_attached = True async def connect_main(self, websocket: WebSocket): @@ -99,6 +203,7 @@ def disconnect_main(self, websocket: WebSocket): self.main_connection = None def disconnect_session(self, session_id: str, websocket: WebSocket): + self.hosted_connection_identities.pop(id(websocket), None) if session_id in self.connections: self.connections[session_id] = [ ws for ws in self.connections[session_id] if ws != websocket @@ -117,19 +222,132 @@ def disconnect_global(self, websocket: WebSocket): ] # Drop this socket's active-dashboard pointer; if it owned the global one, fall back to any window still connected so a closed tab doesn't leave a stale target. self.global_dashboard_ids.pop(id(websocket), None) + self.global_account_ids.pop(id(websocket), None) + self.hosted_connection_identities.pop(id(websocket), None) if self.active_dashboard_id not in self.global_dashboard_ids.values(): self.active_dashboard_id = next(iter(self.global_dashboard_ids.values()), None) + def disconnect_everywhere(self, websocket: WebSocket) -> None: + """Evict one socket from every connection and hosted-identity registry.""" + for session_id in list(self.connections): + self.disconnect_session(session_id, websocket) + self.disconnect_global(websocket) + + def hosted_connection_is_current( + self, + websocket: WebSocket, + identity: HostedConnectionIdentity, + ) -> bool: + """Return whether a hosted socket still has live registered authority.""" + registered = self.hosted_connection_identities.get(id(websocket)) + return registered == identity and not identity.is_expired(self.p_now()) + + def p_connection_candidates(self) -> list[WebSocket]: + sockets: dict[int, WebSocket] = {} + for websocket in ( + *(ws for group in self.connections.values() for ws in group), + *self.global_connections, + ): + sockets[id(websocket)] = websocket + return list(sockets.values()) + + def p_evict_matching_hosted( + self, + predicate: Callable[[HostedConnectionIdentity], bool], + ) -> list[WebSocket]: + sockets: list[WebSocket] = [] + for websocket in self.p_connection_candidates(): + identity = self.hosted_connection_identities.get(id(websocket)) + if identity is not None and predicate(identity): + sockets.append(websocket) + for websocket in sockets: + self.disconnect_everywhere(websocket) + return sockets + + async def p_close_evicted( + self, + sockets: list[WebSocket], + *, + code: int, + reason: str, + ) -> None: + async def p_close_one(websocket: WebSocket) -> None: + try: + await self.p_close_socket( + websocket, + code, + reason, + self.p_close_timeout_s, + ) + except Exception: + logger.debug("hosted socket drain failed", exc_info=True) + + await asyncio.gather(*(p_close_one(websocket) for websocket in sockets)) + + def p_evict_expired(self, candidates: list[WebSocket]) -> list[WebSocket]: + now = self.p_now() + sockets: dict[int, WebSocket] = {} + for websocket in candidates: + identity = self.hosted_connection_identities.get(id(websocket)) + if identity is not None and identity.is_expired(now): + sockets[id(websocket)] = websocket + for websocket in sockets.values(): + self.disconnect_everywhere(websocket) + return list(sockets.values()) + + async def close_hosted_auth_session( + self, + auth_session_key: str, + *, + code: int = HOSTED_AUTH_CLOSE_CODE, + reason: str = HOSTED_AUTH_CLOSE_REASON, + ) -> None: + """Close sockets bound to one hosted login without touching local agent tasks.""" + if not auth_session_key: + return + sockets = self.p_evict_matching_hosted( + lambda identity: identity.auth_session_key == auth_session_key + ) + await self.p_close_evicted(sockets, code=code, reason=reason) + + async def close_hosted_account( + self, + account_id: str, + *, + code: int = HOSTED_AUTH_CLOSE_CODE, + reason: str = HOSTED_ACCOUNT_CLOSE_REASON, + ) -> None: + """Drain every hosted login for one account without touching local sockets.""" + if not account_id: + return + sockets = self.p_evict_matching_hosted( + lambda identity: identity.account_id == account_id + ) + await self.p_close_evicted(sockets, code=code, reason=reason) + async def send_to_session(self, session_id: str, event: str, data: dict): """Broadcast a session event with monotonic sequencing; terminal statuses also persist to disk.""" + account_id = self.p_session_account_id(session_id) data = slim_status_data(event, data) async with seq_log.stamp(session_id, event, data) as (seq, payload_str): + candidates = [ + *self.connections.get(session_id, []), + *self.global_connections, + ] + expired = self.p_evict_expired(candidates) + await self.p_close_evicted( + expired, + code=HOSTED_AUTH_CLOSE_CODE, + reason=HOSTED_EXPIRY_CLOSE_REASON, + ) for ws in list(self.connections.get(session_id, [])): try: await ws.send_text(payload_str) except Exception: logger.debug("send_to_session: send failed (will retry on reconnect)", exc_info=True) for ws in list(self.global_connections): + if not self.p_global_matches(ws, account_id): + continue try: await ws.send_text(payload_str) except Exception: @@ -147,11 +365,31 @@ async def send_to_session(self, session_id: str, event: str, data: dict): logger.debug("agent:message analytics bridge failed", exc_info=True) async def replay_to( - self, session_id: str, websocket: WebSocket, last_seq: int - ) -> dict: + self, + session_id: str, + websocket: WebSocket, + last_seq: int, + identity: HostedConnectionIdentity | None = None, + ) -> dict | None: """Replay buffered events with seq > last_seq; returns ack envelope for the resume handshake.""" oldest, newest, events = seq_log.replay(session_id, last_seq) + async def p_send(payload: str) -> bool: + if identity is not None and not self.hosted_connection_is_current(websocket, identity): + self.disconnect_everywhere(websocket) + await self.p_close_evicted( + [websocket], + code=HOSTED_AUTH_CLOSE_CODE, + reason=( + HOSTED_EXPIRY_CLOSE_REASON + if identity.is_expired(self.p_now()) + else HOSTED_AUTH_CLOSE_REASON + ), + ) + return False + await websocket.send_text(payload) + return True + # Gap-check first: if last_seq predates the buffer, signal REST-refresh; last_seq=0 means fresh client (full replay). if last_seq > 0 and oldest is not None and last_seq < oldest - 1: gap_payload = json.dumps({ @@ -165,7 +403,8 @@ async def replay_to( }, }) try: - await websocket.send_text(gap_payload) + if not await p_send(gap_payload): + return None except Exception: pass return { @@ -181,7 +420,8 @@ async def replay_to( events = self.p_strip_replayed_closes(events) for s in events: try: - await websocket.send_text(s) + if not await p_send(s): + return None except Exception: logger.debug("replay_to: send failed", exc_info=True) break @@ -199,7 +439,8 @@ async def replay_to( terminal = seq_log.load_terminal(session_id) if terminal is not None: try: - await websocket.send_text(terminal) + if not await p_send(terminal): + return None except Exception: pass return {"ok": True, "replayed": 1, "terminal_only": True} @@ -256,11 +497,37 @@ def p_filter_stale_approvals(self, events: list[str]) -> list[str]: out.append(payload_str) return out - async def broadcast_global(self, event: str, data: dict): - """Send to all dashboard connections; bypasses seq_log (dashboard resumes via full state refetch).""" + def p_session_account_id(self, session_id: str | None) -> str | None: + if not session_id: + return None + from backend.apps.agents.agent_manager import agent_manager + session = agent_manager.get_session(session_id) + return session.owner_account_id if session is not None else None + + def p_event_account_id(self, data: dict) -> str | None: + for candidate in (data, data.get("session") or {}, data.get("output") or {}): + owner = candidate.get("owner_account_id") if isinstance(candidate, dict) else None + if owner: + return owner + return self.p_session_account_id(data.get("session_id") or data.get("parent_session_id")) + + def p_global_matches(self, websocket: WebSocket, account_id: str | None) -> bool: + return self.global_account_ids.get(id(websocket)) == account_id + + async def broadcast_global(self, event: str, data: dict, account_id: str | None = None): + """Send an event only to dashboards in its account partition.""" + target_account_id = account_id if account_id is not None else self.p_event_account_id(data) payload = json.dumps({"event": event, "data": slim_status_data(event, data)}) dead: list[WebSocket] = [] + expired = self.p_evict_expired(list(self.global_connections)) + await self.p_close_evicted( + expired, + code=HOSTED_AUTH_CLOSE_CODE, + reason=HOSTED_EXPIRY_CLOSE_REASON, + ) for ws in list(self.global_connections): + if not self.p_global_matches(ws, target_account_id): + continue try: await ws.send_text(payload) except Exception: @@ -279,6 +546,7 @@ async def send_approval_request( """Send an approval request and wait for the user's decision; 10-minute timeout prevents permanent park.""" future = asyncio.get_event_loop().create_future() self.pending_futures[request_id] = future + self.pending_approval_sessions[request_id] = session_id payload: dict = { "request_id": request_id, @@ -299,23 +567,39 @@ async def send_approval_request( return {"behavior": "deny", "message": "Approval timed out"} finally: self.pending_futures.pop(request_id, None) - - def resolve_approval(self, request_id: str, decision: dict): + self.pending_approval_sessions.pop(request_id, None) + + def approval_session_id(self, request_id: str) -> str | None: + return self.pending_approval_sessions.get(request_id) + + def resolve_approval( + self, + request_id: str, + decision: dict, + *, + session_id: str | None = None, + ) -> bool: """Resolve a pending approval Future with the user's decision.""" + if session_id is not None and self.pending_approval_sessions.get(request_id) != session_id: + return False future = self.pending_futures.get(request_id) if future and not future.done(): future.set_result(decision) + return True + return False async def send_browser_command( - self, request_id: str, action: str, browser_id: str, params: dict, tab_id: str = "" + self, request_id: str, action: str, browser_id: str, params: dict, tab_id: str = "", + *, owner: BrowserCommandOwner, ) -> dict: - """Send a browser command to the frontend and wait for the result.""" + """Send a browser command to the frontend and wait for the owner-bound result.""" if not self.global_connections and not await await_reconnect(lambda: bool(self.global_connections)): return {"error": "No dashboard is connected. Open the dashboard to use browser tools."} loop = asyncio.get_event_loop() future = loop.create_future() self.browser_futures[request_id] = future + self.browser_command_owners[request_id] = owner payload = { "request_id": request_id, @@ -351,8 +635,11 @@ async def send_browser_command( return {"error": "No dashboard is connected. Open the dashboard to use browser tools."} finally: self.browser_futures.pop(request_id, None) + self.browser_command_owners.pop(request_id, None) - async def send_main_command(self, request_id: str, action: str, params: dict) -> dict: + async def send_main_command( + self, request_id: str, action: str, params: dict, *, owner: BrowserCommandOwner + ) -> dict: """Send a command straight to the throttle-free Electron MAIN socket (cookie reads only); returns a not-connected error so the caller can fall back to the renderer.""" ws = self.main_connection if ws is None: @@ -360,6 +647,7 @@ async def send_main_command(self, request_id: str, action: str, params: dict) -> loop = asyncio.get_event_loop() future = loop.create_future() self.browser_futures[request_id] = future + self.browser_command_owners[request_id] = owner payload = {"request_id": request_id, "action": action, "browser_id": "", "tab_id": "", "params": params} try: await ws.send_text(json.dumps({"event": "browser:command", "data": payload})) @@ -372,12 +660,31 @@ async def send_main_command(self, request_id: str, action: str, params: dict) -> return {"error": f"Electron main bridge send failed: {e}"} finally: self.browser_futures.pop(request_id, None) - - def resolve_browser_command(self, request_id: str, result: dict): - """Resolve a pending browser command Future with the frontend's result.""" + self.browser_command_owners.pop(request_id, None) + + def resolve_browser_command( + self, request_id: str, result: dict, *, claimant: BrowserCommandOwner + ) -> bool: + """Resolve a pending browser command only for its exact recorded owner. + + `claimant` is built by the ingress handler from the submitting + connection's server-derived state — never from result payload bytes. + Acceptance requires a live owner record whose origin, account, and auth + session all equal the claimant's, and is single-consumption: the first + matching result wins, so replays and the renderer's dedupe-cache + re-sends are refused (the sender's finally clears both records on + every terminal path). Refusal is silent (False) — a legitimate + duplicate must not error a healthy socket — and there is no ownerless + fallback path. + """ + owner = self.browser_command_owners.get(request_id) future = self.browser_futures.get(request_id) - if future and not future.done(): - future.set_result(result) + if owner is None or future is None or future.done(): + return False + if claimant != owner: + return False + future.set_result(result) + return True ws_manager = ConnectionManager() diff --git a/backend/apps/agents/disconnect_subscription.py b/backend/apps/agents/disconnect_subscription.py index 91ab3e92b..c593f4637 100644 --- a/backend/apps/agents/disconnect_subscription.py +++ b/backend/apps/agents/disconnect_subscription.py @@ -39,8 +39,10 @@ class SubscriptionDisconnectResult(BaseModel): def sync_settings_state() -> None: """Push the settings snapshot to the cloud state sync, exactly as connecting does. Imported late: service.client reaches back into this package.""" from backend.apps.service.client import sync + from backend.apps.settings.redaction import redact_settings from backend.apps.settings.settings import load_settings - sync(load_settings().model_dump()) + # F1: credentials never leave the machine through the telemetry/state sync; every sync site redacts first. + sync(redact_settings(load_settings().model_dump())) @typechecked diff --git a/backend/apps/agents/events/AgentEvent.py b/backend/apps/agents/events/AgentEvent.py new file mode 100644 index 000000000..4dbf8c1b7 --- /dev/null +++ b/backend/apps/agents/events/AgentEvent.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Annotated, Literal, Optional, Union +from uuid import uuid4 + +from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, TypeAdapter +from typeguard import typechecked + + +@typechecked +def p_event_id() -> str: + return uuid4().hex + + +@typechecked +def p_now() -> datetime: + return datetime.now(timezone.utc) + + +class AgentEventBase(BaseModel): + model_config = ConfigDict(validate_assignment=True, extra="forbid", frozen=True) + schema_version: Literal["1"] = "1" + event_id: str = Field(default_factory=p_event_id, min_length=32, max_length=32, pattern=r"^[0-9a-f]{32}$") + session_id: str = Field(min_length=1, max_length=128) + turn_id: str = Field(min_length=1, max_length=128) + sequence: int = Field(ge=0) + occurred_at: AwareDatetime = Field(default_factory=p_now) + monotonic_ms: int = Field(ge=0) + + +class TurnStartedEvent(AgentEventBase): + kind: Literal["turn.started"] = "turn.started" + provider: str = Field(min_length=1, max_length=64) + model: str = Field(min_length=1, max_length=128) + + +class TurnFirstTokenEvent(AgentEventBase): + kind: Literal["turn.first_token"] = "turn.first_token" + ttft_ms: int = Field(ge=0) + + +class ToolStartedEvent(AgentEventBase): + kind: Literal["tool.started"] = "tool.started" + tool_call_id: str = Field(min_length=1, max_length=128) + tool_name: str = Field(min_length=1, max_length=128) + + +class ToolCompletedEvent(AgentEventBase): + kind: Literal["tool.completed"] = "tool.completed" + tool_call_id: str = Field(min_length=1, max_length=128) + tool_name: str = Field(min_length=1, max_length=128) + duration_ms: int = Field(ge=0) + status: Literal["success", "error", "cancelled"] + error_type: Optional[str] = Field(default=None, max_length=128) + + +class TurnCompletedEvent(AgentEventBase): + kind: Literal["turn.completed"] = "turn.completed" + duration_ms: int = Field(ge=0) + input_tokens: int = Field(default=0, ge=0) + output_tokens: int = Field(default=0, ge=0) + + +class TurnFailedEvent(AgentEventBase): + kind: Literal["turn.failed"] = "turn.failed" + duration_ms: int = Field(ge=0) + error_type: str = Field(min_length=1, max_length=128) + retryable: bool = False + + +AgentEvent = Annotated[ + Union[ + TurnStartedEvent, + TurnFirstTokenEvent, + ToolStartedEvent, + ToolCompletedEvent, + TurnCompletedEvent, + TurnFailedEvent, + ], + Field(discriminator="kind"), +] + +P_AGENT_EVENT_ADAPTER = TypeAdapter(AgentEvent) + + +@typechecked +def parse_agent_event(value: object) -> AgentEvent: + return P_AGENT_EVENT_ADAPTER.validate_python(value) diff --git a/backend/apps/agents/events/AgentEventSink.py b/backend/apps/agents/events/AgentEventSink.py new file mode 100644 index 000000000..5e964da25 --- /dev/null +++ b/backend/apps/agents/events/AgentEventSink.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import logging +from collections import OrderedDict, deque +from dataclasses import dataclass +from threading import Lock +from typing import Dict, Protocol, Tuple, runtime_checkable + +from pydantic import BaseModel, ConfigDict +from typeguard import typechecked + +from backend.apps.agents.events.AgentEvent import AgentEvent + +logger = logging.getLogger(__name__) + + +@runtime_checkable +class AgentEventSink(Protocol): + @typechecked + def emit(self, event: AgentEvent) -> None: + ... + + +class NullAgentEventSink(BaseModel): + model_config = ConfigDict(validate_assignment=True) + + @typechecked + def emit(self, event: AgentEvent) -> None: + return None + + +@dataclass(frozen=True) +class AgentEventSnapshot: + session_id: str + events: Tuple[AgentEvent, ...] + dropped_events: int + + +class BoundedAgentEventSink: + """Thread-safe per-session event history for debug timeline adapters.""" + + @typechecked + def __init__(self, max_sessions: int = 64, max_events_per_session: int = 512) -> None: + if max_sessions < 1 or max_events_per_session < 1: + raise ValueError("event sink bounds must be at least 1") + self.max_sessions = max_sessions + self.max_events_per_session = max_events_per_session + self.p_events: OrderedDict[str, deque[AgentEvent]] = OrderedDict() + self.p_dropped: Dict[str, int] = {} + self.p_lock = Lock() + + @typechecked + def emit(self, event: AgentEvent) -> None: + with self.p_lock: + session_events = self.p_events.get(event.session_id) + if session_events is None: + if len(self.p_events) >= self.max_sessions: + evicted_session, _ = self.p_events.popitem(last=False) + self.p_dropped.pop(evicted_session, None) + session_events = deque(maxlen=self.max_events_per_session) + self.p_events[event.session_id] = session_events + self.p_dropped[event.session_id] = 0 + else: + self.p_events.move_to_end(event.session_id) + if len(session_events) == self.max_events_per_session: + self.p_dropped[event.session_id] += 1 + session_events.append(event) + + @typechecked + def snapshot(self, session_id: str) -> AgentEventSnapshot: + with self.p_lock: + return AgentEventSnapshot( + session_id=session_id, + events=tuple(self.p_events.get(session_id, ())), + dropped_events=self.p_dropped.get(session_id, 0), + ) + + @typechecked + def clear(self, session_id: str | None = None) -> None: + with self.p_lock: + if session_id is None: + self.p_events.clear() + self.p_dropped.clear() + return + self.p_events.pop(session_id, None) + self.p_dropped.pop(session_id, None) + + +@typechecked +def emit_agent_event(sink: AgentEventSink, event: AgentEvent) -> bool: + try: + sink.emit(event) + return True + except Exception: + logger.debug("agent event sink failed", exc_info=True) + return False diff --git a/backend/apps/agents/events/AgentTurnEventEmitter.py b/backend/apps/agents/events/AgentTurnEventEmitter.py new file mode 100644 index 000000000..a95c04163 --- /dev/null +++ b/backend/apps/agents/events/AgentTurnEventEmitter.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import time +from threading import RLock +from typing import Any, Dict, Literal +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, InstanceOf +from typeguard import typechecked + +from backend.apps.agents.events.AgentEvent import ( + ToolCompletedEvent, + ToolStartedEvent, + TurnCompletedEvent, + TurnFailedEvent, + TurnFirstTokenEvent, + TurnStartedEvent, +) +from backend.apps.agents.events.AgentEventSink import AgentEventSink, emit_agent_event + + +class AgentTurnEventEmitter(BaseModel): + model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True) + + sink: InstanceOf[AgentEventSink] + session_id: str + provider: str + model: str + turn_id: str = Field(default_factory=lambda: uuid4().hex) + sequence: int = 0 + started_monotonic: float = Field(default_factory=time.monotonic) + first_token_emitted: bool = False + tool_starts: Dict[str, float] = Field(default_factory=dict) + tool_names: Dict[str, str] = Field(default_factory=dict) + p_lock: Any = Field(default_factory=RLock, exclude=True) + + @typechecked + def emit_started(self) -> None: + with self.p_lock: + emit_agent_event(self.sink, TurnStartedEvent(**self.next_fields(), provider=self.provider, model=self.model)) + + @typechecked + def emit_first_token(self) -> None: + with self.p_lock: + if self.first_token_emitted: + return + self.first_token_emitted = True + emit_agent_event( + self.sink, + TurnFirstTokenEvent(**self.next_fields(), ttft_ms=self.duration_ms()), + ) + + @typechecked + def emit_tool_started(self, tool_call_id: str, tool_name: str) -> None: + with self.p_lock: + safe_id = tool_call_id[:128] + if not safe_id or safe_id in self.tool_starts: + return + safe_name = (tool_name or "unknown")[:128] + self.tool_starts[safe_id] = time.monotonic() + self.tool_names[safe_id] = safe_name + emit_agent_event( + self.sink, + ToolStartedEvent( + **self.next_fields(), + tool_call_id=safe_id, + tool_name=safe_name, + ), + ) + + @typechecked + def emit_tool_completed( + self, + tool_call_id: str, + tool_name: str, + status: Literal["success", "error", "cancelled"] = "success", + error_type: str | None = None, + ) -> None: + with self.p_lock: + safe_id = tool_call_id[:128] + if not safe_id: + return + if safe_id not in self.tool_starts: + self.emit_tool_started(safe_id, tool_name) + started = self.tool_starts.pop(safe_id, time.monotonic()) + safe_name = self.tool_names.pop(safe_id, (tool_name or "unknown")[:128]) + emit_agent_event( + self.sink, + ToolCompletedEvent( + **self.next_fields(), + tool_call_id=safe_id, + tool_name=safe_name, + duration_ms=max(0, int((time.monotonic() - started) * 1000)), + status=status, + error_type=error_type[:128] if error_type else None, + ), + ) + + @typechecked + def close_open_tools(self, status: Literal["error", "cancelled"] = "cancelled") -> None: + with self.p_lock: + for tool_call_id in list(self.tool_starts): + self.emit_tool_completed( + tool_call_id, + self.tool_names.get(tool_call_id, "unknown"), + status=status, + error_type="turn_ended" if status == "error" else None, + ) + + @typechecked + def emit_completed(self, input_tokens: int = 0, output_tokens: int = 0) -> None: + with self.p_lock: + self.close_open_tools() + emit_agent_event( + self.sink, + TurnCompletedEvent( + **self.next_fields(), + duration_ms=self.duration_ms(), + input_tokens=input_tokens, + output_tokens=output_tokens, + ), + ) + + @typechecked + def emit_failed(self, error_type: str, retryable: bool = False) -> None: + with self.p_lock: + self.close_open_tools(status="error") + emit_agent_event( + self.sink, + TurnFailedEvent( + **self.next_fields(), + duration_ms=self.duration_ms(), + error_type=error_type[:128], + retryable=retryable, + ), + ) + + @typechecked + def next_fields(self) -> Dict[str, Any]: + fields: Dict[str, Any] = { + "session_id": self.session_id, + "turn_id": self.turn_id, + "sequence": self.sequence, + "monotonic_ms": int(time.monotonic() * 1000), + } + self.sequence += 1 + return fields + + @typechecked + def duration_ms(self) -> int: + return max(0, int((time.monotonic() - self.started_monotonic) * 1000)) diff --git a/backend/apps/agents/events/__init__.py b/backend/apps/agents/events/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/apps/agents/manager/AgentLaunch.py b/backend/apps/agents/manager/AgentLaunch.py index 3b5445563..43258d1a3 100644 --- a/backend/apps/agents/manager/AgentLaunch.py +++ b/backend/apps/agents/manager/AgentLaunch.py @@ -17,7 +17,7 @@ ) from backend.apps.agents.core.ws_manager import ws_manager from backend.apps.settings.settings import load_settings -from backend.apps.agents.manager.session.session_store import load_session_data +from backend.apps.agents.manager.session.session_store import load_session_data, save_session from backend.apps.agents.manager.session.apply_context_window import apply_context_window from backend.apps.agents.manager.session.workspace_git import ( detect_git_identity, @@ -81,8 +81,18 @@ async def launch_agent(self, config: AgentConfig) -> AgentSession: or global_settings.default_folder or os.path.expanduser("~") ) + # An owned session lives under its owner's workspace root when the build has one (hosted); the desktop has none. + owned_workspace = False + try: + from backend.apps.hosting.policy import hosting_policy + owner_root = hosting_policy().owned_workspace_root(config.owner_account_id) + if owner_root: + effective_cwd = os.path.join(owner_root, session_id) + owned_workspace = True + except Exception: + logger.exception("owned workspace routing failed; using default cwd") - if config.mode in ("view-builder", "skill-builder") and not config.target_directory: + if config.mode in ("view-builder", "skill-builder") and not config.target_directory and not owned_workspace: effective_cwd = os.path.join(effective_cwd, session_id) os.makedirs(effective_cwd, exist_ok=True) @@ -98,6 +108,7 @@ async def launch_agent(self, config: AgentConfig) -> AgentSession: workspace_id=session_id, folder=effective_cwd, session_id=session_id, + owner_account_id=config.owner_account_id, ) if output_id: # Broadcast the new row so the Apps sidebar lights up immediately, even before the user clicks into it. The row name is still the placeholder ("Untitled App") at this point; the post-session meta-sync below fires a second upsert with the real name once the agent has written meta.json. @@ -138,6 +149,7 @@ async def launch_agent(self, config: AgentConfig) -> AgentSession: repo_url=repo_url, branch=branch_name, dashboard_id=config.dashboard_id, + owner_account_id=config.owner_account_id, workflow_run_id=config.workflow_run_id, workflow_edit_id=config.workflow_edit_id, thinking_level=getattr(global_settings, "default_thinking_level", "auto"), @@ -145,6 +157,18 @@ async def launch_agent(self, config: AgentConfig) -> AgentSession: apply_context_window(session, global_settings) self.sessions[session_id] = session + # Snapshot at birth. Until now a launched-but-quiet session lived only in memory until its + # first turn ended (the turn snapshot), the chat was closed, or the backend shut down + # gracefully (persist_all_sessions). A crash or SIGKILL in between left no file, so the + # respawned backend could not promote the session into its dashboard's list, the renderer + # treated that scoped list as authority, stripped the card, and the debounced layout save + # persisted the loss: the board forgot a chat the user had just opened. With the file on + # disk a respawn finds it (reconcile_on_startup marks it stopped, the card returns as the + # parked chat it was), which is exactly what a graceful shutdown already gave it. + try: + save_session(session_id, session.model_dump(mode="json")) + except Exception: + logger.warning(f"launch: could not snapshot session {session_id}", exc_info=True) await ws_manager.send_to_session(session_id, "agent:status", { "session_id": session_id, diff --git a/backend/apps/agents/manager/AgentManagerProtocol.py b/backend/apps/agents/manager/AgentManagerProtocol.py index 904f531af..1baf2badb 100644 --- a/backend/apps/agents/manager/AgentManagerProtocol.py +++ b/backend/apps/agents/manager/AgentManagerProtocol.py @@ -21,10 +21,14 @@ from backend.apps.agents.manager.run.client_pool import ClientHandle from backend.apps.agents.manager.streaming.HookContext import HookContext from backend.apps.agents.manager.streaming.PartialReply import PartialReply + from backend.apps.agents.manager.session.SessionStore import SessionStore + from backend.apps.agents.events.AgentEventSink import AgentEventSink class AgentManagerProtocol: # State set in AgentManager.__init__. + store: SessionStore + event_sink: AgentEventSink sessions: Dict[str, AgentSession] tasks: Dict[str, asyncio.Task] live_partial: Dict[str, PartialReply] diff --git a/backend/apps/agents/manager/Messaging.py b/backend/apps/agents/manager/Messaging.py index 728618dc9..cb17152b1 100644 --- a/backend/apps/agents/manager/Messaging.py +++ b/backend/apps/agents/manager/Messaging.py @@ -76,7 +76,7 @@ async def send_message( self.sessions[session_id] = session else: raise ValueError(f"Session {session_id} not found") - + existing = self.tasks.get(session_id) if existing and not existing.done(): # A mid-turn message used to be silently dropped here (no bubble, no trace); queue it and the turn task's done callback replays it. @@ -175,7 +175,7 @@ async def send_message( logger.warning(f"[browser-fast-path] gate error, normal path: {e}") if fast_verdict != "no": - task = asyncio.create_task(run_browser_fast_path(session, session_id, prompt, selected_browser_ids, fast_brief, fast_verdict)) + task = asyncio.create_task(run_browser_fast_path(session, session_id, prompt, selected_browser_ids, fast_brief, fast_verdict, event_sink=self.event_sink)) else: task = asyncio.create_task(self.run_agent_loop(session_id, prompt, images=images, context_paths=context_paths, forced_tools=forced_tools, attached_skills=attached_skills, selected_browser_ids=selected_browser_ids, selected_app_output_ids=selected_app_output_ids, selected_setting_ids=selected_setting_ids)) self.register_turn_task(session_id, task) diff --git a/backend/apps/agents/manager/configure_provider_env.py b/backend/apps/agents/manager/configure_provider_env.py index 028263c50..d39d5b6d3 100644 --- a/backend/apps/agents/manager/configure_provider_env.py +++ b/backend/apps/agents/manager/configure_provider_env.py @@ -8,6 +8,10 @@ from typeguard import typechecked from backend.apps.agents.core.models import AgentSession +from backend.apps.agents.manager.provider_runtime import ( + DEFAULT_PROVIDER_RUNTIME, + ProviderRuntime, +) from backend.apps.settings.models import AppSettings from backend.auth import get_auth_token @@ -15,16 +19,17 @@ @typechecked -async def router_available(global_settings: AppSettings) -> bool: +async def router_available( + global_settings: AppSettings, + runtime: ProviderRuntime = DEFAULT_PROVIDER_RUNTIME, +) -> bool: """True when 9Router is up, reviving it first if it died. A dead router must never masquerade as "no provider configured": detection now shares the dispatch path's lazy-start, so a crashed or orphaned router self-heals on the very next send instead of erroring the turn. Revival is gated on EVIDENCE of a provider (a settings key, proxy mode, or an active connection in the router's on-disk db) so a zero-config user keeps the clean no-provider message instead of us booting a router with nothing to route.""" - from backend.apps.nine_router import ensure_running as p_ensure, is_running as p_running - from backend.apps.nine_router.process import has_persisted_connections - if p_running(): + if runtime.router_is_running(): return True p_evidence = any([ getattr(global_settings, "anthropic_api_key", None), @@ -33,13 +38,13 @@ async def router_available(global_settings: AppSettings) -> bool: getattr(global_settings, "openrouter_api_key", None), getattr(global_settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"), bool(getattr(global_settings, "custom_providers", None) or []), - has_persisted_connections(), + runtime.has_persisted_connections(), ]) if not p_evidence: return False logger.info("[MCP-DEBUG] 9Router down at provider detection; reviving before concluding") - await p_ensure() - return p_running() + await runtime.ensure_router_running() + return runtime.router_is_running() @typechecked @@ -49,8 +54,8 @@ async def configure_provider_env( resolved_model: object, api_type: Optional[str], global_settings: AppSettings, + runtime: ProviderRuntime = DEFAULT_PROVIDER_RUNTIME, ) -> None: - from backend.apps.nine_router import is_running as nine_router_running from backend.apps.agents.providers.registry import NINEROUTER_MODEL_PREFIXES as NINEROUTER_MODEL_PREFIXES resolved_is_9router = isinstance(resolved_model, str) and resolved_model.startswith(NINEROUTER_MODEL_PREFIXES) @@ -84,11 +89,10 @@ async def configure_provider_env( logger.info(f"[MCP-DEBUG] Using direct OpenAI API key (route=api) for {session.model} via openai-passthrough") elif is_pinned_api_route and api_route_provider == "custom": # User OpenAI-compatible endpoint (Ollama/Together/LM Studio) via 9Router's synced provider node. - from backend.apps.nine_router import ensure_running as p_9r_ensure_c - if not nine_router_running(): + if not runtime.router_is_running(): logger.info(f"[MCP-DEBUG] custom provider selected but 9Router not running; waiting for startup") - await p_9r_ensure_c() - if not nine_router_running(): + await runtime.ensure_router_running() + if not runtime.router_is_running(): raise ValueError( "9Router could not start. Custom OpenAI-compatible " "providers need 9Router to translate the Anthropic " @@ -104,8 +108,9 @@ async def configure_provider_env( if cp: # Local servers often run auth-disabled; placeholder key since the OpenAI SDK requires non-empty. env["OPENAI_API_KEY"] = (cp.api_key or "").strip() or "no-auth-required" - from backend.apps.nine_router import normalize_openai_compat_base_url as norm_cp_url - env["OPENAI_BASE_URL"] = norm_cp_url(cp.base_url or "") + env["OPENAI_BASE_URL"] = runtime.normalize_openai_compat_base_url( + cp.base_url or "" + ) # Pin subagents or CLI's default Haiku 4.5 404s on the custom provider. if global_settings.anthropic_api_key: env["CLAUDE_CODE_SUBAGENT_MODEL"] = "claude-sonnet-4-6" @@ -130,11 +135,10 @@ async def configure_provider_env( logger.info(f"[MCP-DEBUG] Using direct Google API key (route=api) for {session.model} via local proxy") elif api_type == "openrouter" and getattr(global_settings, "openrouter_api_key", None): # OpenRouter via 9Router; with no Anthropic key/sub, fall back to OR's resold Claude for subagents (incl. WebSearch delegation) so they stay on the same OR billing. - if not nine_router_running(): - from backend.apps.nine_router import ensure_running as nine_router_ensure + if not runtime.router_is_running(): logger.info(f"[MCP-DEBUG] OpenRouter selected but 9Router not running; waiting for startup") - await nine_router_ensure() - if not nine_router_running(): + await runtime.ensure_router_running() + if not runtime.router_is_running(): raise ValueError( "9Router could not start. OpenRouter routing requires " "Node.js, install it and restart the app, or pick a " @@ -157,8 +161,7 @@ async def configure_provider_env( options_kwargs["env"] = env logger.info(f"[MCP-DEBUG] Using OpenRouter for {session.model}") elif api_type == "anthropic" and not resolved_is_9router and getattr(global_settings, "connection_mode", "own_key") in ("openswarm-pro", "free-trial"): - from backend.apps.settings.credentials import proxy_auth - bearer, proxy_url = proxy_auth(global_settings) + bearer, proxy_url = runtime.proxy_auth(global_settings) bearer = bearer or "" options_kwargs["env"] = { "ANTHROPIC_AUTH_TOKEN": bearer, @@ -179,7 +182,7 @@ async def configure_provider_env( elif api_type == "anthropic" and not resolved_is_9router and global_settings.anthropic_api_key: options_kwargs["env"] = {"ANTHROPIC_API_KEY": global_settings.anthropic_api_key} logger.info("[MCP-DEBUG] Using direct Anthropic API key") - elif await router_available(global_settings): + elif await router_available(global_settings, runtime): # Gemini-bound ids go through the local proxy for schema scrubbing; everything else hits 9Router directly. is_gemini_bound = ( isinstance(resolved_model, str) diff --git a/backend/apps/agents/manager/permissions/build_effective_tool_lists.py b/backend/apps/agents/manager/permissions/build_effective_tool_lists.py index 9fc4f1e83..6eb9c1c14 100644 --- a/backend/apps/agents/manager/permissions/build_effective_tool_lists.py +++ b/backend/apps/agents/manager/permissions/build_effective_tool_lists.py @@ -3,12 +3,13 @@ map, and the registered MCP servers; lifted out of the agent loop and covered by the MCP-gate invariant tests. Returns (allowed, disallowed).""" -from typing import Dict, List, Tuple +from typing import Dict, FrozenSet, List, Tuple from typeguard import typechecked from backend.apps.agents.core.models import AgentSession from backend.apps.agents.manager.permissions import path_gate +from backend.apps.hosting.policy import MUTATING_BUILTINS, hosting_policy from backend.apps.agents.manager.prompt.tool_catalog import ( FULL_TOOLS, get_all_known_tool_names, @@ -25,6 +26,20 @@ READ_ONLY_BLOCKED_TOOLS = ("Edit", "Bash", "NotebookEdit") + +def p_apply_builtin_denials( + allowed_tools: List[str], + disallowed_tools: List[str], + denials: FrozenSet[str], +) -> Tuple[List[str], List[str]]: + allowed = [tool for tool in allowed_tools if tool not in denials] + denied = list(disallowed_tools) + for tool in sorted(denials): + if tool not in denied: + denied.append(tool) + return allowed, denied + + @typechecked def build_effective_tool_lists( session: AgentSession, @@ -130,6 +145,19 @@ def build_effective_tool_lists( for wt_name in ("WebSearch", "WebFetch"): if wt_name not in effective_disallowed: effective_disallowed.append(wt_name) + + # The build's per-session denial list; fail closed to the mutating built-ins for an owned session whose policy cannot be consulted. + try: + denials = hosting_policy().builtin_tool_denials(session) + except Exception: + denials = MUTATING_BUILTINS if session.owner_account_id else frozenset() + if denials: + effective_allowed, effective_disallowed = p_apply_builtin_denials( + effective_allowed, + effective_disallowed, + denials, + ) + # With the openswarm-ui server live, the built-in AskUserQuestion is swapped for AskUI (same # Agent->SpawnAgent playbook: prompt nudges lose to the trained prior, a hard deny doesn't). # AskUI's option-list/question-flow cover the flat-choice cases; denying the built-in is what diff --git a/backend/apps/agents/manager/permissions/gate_hooks.py b/backend/apps/agents/manager/permissions/gate_hooks.py index 76864ab3f..f3d9dc2b6 100644 --- a/backend/apps/agents/manager/permissions/gate_hooks.py +++ b/backend/apps/agents/manager/permissions/gate_hooks.py @@ -179,6 +179,8 @@ async def offer_from_prompt(): if decision.behavior == "allow": if tool_use_id: ctx.tool_start_times[tool_use_id] = time.time() + if ctx.event_emitter is not None: + ctx.event_emitter.emit_tool_started(tool_use_id, tool_name) return { "hookSpecificOutput": { "hookEventName": hook_event, @@ -195,4 +197,6 @@ async def offer_from_prompt(): if tool_use_id: ctx.tool_start_times[tool_use_id] = time.time() + if ctx.event_emitter is not None: + ctx.event_emitter.emit_tool_started(tool_use_id, tool_name) return {} diff --git a/backend/apps/agents/manager/prompt/compose_turn_system_prompt.py b/backend/apps/agents/manager/prompt/compose_turn_system_prompt.py index 4c949788e..51afac498 100644 --- a/backend/apps/agents/manager/prompt/compose_turn_system_prompt.py +++ b/backend/apps/agents/manager/prompt/compose_turn_system_prompt.py @@ -59,10 +59,13 @@ def compose_turn_system_prompt( tz_name = tz_name or "UTC" now_local = datetime.now(ZoneInfo(tz_name)) tz_abbr = now_local.strftime("%Z") or tz_name + # No `%-d` / `%-I`: those are glibc extensions. On Windows strftime raises ValueError on + # them, the except below swallowed it, and the agent shipped without a clock on that OS. + hour = now_local.strftime("%I:%M %p").lstrip("0") time_ctx = ( "\n" - f"Today is {now_local.strftime('%A, %B %-d, %Y')}.\n" - f"Local time: {now_local.strftime('%-I:%M %p')} {tz_abbr} ({tz_name}).\n" + f"Today is {now_local.strftime('%A, %B')} {now_local.day}, {now_local.year}.\n" + f"Local time: {hour} {tz_abbr} ({tz_name}).\n" "Use this as ground truth for any date/time/day-of-week question. The timezone also " "gives the user's coarse region; when they say 'here' or 'near me' without a place, " "infer the likely city from it (say you inferred it) instead of claiming you can't know.\n" diff --git a/backend/apps/agents/manager/provider_runtime.py b/backend/apps/agents/manager/provider_runtime.py new file mode 100644 index 000000000..8159728d7 --- /dev/null +++ b/backend/apps/agents/manager/provider_runtime.py @@ -0,0 +1,47 @@ +"""Injected runtime boundary for provider routing.""" + +from __future__ import annotations + +from typing import Protocol, runtime_checkable + +from backend.apps import nine_router +from backend.apps.nine_router import process as nine_router_process +from backend.apps.settings import credentials +from backend.apps.settings.models import AppSettings + + +@runtime_checkable +class ProviderRuntime(Protocol): + """Operations provider routing needs from sibling applications.""" + + def router_is_running(self) -> bool: ... + + async def ensure_router_running(self) -> None: ... + + def has_persisted_connections(self) -> bool: ... + + def normalize_openai_compat_base_url(self, base_url: str) -> str: ... + + def proxy_auth(self, settings: AppSettings) -> tuple[str | None, str | None]: ... + + +class DefaultProviderRuntime: + """Production adapter; dynamic lookups preserve established test seams.""" + + def router_is_running(self) -> bool: + return nine_router.is_running() + + async def ensure_router_running(self) -> None: + await nine_router.ensure_running() + + def has_persisted_connections(self) -> bool: + return nine_router_process.has_persisted_connections() + + def normalize_openai_compat_base_url(self, base_url: str) -> str: + return nine_router.normalize_openai_compat_base_url(base_url) + + def proxy_auth(self, settings: AppSettings) -> tuple[str | None, str | None]: + return credentials.proxy_auth(settings) + + +DEFAULT_PROVIDER_RUNTIME: ProviderRuntime = DefaultProviderRuntime() diff --git a/backend/apps/agents/manager/run/RunOptions.py b/backend/apps/agents/manager/run/RunOptions.py index f2ce7e059..46aff689d 100644 --- a/backend/apps/agents/manager/run/RunOptions.py +++ b/backend/apps/agents/manager/run/RunOptions.py @@ -33,13 +33,11 @@ pre_send_context_guard, set_framework_overhead, register_web_mcp_server, append_web_tools_hint, inject_thinking_options, merge_hard_blocked_tools, ) +from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol logger = logging.getLogger(__name__) -from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol - - class RunOptions(AgentManagerProtocol): # No return annotation: the returned tuple carries an SDK ClaudeAgentOptions, which can't be module-imported here (mock-mode would fail to import the manager); it's lazy-imported below. @typechecked @@ -85,6 +83,9 @@ async def pre_tool_hook(input_data, tool_use_id, context): async def post_tool_hook(input_data, tool_use_id, context): return await post_tool_hook_mod.post_tool_hook(hook_ctx, input_data, tool_use_id, context) + + async def post_tool_failure_hook(input_data, tool_use_id, context): + return await post_tool_hook_mod.post_tool_failure_hook(hook_ctx, input_data, tool_use_id, context) _, mode_sys_prompt, _ = resolve_mode(session.mode, get_all_tool_names) # Reconcile active_mcps against currently-enabled tools (Phase 3). If the user toggled a server off in the Tools page mid-session, drop it from active_mcps automatically so the model isn't told "X is active" while build_mcp_servers silently filters it out. Emit a context_status event so the model and UI both know. @@ -193,6 +194,7 @@ async def stop_hook(input_data, tool_use_id, context): "hooks": { "PreToolUse": [HookMatcher(matcher=None, hooks=[pre_tool_hook])], "PostToolUse": [HookMatcher(matcher=None, hooks=[post_tool_hook])], + "PostToolUseFailure": [HookMatcher(matcher=None, hooks=[post_tool_failure_hook])], "Stop": [HookMatcher(matcher=None, hooks=[stop_hook])], }, "allowed_tools": effective_allowed, diff --git a/backend/apps/agents/manager/run/TurnAdmission.py b/backend/apps/agents/manager/run/TurnAdmission.py new file mode 100644 index 000000000..932508059 --- /dev/null +++ b/backend/apps/agents/manager/run/TurnAdmission.py @@ -0,0 +1,46 @@ +import asyncio +import os +from contextlib import asynccontextmanager +from typing import AsyncIterator + +from typeguard import typechecked + +from backend.apps.agents.core.models import AgentSession +from backend.apps.agents.core.ws_manager import ws_manager + + +# Cap concurrent ROOT agent turns so firing 30 agents at once does not spawn 30 CLIs at once. +# The overflow queues. Env-tunable; 0/blank disables the gate. +MAX_CONCURRENT_TURNS = int(os.environ.get("OSW_MAX_CONCURRENT_TURNS", "8") or "0") + + +class TurnAdmission: + @typechecked + def get_turn_admission(self) -> asyncio.Semaphore: + """Return the admission semaphore for the current running loop.""" + loop = asyncio.get_running_loop() + if self.p_turn_admission_sema is None or self.p_turn_admission_loop is not loop: + self.p_turn_admission_sema = asyncio.Semaphore(MAX_CONCURRENT_TURNS) + self.p_turn_admission_loop = loop + return self.p_turn_admission_sema + + @asynccontextmanager + async def turn_admission_slot(self, session: AgentSession, session_id: str) -> AsyncIterator[None]: + """Hold one concurrency slot for a root turn; child turns bypass to avoid deadlock.""" + if MAX_CONCURRENT_TURNS <= 0 or session.parent_session_id is not None: + yield + return + sema = self.get_turn_admission() + was_queued = sema.locked() + if was_queued: + try: + await ws_manager.send_to_session(session_id, "agent:queued", {"session_id": session_id}) + except Exception: + pass + async with sema: + if was_queued: + try: + await ws_manager.send_to_session(session_id, "agent:admitted", {"session_id": session_id}) + except Exception: + pass + yield diff --git a/backend/apps/agents/manager/run/TurnRunner.py b/backend/apps/agents/manager/run/TurnRunner.py index 633017e41..e9f91aed0 100644 --- a/backend/apps/agents/manager/run/TurnRunner.py +++ b/backend/apps/agents/manager/run/TurnRunner.py @@ -27,13 +27,11 @@ ) from backend.apps.agents.manager.streaming import thinking as thinking_mod from backend.apps.settings.models import AppSettings +from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol logger = logging.getLogger(__name__) -from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol - - class TurnRunner(AgentManagerProtocol): # `options` is the SDK ClaudeAgentOptions, lazy-imported below (so mock-mode can import the manager without the SDK present), so it's left unannotated; everything else is typed. @typechecked @@ -126,13 +124,15 @@ async def p_run_streaming_turn(p_stream=None): if isinstance(message, StreamEvent): await handle_stream_event( - message, session, session_id, turn, thinking, self.live_partial + message, session, session_id, turn, thinking, self.live_partial, + turn.event_emitter, ) elif isinstance(message, AssistantMessage): flight_recorder.crumb(session_id, "assistant-msg") await handle_assistant_message( - message, session, session_id, turn, thinking, self.live_partial, self.sessions + message, session, session_id, turn, thinking, self.live_partial, + self.sessions, turn.event_emitter, ) elif isinstance(message, ResultMessage): flight_recorder.crumb(session_id, "result-msg", subtype=str(getattr(message, "subtype", ""))) @@ -260,6 +260,8 @@ async def p_finalize_interrupted_stream(): wait = 0.0 if wait is not None: capacity_retry_attempt += 1 + if turn.event_emitter is not None: + turn.event_emitter.close_open_tools(status="cancelled") flight_recorder.crumb(session_id, "transient-retry", attempt=capacity_retry_attempt, wait=wait, err=str(e)[:160]) mid_stream = turn.current_turn_emitted logger.warning( @@ -276,4 +278,3 @@ async def p_finalize_interrupted_stream(): options = ClaudeAgentOptions(**options_kwargs) continue raise - diff --git a/backend/apps/agents/manager/run/client_pool.py b/backend/apps/agents/manager/run/client_pool.py index 1f2815274..f16f21cec 100644 --- a/backend/apps/agents/manager/run/client_pool.py +++ b/backend/apps/agents/manager/run/client_pool.py @@ -76,7 +76,7 @@ class ClientHandle(BaseModel): model_config = ConfigDict(validate_assignment=True) fingerprint: str - client: InstanceOf[object] + client: Any lock: InstanceOf[asyncio.Lock] connected_at: float last_used: float @@ -99,6 +99,13 @@ class ClientHandle(BaseModel): # Timer cadence for the background reclaim; the acquire-time sweep is lazy (fires only when some session takes a turn), this one catches an all-quiet pool. SWEEP_INTERVAL_SECONDS = float(os.environ.get("OSW_CLIENT_SWEEP_INTERVAL_SECONDS", "60")) +# Hard ceiling on warm CLIs regardless of idle age: past this, the least-recently-used IDLE sessions are disposed (they respawn ~0.5s on their next message), bounding the "30 chats open" resident-memory case. Kept a SOFT cap: a mid-turn or just-acquired client is never evicted, so a burst of live turns may exceed it rather than kill work. +MAX_LIVE_CLIENTS = int(os.environ.get("OSW_CLIENT_MAX_LIVE", "12")) +# Never cap-evict a client used this recently; far larger than the acquire->lock window, so a just-acquired client can't be reaped before its turn takes the lock. +LRU_GUARD_SECONDS = float(os.environ.get("OSW_CLIENT_LRU_GUARD_SECONDS", "5")) +# Timer cadence for the background reclaim; the acquire-time sweep is lazy (fires only when some session takes a turn), this one catches an all-quiet pool. +SWEEP_INTERVAL_SECONDS = float(os.environ.get("OSW_CLIENT_SWEEP_INTERVAL_SECONDS", "60")) + @typechecked async def evict_idle_clients(pool: Dict[str, "ClientHandle"]) -> None: diff --git a/backend/apps/agents/manager/run_browser_fast_path.py b/backend/apps/agents/manager/run_browser_fast_path.py index f7491b747..0c3f80451 100644 --- a/backend/apps/agents/manager/run_browser_fast_path.py +++ b/backend/apps/agents/manager/run_browser_fast_path.py @@ -14,6 +14,8 @@ from backend.apps.agents.core.ws_manager import ws_manager from backend.apps.agents.manager.session.session_store import save_session from backend.apps.settings.settings import load_settings +from backend.apps.agents.events.AgentEventSink import AgentEventSink, NullAgentEventSink +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter logger = logging.getLogger(__name__) @@ -26,6 +28,7 @@ async def run_browser_fast_path( selected_browser_ids: Optional[List[str]], brief: str = "", verdict: str = "act", + event_sink: Optional[AgentEventSink] = None, ) -> None: """Dispatch the browser sub-agent directly and reply with its outcome; the orchestrator LLM never runs. READ verdicts try one local fetch + @@ -35,6 +38,13 @@ async def run_browser_fast_path( task and the children.""" p_fp_t0 = time.monotonic() p_fp_path = verdict + p_event_emitter = AgentTurnEventEmitter( + sink=event_sink or NullAgentEventSink(), + session_id=session_id, + provider="browser_fast_path", + model=session.model, + ) + p_event_emitter.emit_started() logger.info(f"[browser-fast-path] direct dispatch for session {session_id} ({verdict})") text = "" # The fast-path skips the orchestrator, so the UI never gets the BrowserAgent tool-call that draws the "Browser Agent" bubble. Emit a synthetic tool_call/ tool_result pair (same shape + mcp__ name the orchestrator uses) so the bubble shows here too. None until we actually dispatch a browser (a pure READ answer has no browser, so no bubble). @@ -91,6 +101,7 @@ def p_summary(r: Dict[str, object]) -> str: if not text: # show the "Browser Agent" bubble during the dispatch (it renders as running, then completes when we emit the matching result below) p_bubble_tid = uuid4().hex + p_event_emitter.emit_tool_started(p_bubble_tid, p_browser_tool) p_tc = Message(role="tool_call", branch_id=session.active_branch_id, content={"id": p_bubble_tid, "tool": p_browser_tool, "input": {"task": prompt}}) session.messages.append(p_tc) @@ -128,6 +139,7 @@ def p_summary(r: Dict[str, object]) -> str: if not text: text = "The browser agent couldn't complete this and gave no report." except asyncio.CancelledError: + p_event_emitter.emit_failed("cancelled") raise except Exception as e: logger.warning(f"[browser-fast-path] dispatch failed: {e}") @@ -139,6 +151,7 @@ def p_summary(r: Dict[str, object]) -> str: ) # Close the synthetic bubble (always, even if the dispatch threw) so it never hangs as "running"; the bubble pairs this result with its call positionally. if p_bubble_tid: + p_event_emitter.emit_tool_completed(p_bubble_tid, p_browser_tool) # The bubble carries the same auditable record the sub-agent path shows. It used to close # with the literal string "done", so expanding it on this tier revealed nothing. from backend.apps.agents.browser import browser_trace @@ -156,6 +169,8 @@ def p_summary(r: Dict[str, object]) -> str: await ws_manager.send_to_session(session_id, "agent:message", { "session_id": session_id, "message": p_tr.model_dump(mode="json")}) asst_msg = Message(role="assistant", content=text, branch_id=session.active_branch_id) + if text: + p_event_emitter.emit_first_token() session.messages.append(asst_msg) await ws_manager.send_to_session(session_id, "agent:message", { "session_id": session_id, @@ -163,6 +178,7 @@ def p_summary(r: Dict[str, object]) -> str: }) session.status = "completed" session.closed_at = datetime.now() + p_event_emitter.emit_completed() await ws_manager.send_to_session(session_id, "agent:status", { "session_id": session_id, "status": "completed", diff --git a/backend/apps/agents/manager/session/SessionLifecycle.py b/backend/apps/agents/manager/session/SessionLifecycle.py index 12b998ad5..c8d90f7a3 100644 --- a/backend/apps/agents/manager/session/SessionLifecycle.py +++ b/backend/apps/agents/manager/session/SessionLifecycle.py @@ -4,6 +4,7 @@ import asyncio import logging +import os from datetime import datetime from typing import Dict, List, Optional, Set @@ -17,7 +18,9 @@ save_session, build_search_text, ) +from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol from backend.apps.agents.manager.session.apply_context_window import apply_context_window +from backend.apps.agents.manager.session.workspace_git import ensure_cwd_git_repo from backend.apps.agents.manager.session import resume_and_duplicate from backend.apps.agents.manager.view_builder_state import ( view_builder_render_retry_counts, @@ -31,9 +34,6 @@ P_NON_CHAT_MODES = {"browser-agent", "sub-agent", "invoked-agent", "app-agent"} -from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol - - class SessionLifecycle(AgentManagerProtocol): @staticmethod @typechecked @@ -101,16 +101,11 @@ def purge_session_memory(self, session_id: str) -> None: close or delete can't strand stale per-session state that lives until the process dies. One chokepoint on purpose: a new per-session cache wires its eviction in HERE and both removal paths get it for free.""" - self.sessions.pop(session_id, None) - self.tasks.pop(session_id, None) - self.live_partial.pop(session_id, None) - self.cancel_events.pop(session_id, None) - self.pending_messages.pop(session_id, None) view_builder_render_retry_counts.pop(session_id, None) view_builder_dirty_sessions.discard(session_id) dispose_client_soon(self.client_pool, session_id) - self.hook_ctxs.pop(session_id, None) - self.stderr_buffers.pop(session_id, None) + self.pending_messages.pop(session_id, None) + self.store.purge_session_runtime(session_id) @typechecked async def delete_session(self, session_id: str) -> None: @@ -138,6 +133,35 @@ async def delete_session(self, session_id: str) -> None: delete_session_file(session_id) logger.info(f"Session {session_id} permanently deleted") + @typechecked + async def delete_sessions_for_owner(self, owner_account_id: str) -> int: + """Permanently delete every in-memory or persisted session for an owner.""" + session_ids = { + sid for sid, session in self.sessions.items() + if session.owner_account_id == owner_account_id + } + for sid, data in load_all_session_data(): + if data.get("owner_account_id") == owner_account_id: + session_ids.add(sid) + + deleted = 0 + failures = 0 + for sid in sorted(session_ids): + try: + if sid in self.sessions: + await self.delete_session(sid) + else: + delete_session_file(sid) + deleted += 1 + except Exception: + logger.exception("Failed to delete owned session %s during reset", sid) + failures += 1 + if failures: + raise RuntimeError( + f"failed to delete {failures} owned session(s) during reset" + ) + return deleted + @typechecked async def resume_session(self, session_id: str) -> AgentSession: if session_id in self.sessions: @@ -159,6 +183,7 @@ def get_history( limit: int = 20, offset: int = 0, dashboard_id: Optional[str] = None, + owner_account_id: Optional[str] = None, closed_only: bool = False, ) -> Dict: """Return paginated, optionally filtered summaries of sessions, live ones included.""" @@ -189,9 +214,11 @@ def get_history( # Children are machinery, not chats: a busy user's real history was buried under hundreds of "Browser Agent" rows. if data.get("mode") in P_NON_CHAT_MODES: continue - # The boot fetch wants CLOSED sessions only: open ones landing in the client's history map made its resurrection gate swallow their terminal frames. Search keeps the full pool (open sessions on other dashboards are reachable nowhere else). + # The boot fetch wants CLOSED sessions only: open ones landing in the client's history map made its resurrection gate swallow their terminal frames. if closed_only and not data.get("closed_at"): continue + if owner_account_id and data.get("owner_account_id") != owner_account_id: + continue if dashboard_id and data.get("dashboard_id") != dashboard_id: continue if q_lower: @@ -278,6 +305,35 @@ def p_dashboard_card_ids(self, dashboard_id: str) -> Set[str]: def get_session(self, session_id: str) -> Optional[AgentSession]: return self.sessions.get(session_id) + @typechecked + def is_hosted_owned_workspace(self, session: AgentSession) -> bool: + if not session.owner_account_id or not session.cwd: + return False + try: + from backend.apps.hosting.policy import hosting_policy + + owner_root = hosting_policy().owned_workspace_root(session.owner_account_id) + if not owner_root: + return False + root = os.path.abspath(owner_root) + cwd = os.path.abspath(session.cwd) + return cwd == root or cwd.startswith(root + os.sep) + except Exception: + return False + + @typechecked + def ensure_session_workspace_ready(self, session: AgentSession) -> bool: + """Recreate a missing owned workspace before handing cwd to the SDK.""" + if not session.cwd or os.path.isdir(session.cwd): + return False + if not self.is_hosted_owned_workspace(session): + return False + os.makedirs(session.cwd, exist_ok=True) + ensure_cwd_git_repo(session.cwd) + session.needs_fresh_session = True + logger.info("Recreated missing owned workspace for session %s", session.id) + return True + @typechecked def get_browser_agent_children(self, parent_session_id: str) -> List[dict]: """Return browser-agent sessions for a parent, from memory or disk.""" @@ -302,4 +358,3 @@ def get_browser_agent_children(self, parent_session_id: str) -> List[dict]: results.append(sess.model_dump(mode="json")) return results - diff --git a/backend/apps/agents/manager/session/SessionStore.py b/backend/apps/agents/manager/session/SessionStore.py new file mode 100644 index 000000000..21ae911e7 --- /dev/null +++ b/backend/apps/agents/manager/session/SessionStore.py @@ -0,0 +1,101 @@ +import asyncio +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field +from typeguard import typechecked + + +class SessionStore(BaseModel): + """Runtime-owned per-session maps for AgentManager; JSON persistence stays in session_store.py.""" + model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True) + + sessions: Dict[str, Any] = Field(default_factory=dict) + tasks: Dict[str, Any] = Field(default_factory=dict) + live_partial: Dict[str, Any] = Field(default_factory=dict) + cancel_events: Dict[str, asyncio.Event] = Field(default_factory=dict) + client_pool: Dict[str, Any] = Field(default_factory=dict) + hook_ctxs: Dict[str, Any] = Field(default_factory=dict) + stderr_buffers: Dict[str, List[str]] = Field(default_factory=dict) + + @typechecked + def get_session(self, session_id: str) -> Optional[Any]: + return self.sessions.get(session_id) + + @typechecked + def set_session(self, session_id: str, session: Any) -> None: + self.sessions[session_id] = session + + @typechecked + def pop_session(self, session_id: str) -> Optional[Any]: + return self.sessions.pop(session_id, None) + + @typechecked + def has_session(self, session_id: str) -> bool: + return session_id in self.sessions + + @typechecked + def session_values(self) -> List[Any]: + return list(self.sessions.values()) + + @typechecked + def session_items(self) -> List[tuple[str, Any]]: + return list(self.sessions.items()) + + @typechecked + def get_task(self, session_id: str) -> Optional[Any]: + return self.tasks.get(session_id) + + @typechecked + def set_task(self, session_id: str, task: Any) -> None: + self.tasks[session_id] = task + + @typechecked + def pop_task(self, session_id: str) -> Optional[Any]: + return self.tasks.pop(session_id, None) + + @typechecked + def is_live_task(self, session_id: str, task: Any) -> bool: + return self.tasks.get(session_id) is task + + @typechecked + def set_live_partial(self, session_id: str, value: Any) -> None: + self.live_partial[session_id] = value + + @typechecked + def pop_live_partial(self, session_id: str) -> Optional[Any]: + return self.live_partial.pop(session_id, None) + + @typechecked + def get_or_create_stderr_buffer(self, session_id: str) -> List[str]: + return self.stderr_buffers.setdefault(session_id, []) + + @typechecked + def get_hook_ctx(self, session_id: str) -> Optional[Any]: + return self.hook_ctxs.get(session_id) + + @typechecked + def set_hook_ctx(self, session_id: str, hook_ctx: Any) -> None: + self.hook_ctxs[session_id] = hook_ctx + + @typechecked + def get_cancel_event(self, session_id: str) -> Optional[asyncio.Event]: + return self.cancel_events.get(session_id) + + @typechecked + def set_cancel_event(self, session_id: str, event: asyncio.Event) -> None: + self.cancel_events[session_id] = event + + @typechecked + def purge_session_runtime(self, session_id: str) -> None: + self.sessions.pop(session_id, None) + self.tasks.pop(session_id, None) + self.live_partial.pop(session_id, None) + self.cancel_events.pop(session_id, None) + self.client_pool.pop(session_id, None) + self.hook_ctxs.pop(session_id, None) + self.stderr_buffers.pop(session_id, None) + + @typechecked + def clear_sessions_and_tasks(self) -> None: + self.sessions.clear() + self.tasks.clear() diff --git a/backend/apps/agents/manager/session/session_store.py b/backend/apps/agents/manager/session/session_store.py index a134e5966..5fcd28f9c 100644 --- a/backend/apps/agents/manager/session/session_store.py +++ b/backend/apps/agents/manager/session/session_store.py @@ -25,6 +25,16 @@ def load_session_data(session_id: str) -> Optional[Dict]: return read_json_or_none(os.path.join(sessions_dir(), f"{session_id}.json")) +@typechecked +def load_session_owner_account_id(session_id: str) -> Optional[str]: + """Read persisted ownership without hydrating or publishing the session.""" + data = load_session_data(session_id) + if data is None: + return None + owner_account_id = data.get("owner_account_id") + return owner_account_id if isinstance(owner_account_id, str) else None + + @typechecked def delete_session_file(session_id: str) -> None: path = os.path.join(sessions_dir(), f"{session_id}.json") diff --git a/backend/apps/agents/manager/streaming/HookContext.py b/backend/apps/agents/manager/streaming/HookContext.py index 38c7e91a2..d39f974f3 100644 --- a/backend/apps/agents/manager/streaming/HookContext.py +++ b/backend/apps/agents/manager/streaming/HookContext.py @@ -4,11 +4,12 @@ holds (pydantic keeps the instance, doesn't copy it), so hook-side mutations to status / pending_approvals are visible to the loop.""" -from typing import Dict +from typing import Dict, Optional from pydantic import BaseModel, ConfigDict, InstanceOf from backend.apps.agents.core.models import AgentSession +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter class HookContext(BaseModel): @@ -21,6 +22,7 @@ class HookContext(BaseModel): policy_defaults: Dict[str, str] # The manager's LIVE session registry (InstanceOf keeps the reference, so a sub-agent the post hook spawns is visible to the manager; a plain Dict field pydantic would copy). sessions: InstanceOf[dict] + event_emitter: Optional[AgentTurnEventEmitter] = None # tool_use_id -> wall-clock start (s); pre records it, post pops it for elapsed_ms. tool_start_times: Dict[str, float] = {} # Consecutive ToolSearch calls; a run of these is the "looping on ToolSearch" wedge. diff --git a/backend/apps/agents/manager/streaming/handle_assistant_message.py b/backend/apps/agents/manager/streaming/handle_assistant_message.py index fb9d0c7b6..0286a8cdc 100644 --- a/backend/apps/agents/manager/streaming/handle_assistant_message.py +++ b/backend/apps/agents/manager/streaming/handle_assistant_message.py @@ -17,6 +17,7 @@ from backend.apps.agents.manager.streaming.upsert_message import upsert_message from backend.apps.agents.manager.streaming.PartialReply import PartialReply from backend.apps.agents.manager.streaming import thinking as thinking_mod +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter # The block types drive isinstance DISPATCH, so they must be real at runtime; imported inside the handler because by stream time the SDK is already resident (the turn's presence check imported it), keeping the 350ms sdk+mcp chain off the boot graph. from typing import TYPE_CHECKING @@ -36,6 +37,7 @@ async def handle_assistant_message( thinking: ThinkingState, live_partial: Dict[str, PartialReply], sessions: Dict[str, AgentSession], + event_emitter: Optional[AgentTurnEventEmitter] = None, ) -> None: from claude_agent_sdk.types import ThinkingBlock, TextBlock, ToolUseBlock @@ -66,6 +68,9 @@ async def handle_assistant_message( "input": block.input, }) + if content_parts and any(content_parts) and event_emitter is not None: + event_emitter.emit_first_token() + # Accumulate this AssistantMessage's contributions into the turn-level thinking pill. We re-emit the SAME message id each time so the frontend dedupes (addMessage replaces by id) and the bubble updates live as more thought / tools arrive. This is what gives us "Thought for 18s · 412 tokens · 3 tools used" reflecting the whole turn rather than just one think-step. NOTE: tool count is incremented in the content_block_start (block_type=="tool_use") branch above, NOT here. That path fires for both Anthropic and 9Router-translated providers; counting again here would double. If a provider somehow doesn't surface content_block_start for tool blocks but DOES surface them in the AssistantMessage envelope (defensive case), the max() in the consolidated emit will still pick up the higher count. if new_thinking_parts: thinking.text_parts.extend(new_thinking_parts) @@ -183,4 +188,3 @@ async def handle_assistant_message( turn.stream_text_msg_id = None turn.stream_tool_msg_ids_ordered = [] turn.stream_block_index_map = {} - diff --git a/backend/apps/agents/manager/streaming/handle_stream_event.py b/backend/apps/agents/manager/streaming/handle_stream_event.py index 33d911575..54fb10ae0 100644 --- a/backend/apps/agents/manager/streaming/handle_stream_event.py +++ b/backend/apps/agents/manager/streaming/handle_stream_event.py @@ -5,7 +5,7 @@ import time from datetime import datetime -from typing import Dict +from typing import Dict, Optional from uuid import uuid4 from typeguard import typechecked @@ -14,6 +14,7 @@ from backend.apps.agents.core.ws_manager import ws_manager from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState from backend.apps.agents.manager.streaming.PartialReply import PartialReply +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter # Runtime annotation stays `object` (the old ImportError fallback already admitted that); the real # type lives behind TYPE_CHECKING so importing this module stops paying the 350ms claude_agent_sdk+mcp chain at boot. @@ -33,6 +34,7 @@ async def handle_stream_event( turn: TurnState, thinking: ThinkingState, live_partial: Dict[str, PartialReply], + event_emitter: Optional[AgentTurnEventEmitter] = None, ) -> None: event = message.event event_type = event.get("type") @@ -89,6 +91,8 @@ async def handle_stream_event( if msg_id and delta_type == "text_delta": text_chunk = delta.get("text", "") + if text_chunk and event_emitter is not None: + event_emitter.emit_first_token() turn.assistant_text_chars += len(text_chunk) turn.stream_text_accum += text_chunk live_partial[session_id] = PartialReply( diff --git a/backend/apps/agents/manager/streaming/post_tool_hook.py b/backend/apps/agents/manager/streaming/post_tool_hook.py index 01bdae839..be652ef5d 100644 --- a/backend/apps/agents/manager/streaming/post_tool_hook.py +++ b/backend/apps/agents/manager/streaming/post_tool_hook.py @@ -25,6 +25,21 @@ logger = logging.getLogger(__name__) +@typechecked +async def post_tool_failure_hook( + ctx: HookContext, input_data: dict, tool_use_id, context +) -> Dict[str, object]: + if tool_use_id and ctx.event_emitter is not None: + ctx.tool_start_times.pop(tool_use_id, None) + ctx.event_emitter.emit_tool_completed( + str(tool_use_id), + input_data.get("tool_name", ""), + status="cancelled" if input_data.get("is_interrupt") else "error", + error_type="ToolExecutionError", + ) + return {} + + @typechecked async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, context) -> Dict[str, object]: session = ctx.session @@ -38,6 +53,14 @@ async def post_tool_hook(ctx: HookContext, input_data: dict, tool_use_id, contex # Accumulate per-tool latency on the session. Lets the cloud aggregate a tool-latency distribution into the existing daily.summary without firing per-tool events. hook_tool_name_early = input_data.get("tool_name", "") + if tool_use_id and ctx.event_emitter is not None: + is_error = bool(input_data.get("is_error")) + ctx.event_emitter.emit_tool_completed( + str(tool_use_id), + hook_tool_name_early, + status="error" if is_error else "success", + error_type="tool_error" if is_error else None, + ) if hook_tool_name_early and elapsed_ms is not None and elapsed_ms >= 0: latencies = getattr(session, "tool_latencies", None) if latencies is None: diff --git a/backend/apps/agents/manager/streaming/state.py b/backend/apps/agents/manager/streaming/state.py index 77afc4fdb..ee77be051 100644 --- a/backend/apps/agents/manager/streaming/state.py +++ b/backend/apps/agents/manager/streaming/state.py @@ -6,6 +6,7 @@ from typing import Dict, List, Optional from pydantic import BaseModel, ConfigDict, InstanceOf +from backend.apps.agents.events.AgentTurnEventEmitter import AgentTurnEventEmitter class ThinkingState(BaseModel): @@ -35,6 +36,7 @@ class TurnState(BaseModel): model_config = ConfigDict(validate_assignment=True) + event_emitter: Optional[AgentTurnEventEmitter] = None stream_text_msg_id: Optional[str] = None stream_tool_msg_ids_ordered: List[str] = [] stream_block_index_map: Dict[int, str] = {} diff --git a/backend/apps/agents/manager/streaming/thinking.py b/backend/apps/agents/manager/streaming/thinking.py index 8d4c46354..884222219 100644 --- a/backend/apps/agents/manager/streaming/thinking.py +++ b/backend/apps/agents/manager/streaming/thinking.py @@ -177,4 +177,3 @@ async def ticker_loop(thinking: ThinkingState, turn: TurnState, session: AgentSe await emit_consolidated_thinking(thinking, turn, session, session_id, sessions) except asyncio.CancelledError: pass - diff --git a/backend/apps/agents/providers/registry.py b/backend/apps/agents/providers/registry.py index ae89e774d..b5e97c4db 100644 --- a/backend/apps/agents/providers/registry.py +++ b/backend/apps/agents/providers/registry.py @@ -271,6 +271,8 @@ def resolve_model_id_for_sdk(short_name: str, settings: AppSettings) -> str: return short_name if entry.get("route") == "cc": return entry.get("router_model_id", entry.get("model_id", short_name)) + if entry.get("route") == "api" and entry.get("api") == "openai": + return entry.get("router_model_id", entry.get("model_id", short_name)) if entry.get("route") == "api": # OpenAI own-key still rides 9Router (the cp-openai node fixes max_tokens + translates Anthropic->OpenAI), so it MUST keep its cp-openai/ routing prefix or 9Router has no node to dispatch to. Anthropic own-key goes straight to api.anthropic.com and Gemini own-key via the local proxy, both on the bare id. if entry.get("api") == "openai": diff --git a/backend/apps/agents/tools/ssrf_guard.py b/backend/apps/agents/tools/ssrf_guard.py index d5272fce7..262790886 100644 --- a/backend/apps/agents/tools/ssrf_guard.py +++ b/backend/apps/agents/tools/ssrf_guard.py @@ -4,11 +4,10 @@ incl. cloud metadata, CGNAT, multicast, ULA v6, etc). Resolution is async (non-blocking) and covers both IPv4 AND IPv6 via getaddrinfo. -Loopback (127/8, ::1) is INTENTIONALLY allowed because the desktop app's App -Builder previews servers on 127.0.0.1: and the agent needs to be able -to verify the built app actually runs. The user owns the loopback surface on -their own machine; the realistic SSRF threat for a desktop app is cloud -metadata (169.254.169.254) + internal corporate LANs, not localhost. +Direct and IPv4-mapped loopback are allowed for desktop deployments because +App Builder previews servers on 127.0.0.1: and the agent needs to +verify the built app actually runs. Hosted deployments block loopback, and +transition/local-use ranges never inherit the desktop exception. """ from __future__ import annotations @@ -20,6 +19,8 @@ import httpx +from backend.apps.hosting.policy import hosting_policy + logger = logging.getLogger(__name__) @@ -47,16 +48,23 @@ class DomainUnreachable(SSRFBlocked): ipaddress.ip_network("169.254.0.0/16"), # link-local incl. cloud metadata ipaddress.ip_network("100.64.0.0/10"), # CGNAT ipaddress.ip_network("224.0.0.0/4"), # multicast + ipaddress.ip_network("240.0.0.0/4"), # reserved + limited broadcast ipaddress.ip_network("0.0.0.0/8"), # "this network" ipaddress.ip_network("198.18.0.0/15"), # benchmarking + ipaddress.ip_network("192.0.2.0/24"), # TEST-NET-1 + ipaddress.ip_network("198.51.100.0/24"), # TEST-NET-2 + ipaddress.ip_network("203.0.113.0/24"), # TEST-NET-3 ] P_BLOCKED_V6_NETS = [ ipaddress.ip_network("fe80::/10"), # link-local + ipaddress.ip_network("fec0::/10"), # deprecated site-local ipaddress.ip_network("fc00::/7"), # ULA + ipaddress.ip_network("64:ff9b:1::/48"), # local-use NAT64 ipaddress.ip_network("ff00::/8"), # multicast ipaddress.ip_network("::/128"), # unspecified ] +P_NAT64_WELL_KNOWN = ipaddress.ip_network("64:ff9b::/96") async def p_resolve_host_async(host: str) -> list[str]: @@ -70,21 +78,39 @@ async def p_resolve_host_async(host: str) -> list[str]: def p_is_forbidden_ip(ip_str: str) -> bool: - """True iff this IP is in a blocked range. Loopback is allowed (see module docstring).""" + """True iff this IP is blocked for the current deployment.""" try: ip = ipaddress.ip_address(ip_str) except ValueError: return True # unparseable -> block - # v6 can carry a v4 target (v4-mapped ::ffff:, 6to4 2002::) and routes to it; judge by the embedded v4 or a private host slips past the v6 list. - if ip.version == 6: - embedded = ip.ipv4_mapped or ip.sixtofour - if embedded is not None: - ip = embedded - if ip.is_loopback: - return False - if ip.version == 4: + + if isinstance(ip, ipaddress.IPv4Address): + if ip.is_loopback: + return hosting_policy().blocks_loopback_targets() return any(ip in net for net in P_BLOCKED_V4_NETS) - return any(ip in net for net in P_BLOCKED_V6_NETS) + + # Only direct v6 and v4-mapped loopback belong to desktop previews. Other + # transition encodings must be judged as network targets, even when their + # embedded v4 address is loopback. + if ip.is_loopback: + return hosting_policy().blocks_loopback_targets() + mapped = ip.ipv4_mapped + if mapped is not None: + if mapped.is_loopback: + return hosting_policy().blocks_loopback_targets() + return any(mapped in net for net in P_BLOCKED_V4_NETS) + + if any(ip in net for net in P_BLOCKED_V6_NETS): + return True + + embedded = ip.sixtofour + if embedded is None and ip in P_NAT64_WELL_KNOWN: + embedded = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF) + if embedded is not None: + if embedded.is_loopback: + return True + return any(embedded in net for net in P_BLOCKED_V4_NETS) + return False async def assert_safe_url(url: str) -> str: diff --git a/backend/apps/health/health.py b/backend/apps/health/health.py index ed555db03..cbca53883 100644 --- a/backend/apps/health/health.py +++ b/backend/apps/health/health.py @@ -1,19 +1,38 @@ -from backend.config.Apps import SubApp from contextlib import asynccontextmanager +from collections.abc import Callable + +from fastapi import BackgroundTasks, status from fastapi.responses import PlainTextResponse from pydantic import BaseModel, ConfigDict from typeguard import typechecked -from fastapi import status, HTTPException + +from backend.config.Apps import SubApp + + +ready_background_task: Callable[[], None] | None = None + + +def set_ready_background_task(task: Callable[[], None] | None) -> None: + global ready_background_task + ready_background_task = task + @asynccontextmanager async def health_lifespan(): yield + health = SubApp("health", health_lifespan) + @health.router.get("/check") @typechecked -async def check() -> PlainTextResponse: +async def check(background_tasks: BackgroundTasks) -> PlainTextResponse: + if ready_background_task is not None: + # FastAPI runs this after the response body is sent. The Electron shell + # can mark the backend ready before cache population starts competing + # for disk, Defender scans, or the bundled Python interpreter. + background_tasks.add_task(ready_background_task) return PlainTextResponse( content="OK", status_code=status.HTTP_200_OK, diff --git a/backend/apps/hosting/__init__.py b/backend/apps/hosting/__init__.py new file mode 100644 index 000000000..f8cf4085c --- /dev/null +++ b/backend/apps/hosting/__init__.py @@ -0,0 +1,2 @@ +"""The hosting seam: the one place the app asks "who owns this request, and what does this +build allow?". See policy.py.""" diff --git a/backend/apps/hosting/policy.py b/backend/apps/hosting/policy.py new file mode 100644 index 000000000..95cf53bae --- /dev/null +++ b/backend/apps/hosting/policy.py @@ -0,0 +1,186 @@ +"""Hosting policy seam. + +On the desktop every request is the local user: nothing is scoped by owner, and every process-wide +answer below is the permissive default. A hosted (multi-tenant) build supplies its own policy through +`p_provider`; the rest of the app only ever talks to `REQUEST_SCOPE` (a FastAPI dependency) and +`hosting_policy()`, so no route or manager knows which build it is running in. + +Everything here is a plain default: the desktop scope allows, filters nothing, stamps no owner, and +`hosting_policy()` answers "not hosted" to every question. +""" +from __future__ import annotations + +from typing import Any, Awaitable, Callable, Iterable, List, Optional, Tuple, TypeVar + +from fastapi import Request, params +from typeguard import typechecked + +T = TypeVar("T") + +#: The built-in tools that change the machine or spawn work. A hosted build may deny them to trial +#: callers; they are also the fail-closed answer when an owned session's policy cannot be consulted. +MUTATING_BUILTINS: frozenset = frozenset({ + "Agent", + "Bash", + "CronCreate", + "CronDelete", + "Edit", + "EnterWorktree", + "InvokeAgent", + "NotebookEdit", + "TodoWrite", + "Write", +}) + + +class RequestScope: + """Who owns the current request, and what it may do. Desktop: nobody owns anything and + everything is allowed. A hosted build returns a subclass bound to the caller's account.""" + + #: True in a hosted build once the caller is resolved; the desktop is never hosted. + hosted: bool = False + #: The caller's account id in a hosted build; None on the desktop. + owner_id: Optional[str] = None + + # ---- ownership ----------------------------------------------------------------------- + @typechecked + def require_owner_of(self, owner_account_id: Optional[str]) -> None: + """Raise unless the caller owns the resource stamped with `owner_account_id`.""" + + @typechecked + def filter_owned(self, items: Iterable[T]) -> List[T]: + """Keep only the caller's items (items carry `owner_account_id`).""" + return list(items) + + @typechecked + def stamp_owner(self, item: Any) -> None: + """Mark a freshly created resource as the caller's (no-op on the desktop).""" + + @typechecked + def owner_for_new_resource(self, requested: Optional[str]) -> Optional[str]: + """The owner a new resource is created under: whatever the caller asked for on the desktop, + always the caller itself in a hosted build.""" + return requested + + @typechecked + def require_local_operator(self, what: str) -> None: + """Raise unless the caller is the machine's own operator (always, on the desktop). `what` + names the operation for the refusal message.""" + + # ---- agents --------------------------------------------------------------------------- + @typechecked + def sanitize_launch_config(self, config: T) -> T: + return config + + @typechecked + async def admit_launch( + self, config: Any, launch: Callable[[Any], Awaitable[Any]], + ) -> Tuple[Any, bool]: + """Run `launch(config)` under this scope's admission rules. Returns the session and whether + the caller should still run the launch's first turn (a hosted build may already have + recorded the launch as the durable side effect and answers False).""" + return await launch(config), True + + @typechecked + def admit_prompt( + self, + session: Any, + *, + requested_mode: Optional[str], + forced_tools: Optional[List[str]], + side_effect_payload: dict, + ) -> bool: + """Admit a prompt/edit for `session`. Returns True when the request was a replay of an + already-admitted one (the caller then answers `replayed` instead of running it).""" + return False + + @typechecked + async def authorize_approval( + self, + approval_session_id: Optional[str], + session_lookup: Callable[[str], Awaitable[Any]], + ) -> None: + """Raise unless the caller may answer the approval that `approval_session_id` is waiting on; + `session_lookup` resolves the owning session when the scope needs to check it.""" + + @typechecked + def resolve_approval(self, request_id: str, decision: dict, approval_session_id: Optional[str]) -> bool: + """Deliver an approval decision through the scope's own channel. Returns True when it did; + False means the caller delivers it through the desktop path.""" + return False + + # ---- outputs -------------------------------------------------------------------------- + @typechecked + def require_app_builder_enabled(self) -> None: + """Raise unless this caller may use the App Builder (always allowed on the desktop).""" + + @typechecked + def register_seeded_workspace(self, workspace_id: str, meta: Optional[dict]) -> Optional[str]: + """Record a seeded workspace as the caller's output when the build tracks ownership; returns + the output id it now maps to, or None when nothing is recorded (the desktop).""" + return None + + +class HostingPolicy: + """Process-wide answers. Desktop defaults throughout.""" + + enabled: bool = False + + @typechecked + def request_scope(self, request: Request) -> RequestScope: + return DESKTOP_SCOPE + + @typechecked + def owned_workspace_root(self, owner_account_id: Optional[str]) -> Optional[str]: + """The per-owner workspace root a session's cwd must live under, or None when sessions use + the ordinary cwd.""" + return None + + @typechecked + def builtin_tool_denials(self, session: Any) -> frozenset: + """Built-in tools this session may not use (empty on the desktop).""" + return frozenset() + + @typechecked + def tool_update_restricted(self, tool: Any, body: Any) -> bool: + """True when a tools-library update must be refused for this build.""" + return False + + @typechecked + def workflows_disabled(self) -> bool: + return False + + @typechecked + def blocks_loopback_targets(self) -> bool: + """True when SSRF checks must treat loopback as a network target (never on the desktop, + where local previews are the point).""" + return False + + +DESKTOP_SCOPE = RequestScope() +DESKTOP_POLICY = HostingPolicy() + +# A build that hosts replaces this with a provider that resolves its own policy. +p_provider: Callable[[], HostingPolicy] = lambda: DESKTOP_POLICY + + +@typechecked +def hosting_policy() -> HostingPolicy: + return p_provider() + + +@typechecked +def request_scope(request: Request) -> RequestScope: + """Resolve the caller's scope for a request (what REQUEST_SCOPE injects).""" + return hosting_policy().request_scope(request) + + +class p_RequestScopeDependency(RequestScope, params.Depends): + """The route default `scope: RequestScope = REQUEST_SCOPE`. FastAPI sees a dependency and injects + the caller's scope; a route called directly (tests do) gets this object, which is the desktop scope.""" + + def __init__(self) -> None: + params.Depends.__init__(self, dependency=request_scope) + + +REQUEST_SCOPE: RequestScope = p_RequestScopeDependency() diff --git a/backend/apps/nine_router/process.py b/backend/apps/nine_router/process.py index 8df0c34a0..9e171b729 100644 --- a/backend/apps/nine_router/process.py +++ b/backend/apps/nine_router/process.py @@ -567,7 +567,11 @@ async def p_ensure_running_impl(): env["ELECTRON_RUN_AS_NODE"] = "1" else: # Dev: install the pinned npm package into a local cache once, then spawn `node app/server.js` directly (bypasses the package cli.js tray icon users confusingly quit, its update-check spinner, and the TUI). - cached_server = p_ensure_router_cached() + # A cold npm install can take minutes on Windows. This path is started + # as a background task during app lifespan, so running the synchronous + # installer on the event loop would still block the HTTP server from + # binding and make the whole backend appear hung. + cached_server = await asyncio.to_thread(p_ensure_router_cached) if not cached_server: return node = p_find_node() diff --git a/backend/apps/nine_router/sync_custom.py b/backend/apps/nine_router/sync_custom.py index 44e854e8a..432669505 100644 --- a/backend/apps/nine_router/sync_custom.py +++ b/backend/apps/nine_router/sync_custom.py @@ -17,11 +17,14 @@ find_keyed_connection, nr, ) +from backend.apps.settings.credentials import proxy_auth logger = logging.getLogger(__name__) # We mirror settings.custom_providers[] with prefix `cp-` so they don't collide with the user's primary OpenAI key. NINE_ROUTER_CUSTOM_NAME_SUFFIX = " (OpenSwarm-managed)" +NINE_ROUTER_OPENAI_COMPAT_NAME = f"OpenAI{NINE_ROUTER_CUSTOM_NAME_SUFFIX}" +P_RESERVED_MANAGED_PREFIXES = {NINE_ROUTER_OPENAI_KEYED_PREFIX} async def sync_openai_compat_node(api_key: str | None) -> None: @@ -32,7 +35,7 @@ async def sync_openai_compat_node(api_key: str | None) -> None: import os as p_os port = p_os.environ.get("OPENSWARM_PORT", "8324") base_url = f"http://127.0.0.1:{port}/api/openai-passthrough/v1" - managed_name = f"OpenAI{NINE_ROUTER_CUSTOM_NAME_SUFFIX}" + managed_name = NINE_ROUTER_OPENAI_COMPAT_NAME try: async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client: @@ -186,6 +189,8 @@ async def sync_custom_providers(providers: list) -> None: api_key = api_key.strip() or "no-auth-required" slug = p_custom_provider_slug(name) prefix = f"cp-{slug}" + if prefix in P_RESERVED_MANAGED_PREFIXES: + prefix = f"cp-custom-{slug}" seen_prefixes.add(prefix) managed_name = f"{name.strip()}{NINE_ROUTER_CUSTOM_NAME_SUFFIX}" @@ -260,7 +265,7 @@ async def sync_custom_providers(providers: list) -> None: return for prefix, node in managed_by_prefix.items(): # cp-openai wears the same managed suffix but belongs to sync_openai_compat_node; reaping it here killed every gpt-*-api request with "No credentials". - if prefix in seen_prefixes or prefix == NINE_ROUTER_OPENAI_KEYED_PREFIX: + if prefix in P_RESERVED_MANAGED_PREFIXES or prefix in seen_prefixes or prefix == NINE_ROUTER_OPENAI_KEYED_PREFIX: continue try: async with nr().httpx.AsyncClient(timeout=5.0, headers=cli_auth_headers()) as client: @@ -331,7 +336,6 @@ async def sync_pro_routing(settings_obj) -> None: the bearer (activate, sign-in, sign-out, disconnect, free-trial arm/clear). Never raises.""" try: - from backend.apps.settings.credentials import proxy_auth bearer, base = proxy_auth(settings_obj) active = bool(bearer) await sync_openswarm_pro_as_claude( diff --git a/backend/apps/outputs/execution/ExecutionContract.py b/backend/apps/outputs/execution/ExecutionContract.py new file mode 100644 index 000000000..aa71fc1ad --- /dev/null +++ b/backend/apps/outputs/execution/ExecutionContract.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from typing import Any, Dict, Literal, Optional, Protocol, runtime_checkable +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field +from typeguard import typechecked + + +ExecutionStatus = Literal[ + "success", + "rejected", + "timed_out", + "cancelled", + "resource_exhausted", + "internal_failure", +] + + +class ExecutionLimits(BaseModel): + model_config = ConfigDict(validate_assignment=True, extra="forbid") + + wall_time_ms: int = Field(default=30_000, ge=1, le=300_000) + cpu_time_ms: int = Field(default=30_000, ge=1, le=300_000) + memory_bytes: int = Field(default=268_435_456, ge=16_777_216, le=2_147_483_648) + process_count: int = Field(default=16, ge=1, le=256) + disk_bytes: int = Field(default=67_108_864, ge=0, le=2_147_483_648) + output_bytes: int = Field(default=1_048_576, ge=1, le=16_777_216) + + +class ExecutionRequest(BaseModel): + model_config = ConfigDict(validate_assignment=True, extra="forbid") + + execution_id: str = Field(default_factory=lambda: uuid4().hex, min_length=32, max_length=32) + policy_version: str = Field(default="f2a-v1", min_length=1, max_length=64) + code: str = Field(min_length=1, max_length=1_000_000) + input_data: Dict[str, Any] = Field(default_factory=dict) + validation_mode: Literal["strict", "user_approved"] = "strict" + egress_policy: Literal["deny", "provider_defined"] = "deny" + limits: ExecutionLimits = Field(default_factory=ExecutionLimits) + + +class ExecutionResult(BaseModel): + model_config = ConfigDict(validate_assignment=True, extra="forbid") + + execution_id: str + status: ExecutionStatus + result: Dict[str, Any] = Field(default_factory=dict) + stdout: str = Field(default="", max_length=1_048_576) + stderr: str = Field(default="", max_length=1_048_576) + duration_ms: int = Field(default=0, ge=0) + provider: str = Field(min_length=1, max_length=128) + runtime_identity: str = Field(min_length=1, max_length=256) + isolation_enforced: bool + resource_limit_reason: Optional[str] = Field(default=None, max_length=128) + error: Optional[str] = Field(default=None, max_length=1_000) + + +@runtime_checkable +class ExecutorPort(Protocol): + @typechecked + async def execute(self, request: ExecutionRequest) -> ExecutionResult: + ... diff --git a/backend/apps/outputs/execution/SubprocessExecutor.py b/backend/apps/outputs/execution/SubprocessExecutor.py new file mode 100644 index 000000000..724265420 --- /dev/null +++ b/backend/apps/outputs/execution/SubprocessExecutor.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import sys +import time + +from pydantic import BaseModel, ConfigDict +from typeguard import typechecked + +from backend.apps.outputs.execution.ExecutionContract import ExecutionRequest, ExecutionResult, ExecutionStatus +from backend.apps.outputs.code_safety import UnsafeCodeError +from backend.apps.outputs.executor import execute_backend_code + + +class SubprocessExecutor(BaseModel): + model_config = ConfigDict(validate_assignment=True) + + provider_name: str = "desktop-legacy-subprocess" + + @typechecked + async def execute(self, request: ExecutionRequest) -> ExecutionResult: + started = time.monotonic() + try: + result = await execute_backend_code( + request.code, + request.input_data, + approved=request.validation_mode == "user_approved", + ) + return ExecutionResult( + execution_id=request.execution_id, + status="success", + result=result.result, + stdout=result.stdout, + stderr=result.stderr, + duration_ms=int((time.monotonic() - started) * 1000), + provider=self.provider_name, + runtime_identity=sys.version, + isolation_enforced=False, + ) + except UnsafeCodeError as exc: + return self.failure(request, "rejected", started, str(exc)) + except Exception as exc: + status: ExecutionStatus = "timed_out" if "timed out" in str(exc).lower() else "internal_failure" + return self.failure(request, status, started, str(exc)) + + @typechecked + def failure( + self, + request: ExecutionRequest, + status: ExecutionStatus, + started: float, + error: str, + ) -> ExecutionResult: + return ExecutionResult( + execution_id=request.execution_id, + status=status, + duration_ms=int((time.monotonic() - started) * 1000), + provider=self.provider_name, + runtime_identity=sys.version, + isolation_enforced=False, + error=error[:1_000], + ) diff --git a/backend/apps/outputs/execution/__init__.py b/backend/apps/outputs/execution/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/apps/outputs/executor.py b/backend/apps/outputs/executor.py index c79da34b9..5d9600efe 100644 --- a/backend/apps/outputs/executor.py +++ b/backend/apps/outputs/executor.py @@ -6,11 +6,12 @@ import tempfile from dataclasses import dataclass -from backend.apps.outputs.code_safety import ALLOWED_MODULES, validate_code_safety +from backend.sandbox_policy import SANDBOX_POLICY +from backend.apps.outputs.code_safety import validate_code_safety logger = logging.getLogger(__name__) -TIMEOUT_SECONDS = 30 +TIMEOUT_SECONDS = SANDBOX_POLICY.timeout_seconds # Env vars we always scrub from the subprocess, approved or not. These are the keys an attacker would actually want; install token, provider API keys, cloud credentials. Everything else is local-machine convenience. P_SCRUBBED_ENV_KEYS = frozenset({ @@ -27,7 +28,18 @@ "STRIPE_API_KEY", "STRIPE_SECRET_KEY", "GITHUB_TOKEN", + # Hosted deployment secrets (control store, edge storage, database): approved code on a hosted backend must never see them either. + "DATABASE_URL", + "PGPASSWORD", + "POSTGRES_PASSWORD", + "OPENSWARM_EDGE_STORAGE_PATH", }) +# Whole families a hosting deployment injects (lent provider keys, demo passwords, edge auth): scrubbed by prefix so a new secret in the family never needs a list entry. +P_SCRUBBED_ENV_PREFIXES = ("OPENSWARM_DEMO_", "OPENSWARM_EDGE_", "EDGE_AUTH_") + + +def p_env_is_secret(key: str) -> bool: + return key in P_SCRUBBED_ENV_KEYS or key.startswith(P_SCRUBBED_ENV_PREFIXES) def exec_env(approved: bool = False) -> dict: @@ -47,7 +59,7 @@ def exec_env(approved: bool = False) -> dict: install token or provider API keys. """ if approved: - env = {k: v for k, v in os.environ.items() if k not in P_SCRUBBED_ENV_KEYS} + env = {k: v for k, v in os.environ.items() if not p_env_is_secret(k)} env["PYTHONDONTWRITEBYTECODE"] = "1" # Force UTF-8 even if the parent somehow lacked it (dev mode where Electron didn't inject PYTHONUTF8). Without this, a child reading non-ASCII stdin/files on a cp1252 Windows machine raises UnicodeDecodeError, the "works on my laptop, not theirs" failure. env["PYTHONUTF8"] = "1" @@ -69,28 +81,6 @@ def exec_env(approved: bool = False) -> dict: return env -# The subprocess bootstrap: capture stdout, read the input, and hand the code a `result` to fill. Bound objects rather than modules carry the answer back out, so the hardening below can take the modules away. -P_PREAMBLE = ( - "import json, sys, io, builtins\n" - "p_stdout = sys.stdout\n" - "p_capture = io.StringIO()\n" - "sys.stdout = p_capture\n" - "input_data = json.loads(sys.stdin.read())\n" - "result = {}\n" -) -# Warm the allowlist BEFORE scrubbing builtins, because half the stdlib borrows the builtins the scrub deletes while it loads (tokenize does `from builtins import open`, which is how `import dataclasses` dies). Warm imports are cache hits, so gate-passing code never touches the loader again. Then the module handles go: leaving `sys` bound handed gate-passing code a live `sys.modules['os']` with no import statement in sight. exec/eval/compile/__import__ stay put whatever we'd like: the import statement, the loader and namedtuple all run on them, and calling them by name is a static-gate warning anyway. -P_SANDBOX_HARDENING = ( - f"for p_name in {tuple(sorted(ALLOWED_MODULES))!r}:\n" - " try: __import__(p_name)\n" - " except ImportError: pass\n" - "for p_name in ('open','input','breakpoint','exit','quit'):\n" - " try: delattr(builtins, p_name)\n" - " except AttributeError: pass\n" - "del sys, io, builtins, p_name\n" -) -P_POSTAMBLE = ( - "\np_stdout.write(json.dumps({\"__stdout__\": p_capture.getvalue(), \"__result__\": result}))\n" -) @dataclass @@ -112,21 +102,23 @@ async def execute_backend_code( Security boundaries (defense in depth; none alone is sufficient): 1. The static gate in code_safety.py, on every run that isn't approved. 2. Subprocess cwd = fresh temp dir (not the OpenSwarm process cwd). - 3. Subprocess env strips PATH, all *TOKEN / *_API_KEY inheritance. - 4. Preamble scrubs the I/O builtins and drops the module handles it - needed, so gate-passing code starts with no reachable module. + 3. Subprocess env is built from an explicit runtime allowlist. + 4. SANDBOX_POLICY's preamble warms the allowlist, scrubs the I/O builtins and + drops the module handles it needed, so gate-passing code starts with no + reachable module (AST-bypass tricks land on nothing). 5. 30s wall-clock timeout, killed on overrun. `approved=True` means a user saw the warnings and clicked Run Anyway, and it relaxes 1, 3 and 4 together. It is the ONLY thing that relaxes them: a caller that has already run the gate itself still gets the sandbox, because - "we checked" must never be the reason the walls come down. + "we checked" must never be the reason the walls come down. (Decision + 2026-08-16: upstream's App Builder semantics; the edge never approves.) """ if not approved: validate_code_safety(code) - wrapper = P_PREAMBLE + ("" if approved else P_SANDBOX_HARDENING) + code + P_POSTAMBLE + wrapper = SANDBOX_POLICY.wrap(code, approved=approved) with tempfile.TemporaryDirectory(prefix="openswarm-exec-") as workdir: proc = await asyncio.create_subprocess_exec( diff --git a/backend/apps/outputs/models.py b/backend/apps/outputs/models.py index 9a65f53c4..e6cd45ece 100644 --- a/backend/apps/outputs/models.py +++ b/backend/apps/outputs/models.py @@ -1,15 +1,19 @@ -import os - -from pydantic import BaseModel, Field, computed_field, model_validator +from pydantic import BaseModel, Field, computed_field, field_validator, model_validator from typing import Literal, Optional, Any from uuid import uuid4 from datetime import datetime +from backend.apps.outputs.path_security import ( + validate_output_id, + validate_workspace_id, + workspace_directory, +) from backend.config.paths import OUTPUTS_WORKSPACE_DIR class Output(BaseModel): id: str = Field(default_factory=lambda: uuid4().hex) + owner_account_id: Optional[str] = None name: str description: str = "" icon: str = "view_quilt" @@ -33,6 +37,16 @@ class Output(BaseModel): publish_status: Optional[Literal["publishing", "published", "error"]] = None publish_error: Optional[str] = None + @field_validator("id") + @classmethod + def p_validate_id(cls, value: str) -> str: + return validate_output_id(value) + + @field_validator("workspace_id") + @classmethod + def p_validate_workspace_id(cls, value: Optional[str]) -> Optional[str]: + return validate_workspace_id(value) if value is not None else None + @model_validator(mode="before") @classmethod def p_migrate_flat_fields(cls, data: Any) -> Any: @@ -59,7 +73,7 @@ def workspace_path(self) -> str: """Absolute on-disk folder for this app, resolved from workspace_id. API-only (excluded from the saved JSON since it's machine-specific and re-derivable); lets the frontend show the real edit path when an App card is selected.""" if not self.workspace_id: return "" - return os.path.abspath(os.path.join(OUTPUTS_WORKSPACE_DIR, self.workspace_id)) + return workspace_directory(OUTPUTS_WORKSPACE_DIR, self.workspace_id) @property def frontend_code(self) -> str: @@ -84,6 +98,7 @@ class OutputVersion(BaseModel): class OutputCreate(BaseModel): + owner_account_id: Optional[str] = None name: str description: str = "" icon: str = "view_quilt" @@ -97,6 +112,11 @@ class OutputCreate(BaseModel): session_id: Optional[str] = None workspace_id: Optional[str] = None + @field_validator("workspace_id") + @classmethod + def p_validate_workspace_id(cls, value: Optional[str]) -> Optional[str]: + return validate_workspace_id(value) if value is not None else None + @model_validator(mode="before") @classmethod def p_migrate_flat_fields(cls, data: Any) -> Any: @@ -118,6 +138,7 @@ def p_migrate_flat_fields(cls, data: Any) -> Any: class OutputUpdate(BaseModel): + owner_account_id: Optional[str] = None name: Optional[str] = None description: Optional[str] = None icon: Optional[str] = None @@ -127,6 +148,11 @@ class OutputUpdate(BaseModel): session_id: Optional[str] = None workspace_id: Optional[str] = None + @field_validator("workspace_id") + @classmethod + def p_validate_workspace_id(cls, value: Optional[str]) -> Optional[str]: + return validate_workspace_id(value) if value is not None else None + @model_validator(mode="before") @classmethod def p_migrate_flat_fields(cls, data: Any) -> Any: @@ -183,6 +209,11 @@ class WorkspaceSeedRequest(BaseModel): # "webapp_template" (default) → seed the vendored openswarm-ai/webapp-template snapshot (React + Vite + TS frontend with optional FastAPI backend), allocate a free FRONTEND_PORT, leave BACKEND_PORT=NONE. Runtime spawns `bash run.sh`; preview pane points at `http://localhost:{FRONTEND_PORT}/`. "flat" → legacy single-`index.html` workspace, kept for explicit opt-in (migration helper, regression tests). Workspaces predating this flip continue to work in old-mode automatically since the runtime detects mode via the presence of `run.sh`. template_mode: Literal["flat", "webapp_template"] = "webapp_template" + @field_validator("workspace_id") + @classmethod + def p_validate_workspace_id(cls, value: str) -> str: + return validate_workspace_id(value) + @model_validator(mode="before") @classmethod def p_migrate_flat_fields(cls, data: Any) -> Any: diff --git a/backend/apps/outputs/outputs.py b/backend/apps/outputs/outputs.py index 6af4b3477..cadbb87cf 100644 --- a/backend/apps/outputs/outputs.py +++ b/backend/apps/outputs/outputs.py @@ -1,26 +1,32 @@ import asyncio import json +import importlib.util import os import logging import mimetypes +import ntpath from datetime import datetime -from typing import Optional +from typing import Annotated, Optional from contextlib import asynccontextmanager -from fastapi import HTTPException, Query +from urllib.parse import unquote +from fastapi import HTTPException, Query, Request from fastapi.responses import Response from backend.auth import get_auth_token from backend.config.Apps import SubApp +from backend.apps.hosting.policy import REQUEST_SCOPE, RequestScope, hosting_policy from backend.apps.outputs.models import ( Output, OutputCreate, OutputUpdate, OutputExecute, OutputExecuteResult, VibeCodeRequest, WorkspaceSeedRequest, AgentCreateAppRequest, PublishPreflightRequest, PublishRequest, PublishPreflightResponse, - PublishResult, PublishReview, + PublishResult, ) +from backend.apps.outputs.path_security import contained_path, workspace_directory from backend.apps.outputs.code_safety import get_code_warnings -from backend.apps.outputs.executor import execute_backend_code -from backend.apps.outputs.publish_capability import check_publish_capability +from backend.apps.outputs.execution.ExecutionContract import ExecutionRequest +from backend.apps.outputs.execution.SubprocessExecutor import SubprocessExecutor from backend.apps.outputs.publish_common import slugify, PublishError -from backend.apps.outputs.publish_scan import scan_for_publish, quick_ast_gate +from backend.apps.outputs.publish_scan import scan_for_publish +from backend.apps.outputs.publish_capability import check_publish_capability from backend.apps.outputs.publish_build import build_static, collect_bundle from backend.apps.outputs.publish_cloud import upload_to_cloud, unpublish_from_cloud from backend.apps.outputs.view_builder_templates import ( @@ -33,7 +39,6 @@ from backend.apps.outputs.html_inject import ( get_anthropic_client, validate_against_schema, - build_data_injection, inject_data_into_html, backend_url_for_workspace, inject_token_into_relative_urls, @@ -43,8 +48,7 @@ load_all, save, load, - load_output, - resolve_in_workspace, + output_metadata_path, walk_directory, workspace_root, would_shrink_oversize_file, @@ -52,6 +56,46 @@ from backend.apps.outputs.prompts import VIBE_CODE_SYSTEM_PROMPT logger = logging.getLogger(__name__) +p_executor = SubprocessExecutor() + +AttachmentId = Annotated[ + Optional[str], + Query( + min_length=1, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", + ), +] + + +def workspace_folder(workspace_id: str) -> str: + try: + return workspace_directory(WORKSPACE_DIR, workspace_id) + except (TypeError, ValueError): + raise HTTPException(status_code=403, detail="Invalid workspace ID") + + +def p_workspace_file_path(folder: str, filepath: str) -> str: + decoded = filepath + for _ in range(8): + next_value = unquote(decoded) + if next_value == decoded: + break + decoded = next_value + else: + raise HTTPException(status_code=403, detail="Path traversal not allowed") + + drive, _ = ntpath.splitdrive(decoded) + if "\x00" in decoded or drive or ntpath.isabs(decoded): + raise HTTPException(status_code=403, detail="Path traversal not allowed") + platform_path = decoded.replace("\\", os.sep).replace("/", os.sep) + try: + full_path = contained_path(folder, platform_path) + except ValueError: + raise HTTPException(status_code=403, detail="Path traversal not allowed") + if os.path.normcase(os.path.realpath(full_path)) == os.path.normcase(os.path.realpath(folder)): + raise HTTPException(status_code=403, detail="Path traversal not allowed") + return full_path @asynccontextmanager @@ -108,15 +152,34 @@ async def p_periodic_sweep() -> None: outputs = SubApp("outputs", outputs_lifespan) -# --------------------------------------------------------------------------- File-serving endpoints (for iframe preview with multi-file support) --------------------------------------------------------------------------- +def p_workspace_output(workspace_id: str) -> Output | None: + return next((output for output in load_all() if output.workspace_id == workspace_id), None) + + +def p_require_output_owner(output: Output, scope: RequestScope) -> None: + scope.require_owner_of(output.owner_account_id) + + +def p_require_workspace_owner(workspace_id: str, scope: RequestScope) -> None: + # Only a hosted build owns workspaces; on the desktop this must not cost a load_all() per request. + if not scope.hosted: + return + output = p_workspace_output(workspace_id) + if not output: + raise HTTPException(status_code=404, detail="Workspace not found") + scope.require_owner_of(output.owner_account_id) + + +# --------------------------------------------------------------------------- +# File-serving endpoints (for iframe preview with multi-file support) +# --------------------------------------------------------------------------- @outputs.router.get("/workspace/{workspace_id}/serve/{filepath:path}") -async def serve_workspace_file(workspace_id: str, filepath: str, p_d: str = ""): +async def serve_workspace_file(workspace_id: str, filepath: str, p_d: str = "", scope: RequestScope = REQUEST_SCOPE): """Serve a file from a workspace folder. For index.html, inject OUTPUT data.""" - folder = os.path.join(WORKSPACE_DIR, workspace_id) - full_path = resolve_in_workspace(folder, filepath) - if full_path is None: - raise HTTPException(status_code=403, detail="Path traversal not allowed") + p_require_workspace_owner(workspace_id, scope) + folder = workspace_folder(workspace_id) + full_path = p_workspace_file_path(folder, filepath) if not os.path.isfile(full_path): raise HTTPException(status_code=404, detail="File not found") @@ -124,11 +187,13 @@ async def serve_workspace_file(workspace_id: str, filepath: str, p_d: str = ""): if filepath.endswith("index.html"): with open(full_path) as f: content = f.read() - input_json, result_json = decode_data_param(p_d) if p_d else ("{}", "null") + data_param = p_d + input_json, result_json = decode_data_param(data_param) if data_param else ("{}", "null") backend_url_json = backend_url_for_workspace(workspace_id) content = inject_data_into_html(content, input_json, result_json, backend_url_json, with_runtime=True) # Iframe sub-resource fetches (,