fix: widen telemetry fields and require complete batch acknowledgement (#2652)

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,

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

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')

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

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)