diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md
index 976dabffe..ceec4d2ec 100644
--- a/skills/skills/langbot-mcp-ops/SKILL.md
+++ b/skills/skills/langbot-mcp-ops/SKILL.md
@@ -177,6 +177,11 @@ logs and action results. These times are not internal plugin profiling data.
## Unified execution monitoring
+Use `get_inflight_executions` for a bounded current-workspace snapshot of active
+and recently started executions, including `progress_event`. Progress is a
+reported stage, not an estimated completion percentage. The UI uses the shared
+`GET /api/v1/monitoring/in-flight/stream` SSE feed instead of polling per viewer.
+
Use `get_monitoring_executions` for the execution list and
its legacy `pipeline_ids` parameter to filter any processor kind (Agent,
Pipeline or event processor); the summary uses the same processor scope. Use
diff --git a/src/langbot/pkg/agent/runner/run_ledger_store.py b/src/langbot/pkg/agent/runner/run_ledger_store.py
index 1cb3d476d..e34a632c6 100644
--- a/src/langbot/pkg/agent/runner/run_ledger_store.py
+++ b/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:
diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py
index 9b00e356a..3f9dae245 100644
--- a/src/langbot/pkg/api/http/controller/groups/monitoring.py
+++ b/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."""
diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py
index ad5b8858a..d4ec46591 100644
--- a/src/langbot/pkg/api/http/service/monitoring.py
+++ b/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,
diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py
index 6c2367a37..030c89f4e 100644
--- a/src/langbot/pkg/api/mcp/server.py
+++ b/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.'
)
diff --git a/src/langbot/pkg/utils/inflight.py b/src/langbot/pkg/utils/inflight.py
new file mode 100644
index 000000000..6a0f1a0e0
--- /dev/null
+++ b/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()
diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py
index ce3d79f34..eeb859c81 100644
--- a/tests/unit_tests/api/service/test_monitoring_executions.py
+++ b/tests/unit_tests/api/service/test_monitoring_executions.py
@@ -736,3 +736,60 @@ async def test_pipeline_debug_is_consistent_across_rows_rollups_and_telemetry(en
assert real_calls == 1 and real_rows[0]['id'] == 'llm-2'
traffic = await get_traffic_series(service.ap, WORKSPACE, mode='debug', pipeline_ids=['pipeline-1'])
assert sum(p['llm_calls'] for p in traffic['points']) == 1
+
+
+@pytest.mark.asyncio
+async def test_inflight_snapshot_is_scoped_and_keeps_terminal_transition(engine, service):
+ import sqlalchemy as sa
+ from unittest.mock import AsyncMock
+
+ service._get_execution_summary = AsyncMock(side_effect=AssertionError('No dashboard aggregates'))
+ service._get_token_coverage = AsyncMock(side_effect=AssertionError('No dashboard aggregates'))
+ async with AsyncSession(engine) as session:
+ for workspace, run_id in [(WORKSPACE, 'flight-visible'), ('other-workspace', 'flight-secret')]:
+ session.add(persistence_agent_run.AgentRun(
+ run_id=run_id, workspace_id=workspace, runner_id='runner', status='running', created_at=_dt(1),
+ ))
+ session.add(persistence_agent_run.AgentRunEvent(run_id='flight-visible', sequence=1, type='tool.started'))
+ await session.commit()
+ snapshot = await service.get_inflight_snapshot(WORKSPACE)
+ rows = {r['id']: r for r in snapshot['items']}
+ assert 'flight-secret' not in rows
+ assert rows['flight-visible']['progress_event'] == 'tool.started'
+ async with AsyncSession(engine) as session:
+ await session.execute(sa.update(persistence_agent_run.AgentRun).where(
+ persistence_agent_run.AgentRun.run_id == 'flight-visible'
+ ).values(status='completed'))
+ await session.commit()
+ next_snapshot = await service.get_inflight_snapshot(WORKSPACE, [rows['flight-visible']])
+ assert next(r for r in next_snapshot['items'] if r['id'] == 'flight-visible')['status_group'] == 'completed'
+
+
+@pytest.mark.asyncio
+async def test_inflight_viewers_share_work_and_cleanup():
+ import asyncio
+ from unittest.mock import AsyncMock
+ from langbot.pkg.utils.inflight import InflightHub
+
+ hub = InflightHub()
+ loader = AsyncMock(return_value={'items': []})
+ first = hub.watch('one', loader)
+ second = hub.watch('one', loader)
+ assert (await anext(first))['kind'] == 'snapshot'
+ assert (await anext(second))['kind'] == 'snapshot'
+ assert loader.await_count == 1
+ other_loader = AsyncMock(return_value={'items': ['private']})
+ other = hub.watch('two', other_loader)
+ assert (await anext(other))['data']['items'] == ['private']
+ hub.track('one', ['run'])
+ hub.notify_run('run', 'tool.started')
+ loader.return_value = {'items': ['updated']}
+ frame = await asyncio.wait_for(anext(first), 2)
+ assert frame['data']['items'] == ['updated']
+ assert (await anext(second)) == frame
+ await first.aclose()
+ assert 'one' in hub.channels
+ await second.aclose()
+ assert 'one' not in hub.channels and 'run' not in hub.run_workspaces
+ await other.aclose()
+ assert not hub.channels
diff --git a/web/src/app/home/bots/BotDetailContent.tsx b/web/src/app/home/bots/BotDetailContent.tsx
index bcce0e9da..de9e9bdf1 100644
--- a/web/src/app/home/bots/BotDetailContent.tsx
+++ b/web/src/app/home/bots/BotDetailContent.tsx
@@ -1,6 +1,6 @@
import EntityLoadState from '@/components/EntityLoadState';
import { useState, useEffect, useRef, useCallback } from 'react';
-import { useNavigate } from 'react-router-dom';
+import { useNavigate, useSearchParams } from 'react-router-dom';
import { Tabs, TabsList, TabsTrigger, TabsContent } from '@/components/ui/tabs';
import { Button } from '@/components/ui/button';
import { Switch } from '@/components/ui/switch';
@@ -22,7 +22,14 @@ import type { BotSessionMonitorHandle } from '@/app/home/bots/components/bot-ses
import { httpClient } from '@/app/infra/http/HttpClient';
import { useSidebarData } from '@/app/home/components/home-sidebar/SidebarDataContext';
import { useTranslation } from 'react-i18next';
-import { Settings, FileText, Users, RefreshCw, Trash2, Loader2 } from 'lucide-react';
+import {
+ Settings,
+ FileText,
+ Users,
+ RefreshCw,
+ Trash2,
+ Loader2,
+} from 'lucide-react';
import { cn } from '@/lib/utils';
import { toast } from 'sonner';
import { showBotError } from './bot-error';
@@ -56,7 +63,23 @@ export default function BotDetailContent({ id }: { id: string }) {
return () => setDetailEntityName(null);
}, [id, isCreateMode, bots, setDetailEntityName, t]);
- const [activeTab, setActiveTab] = useState('config');
+ const [searchParams, setSearchParams] = useSearchParams();
+ const requestedTab = searchParams.get('tab');
+ const activeTab =
+ !isCreateMode &&
+ canViewMonitoring &&
+ ['logs', 'sessions'].includes(requestedTab || '')
+ ? requestedTab!
+ : 'config';
+ const setActiveTab = (tab: string) =>
+ setSearchParams(
+ (previous) => {
+ const next = new URLSearchParams(previous);
+ next.set('tab', tab);
+ return next;
+ },
+ { replace: true },
+ );
const [adapterLabel, setAdapterLabel] = useState('');
const [showDeleteConfirm, setShowDeleteConfirm] = useState(false);
const [basicInfoOpen, setBasicInfoOpen] = useState(false);
@@ -101,7 +124,9 @@ export default function BotDetailContent({ id }: { id: string }) {
setBot((current) =>
current ? { ...current, enable: checked } : current,
);
- toast.success(t(checked ? 'bots.enableConfirmed' : 'bots.disableConfirmed'));
+ toast.success(
+ t(checked ? 'bots.enableConfirmed' : 'bots.disableConfirmed'),
+ );
refreshBots();
} catch (error) {
setBotEnabled(prev);
@@ -217,17 +242,34 @@ export default function BotDetailContent({ id }: { id: string }) {
{enableLoaded && (
- {isTogglingEnable
- ?
- : }
- {t(isTogglingEnable
- ? botEnabled ? 'bots.enabling' : 'bots.disabling'
- : botEnabled ? 'bots.enableConfirmed' : 'bots.disableConfirmed')}
+ {isTogglingEnable ? (
+
+ ) : (
+
+ )}
+ {t(
+ isTogglingEnable
+ ? botEnabled
+ ? 'bots.enabling'
+ : 'bots.disabling'
+ : botEnabled
+ ? 'bots.enableConfirmed'
+ : 'bots.disableConfirmed',
+ )}
- {t(formDirty && !botEnabled ? 'bots.saveAndEnable' : 'common.save')}
+ {t(
+ formDirty && !botEnabled
+ ? 'bots.saveAndEnable'
+ : 'common.save',
+ )}