diff --git a/src/langbot/pkg/telemetry/execution.py b/src/langbot/pkg/telemetry/execution.py index d914deadd..80123d4b7 100644 --- a/src/langbot/pkg/telemetry/execution.py +++ b/src/langbot/pkg/telemetry/execution.py @@ -255,7 +255,7 @@ class ExecutionCounters: if family not in FAMILIES or mode not in MODES or outcome not in OUTCOMES: return # Only code-defined identifiers are accepted; never pass user values. - if any(not isinstance(v, str) or len(v) > 160 for v in (operation, adapter, runner, node)): + if any(not isinstance(v, str) or len(v.encode('utf-8')) > 512 for v in (operation, adapter, runner, node)): return state = trace_mod.current() if state is None: @@ -602,7 +602,7 @@ def ingress( ap, context, family='platform_event', - operation=str(synthetic_event)[:160], + operation=str(synthetic_event)[:512], outcome='success', synthetic=True, node=node, diff --git a/src/langbot/pkg/telemetry/resources.py b/src/langbot/pkg/telemetry/resources.py index 5164e090e..ff959fd7c 100644 --- a/src/langbot/pkg/telemetry/resources.py +++ b/src/langbot/pkg/telemetry/resources.py @@ -29,8 +29,8 @@ import json import typing MAX_ITEMS_PER_LIST = 64 -MAX_NAME_CHARS = 160 -MAX_RESOURCES_BYTES = 8 * 1024 +MAX_NAME_CHARS = 512 +MAX_RESOURCES_BYTES = 32 * 1024 _resources: contextvars.ContextVar[dict[str, typing.Any] | None] = contextvars.ContextVar( 'telemetry_resources', default=None diff --git a/src/langbot/pkg/telemetry/telemetry.py b/src/langbot/pkg/telemetry/telemetry.py index 5c9a41d22..22ac71ae4 100644 --- a/src/langbot/pkg/telemetry/telemetry.py +++ b/src/langbot/pkg/telemetry/telemetry.py @@ -159,6 +159,7 @@ class TelemetryManager: else: # Detect application-level errors inside HTTP 200 responses app_err = False + j = None try: j = await httpclient.parse_json_response(resp) app_code = j.get('code') if isinstance(j, dict) else None @@ -181,6 +182,19 @@ class TelemetryManager: f'Telemetry posted to {url}, status {resp.status_code} - response: {body}' ) if not app_err: + if isinstance(sanitized.get('records'), list): + ack = j.get('data') if isinstance(j, dict) else None + if ( + not isinstance(ack, dict) + or type(ack.get('stored')) is not int + or type(ack.get('rejected')) is not int + or ack['rejected'] != 0 + or ack['stored'] != len(sanitized['records']) + ): + self.ap.logger.warning( + 'Telemetry batch was not fully acknowledged; retaining records' + ) + return False return True except asyncio.TimeoutError: self.ap.logger.debug(f'Telemetry post to {url} timed out') diff --git a/tests/unit_tests/telemetry/test_batch_ack.py b/tests/unit_tests/telemetry/test_batch_ack.py new file mode 100644 index 000000000..0aeff5448 --- /dev/null +++ b/tests/unit_tests/telemetry/test_batch_ack.py @@ -0,0 +1,29 @@ +from unittest.mock import Mock + +import httpx +import pytest + +from langbot.pkg.telemetry.telemetry import TelemetryManager + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + 'ack,expected', + [ + ({'stored': 1, 'rejected': 0}, True), + ({'stored': 0, 'rejected': 1}, False), + ({'stored': 1, 'rejected': 1}, False), + ({'stored': 0, 'rejected': 0}, False), + ({}, False), + ({'stored': True, 'rejected': 0}, False), + ], +) +async def test_batch_requires_complete_ack(ack, expected): + app = Mock() + manager = TelemetryManager(app) + manager.telemetry_config = {'url': 'https://telemetry.test'} + async with httpx.AsyncClient( + transport=httpx.MockTransport(lambda request: httpx.Response(200, json={'code': 0, 'data': ack})) + ) as client: + manager._client = client + assert await manager.send({'records': [{'event_id': 'a' * 163}]}) is expected diff --git a/tests/unit_tests/telemetry/test_trace.py b/tests/unit_tests/telemetry/test_trace.py index 381b0ee6b..cf7e63701 100644 --- a/tests/unit_tests/telemetry/test_trace.py +++ b/tests/unit_tests/telemetry/test_trace.py @@ -230,7 +230,7 @@ class TestTraceRecording: _, counters = make_counters(trace_config()) binding = trace.bind() try: - counters.record(CONTEXT, **{**STAGE, 'operation': 'x' * 200}) + counters.record(CONTEXT, **{**STAGE, 'operation': 'x' * 513}) counters.record(CONTEXT, **{**STAGE, 'family': 'not-a-family'}) finally: trace.unbind_root(binding)