From f93b7204bb1e05ade23e46d56a29cb02e99ab6fd Mon Sep 17 00:00:00 2001 From: Miguel Pereira Date: Mon, 12 Oct 2026 17:17:55 +0100 Subject: [PATCH 1/3] Add CC_LANGFUSE_CAPTURE_HISTORY to stop quadratic generation inputs Since v1.2 every generation's input carries the whole conversation so far. On a long session that is quadratic: each LLM call repeats every earlier message, and one span batch can exceed the OTLP exporter's 64 MiB request limit, which then drops the entire batch with only a stderr line. Measured re-sending one 165-turn session: 51 batches of 67-425 MB dropped. CC_LANGFUSE_CAPTURE_HISTORY (default true, upstream behaviour unchanged) set to false skips building the session history, so each generation carries its own new input only, as before v1.2. Co-Authored-By: Claude Opus 5.5 --- .claude-plugin/plugin.json | 6 +++++ hooks/langfuse_hook.py | 8 +++++- tests/conftest.py | 2 ++ tests/integration/test_generation_history.py | 26 ++++++++++++++++++++ 4 files changed, 41 insertions(+), 1 deletion(-) diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 3583d68..50018ef 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -61,6 +61,12 @@ "description": "Upload images to Langfuse and show them in the trace. This includes pasted images and screenshots from tool results. Your Langfuse deployment must support media upload. Set to false to show short text markers instead.", "default": true }, + "CC_LANGFUSE_CAPTURE_HISTORY": { + "type": "boolean", + "title": "Capture conversation history", + "description": "Send the conversation so far as each LLM call's input. On long sessions this grows quadratically and span batches can exceed the exporter's 64 MiB limit, which drops them. Set to false to send each call's own new input only.", + "default": true + }, "CC_LANGFUSE_STATE_DIR": { "type": "string", "title": "State directory", diff --git a/hooks/langfuse_hook.py b/hooks/langfuse_hook.py index a69b9af..7adf2a9 100644 --- a/hooks/langfuse_hook.py +++ b/hooks/langfuse_hook.py @@ -41,6 +41,12 @@ def _opt(name: str) -> str: AGENT_TYPE_TAGS = (_opt("CC_LANGFUSE_AGENT_TYPE_TAGS") or "true").lower() == "true" CAPTURE_SKILL_CONTENT = _opt("CC_LANGFUSE_CAPTURE_SKILL_CONTENT").lower() == "true" CAPTURE_IMAGES = (_opt("CC_LANGFUSE_CAPTURE_IMAGES") or "true").lower() == "true" +# Each generation's input carries the conversation so far. On a long session that is +# quadratic in size: every LLM call repeats every earlier message, a single span batch +# can exceed the OTLP exporter's 64 MiB request limit, and the exporter then drops the +# whole batch (measured: 51 batches of 67-425 MB dropped re-sending one 165-turn +# session). "false" sends each generation's own new input only, as before history. +CAPTURE_HISTORY = (_opt("CC_LANGFUSE_CAPTURE_HISTORY") or "true").lower() == "true" OPERATOR_TAGS_VAR = "CC_LANGFUSE_TRACE_TAGS" try: MAX_CHARS = int(_opt("CC_LANGFUSE_MAX_CHARS") or "20000") @@ -3497,7 +3503,7 @@ def emit_new_turns_from_transcript( # One full-file pass per firing serves every turn emitted below. session_history = None - if turns or session_state.open_turn: + if CAPTURE_HISTORY and (turns or session_state.open_turn): session_history = build_session_history( transcript_path, get_task_id_to_tool_use_id(subagent_transcripts_by_tool_use_id), diff --git a/tests/conftest.py b/tests/conftest.py index c6ceff1..0dc1020 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -24,6 +24,8 @@ os.environ.pop("CLAUDE_PLUGIN_OPTION_CC_LANGFUSE_STATE_DIR", None) os.environ.pop("CC_LANGFUSE_CAPTURE_IMAGES", None) os.environ.pop("CLAUDE_PLUGIN_OPTION_CC_LANGFUSE_CAPTURE_IMAGES", None) +os.environ.pop("CC_LANGFUSE_CAPTURE_HISTORY", None) +os.environ.pop("CLAUDE_PLUGIN_OPTION_CC_LANGFUSE_CAPTURE_HISTORY", None) # Mirrors OTel's active-span context: the stubbed use_span pushes the parent # span here so FakeOtelSpan can record which span it was started under, diff --git a/tests/integration/test_generation_history.py b/tests/integration/test_generation_history.py index 36b738f..37ddca5 100644 --- a/tests/integration/test_generation_history.py +++ b/tests/integration/test_generation_history.py @@ -477,3 +477,29 @@ def test_a_tool_use_without_a_name_keeps_the_nested_shape_in_the_history( assert generation_inputs(fake_langfuse)[1][1]["tool_calls"] == [ {"id": "toolu-x", "type": "function", "function": {"name": None}} ] + + +def test_history_opt_out_sends_only_each_turns_own_input( + hook_module, fake_langfuse, isolated_hook_state, tmp_path, monkeypatch +): + monkeypatch.setattr(hook_module, "CAPTURE_HISTORY", False) + transcript = tmp_path / f"{SESSION}.jsonl" + rows: list[dict[str, Any]] = [] + for index in (1, 2, 3): + rows.extend(simple_turn_rows(SESSION, index)) + write_rows(transcript, rows) + + hook_module.emit_new_turns_from_transcript( + fake_langfuse, config_for(hook_module), SESSION, transcript + ) + + inputs = generation_inputs(fake_langfuse) + assert len(inputs) == 3 + # No prior turns are repeated: turn 3 carries only its own question. + for index, generation_input in zip((1, 2, 3), inputs): + text = json.dumps(generation_input) + assert f"Question {index}?" in text + for other in (1, 2, 3): + if other != index: + assert f"Question {other}?" not in text + assert f"Answer {other}." not in text From 415d561d9eac37b923582793c894d8a3e92b54c4 Mon Sep 17 00:00:00 2001 From: Miguel Pereira Date: Mon, 12 Oct 2026 17:24:37 +0100 Subject: [PATCH 2/3] Trace workflow agents spawned with a named agent type get_workflow_agents_in_run_dir kept only metas whose agentType was "workflow-subagent", which is what an untyped spawn records. A workflow script that names a type (agent(..., {agentType: 'round-writer'})) records that type instead, so every such agent was dropped and the Workflow tool span came out with no children. Measured on one session: 0 of ~1,000 round agents traced, $990 of cost missing. A workflow agent is now identified by living in the run directory without a toolUseId (classic subagent) or taskKind (teammate), whatever its type. Co-Authored-By: Claude Opus 5.5 --- hooks/langfuse_hook.py | 11 ++++++++--- tests/unit/test_subagent_discovery.py | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/hooks/langfuse_hook.py b/hooks/langfuse_hook.py index 7adf2a9..f18e32b 100644 --- a/hooks/langfuse_hook.py +++ b/hooks/langfuse_hook.py @@ -1769,8 +1769,8 @@ def get_workflow_agents_in_run_dir(run_dir: Path) -> List[Dict[str, Any]]: """The workflow-spawned agent transcripts of one Workflow run directory. Workflow-tool agents live under /subagents/workflows//; - their meta.json carries agentType=="workflow-subagent" and — unlike - classic subagents — no toolUseId, so a run is identified by its directory + their meta.json carries — unlike classic subagents — no toolUseId (agentType is + "workflow-subagent" only for an untyped spawn), so a run is identified by its directory name instead. The launching tool_use is linked via toolUseResult.runId on the parent transcript's tool_result row (see get_workflow_launch_marker_from_row). @@ -1785,7 +1785,12 @@ def get_workflow_agents_in_run_dir(run_dir: Path) -> List[Dict[str, Any]]: metadata = json.loads(meta_path.read_text(encoding="utf-8")) except Exception: continue - if not isinstance(metadata, dict) or metadata.get("agentType") != "workflow-subagent": + # A workflow agent's agentType is "workflow-subagent" only when the script + # spawned it untyped; a script that names an agent type (agent(..., {agentType: + # 'round-writer'})) records that type instead. What identifies a workflow agent + # is where it lives plus the absence of the classic-subagent launch link + # (toolUseId) and of the teammate marker (taskKind), never its type. + if not isinstance(metadata, dict) or metadata.get("toolUseId") or metadata.get("taskKind"): continue resolved = resolve_agent_jsonl_and_id(meta_path) diff --git a/tests/unit/test_subagent_discovery.py b/tests/unit/test_subagent_discovery.py index 6a9355f..c3c6ee7 100644 --- a/tests/unit/test_subagent_discovery.py +++ b/tests/unit/test_subagent_discovery.py @@ -467,3 +467,22 @@ def test_ignores_bad_or_incomplete_subagent_metadata(hook_module, tmp_path): (subagent_dir / "agent-missing-tool-id.jsonl").write_text("", encoding="utf-8") assert hook_module.get_subagent_transcripts_by_tool_use_id(transcript) == {} + + +def test_workflow_discovery_keeps_agents_spawned_with_a_named_agent_type(hook_module, tmp_path): + # A workflow script that passes agentType records that type in the meta, not + # "workflow-subagent"; those agents are still the run's agents. + transcript = tmp_path / "transcript.jsonl" + transcript.write_text("", encoding="utf-8") + run_dir = tmp_path / "transcript" / "subagents" / "workflows" / "wf_typed" + run_dir.mkdir(parents=True) + for agent_id, agent_type in (("a1", "round-writer"), ("a2", "round-runner")): + (run_dir / f"agent-{agent_id}.meta.json").write_text( + json.dumps({"agentType": agent_type, "workflowPhase": "Write", "spawnDepth": 1}), + encoding="utf-8", + ) + (run_dir / f"agent-{agent_id}.jsonl").write_text("", encoding="utf-8") + + agents = hook_module.WorkflowAgentTranscriptsByRunId(transcript).get("wf_typed") + assert agents is not None + assert sorted(agent["agent_type"] for agent in agents) == ["round-runner", "round-writer"] From 58e94facb67d87dec3571973c965d2d97afd1adb Mon Sep 17 00:00:00 2001 From: Miguel Pereira Date: Mon, 12 Oct 2026 17:29:24 +0100 Subject: [PATCH 3/3] Cap spans per export request at 128 so a batch stays under the 64 MiB limit The SDK default of 512 spans per OTLP request, with spans carrying up to MAX_CHARS of input and output each, reached the exporter's 64 MiB max_request_size; an over-limit batch is dropped whole (measured: five 67 MB batches dropped re-sending one session with history off). LANGFUSE_FLUSH_AT now defaults to 128 unless the user set it. Co-Authored-By: Claude Opus 5.5 --- hooks/langfuse_hook.py | 7 +++++++ tests/unit/test_backlog_flush.py | 15 +++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/hooks/langfuse_hook.py b/hooks/langfuse_hook.py index f18e32b..b5da67c 100644 --- a/hooks/langfuse_hook.py +++ b/hooks/langfuse_hook.py @@ -174,6 +174,12 @@ def _resolve_state_dir() -> Tuple[Path, str]: # after every turn of a multi-turn firing so the queue never holds more than one turn. DEFAULT_OTEL_BSP_MAX_QUEUE_SIZE = "65536" +# Spans per export request. The SDK default (512) times spans carrying up to +# MAX_CHARS of input and output each reaches the OTLP exporter's 64 MiB request +# limit, and an over-limit batch is dropped whole (measured: 67 MB batches dropped). +# 128 keeps a full batch near a quarter of the limit. +DEFAULT_LANGFUSE_FLUSH_AT = "128" + @dataclass class LangfuseConfig: public_key: str @@ -394,6 +400,7 @@ def create_langfuse_client(config: LangfuseConfig) -> Optional[Langfuse]: os.environ["LANGFUSE_MEDIA_UPLOAD_ENABLED"] = "false" # Read by the BatchSpanProcessor at construction, so it must be set before Langfuse(). os.environ.setdefault("OTEL_BSP_MAX_QUEUE_SIZE", DEFAULT_OTEL_BSP_MAX_QUEUE_SIZE) + os.environ.setdefault("LANGFUSE_FLUSH_AT", DEFAULT_LANGFUSE_FLUSH_AT) try: return Langfuse( public_key=config.public_key, diff --git a/tests/unit/test_backlog_flush.py b/tests/unit/test_backlog_flush.py index 674067f..0531f8c 100644 --- a/tests/unit/test_backlog_flush.py +++ b/tests/unit/test_backlog_flush.py @@ -62,3 +62,18 @@ def test_client_creation_keeps_user_queue_ceiling(hook_module: Any, monkeypatch: hook_module.create_langfuse_client(config) import os assert os.environ["OTEL_BSP_MAX_QUEUE_SIZE"] == "123" + + +def test_client_creation_caps_spans_per_export_request(hook_module: Any, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("LANGFUSE_FLUSH_AT", raising=False) + seen: dict[str, Any] = {} + + class RecordingLangfuse: + def __init__(self, **_: Any) -> None: + import os + seen["flush_at"] = os.environ.get("LANGFUSE_FLUSH_AT") + + monkeypatch.setattr(hook_module, "Langfuse", RecordingLangfuse) + config = hook_module.LangfuseConfig(public_key="pk", secret_key="sk", host="h", user_id=None) + hook_module.create_langfuse_client(config) + assert seen["flush_at"] == hook_module.DEFAULT_LANGFUSE_FLUSH_AT