diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index aae227e25..5cbdf56c6 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/src/langbot/pkg/api/http/controller/groups/monitoring.py @@ -24,12 +24,21 @@ def parse_iso_datetime(datetime_str: str | None) -> datetime.datetime | None: return dt -def validate_monitoring_window(start_time: datetime.datetime | None, end_time: datetime.datetime | None) -> None: +def validate_monitoring_window( + start_time: datetime.datetime | None, + end_time: datetime.datetime | None, + *, + allow_point_window: bool = False, +) -> None: """Bound explicit dashboard windows to one year without silently truncating them.""" if start_time is None: return end = end_time or datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) - if start_time >= end or end - start_time > datetime.timedelta(days=365): + if ( + start_time > end + or (start_time == end and not allow_point_window) + or end - start_time > datetime.timedelta(days=365) + ): quart.abort(400, description='Monitoring time range must be positive and no longer than 365 days') @@ -580,7 +589,8 @@ class MonitoringRouterGroup(group.RouterGroup): """Get detailed analysis for a specific session""" start_time = parse_iso_datetime(quart.request.args.get('startTime')) end_time = parse_iso_datetime(quart.request.args.get('endTime')) - validate_monitoring_window(start_time, end_time) + # Message pages can contain one timestamp; tool-call bounds are inclusive. + validate_monitoring_window(start_time, end_time, allow_point_window=True) analysis = await self.ap.monitoring_service.get_session_analysis( request_context, session_id, diff --git a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py index 23a69bea8..e927a3baf 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/litellmchat.py @@ -16,7 +16,11 @@ import langbot_plugin.api.entities.builtin.provider.message as provider_message class _ThinkStripState: - """Stateful filter that drops think blocks across chunks.""" + """Drop explicit think blocks, buffering only possible split tag prefixes. + + Untagged text must stream immediately: an orphan closing tag cannot identify + earlier text as reasoning without delaying every ordinary answer until EOF. + """ _THINK_OPEN = '' _THINK_CLOSE = '' @@ -31,7 +35,6 @@ class _ThinkStripState: self._open_tags = tuple(open_tag for open_tag, _close_tag in self._pairs) self._buf = '' self._close_tag: str | None = None - self._pending_initial = True def feed(self, chunk: str) -> str: """Feed a streaming delta and return user-visible content.""" @@ -64,7 +67,6 @@ class _ThinkStripState: if close_idx != -1: self._close_tag = None self._buf = '' - self._pending_initial = False return self._process_visible_text(text[close_idx + len(close_tag) :]) self._buf = self._close_prefix(text, close_tag) @@ -75,37 +77,14 @@ class _ThinkStripState: index = 0 while index < len(text): - if self._pending_initial: - open_idx, open_tag, close_tag = self._find_next_open(text, index) - orphan_close_idx, orphan_close_tag = self._find_next_close(text, index) - - if orphan_close_idx != -1 and (open_idx == -1 or orphan_close_idx < open_idx): - self._pending_initial = False - index = orphan_close_idx + len(orphan_close_tag) - continue - - if open_idx == -1: - self._buf = text[index:] - return ''.join(out) - - if open_idx > index: - self._pending_initial = False - out.append(text[index:open_idx]) - index = open_idx - continue - open_idx, open_tag, close_tag = self._find_next_open(text, index) if open_idx == -1: emit_end = self._visible_emit_end(text, index) out.append(text[index:emit_end]) - if emit_end > index: - self._pending_initial = False self._buf = text[emit_end:] return ''.join(out) out.append(text[index:open_idx]) - if open_idx > index: - self._pending_initial = False body_start = open_idx + len(open_tag) close_idx = text.find(close_tag, body_start) if close_idx == -1: @@ -113,7 +92,6 @@ class _ThinkStripState: self._buf = self._close_prefix(text[body_start:], close_tag) return ''.join(out) - self._pending_initial = False index = close_idx + len(close_tag) self._buf = '' @@ -131,16 +109,6 @@ class _ThinkStripState: best_close = close_tag return best_idx, best_open, best_close - def _find_next_close(self, text: str, start: int) -> tuple[int, str]: - best_idx = -1 - best_close = '' - for _open_tag, close_tag in self._pairs: - idx = text.find(close_tag, start) - if idx != -1 and (best_idx == -1 or idx < best_idx): - best_idx = idx - best_close = close_tag - return best_idx, best_close - def _visible_emit_end(self, text: str, start: int) -> int: visible = text[start:] limit = min(len(visible), max(len(open_tag) for open_tag in self._open_tags) - 1) diff --git a/tests/unit_tests/api/service/test_monitoring_sessions.py b/tests/unit_tests/api/service/test_monitoring_sessions.py index 7451c9124..868f51589 100644 --- a/tests/unit_tests/api/service/test_monitoring_sessions.py +++ b/tests/unit_tests/api/service/test_monitoring_sessions.py @@ -183,6 +183,31 @@ async def test_analysis_fails_closed_and_scopes_statistics(service): assert result['session']['bot_id'] == 'b' +async def test_single_message_analysis_without_tool_calls(service): + from langbot.pkg.api.http.controller.groups.monitoring import validate_monitoring_window + from werkzeug.exceptions import BadRequest + + await service.record_session_start(context('a'), session_id='person_42', **resource('a')) + await service.record_message(context('a'), session_id='person_42', message_content='hi', **resource('a')) + row = (await service.ap.persistence_mgr.execute_async(sa.select(models.MonitoringMessage.timestamp))).one() + timestamp = row.timestamp + validate_monitoring_window(timestamp, timestamp, allow_point_window=True) + result = await service.get_session_analysis( + context(), 'person_42', bot_id='a', start_time=timestamp, end_time=timestamp + ) + assert result['found'] is True + assert result['message_stats']['total'] == 1 + assert result['tool_calls'] == [] + for start, end in [ + (timestamp + dt.timedelta(seconds=1), timestamp), + (timestamp, timestamp + dt.timedelta(days=366)), + ]: + with pytest.raises(BadRequest): + validate_monitoring_window(start, end, allow_point_window=True) + with pytest.raises(BadRequest): + validate_monitoring_window(timestamp, timestamp) + + async def test_activity_requires_bot_and_upsert_counts_racing_first_queries(service): for _ in range(2): await service.record_session_start(context('a'), session_id='person_42', **resource('a')) diff --git a/tests/unit_tests/provider/test_litellmchat.py b/tests/unit_tests/provider/test_litellmchat.py index 4c089816d..f73fac2e4 100644 --- a/tests/unit_tests/provider/test_litellmchat.py +++ b/tests/unit_tests/provider/test_litellmchat.py @@ -379,43 +379,33 @@ class TestInvokeLLMStreamUsage: assert ''.join(chunk.content or '' for chunk in collected) == 'Visible answer' @pytest.mark.asyncio - async def test_stream_removes_initial_orphan_think_close(self): - """Initial reasoning content without an open tag is removed until .""" - import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query - import langbot_plugin.api.entities.builtin.provider.message as provider_message - + @pytest.mark.parametrize('separate_reasoning', [False, True]) + async def test_remove_think_preserves_incremental_plain_text(self, separate_reasoning): + """Visible text must arrive before the provider produces the next delta.""" mock_ap = Mock() - mock_ap.tool_mgr = Mock() mock_ap.tool_mgr.generate_tools_for_openai = AsyncMock(return_value=None) requester = litellmchat.LiteLLMRequester(ap=mock_ap, config={}) - model = MockRuntimeModel('minimax-m3', 'test-api-key') + model = MockRuntimeModel('gpt-4o', 'test-api-key') + consumed = [] - chunks = [ - self._make_chunk(content='hidden reasoning'), - self._make_chunk(content=' still hidden', ''), + ('CRETIRE_REASONING_BEGINk', 'CRETIRE_REASONING_ENDk'), + ], + ) + def test_split_tags_and_repeated_blocks(self, opening, closing): + state = litellmchat._ThinkStripState() + assert state.feed('Before ') == 'Before ' + for char in opening + 'hidden' + closing: + assert state.feed(char) == '' + assert state.feed('Answer ') == 'Answer ' + assert state.feed(opening + 'more hidden' + closing + 'end') == 'end' + assert state.flush() == '' + + def test_unclosed_think_is_discarded(self): + state = litellmchat._ThinkStripState() + assert state.feed('hidden') == '' + assert state.flush() == '' + + def test_partial_non_tag_is_released(self): + state = litellmchat._ThinkStripState() + assert state.feed('Answer ') == '' + assert state.feed(' <') == ' ' + assert state.flush() == '<' diff --git a/web/src/app/home/pipelines/PipelineMigration.tsx b/web/src/app/home/pipelines/PipelineMigration.tsx index 892022f37..d85b3931f 100644 --- a/web/src/app/home/pipelines/PipelineMigration.tsx +++ b/web/src/app/home/pipelines/PipelineMigration.tsx @@ -340,16 +340,12 @@ export default function PipelineMigration({ return ( <> - {(count > 0 || previewError || results.length > 0) && ( + {(count > 0 || results.length > 0) && (