feat: support Claude Code transcripts

server.py

@@ -795,6 +795,71 @@ def _parse_antigravity_transcript(text: str) -> list[dict[str, str]]:
     return pairs
 
 
+def _content_to_text(content: Any) -> str:
+    """Extract readable text from Claude Code message content blocks."""
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        chunks: list[str] = []
+        for item in content:
+            if isinstance(item, str):
+                chunks.append(item)
+            elif isinstance(item, dict) and item.get("type") == "text":
+                chunks.append(str(item.get("text", "")))
+        return "\n".join(chunk.strip() for chunk in chunks if chunk.strip()).strip()
+    return ""
+
+
+def _parse_claude_code_transcript(text: str) -> list[dict[str, str]]:
+    """Parse Claude Code JSONL transcripts into message pairs."""
+    pairs: list[dict[str, str]] = []
+    current_user_query: str | None = None
+    assistant_chunks: list[str] = []
+
+    for raw_line in text.splitlines():
+        raw_line = raw_line.strip()
+        if not raw_line:
+            continue
+
+        try:
+            event = json.loads(raw_line)
+        except json.JSONDecodeError:
+            continue
+
+        if not isinstance(event, dict):
+            continue
+
+        has_nested_message = isinstance(event.get("message"), dict)
+        message = event["message"] if has_nested_message else event
+        role = message.get("role") or (event.get("type") if not has_nested_message else None)
+        content = _content_to_text(message.get("content"))
+        if not content:
+            continue
+
+        if role == "user":
+            if current_user_query and assistant_chunks:
+                pairs.append({
+                    "user_query": current_user_query,
+                    "agent_response": "\n\n".join(assistant_chunks).strip(),
+                })
+                current_user_query = content
+                assistant_chunks = []
+            elif current_user_query:
+                current_user_query = f"{current_user_query}\n\n{content}"
+            else:
+                current_user_query = content
+        elif role == "assistant" and current_user_query:
+            assistant_chunks.append(content)
+
+    if current_user_query and assistant_chunks:
+        pairs.append({
+            "user_query": current_user_query,
+            "agent_response": "\n\n".join(assistant_chunks).strip(),
+        })
+
+    return pairs
+
+
 async def _parse_transcript_with_llm(text: str) -> list[dict[str, str]]:
     """Use an LLM to parse transcript text when format detection fails."""
     from src.models import get_model
@@ -854,6 +919,10 @@ def _parse_transcript_text(text: str) -> tuple[str, list[dict[str, str]]]:
         if pairs:
             return "antigravity", pairs
 
+    pairs = _parse_claude_code_transcript(text)
+    if pairs:
+        return "claude_code", pairs
+
     return "unknown", []
 
 

src/api/routes/memory.py

@@ -447,6 +447,77 @@ def _parse_antigravity_transcript(text: str) -> List[MessagePair]:
     return pairs
 
 
+def _content_to_text(content: Any) -> str:
+    """Extract readable text from Claude Code message content blocks."""
+    if isinstance(content, str):
+        return content.strip()
+    if isinstance(content, list):
+        chunks: List[str] = []
+        for item in content:
+            if isinstance(item, str):
+                chunks.append(item)
+            elif isinstance(item, dict) and item.get("type") == "text":
+                chunks.append(str(item.get("text", "")))
+        return "\n".join(chunk.strip() for chunk in chunks if chunk.strip()).strip()
+    return ""
+
+
+def _parse_claude_code_transcript(text: str) -> List[MessagePair]:
+    """Parse Claude Code JSONL transcripts into message pairs.
+
+    Claude Code stores/export transcripts as newline-delimited JSON objects. User
+    and assistant turns live under ``message.role`` and ``message.content``.
+    Tool calls/results are intentionally ignored so only conversational text is
+    sent to the memory pipeline.
+    """
+    pairs: List[MessagePair] = []
+    current_user_query: str | None = None
+    assistant_chunks: List[str] = []
+
+    for raw_line in text.splitlines():
+        raw_line = raw_line.strip()
+        if not raw_line:
+            continue
+
+        try:
+            event = json.loads(raw_line)
+        except json.JSONDecodeError:
+            continue
+
+        if not isinstance(event, dict):
+            continue
+
+        has_nested_message = isinstance(event.get("message"), dict)
+        message = event["message"] if has_nested_message else event
+        role = message.get("role") or (event.get("type") if not has_nested_message else None)
+        content = _content_to_text(message.get("content"))
+        if not content:
+            continue
+
+        if role == "user":
+            if current_user_query and assistant_chunks:
+                pairs.append(MessagePair(
+                    user_query=current_user_query,
+                    agent_response="\n\n".join(assistant_chunks).strip(),
+                ))
+                current_user_query = content
+                assistant_chunks = []
+            elif current_user_query:
+                current_user_query = f"{current_user_query}\n\n{content}"
+            else:
+                current_user_query = content
+        elif role == "assistant" and current_user_query:
+            assistant_chunks.append(content)
+
+    if current_user_query and assistant_chunks:
+        pairs.append(MessagePair(
+            user_query=current_user_query,
+            agent_response="\n\n".join(assistant_chunks).strip(),
+        ))
+
+    return pairs
+
+
 async def _parse_transcript_with_llm(text: str) -> List[MessagePair]:
     """Use an LLM to parse transcript text when format detection fails."""
     from src.models import get_model
@@ -507,6 +578,10 @@ def _parse_transcript_text(text: str) -> tuple[str, List[MessagePair]]:
         if pairs:
             return "antigravity", pairs
 
+    pairs = _parse_claude_code_transcript(text)
+    if pairs:
+        return "claude_code", pairs
+
     return "unknown", []
 
 

tests/test_claude_code_transcript.py

@@ -0,0 +1,73 @@
+import ast
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, List
+
+
+@dataclass
+class MessagePair:
+    user_query: str
+    agent_response: str
+
+
+def load_parser():
+    source = Path("src/api/routes/memory.py").read_text()
+    tree = ast.parse(source)
+    wanted = {
+        "_parse_cursor_transcript",
+        "_parse_antigravity_transcript",
+        "_content_to_text",
+        "_parse_claude_code_transcript",
+        "_parse_transcript_text",
+    }
+    module = ast.Module(
+        body=[node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name in wanted],
+        type_ignores=[],
+    )
+    ast.fix_missing_locations(module)
+    namespace = {
+        "Any": Any,
+        "List": List,
+        "MessagePair": MessagePair,
+        "json": json,
+        "re": re,
+    }
+    exec(compile(module, "memory_parser_subset", "exec"), namespace)
+    return namespace["_parse_transcript_text"]
+
+
+def test_parse_claude_code_jsonl_transcript():
+    parse_transcript_text = load_parser()
+    transcript = "\n".join([
+        '{"type":"user","message":{"role":"user","content":"Add tests for login"}}',
+        '{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"I added the login tests."},{"type":"tool_use","name":"Bash"}]}}',
+        '{"type":"user","message":{"role":"user","content":[{"type":"text","text":"Run them"}]}}',
+        '{"type":"assistant","message":{"role":"assistant","content":"All tests passed."}}',
+    ])
+
+    format_detected, pairs = parse_transcript_text(transcript)
+
+    assert format_detected == "claude_code"
+    assert len(pairs) == 2
+    assert pairs[0].user_query == "Add tests for login"
+    assert pairs[0].agent_response == "I added the login tests."
+    assert pairs[1].user_query == "Run them"
+    assert pairs[1].agent_response == "All tests passed."
+
+
+def test_parse_claude_code_ignores_tool_only_blocks():
+    parse_transcript_text = load_parser()
+    transcript = "\n".join([
+        '{"message":{"role":"user","content":"Inspect the repo"}}',
+        '{"message":{"role":"assistant","content":[{"type":"tool_use","name":"Read"}]}}',
+        '{"message":{"role":"assistant","content":[{"type":"text","text":"The repo uses FastAPI."}]}}',
+    ])
+
+    format_detected, pairs = parse_transcript_text(transcript)
+
+    assert format_detected == "claude_code"
+    assert len(pairs) == 1
+    assert pairs[0].user_query == "Inspect the repo"
+    assert pairs[0].agent_response == "The repo uses FastAPI."