feat(hook): name traces after the session title

hooks/langfuse_hook.py

@@ -537,6 +537,10 @@ class SessionState:
     # partially emitted turn across firings and across the open -> closed ->
     # deferred transitions; entries are dropped once the turn is finalized.
     turn_progress: Dict[str, Dict[str, Any]] = field(default_factory=dict)
+    # Session titles seen in the transcript, keyed by the row type that
+    # carried them ("custom-title", "ai-title"). Only new rows are read per
+    # firing, so the titles are kept here to stay available to later ones.
+    titles_by_type: Dict[str, str] = field(default_factory=dict)
 
 def get_session_state(global_state: Dict[str, Any], key: str) -> SessionState:
     s = global_state.get(key, {})
@@ -555,6 +559,9 @@ def get_session_state(global_state: Dict[str, Any], key: str) -> SessionState:
     turn_progress = s.get("turn_progress")
     if not isinstance(turn_progress, dict):
         turn_progress = {}
+    titles_by_type = s.get("titles_by_type")
+    if not isinstance(titles_by_type, dict):
+        titles_by_type = {}
     return SessionState(
         offset=int(s.get("offset", 0)),
         buffer=str(s.get("buffer", "")),
@@ -564,6 +571,7 @@ def get_session_state(global_state: Dict[str, Any], key: str) -> SessionState:
         open_turn=open_turn,
         turn_numbers=turn_numbers,
         turn_progress=turn_progress,
+        titles_by_type=titles_by_type,
     )
 
 def update_session_state(global_state: Dict[str, Any], key: str, session_state: SessionState) -> None:
@@ -576,6 +584,7 @@ def update_session_state(global_state: Dict[str, Any], key: str, session_state:
         "open_turn": session_state.open_turn or {},
         "turn_numbers": session_state.turn_numbers or {},
         "turn_progress": session_state.turn_progress or {},
+        "titles_by_type": session_state.titles_by_type or {},
         "updated": datetime.now(timezone.utc).isoformat(),
     }
 
@@ -915,6 +924,52 @@ def read_new_jsonl(transcript_path: Path, session_state: SessionState) -> Tuple[
     return msgs, session_state
 
 
+# ----------------- Session title -----------------
+# Claude Code writes the session title into the transcript as its own row:
+# "custom-title" (set by /rename) or "ai-title" (auto-generated). A row is
+# appended again whenever the title changes, so the last occurrence of a type
+# is that type's current title. Listed in preference order.
+SESSION_TITLE_ROW_FIELDS: Tuple[Tuple[str, str], ...] = (
+    ("custom-title", "customTitle"),
+    ("ai-title", "aiTitle"),
+)
+MAX_SESSION_TITLE_CHARS = 200
+
+
+def clean_session_title(value: Any) -> Optional[str]:
+    """Trimmed, length-bounded title, or None for anything unusable."""
+    if not isinstance(value, str):
+        return None
+    title = value.strip()
+    if not title:
+        return None
+    return title[:MAX_SESSION_TITLE_CHARS]
+
+
+def update_session_titles(session_state: SessionState, rows: List[Dict[str, Any]]) -> None:
+    """Record the session titles carried by these rows."""
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        for row_type, title_field in SESSION_TITLE_ROW_FIELDS:
+            if row.get("type") != row_type:
+                continue
+            title = clean_session_title(row.get(title_field))
+            if title is not None:
+                session_state.titles_by_type[row_type] = title
+                debug(f"Session title from {row_type} row")
+
+
+def get_session_title(session_state: SessionState) -> Optional[str]:
+    """The session's title: the operator's own rename wins over the
+    auto-generated one, whichever row came last in the transcript."""
+    for row_type, _ in SESSION_TITLE_ROW_FIELDS:
+        title = clean_session_title(session_state.titles_by_type.get(row_type))
+        if title is not None:
+            return title
+    return None
+
+
 # ----------------- Turn assembly -----------------
 @dataclass
 class Turn:
@@ -1574,6 +1629,8 @@ def get_new_turns_from_transcript(
                 f"{session_id}; not checking for copied history"
             )
 
+    update_session_titles(session_state, rows)
+
     # Re-attach the trailing open turn from the previous run. Stop fires
     # multiple times within one logical turn, so a batch can begin with
     # user-less continuation rows (task notifications, assistant rows); with
@@ -2029,7 +2086,8 @@ def collect_subagent_skill_tags(
                 add_skill_tags_from_rows(rows, names, "subagent-skill:")
     return names
 
-# Constant on purpose: a shared trace name keeps Langfuse name-based grouping usable.
+# Trace name for sessions without a title; a shared name keeps Langfuse
+# name-based grouping usable for them.
 TRACE_NAME = "Claude Code Turn"
 
 def get_trace_tags(
@@ -3198,7 +3256,8 @@ def emit_turn(langfuse: Langfuse, session_id: str, turn_num: int,
               trace_seed: Optional[str] = None,
               parent_context: Optional[Tuple[str, str]] = None,
               workflow_agent_transcripts_by_run_id: Optional[WorkflowAgentTranscriptsByRunId] = None,
-              history_prefix: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
+              history_prefix: Optional[List[Dict[str, Any]]] = None,
+              session_title: Optional[str] = None) -> Dict[str, Any]:
     """Emit a turn, resuming from prior firings' progress.
 
     With no progress and close=True this is the classic one-shot emission.
@@ -3220,7 +3279,7 @@ def emit_turn(langfuse: Langfuse, session_id: str, turn_num: int,
         attribute_propagation = propagate_attributes(
             session_id=session_id,
             user_id=user_id,
-            trace_name=TRACE_NAME,
+            trace_name=session_title or TRACE_NAME,
             tags=get_trace_tags(turn, subagent_transcripts_by_tool_use_id),
         )
     else:
@@ -3328,6 +3387,7 @@ def emit_and_close_ready_turns(
                     if session_history is not None
                     else None
                 ),
+                session_title=get_session_title(session_state),
             )
         except Exception as e:
             # Log at INFO so SDK incompatibilities (and other emit failures)
@@ -3398,6 +3458,7 @@ def emit_ready_observations_of_open_turn(
                 if session_history is not None
                 else None
             ),
+            session_title=get_session_title(session_state),
         )
         session_state.turn_progress[user_row_uuid] = progress
     except Exception as e:

tests/integration/test_state_lock_contention.py

@@ -111,7 +111,7 @@ def test_entry_of_another_session_survives_the_emit(
     def other_session_saves() -> None:
         with hook_module.FileLock(hook_module.LOCK_FILE):
             state = hook_module.load_hook_state()
-            state["other-session-key"] = {"offset": 42, "updated": "2026-09-10T00:00:00+00:00"}
+            state["other-session-key"] = {"offset": 42, "updated": _now_iso()}
             hook_module.save_hook_state(state)
 
     _during_emit(hook_module, monkeypatch, other_session_saves)

tests/unit/test_transcript_reader.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
+import contextlib
 import json
+from pathlib import Path
+
+import pytest
 
 
 def test_read_new_jsonl_reads_once_then_advances_offset(
@@ -101,3 +105,161 @@ def write(rows: list[dict]) -> None:
     assert state.pending_agent_turns == []
     assert state.pending_task_notifications == []
     assert state.open_turn == {}
+
+
+# ----------------- Session title -----------------
+
+TITLED_SESSION_ID = "session-titled"
+
+
+def custom_title_row(value: object) -> dict:
+    return {"type": "custom-title", "customTitle": value, "sessionId": TITLED_SESSION_ID}
+
+
+def ai_title_row(value: object) -> dict:
+    return {"type": "ai-title", "aiTitle": value, "sessionId": TITLED_SESSION_ID}
+
+
+def title_of(hook_module, rows: list[dict]) -> str | None:
+    state = hook_module.SessionState()
+    hook_module.update_session_titles(state, rows)
+    return hook_module.get_session_title(state)
+
+
+def conversation_rows(name: str) -> list[dict]:
+    return [
+        {"type": "user", "timestamp": "2026-01-01T00:00:00.000Z", "uuid": f"{name}-user",
+         "sessionId": TITLED_SESSION_ID, "message": {"role": "user", "content": "Question."}},
+        {"type": "assistant", "timestamp": "2026-01-01T00:00:01.000Z", "uuid": f"{name}-assistant",
+         "sessionId": TITLED_SESSION_ID,
+         "message": {"id": f"{name}-msg", "role": "assistant", "model": "m",
+                     "content": [{"type": "text", "text": "Answer."}]}},
+    ]
+
+
+def write_transcript(directory: Path, rows: list[dict]) -> Path:
+    transcript = directory / f"{TITLED_SESSION_ID}.jsonl"
+    transcript.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8")
+    return transcript
+
+
+def record_propagated_trace_names(hook_module, monkeypatch: pytest.MonkeyPatch) -> list[str | None]:
+    names: list[str | None] = []
+
+    @contextlib.contextmanager
+    def recording_propagate(**kwargs):
+        names.append(kwargs.get("trace_name"))
+        yield
+
+    monkeypatch.setattr(hook_module, "propagate_attributes", recording_propagate)
+    return names
+
+
+def test_a_custom_title_wins_over_an_ai_title_whichever_row_came_last(hook_module):
+    assert title_of(hook_module, [
+        custom_title_row("Meeting Module"),
+        ai_title_row("Langfuse port 5432 conflict"),
+    ]) == "Meeting Module"
+    assert title_of(hook_module, [
+        ai_title_row("Langfuse port 5432 conflict"),
+        custom_title_row("Meeting Module"),
+    ]) == "Meeting Module"
+
+
+def test_the_last_title_row_of_a_type_replaces_the_earlier_one(hook_module):
+    assert title_of(hook_module, [
+        custom_title_row("First name"),
+        custom_title_row("Second name"),
+    ]) == "Second name"
+    assert title_of(hook_module, [
+        ai_title_row("First guess"),
+        ai_title_row("Second guess"),
+    ]) == "Second guess"
+
+
+def test_an_ai_title_is_used_until_the_session_is_renamed(hook_module):
+    state = hook_module.SessionState()
+    hook_module.update_session_titles(state, [ai_title_row("Langfuse port 5432 conflict")])
+    assert hook_module.get_session_title(state) == "Langfuse port 5432 conflict"
+
+    hook_module.update_session_titles(state, [custom_title_row("Meeting Module")])
+    assert hook_module.get_session_title(state) == "Meeting Module"
+
+
+def test_unusable_title_values_are_ignored(hook_module):
+    assert title_of(hook_module, [custom_title_row("")]) is None
+    assert title_of(hook_module, [custom_title_row("   ")]) is None
+    assert title_of(hook_module, [custom_title_row(None)]) is None
+    assert title_of(hook_module, [custom_title_row(42)]) is None
+    assert title_of(hook_module, [{"type": "custom-title", "sessionId": TITLED_SESSION_ID}]) is None
+    assert title_of(hook_module, [{"type": "user", "customTitle": "Not a title row"}]) is None
+
+    # A usable earlier title is kept, not replaced by the unusable one.
+    assert title_of(hook_module, [
+        custom_title_row("Meeting Module"),
+        custom_title_row("  "),
+    ]) == "Meeting Module"
+
+
+def test_a_title_is_trimmed_and_capped(hook_module):
+    assert title_of(hook_module, [custom_title_row("  Meeting Module \n")]) == "Meeting Module"
+
+    capped = title_of(hook_module, [custom_title_row("x" * 5000)])
+    assert capped == "x" * hook_module.MAX_SESSION_TITLE_CHARS
+    assert len(capped) == 200
+
+
+def test_titles_round_trip_through_the_state_file_representation(hook_module):
+    state = hook_module.SessionState()
+    hook_module.update_session_titles(state, [
+        ai_title_row("Langfuse port 5432 conflict"),
+        custom_title_row("Meeting Module"),
+    ])
+
+    global_state: dict = {}
+    hook_module.update_session_state(global_state, "key", state)
+    assert global_state["key"]["titles_by_type"] == {
+        "ai-title": "Langfuse port 5432 conflict",
+        "custom-title": "Meeting Module",
+    }
+
+    restored = hook_module.get_session_state(global_state, "key")
+    assert hook_module.get_session_title(restored) == "Meeting Module"
+
+
+def test_the_session_title_is_the_trace_name_and_outlives_the_firing_that_saw_it(
+    hook_module, fake_langfuse, isolated_hook_state, tmp_path, monkeypatch: pytest.MonkeyPatch
+):
+    transcript = write_transcript(
+        tmp_path,
+        conversation_rows("turn-a") + [custom_title_row("Meeting Module")] + conversation_rows("turn-b"),
+    )
+    trace_names = record_propagated_trace_names(hook_module, monkeypatch)
+    config = hook_module.LangfuseConfig("public", "secret", "https://example.test", "user-1")
+
+    assert hook_module.emit_new_turns_from_transcript(
+        fake_langfuse, config, TITLED_SESSION_ID, transcript) == 1
+
+    # Second firing: the title row sits behind the read offset, so only the
+    # title persisted in session state can name the trace.
+    assert hook_module.emit_new_turns_from_transcript(
+        fake_langfuse, config, TITLED_SESSION_ID, transcript,
+        flush_deferred_agent_turns=True) == 1
+
+    assert len(trace_names) >= 2
+    assert set(trace_names) == {"Meeting Module"}
+
+
+def test_the_trace_name_falls_back_to_the_constant_without_a_title_row(
+    hook_module, fake_langfuse, isolated_hook_state, tmp_path, monkeypatch: pytest.MonkeyPatch
+):
+    transcript = write_transcript(tmp_path, conversation_rows("turn-a") + conversation_rows("turn-b"))
+    trace_names = record_propagated_trace_names(hook_module, monkeypatch)
+    config = hook_module.LangfuseConfig("public", "secret", "https://example.test", "user-1")
+
+    hook_module.emit_new_turns_from_transcript(fake_langfuse, config, TITLED_SESSION_ID, transcript)
+    hook_module.emit_new_turns_from_transcript(
+        fake_langfuse, config, TITLED_SESSION_ID, transcript, flush_deferred_agent_turns=True)
+
+    assert trace_names
+    assert set(trace_names) == {hook_module.TRACE_NAME}