[eric] events: background check sessions (invisible, own admission lane, never queue user turns)

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:

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():

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

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, [])):

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
 

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:

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())