Stop dropping long-session spans and typed workflow agents

.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",

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")
@@ -168,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
@@ -388,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,
@@ -1763,8 +1776,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 <stem>/subagents/workflows/<runId>/;
-    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).
@@ -1779,7 +1792,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)
@@ -3497,7 +3515,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),

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,

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

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

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"]