diff --git a/backend/apps/agents/agent_manager.py b/backend/apps/agents/agent_manager.py index 0195e04a..8447c8d4 100644 --- a/backend/apps/agents/agent_manager.py +++ b/backend/apps/agents/agent_manager.py @@ -37,6 +37,7 @@ 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") +MAX_BACKGROUND_TURNS = int(os.environ.get("OSW_MAX_BACKGROUND_TURNS", "2") or "0") class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionControl, AgentLaunch, SpawnAgentRun, MockAgent, TurnRunner, RunOptions, RunSupport): @@ -56,6 +57,9 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr # 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 + # Background lane: event-trigger checks admit here so they can never queue the user's own turns. + self.p_bg_admission_sema: Optional[asyncio.Semaphore] = None + self.p_bg_admission_loop: Optional[asyncio.AbstractEventLoop] = None @typechecked @@ -68,6 +72,14 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr self.p_turn_admission_loop = loop return self.p_turn_admission_sema + @typechecked + def get_background_admission(self) -> asyncio.Semaphore: + loop = asyncio.get_running_loop() + if self.p_bg_admission_sema is None or self.p_bg_admission_loop is not loop: + self.p_bg_admission_sema = asyncio.Semaphore(max(1, MAX_BACKGROUND_TURNS)) + self.p_bg_admission_loop = loop + return self.p_bg_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 @@ -79,6 +91,11 @@ class AgentManager(SessionLifecycle, SessionPersistence, Messaging, SessionContr if MAX_CONCURRENT_TURNS <= 0 or session.parent_session_id is not None: yield return + if session.background: + # Checks queue among themselves in the small background lane, never against the user's turns; no queued/admitted frames since these sessions are invisible. + async with self.get_background_admission(): + yield + return sema = self.get_turn_admission() was_queued = sema.locked() if was_queued: diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py index e1688c4c..e842f6bc 100644 --- a/backend/apps/agents/agents.py +++ b/backend/apps/agents/agents.py @@ -65,7 +65,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) - return {"sessions": [p_session_list_item(s) for s in sessions]} + # Background plumbing sessions (event-trigger checks) never surface as cards. + return {"sessions": [p_session_list_item(s) for s in sessions if not s.background]} @agents.router.get("/activity") async def agent_activity(): diff --git a/backend/apps/agents/core/models.py b/backend/apps/agents/core/models.py index 48bc5bc3..15a53184 100644 --- a/backend/apps/agents/core/models.py +++ b/backend/apps/agents/core/models.py @@ -17,6 +17,8 @@ class AgentConfig(BaseModel): 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". selected_app_output_ids: Optional[list[str]] = None + # Background plumbing (event-trigger checks): never surfaces cards or WS frames, and admits through the small background lane instead of competing with the user's turns. + background: bool = False class ApprovalRequest(BaseModel): id: str = Field(default_factory=lambda: uuid4().hex) @@ -82,6 +84,8 @@ class AgentSession(BaseModel): model: str = "sonnet" mode: str = "agent" sdk_session_id: Optional[str] = None + # Background plumbing session (see AgentConfig.background): invisible and background-lane admitted. + background: bool = False system_prompt: Optional[str] = None allowed_tools: list[str] = Field(default_factory=list) max_turns: Optional[int] = None diff --git a/backend/apps/agents/core/ws_manager.py b/backend/apps/agents/core/ws_manager.py index 6996e5cc..a78452b4 100644 --- a/backend/apps/agents/core/ws_manager.py +++ b/backend/apps/agents/core/ws_manager.py @@ -73,6 +73,14 @@ class ConnectionManager: self.browser_futures: dict[str, asyncio.Future] = {} # 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 + # Background plumbing sessions (event-trigger checks): no frames ever leave for these, so no card can flash on the canvas. + self.background_session_ids: set[str] = set() + + def mark_background(self, session_id: str) -> None: + self.background_session_ids.add(session_id) + + def unmark_background(self, session_id: str) -> None: + self.background_session_ids.discard(session_id) async def connect_session(self, session_id: str, websocket: WebSocket): await websocket.accept() @@ -117,6 +125,8 @@ class ConnectionManager: 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.""" + if session_id in self.background_session_ids: + return data = slim_status_data(event, data) async with seq_log.stamp(session_id, event, data) as (seq, payload_str): for ws in list(self.connections.get(session_id, [])): diff --git a/backend/apps/agents/manager/AgentLaunch.py b/backend/apps/agents/manager/AgentLaunch.py index 83c5057d..3faccb59 100644 --- a/backend/apps/agents/manager/AgentLaunch.py +++ b/backend/apps/agents/manager/AgentLaunch.py @@ -117,11 +117,15 @@ class AgentLaunch(AgentManagerProtocol): dashboard_id=config.dashboard_id, workflow_run_id=config.workflow_run_id, workflow_edit_id=config.workflow_edit_id, + background=config.background, thinking_level=getattr(global_settings, "default_thinking_level", "auto"), ) apply_context_window(session, global_settings) self.sessions[session_id] = session + # Background sessions are invisible plumbing: mark BEFORE any frame so no card ever flashes. + if session.background: + ws_manager.mark_background(session_id) await ws_manager.send_to_session(session_id, "agent:status", { "session_id": session_id, @@ -129,11 +133,12 @@ class AgentLaunch(AgentManagerProtocol): "session": session.model_dump(mode="json"), }) - try: - from backend.apps.service.analytics.client import track_agent_created - track_agent_created(id=session.id, dashboard_id=session.dashboard_id) - except Exception: - pass + if not session.background: + try: + from backend.apps.service.analytics.client import track_agent_created + track_agent_created(id=session.id, dashboard_id=session.dashboard_id) + except Exception: + pass return session diff --git a/backend/apps/events/adapters/agent_check.py b/backend/apps/events/adapters/agent_check.py index adad0a94..0b0554f4 100644 --- a/backend/apps/events/adapters/agent_check.py +++ b/backend/apps/events/adapters/agent_check.py @@ -108,7 +108,7 @@ async def run_check_turn( from backend.apps.agents.manager.session.session_store import delete_session_file session = await agent_manager.launch_agent(AgentConfig( - name="Event check", model=model, mode="agent", dashboard_id=dashboard_id, + name="Event check", model=model, mode="agent", dashboard_id=dashboard_id, background=True, )) if active_mcps: session.active_mcps = list(active_mcps) @@ -124,6 +124,11 @@ async def run_check_turn( await agent_manager.send_message(session.id, prompt) return await p_await_reply(session.id) finally: + try: + from backend.apps.agents.core.ws_manager import ws_manager + ws_manager.unmark_background(session.id) + except Exception: + pass try: clear_workflow_approval_memory(session.id) except Exception: diff --git a/backend/tests/test_turn_admission.py b/backend/tests/test_turn_admission.py index 5186ce2c..a3ed0dfa 100644 --- a/backend/tests/test_turn_admission.py +++ b/backend/tests/test_turn_admission.py @@ -143,3 +143,56 @@ def test_semaphore_rebuilds_per_loop(): s1 = asyncio.run(get_sema()) s2 = asyncio.run(get_sema()) assert s1 is not s2 + + +def make_background_session() -> AgentSession: + s = AgentSession(name="Event check", model="haiku", mode="agent") + s.background = True + return s + + +def test_background_lane_never_queues_user_turns(): + """The lane split: checks saturating their own small pool must not block a root turn, and + background turns never occupy main slots.""" + async def run(): + with Harness(cap=1) as h: + old_bg = am.MAX_BACKGROUND_TURNS + am.MAX_BACKGROUND_TURNS = 1 + try: + bg_entered = asyncio.Event() + bg_release = asyncio.Event() + + async def one_background() -> None: + async with h.mgr.turn_admission_slot(make_background_session(), "bg1"): + bg_entered.set() + await bg_release.wait() + + bg_task = asyncio.create_task(one_background()) + await bg_entered.wait() + + # Background pool is saturated (cap 1); a user root still admits instantly. + user_ran = False + async with h.mgr.turn_admission_slot(make_session(), "user1"): + user_ran = True + assert user_ran + assert not any(e == ("user1", "agent:queued") for e in h.events) + + # A second background turn queues behind the first, bounded by ITS lane. + second_done = asyncio.Event() + + async def second_background() -> None: + async with h.mgr.turn_admission_slot(make_background_session(), "bg2"): + second_done.set() + + second_task = asyncio.create_task(second_background()) + await asyncio.sleep(0.05) + assert not second_done.is_set() # held by the background lane, not running + bg_release.set() + await asyncio.wait_for(second_task, timeout=2) + await asyncio.wait_for(bg_task, timeout=2) + # Invisible plumbing: background turns emit no queue frames at all. + assert not any(e[0].startswith("bg") for e in h.events) + finally: + am.MAX_BACKGROUND_TURNS = old_bg + + asyncio.run(run())