Break Lapdog payloads into chunks of 20 spans each (#404)

* Make Lapdog respect DD_TAGS

* Replace DD_TAGS with session-scoped Lapdog tags

Allow Claude to tag its active Lapdog session through an explicit CLI command, avoiding detached-process environment limitations while updating existing and future spans.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Add a release note and extend the change to Codex

* Correctly handle parallel Codex sessions

* Also add support for Pi and address remaining inline comments

* Add one more comment

* Make Lapdog split large traces into chunks of at most 20 spans before sending

* Add a release note

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: yshapiro-57

ddapm_test_agent/claude_hooks.py

@@ -71,6 +71,7 @@
         "user_name",
     }
 )
+_MAX_SPANS_PER_BACKEND_REQUEST = 20
 
 # Models with 1M token context windows (native, no beta header needed).
 # All other models default to 200k.
@@ -1768,6 +1769,36 @@ async def _post_to_backend(self, url: str, headers: Dict[str, str], data: bytes,
         except Exception as e:
             log.warning("Error trying to %s: %s", description, e)
 
+    async def _post_spans_to_backend(
+        self,
+        url: str,
+        headers: Dict[str, str],
+        spans: List[Dict[str, Any]],
+        description: str,
+    ) -> None:
+        """POST spans in ordered chunks that stay within the backend request limit."""
+        chunk_count = (len(spans) + _MAX_SPANS_PER_BACKEND_REQUEST - 1) // _MAX_SPANS_PER_BACKEND_REQUEST
+        if chunk_count > 1:
+            log.info(
+                "Splitting %d spans into %d backend requests of at most %d spans",
+                len(spans),
+                chunk_count,
+                _MAX_SPANS_PER_BACKEND_REQUEST,
+            )
+
+        for chunk_index, start in enumerate(range(0, len(spans), _MAX_SPANS_PER_BACKEND_REQUEST), start=1):
+            chunk = spans[start : start + _MAX_SPANS_PER_BACKEND_REQUEST]
+            payload = {
+                "_dd.stage": "raw",
+                "event_type": "span",
+                "spans": chunk,
+            }
+            data = gzip.compress(msgpack.packb(payload))
+            chunk_description = description
+            if chunk_count > 1:
+                chunk_description = f"{description} (chunk {chunk_index}/{chunk_count}, {len(chunk)} spans)"
+            await self._post_to_backend(url, headers, data, chunk_description)
+
     async def _forward_span_update(self, spans: List[Dict[str, Any]]) -> None:
         """Forward span updates to the DD backend via the update endpoint."""
         if not spans:
@@ -1778,13 +1809,7 @@ async def _forward_span_update(self, spans: List[Dict[str, Any]]) -> None:
             return
         url, headers = target
 
-        payload = {
-            "_dd.stage": "raw",
-            "event_type": "span",
-            "spans": spans,
-        }
-        data = gzip.compress(msgpack.packb(payload))
-        await self._post_to_backend(url, headers, data, f"forward {len(spans)} span updates")
+        await self._post_spans_to_backend(url, headers, spans, f"forward {len(spans)} span updates")
 
     async def _forward_trace_to_backend(
         self, session_id: str, trace_id: Optional[str] = None, span_source: str = "Claude hooks"
@@ -1817,14 +1842,11 @@ async def _forward_trace_to_backend(
                 span["tags"] = tags + ["lapdog_forwarded:true"]
             forwarded_spans.append(span)
 
-        payload = {
-            "_dd.stage": "raw",
-            "event_type": "span",
-            "spans": forwarded_spans,
-        }
-        data = gzip.compress(msgpack.packb(payload))
-        await self._post_to_backend(
-            url, headers, data, f"forward {len(spans)} {span_source} spans for trace {trace_id}"
+        await self._post_spans_to_backend(
+            url,
+            headers,
+            forwarded_spans,
+            f"forward {len(spans)} {span_source} spans for trace {trace_id}",
         )
 
     async def _forward_eval_metrics_to_backend(self, session_id: str, trace_id: Optional[str] = None) -> None:

releasenotes/notes/break-large-payloads-into-chunks-fb10f61dae242e88.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    When forwarding payloads to the Datadog backend, send them in chunks
+    of at most 20 spans each to get around payload size limits on the backend.

tests/test_claude_hooks.py

@@ -1,8 +1,11 @@
+import gzip
 import json
 import os
 import subprocess
 import tempfile
 
+import msgpack
+
 from ddapm_test_agent.claude_hooks import ClaudeHooksAPI
 from ddapm_test_agent.claude_link_tracker import ClaudeLinkTracker
 from ddapm_test_agent.claude_proxy import ClaudeProxyAPI
@@ -37,6 +40,60 @@ async def test_hook_missing_session_id(agent):
     assert "session_id" in body["error"]
 
 
+async def test_trace_forwarding_chunks_payloads_at_twenty_spans(monkeypatch):
+    cases = [
+        (20, [20]),
+        (21, [20, 1]),
+        (45, [20, 20, 5]),
+    ]
+    for span_count, expected_chunk_sizes in cases:
+        hooks = ClaudeHooksAPI()
+        session = hooks._get_or_create_session(f"chunk-session-{span_count}")
+        trace_id = f"chunk-trace-{span_count}"
+        session.trace_id = trace_id
+        hooks._assembled_spans = [
+            {
+                "span_id": str(index),
+                "trace_id": trace_id,
+                "tags": [],
+            }
+            for index in range(span_count)
+        ]
+        posted_payloads = []
+        descriptions = []
+
+        monkeypatch.setattr(
+            hooks,
+            "_resolve_backend_target",
+            lambda *args, **kwargs: ("http://backend.example/api/v2/llmobs", {"Content-Type": "application/msgpack"}),
+        )
+
+        async def fake_post_to_backend(url, headers, data, description):
+            assert url == "http://backend.example/api/v2/llmobs"
+            assert headers["Content-Type"] == "application/msgpack"
+            posted_payloads.append(msgpack.unpackb(gzip.decompress(data), raw=False))
+            descriptions.append(description)
+
+        monkeypatch.setattr(hooks, "_post_to_backend", fake_post_to_backend)
+
+        await hooks._forward_trace_to_backend(session.session_id)
+
+        assert [len(payload["spans"]) for payload in posted_payloads] == expected_chunk_sizes
+        assert all(payload["_dd.stage"] == "raw" for payload in posted_payloads)
+        assert all(payload["event_type"] == "span" for payload in posted_payloads)
+        forwarded_spans = [span for payload in posted_payloads for span in payload["spans"]]
+        assert [span["span_id"] for span in forwarded_spans] == [str(index) for index in range(span_count)]
+        assert all("lapdog_forwarded:true" in span["tags"] for span in forwarded_spans)
+        assert all("lapdog_forwarded:true" not in span["tags"] for span in hooks._assembled_spans)
+        if span_count <= 20:
+            assert all("(chunk " not in description for description in descriptions)
+        else:
+            assert all(
+                f"chunk {index + 1}/{len(expected_chunk_sizes)}" in description
+                for index, description in enumerate(descriptions)
+            )
+
+
 async def test_session_tags_apply_to_existing_and_future_spans(agent):
     session_id = "sess-custom-tags"
     session_token = "launch-token"