diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index a84aa0543..887a028fe 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -1691,9 +1691,14 @@ class MonitoringService(ExecutionDetailsMixin): workspace = require_workspace_uuid(context) active = await self.get_executions(context, statuses=['running', 'queued'], limit=100, include_summary=False) + # Execution timestamps are stored as UTC without a timezone. asyncpg + # rejects aware datetimes for these columns, even though SQLite accepts them. + recent_start = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - datetime.timedelta( + seconds=30 + ) recent = await self.get_executions( context, - start_time=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=30), + start_time=recent_start, limit=100, include_summary=False, ) diff --git a/src/langbot/pkg/utils/inflight.py b/src/langbot/pkg/utils/inflight.py index 6a0f1a0e0..8d6b8c0b4 100644 --- a/src/langbot/pkg/utils/inflight.py +++ b/src/langbot/pkg/utils/inflight.py @@ -7,9 +7,13 @@ No payloads or subscriptions survive the final viewer disconnecting. import asyncio import contextlib +import logging import time +logger = logging.getLogger(__name__) + + class InflightHub: def __init__(self): self.channels = {} @@ -75,6 +79,8 @@ class InflightHub: snapshot = await loader() frame = {'kind': 'snapshot', 'data': snapshot} except Exception: + if previous is None or previous['kind'] != 'unavailable': + logger.exception('Failed to load in-flight monitoring snapshot') frame = {'kind': 'unavailable'} # Heartbeats are sent separately; unchanged snapshots generate no traffic. if frame != previous: diff --git a/tests/integration/persistence/test_monitoring_postgres.py b/tests/integration/persistence/test_monitoring_postgres.py index ebfee7a7e..d46ff2107 100644 --- a/tests/integration/persistence/test_monitoring_postgres.py +++ b/tests/integration/persistence/test_monitoring_postgres.py @@ -7,6 +7,7 @@ no existing tables are dropped. Without that URL these tests are skipped. from __future__ import annotations +import datetime import logging import os import uuid @@ -19,6 +20,7 @@ from sqlalchemy.ext.asyncio import create_async_engine from langbot.pkg.api.http.context import ExecutionContext from langbot.pkg.api.http.service.monitoring import MonitoringService +from langbot.pkg.entity.persistence.agent_run import AgentRun from langbot.pkg.entity.persistence import monitoring as models from langbot.pkg.entity.persistence.workspace import Workspace from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode @@ -158,6 +160,56 @@ async def _read(service, method, context, *args, **kwargs): return await getattr(service, method)(context, *args, **kwargs) +async def test_inflight_snapshot_reads_recent_utc_rows_and_terminal_transitions(service, cloud_database): + _, admin = cloud_database + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + old = now - datetime.timedelta(hours=1) + async with admin.begin() as conn: + for workspace, run_id, status, created_at in [ + (WORKSPACE_A, 'inflight-running', 'running', old), + (WORKSPACE_A, 'inflight-recent', 'completed', now), + (WORKSPACE_B, 'inflight-private', 'running', now), + ]: + await conn.execute( + sa.insert(AgentRun).values( + run_id=run_id, workspace_id=workspace, runner_id='runner', status=status, created_at=created_at + ) + ) + for message_id, timestamp in [('inflight-message', now), ('inflight-old-message', old)]: + await conn.execute( + sa.insert(models.MonitoringMessage).values( + id=message_id, + workspace_uuid=WORKSPACE_A, + timestamp=timestamp, + **RESOURCE, + message_content='hello', + session_id='session', + status='success', + level='info', + ) + ) + + try: + # The SSE producer owns its tenant scope after the HTTP handler has returned. + snapshot = await service.get_inflight_snapshot(_context(WORKSPACE_A)) + assert snapshot['active_total'] == 1 + assert {row['id'] for row in snapshot['items']} == {'inflight-running', 'inflight-recent', 'inflight-message'} + async with admin.begin() as conn: + await conn.execute( + sa.update(AgentRun).where(AgentRun.run_id == 'inflight-running').values(status='completed') + ) + snapshot = await service.get_inflight_snapshot(_context(WORKSPACE_A), snapshot['items']) + assert snapshot['active_total'] == 0 + assert next(row for row in snapshot['items'] if row['id'] == 'inflight-running')['status_group'] == 'completed' + finally: + async with admin.begin() as conn: + await conn.execute( + sa.delete(AgentRun).where( + AgentRun.run_id.in_(['inflight-running', 'inflight-recent', 'inflight-private']) + ) + ) + + def _query(context, sender_id): return SimpleNamespace( _execution_context=context, diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index 5d2ba0168..63fbd32b5 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/tests/unit_tests/api/service/test_monitoring_executions.py @@ -820,3 +820,27 @@ async def test_inflight_viewers_share_work_and_cleanup(): assert 'one' not in hub.channels and 'run' not in hub.run_workspaces await other.aclose() assert not hub.channels + + +@pytest.mark.asyncio +async def test_inflight_snapshot_failure_is_logged_and_recovers(caplog): + import asyncio + from unittest.mock import AsyncMock + from langbot.pkg.utils.inflight import InflightHub + + hub = InflightHub() + snapshot = {'items': [], 'active_total': 0, 'truncated': False} + loader = AsyncMock(side_effect=[RuntimeError('snapshot database failure'), snapshot]) + stream = hub.watch('workspace', loader) + try: + assert await asyncio.wait_for(anext(stream), 2) == {'kind': 'unavailable'} + errors = [ + record for record in caplog.records if record.message == 'Failed to load in-flight monitoring snapshot' + ] + assert len(errors) == 1 + assert errors[0].exc_info[0] is RuntimeError + hub.notify('workspace') + assert await asyncio.wait_for(anext(stream), 2) == {'kind': 'snapshot', 'data': snapshot} + finally: + await stream.aclose() + assert not hub.channels diff --git a/web/src/app/home/components/inflight/InFlightMonitor.tsx b/web/src/app/home/components/inflight/InFlightMonitor.tsx index 13425cf28..3e831ee23 100644 --- a/web/src/app/home/components/inflight/InFlightMonitor.tsx +++ b/web/src/app/home/components/inflight/InFlightMonitor.tsx @@ -166,7 +166,7 @@ function WorkspaceMonitor() {