feat(monitoring): add shared SSE in-flight monitor with draggable dock

src/langbot/pkg/agent/runner/run_ledger_store.py

@@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 from sqlalchemy.orm import sessionmaker
 
 from ...persistence.datetime_utils import as_naive_utc
+from ...utils.inflight import inflight_hub
 
 from ...entity.persistence.agent_run import AgentRun, AgentRunEvent, AgentRuntime
 
@@ -149,6 +150,7 @@ class RunLedgerStore:
             )
             session.add(run)
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run)
 
     async def claim_next_run(
@@ -206,6 +208,7 @@ class RunLedgerStore:
             run.last_claimed_at = as_naive_utc(now)
             run.updated_at = as_naive_utc(now)
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run, include_claim_token=True)
 
     async def renew_claim(
@@ -226,6 +229,7 @@ class RunLedgerStore:
             run.claim_lease_expires_at = as_naive_utc(now + datetime.timedelta(seconds=max(int(lease_seconds), 1)))
             run.updated_at = as_naive_utc(now)
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run)
 
     async def release_claim(
@@ -254,6 +258,7 @@ class RunLedgerStore:
             if status in TERMINAL_STATUSES:
                 run.finished_at = run.finished_at or as_naive_utc(now)
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run)
 
     async def release_expired_claims(
@@ -332,6 +337,7 @@ class RunLedgerStore:
             )
             session.add(row)
             await session.commit()
+            inflight_hub.notify_run(run_id, event_type)
             return self._event_to_dict(row)
 
     async def append_audit_event(
@@ -366,6 +372,7 @@ class RunLedgerStore:
             )
             session.add(row)
             await session.commit()
+            inflight_hub.notify_run(run_id, event_type)
             return self._event_to_dict(row)
 
     async def finalize_run(
@@ -410,6 +417,7 @@ class RunLedgerStore:
                 else:
                     run.metadata_json = _json_dumps(metadata)
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run)
 
     async def validate_active_claim(
@@ -443,6 +451,7 @@ class RunLedgerStore:
             run.updated_at = as_naive_utc(now)
             run.status_reason = status_reason or run.status_reason
             await session.commit()
+            inflight_hub.notify(run.workspace_id)
             return self._run_to_dict(run)
 
     async def get_run(self, run_id: str) -> dict[str, typing.Any] | None:

src/langbot/pkg/api/http/controller/groups/monitoring.py

@@ -59,6 +59,31 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             return self.success(data=metrics)
 
+        @self.route('/in-flight/stream', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        async def stream_inflight(request_context: RequestContext):
+            from .....utils.inflight import inflight_hub
+            from ...service.tenant import require_workspace_uuid
+            import json
+
+            workspace = require_workspace_uuid(request_context)
+            previous = []
+
+            async def snapshot():
+                nonlocal previous
+                result = await self.ap.monitoring_service.get_inflight_snapshot(request_context, previous)
+                previous = [row for row in result['items'] if row['status_group'] in {'running', 'queued'}]
+                return result
+
+            async def frames():
+                async for frame in inflight_hub.watch(workspace, snapshot):
+                    yield 'data: ' + json.dumps(frame, ensure_ascii=False) + '\n\n'
+
+            response = quart.Response(frames(), content_type='text/event-stream')
+            response.headers['Cache-Control'] = 'no-cache, no-store'
+            response.headers['X-Accel-Buffering'] = 'no'
+            response.timeout = None
+            return response
+
         @self.route('/executions', methods=['GET'], permission=Permission.RESOURCE_VIEW)
         async def get_executions(request_context: RequestContext) -> str:
             """Unified execution list across agent runs and pipeline queries."""

src/langbot/pkg/api/http/service/monitoring.py

@@ -127,8 +127,13 @@ def _workspace_transaction(method):
         tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
         if callable(tenant_uow):
             async with tenant_uow(workspace_uuid):
-                return await method(self, context, *args, **kwargs)
-        return await method(self, context, *args, **kwargs)
+                result = await method(self, context, *args, **kwargs)
+        else:
+            result = await method(self, context, *args, **kwargs)
+        if method.__name__ in {'record_message', 'update_message_status'}:
+            from ....utils.inflight import inflight_hub
+            inflight_hub.notify(workspace_uuid)
+        return result
 
     return wrapped
 
@@ -1533,6 +1538,7 @@ class MonitoringService(ExecutionDetailsMixin):
         end_time: datetime.datetime | None = None,
         limit: int = 50,
         offset: int = 0,
+        include_summary: bool = True,
     ) -> dict:
         """List executions from both object types (agent runs and pipeline queries)."""
         workspace_uuid = require_workspace_uuid(context)
@@ -1633,6 +1639,8 @@ class MonitoringService(ExecutionDetailsMixin):
                     by_key[(kind, item['id'])] = item
             page = [by_key[(ref.source, ref.id)] for ref in refs if (ref.source, ref.id) in by_key]
             await self.enrich_execution_rows(workspace_uuid, page)
+        if not include_summary:
+            return {'items': page, 'total': total, 'has_more': total > offset + len(page)}
         return {
             'items': page,
             'total': total,
@@ -1661,6 +1669,54 @@ class MonitoringService(ExecutionDetailsMixin):
             },
         }
 
+    async def get_inflight_snapshot(self, context: TenantContext, previous: list[dict] | None = None) -> dict:
+        """Bounded Workspace snapshot without dashboard aggregates or token payloads."""
+        from ....utils.inflight import inflight_hub
+
+        workspace = require_workspace_uuid(context)
+        active = await self.get_executions(context, statuses=['running', 'queued'], limit=100, include_summary=False)
+        recent = await self.get_executions(
+            context, start_time=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=30),
+            limit=100, include_summary=False,
+        )
+        items = {(r['source'], r['id']): r for r in recent['items'] + active['items'] if r['source'] != 'event'}
+        # Keep the actual terminal state of older runs that just left the active set.
+        for kind, model, column, scope in [
+            ('agent', persistence_agent_run.AgentRun, persistence_agent_run.AgentRun.run_id, persistence_agent_run.AgentRun.workspace_id),
+            ('pipeline', persistence_monitoring.MonitoringMessage, persistence_monitoring.MonitoringMessage.id, persistence_monitoring.MonitoringMessage.workspace_uuid),
+        ]:
+            missing = [r['id'] for r in (previous or [])[:100] if r['source'] == kind and (kind, r['id']) not in items]
+            if missing:
+                records = await self._execution_rows(workspace, sqlalchemy.select(model).where(scope == workspace, column.in_(missing)))
+                names = await self._agent_names(workspace, records) if kind == 'agent' else {}
+                for record in records:
+                    row = self._serialize_agent_execution(record, names) if kind == 'agent' else self._serialize_pipeline_execution(record)
+                    items[(kind, row['id'])] = row
+        rows = list(items.values())
+        Run = persistence_agent_run.AgentRun
+        Message = persistence_monitoring.MonitoringMessage
+        agent_ids = [r['id'] for r in rows if r['source'] == 'agent']
+        pipeline_ids = [r['id'] for r in rows if r['source'] == 'pipeline']
+        linked = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(Run.run_id, Message.id.label('message_id')).outerjoin(
+                Message, sqlalchemy.and_(Message.workspace_uuid == workspace, Message.run_id == Run.run_id)
+            ).where(Run.workspace_id == workspace, sqlalchemy.or_(Run.run_id.in_(agent_ids), Message.id.in_(pipeline_ids)))
+        )
+        identities = {r.run_id: ('pipeline', r.message_id) if r.message_id in pipeline_ids else ('agent', r.run_id) for r in linked.all()}
+        inflight_hub.track(workspace, identities)
+        Event = persistence_agent_run.AgentRunEvent
+        latest = sqlalchemy.select(Event.run_id, sqlalchemy.func.max(Event.sequence).label('seq')).where(
+            Event.run_id.in_(identities)
+        ).group_by(Event.run_id).subquery()
+        phases = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(Event.run_id, Event.type).join(latest, sqlalchemy.and_(Event.run_id == latest.c.run_id, Event.sequence == latest.c.seq))
+        )
+        for run_id, event_type in phases.all():
+            row = items.get(identities[run_id])
+            if row is not None:
+                row['progress_event'] = event_type
+        return {'items': rows, 'active_total': active['total'], 'truncated': active['has_more'] or recent['has_more']}
+
     async def get_token_statistics(
         self,
         context: TenantContext,

src/langbot/pkg/api/mcp/server.py

@@ -207,6 +207,10 @@ class LangBotMCPServer:
                 return _dump(await ap.monitoring_service.get_message_details(context, identifier))
             return _dump(await ap.monitoring_service.get_session_analysis(context, identifier))
 
+        @mcp.tool(description='Read a bounded snapshot of active and recently started Workspace executions, including the latest progress event. No percentage is estimated.')
+        async def get_inflight_executions() -> str:
+            return _dump(await ap.monitoring_service.get_inflight_snapshot(_authorized(Permission.RESOURCE_VIEW)))
+
         @mcp.tool(
             description='List Workspace executions, including events that did not start a processor. The legacy pipeline_ids filter accepts Agent, Pipeline and event processor IDs. Returns bounded rows and summary metrics.'
         )

src/langbot/pkg/utils/inflight.py

@@ -0,0 +1,94 @@
+"""Bounded, process-local fan-out. One snapshot producer per watched Workspace.
+
+Writes wake the producer; quiet workspaces reconcile every 15 seconds to cover
+external workers. A shared broker can replace notify without changing clients.
+No payloads or subscriptions survive the final viewer disconnecting.
+"""
+
+import asyncio
+import contextlib
+import time
+
+
+class InflightHub:
+    def __init__(self):
+        self.channels = {}
+        self.run_workspaces = {}
+
+    def notify(self, workspace):
+        channel = self.channels.get(workspace)
+        if channel:
+            channel['dirty'].set()
+
+    def notify_run(self, run_id, event_type):
+        workspace = self.run_workspaces.get(run_id)
+        channel = self.channels.get(workspace)
+        if channel and channel['phases'].get(run_id) != event_type:
+            channel['phases'][run_id] = event_type
+            channel['dirty'].set()
+
+    def track(self, workspace, run_ids):
+        self.run_workspaces = {key: value for key, value in self.run_workspaces.items() if value != workspace}
+        if workspace not in self.channels:
+            return
+        self.run_workspaces.update({run_id: workspace for run_id in run_ids})
+        if workspace in self.channels:
+            phases = self.channels[workspace]['phases']
+            self.channels[workspace]['phases'] = {key: value for key, value in phases.items() if key in run_ids}
+
+    async def watch(self, workspace, loader):
+        if sum(len(c['queues']) for c in self.channels.values()) >= 1024:
+            raise RuntimeError('Too many monitoring connections')
+        channel = self.channels.get(workspace)
+        if channel is None:
+            channel = dict(queues=set(), dirty=asyncio.Event(), phases={}, snapshot=None)
+            self.channels[workspace] = channel
+            channel['task'] = asyncio.create_task(self._produce(channel, loader))
+        if len(channel['queues']) >= 64:
+            raise RuntimeError('Too many Workspace monitoring connections')
+        queue = asyncio.Queue(maxsize=1)
+        channel['queues'].add(queue)
+        if channel['snapshot'] is not None:
+            queue.put_nowait(channel['snapshot'])
+        try:
+            # Reconnect periodically so HTTP authentication/authorization is renewed.
+            deadline = time.monotonic() + 60
+            while time.monotonic() < deadline:
+                try:
+                    yield await asyncio.wait_for(queue.get(), timeout=min(10, deadline - time.monotonic()))
+                except asyncio.TimeoutError:
+                    yield {'kind': 'heartbeat'}
+        finally:
+            channel['queues'].discard(queue)
+            if not channel['queues']:
+                self.channels.pop(workspace, None)
+                self.track(workspace, [])
+                channel['task'].cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await channel['task']
+
+    async def _produce(self, channel, loader):
+        previous = None
+        while True:
+            channel['dirty'].clear()
+            try:
+                snapshot = await loader()
+                frame = {'kind': 'snapshot', 'data': snapshot}
+            except Exception:
+                frame = {'kind': 'unavailable'}
+            # Heartbeats are sent separately; unchanged snapshots generate no traffic.
+            if frame != previous:
+                channel['snapshot'] = frame
+                for queue in tuple(channel['queues']):
+                    if queue.full():
+                        queue.get_nowait()
+                    queue.put_nowait(frame)
+                previous = frame
+            await asyncio.sleep(0.75)
+            try:
+                await asyncio.wait_for(channel['dirty'].wait(), timeout=15)
+            except asyncio.TimeoutError:
+                pass
+
+
+inflight_hub = InflightHub()