Add a Claude Code session-transcript adapter

skillloop/adapters/claude_code.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from skillloop.schema import AgentMessage, AgentTrace, ToolCall, sha256_text
+
+ADAPTER_NAME = "claude_code"
+ADAPTER_VERSION = "1.0"
+
+_VALID_ROLES = {"system", "user", "assistant", "tool"}
+
+
+def _message_obj(line: dict[str, Any]) -> dict[str, Any] | None:
+    """Return the message dict for a transcript line, or None for meta lines.
+
+    Claude Code session transcripts interleave message lines
+    (``{"message": {"role", "content": [...]}, ...}``) with non-message meta
+    lines (operation events, summaries) that carry no role/content.
+    """
+    msg = line.get("message")
+    if isinstance(msg, dict) and msg.get("role"):
+        return msg
+    if line.get("role") and "content" in line:
+        return line
+    return None
+
+
+def _text_from_blocks(blocks: list[Any]) -> str:
+    parts: list[str] = []
+    for block in blocks:
+        if isinstance(block, dict) and block.get("type") == "text" and block.get("text"):
+            parts.append(str(block["text"]))
+    return "\n".join(parts)
+
+
+def _tool_result_text(content: Any) -> str:
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for block in content:
+            if isinstance(block, dict):
+                if block.get("type") == "text" and block.get("text") is not None:
+                    parts.append(str(block["text"]))
+                elif block.get("content") is not None:
+                    parts.append(str(block["content"]))
+            else:
+                parts.append(str(block))
+        return "\n".join(parts)
+    return str(content)
+
+
+def _parse_ts(value: Any) -> datetime | None:
+    if not value:
+        return None
+    try:
+        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def normalize_claude_code_session(raw_text: str, *, include_sidechains: bool = True) -> tuple[list[AgentMessage], dict[str, Any]]:
+    lines: list[dict[str, Any]] = []
+    for raw in raw_text.splitlines():
+        raw = raw.strip()
+        if not raw:
+            continue
+        try:
+            parsed = json.loads(raw)
+        except json.JSONDecodeError:
+            continue  # tolerate a partial trailing line on a live session
+        if isinstance(parsed, dict):
+            lines.append(parsed)
+    if not include_sidechains:
+        # Drop subagent sidechain turns before both passes so tool_use/tool_result
+        # matching stays internally consistent.
+        lines = [line for line in lines if not line.get("isSidechain")]
+
+    # Pass 1: index tool_result blocks by tool_use_id. In the Anthropic message
+    # format these arrive in a later user turn, so results are matched back to
+    # the assistant's originating tool_use across messages.
+    results: dict[str, dict[str, Any]] = {}
+    for line in lines:
+        msg = _message_obj(line)
+        if not msg or not isinstance(msg.get("content"), list):
+            continue
+        for block in msg["content"]:
+            if isinstance(block, dict) and block.get("type") == "tool_result" and block.get("tool_use_id"):
+                results[str(block["tool_use_id"])] = {
+                    "result": _tool_result_text(block.get("content")),
+                    "is_error": bool(block.get("is_error")),
+                    "ended_at": line.get("timestamp"),
+                }
+
+    # Pass 2: build normalized messages.
+    messages: list[AgentMessage] = []
+    session_id: Any = None
+    for line in lines:
+        if session_id is None and line.get("sessionId"):
+            session_id = line.get("sessionId")
+        msg = _message_obj(line)
+        if not msg:
+            continue
+        role = str(msg.get("role") or "")
+        if role not in _VALID_ROLES:
+            continue
+        content = msg.get("content")
+        ts = line.get("timestamp")
+
+        text = ""
+        thinking_parts: list[str] = []
+        redacted_thinking = False
+        tool_calls: list[ToolCall] = []
+        if isinstance(content, str):
+            text = content
+        elif isinstance(content, list):
+            text = _text_from_blocks(content)
+            for block in content:
+                if not isinstance(block, dict):
+                    continue
+                btype = block.get("type")
+                if btype == "thinking":
+                    # Forward-compatible: capture extended-thinking text when the
+                    # provider includes it. NOTE: Claude Code persists thinking
+                    # blocks with an empty text field plus a signature, so the
+                    # reasoning text is stripped from the transcript and there is
+                    # nothing to preserve from a Claude Code session.
+                    if block.get("thinking"):
+                        thinking_parts.append(str(block["thinking"]))
+                elif btype == "redacted_thinking":
+                    redacted_thinking = True
+                elif btype == "tool_use":
+                    tuid = str(block.get("id") or "")
+                    res = results.get(tuid, {})
+                    raw_args = block.get("input")
+                    arguments = raw_args if isinstance(raw_args, dict) else {}
+                    started, ended = _parse_ts(ts), _parse_ts(res.get("ended_at"))
+                    duration_ms = int((ended - started).total_seconds() * 1000) if (started and ended) else None
+                    status = ("error" if res.get("is_error") else "success") if res else "unknown"
+                    tool_calls.append(
+                        ToolCall(
+                            name=str(block.get("name") or "unknown"),
+                            arguments=arguments,
+                            result=res.get("result"),
+                            id=tuid or None,
+                            started_at=ts,
+                            ended_at=res.get("ended_at"),
+                            duration_ms=duration_ms,
+                            status=status,
+                        )
+                    )
+
+        # Skip turns with no usable signal (e.g. a pure tool_result user turn).
+        if not text.strip() and not tool_calls and not thinking_parts:
+            continue
+
+        metadata = {
+            k: line[k]
+            for k in ("uuid", "parentUuid", "isSidechain", "timestamp")
+            if line.get(k) is not None
+        }
+        # Preserve extended-thinking reasoning out of band so it is not lost,
+        # while keeping the human-readable content field clean.
+        if thinking_parts:
+            metadata["thinking"] = "\n".join(thinking_parts)
+        if redacted_thinking:
+            metadata["thinking_redacted"] = True
+
+        messages.append(
+            AgentMessage(role=role, content=text, tool_calls=tool_calls, metadata=metadata)
+        )
+
+    meta = {"session_id": session_id, "line_count": len(lines), "message_count": len(messages)}
+    return messages, meta
+
+
+def load_claude_code_session(path: str | Path, *, include_sidechains: bool = True) -> AgentTrace:
+    source_path = Path(path).expanduser()
+    raw_text = source_path.read_text(encoding="utf-8")
+    messages, meta = normalize_claude_code_session(raw_text, include_sidechains=include_sidechains)
+    if not messages:
+        raise ValueError(f"No usable messages found in Claude Code session: {source_path}")
+    return AgentTrace(
+        source="claude_code",
+        messages=messages,
+        adapter={"name": ADAPTER_NAME, "version": ADAPTER_VERSION},
+        runtime={"name": "claude_code"},
+        metadata={"path": str(source_path), "project": source_path.parent.name, **meta},
+        raw_artifact_ref=str(source_path),
+        raw_trace_sha256=sha256_text(raw_text),
+    )
+
+
+def latest_claude_code_session(projects_dir: str | Path | None = None, project: str | None = None) -> Path:
+    """Newest Claude Code session transcript, without mutating anything.
+
+    Session files live one level under each project: ``<projects_dir>/<slug>/<id>.jsonl``.
+    Subagent transcripts under ``<slug>/subagents/`` are intentionally excluded.
+    """
+    base = Path(projects_dir).expanduser() if projects_dir else (Path.home() / ".claude" / "projects")
+    if not base.exists():
+        raise FileNotFoundError(f"Claude Code projects dir not found: {base}")
+    candidates = list((base / project).glob("*.jsonl")) if project else list(base.glob("*/*.jsonl"))
+    if not candidates:
+        raise FileNotFoundError(f"No Claude Code session transcripts found under {base}")
+    return max(candidates, key=lambda p: p.stat().st_mtime)

skillloop/cli.py

@@ -5,6 +5,7 @@
 import sys
 from pathlib import Path
 
+from skillloop.adapters.claude_code import latest_claude_code_session, load_claude_code_session
 from skillloop.adapters.generic_jsonl import load_generic_jsonl
 from skillloop.adapters.hermes import load_hermes_export, load_hermes_state_db
 from skillloop.apply.filesystem import export_approved
@@ -66,6 +67,14 @@ def cmd_ingest(args: argparse.Namespace) -> int:
         trace = load_hermes_export(args.input)
     elif args.adapter == "hermes-db":
         trace = load_hermes_state_db(args.db_path, session_id=args.session_id, latest=args.latest)
+    elif args.adapter == "claude-code":
+        if args.latest:
+            source = latest_claude_code_session(args.projects_dir, project=args.project)
+        elif args.input:
+            source = args.input
+        else:
+            raise SystemExit("claude-code ingest requires an input .jsonl path or --latest")
+        trace = load_claude_code_session(source, include_sidechains=not args.no_sidechains)
     else:
         raise SystemExit(f"Unsupported adapter: {args.adapter}")
     trace_id = store.save_trace(trace)
@@ -355,11 +364,14 @@ def build_parser() -> argparse.ArgumentParser:
     p_init.set_defaults(func=cmd_init)
 
     p_ingest = sub.add_parser("ingest", help="Ingest a trace")
-    p_ingest.add_argument("adapter", choices=["generic", "hermes", "hermes-db"])
-    p_ingest.add_argument("input", nargs="?", help="Input JSONL/JSON path for generic or hermes adapters")
+    p_ingest.add_argument("adapter", choices=["generic", "hermes", "hermes-db", "claude-code"])
+    p_ingest.add_argument("input", nargs="?", help="Input JSONL/JSON path for generic, hermes, or claude-code adapters")
     p_ingest.add_argument("--db-path", default=str(Path.home() / ".hermes" / "state.db"), help="Hermes state.db path for hermes-db adapter")
     p_ingest.add_argument("--session-id", default=None, help="Hermes session id for hermes-db adapter")
-    p_ingest.add_argument("--latest", action="store_true", help="Use latest Hermes session with messages for hermes-db adapter")
+    p_ingest.add_argument("--latest", action="store_true", help="Use the latest session (hermes-db: newest session with messages; claude-code: newest transcript)")
+    p_ingest.add_argument("--projects-dir", default=str(Path.home() / ".claude" / "projects"), help="Claude Code projects dir for the claude-code adapter")
+    p_ingest.add_argument("--project", default=None, help="Claude Code project slug to scope --latest to one project")
+    p_ingest.add_argument("--no-sidechains", action="store_true", help="Exclude subagent sidechain turns (claude-code adapter)")
     p_ingest.set_defaults(func=cmd_ingest)
 
     p_traces = sub.add_parser("traces", help="List/show traces")

tests/test_claude_code.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import json
+import os
+
+from skillloop.adapters.claude_code import latest_claude_code_session, load_claude_code_session
+
+
+def _write_session(path, lines):
+    path.write_text("\n".join(json.dumps(line) for line in lines) + "\n", encoding="utf-8")
+
+
+def test_claude_code_adapter_parses_blocks_and_tool_results(tmp_path):
+    lines = [
+        {"type": "summary", "summary": "meta line, no message"},
+        {"type": "user", "uuid": "u1", "timestamp": "2026-06-13T10:00:00Z",
+         "message": {"role": "user", "content": [{"type": "text", "text": "fix the bug"}]}},
+        {"type": "assistant", "uuid": "a1", "timestamp": "2026-06-13T10:00:01Z",
+         "message": {"role": "assistant", "content": [
+             {"type": "text", "text": "running tests"},
+             {"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": {"command": "pytest"}},
+         ]}},
+        {"type": "user", "uuid": "u2", "timestamp": "2026-06-13T10:00:03Z",
+         "message": {"role": "user", "content": [
+             {"type": "tool_result", "tool_use_id": "toolu_1", "content": "2 passed", "is_error": False},
+         ]}},
+        {"type": "assistant", "uuid": "a2", "timestamp": "2026-06-13T10:00:04Z",
+         "message": {"role": "assistant", "content": [{"type": "text", "text": "done, tests pass"}]}},
+    ]
+    session = tmp_path / "session.jsonl"
+    _write_session(session, lines)
+
+    trace = load_claude_code_session(session)
+
+    assert trace.source == "claude_code"
+    assert trace.adapter["name"] == "claude_code"
+    # the meta line and the pure tool_result turn carry no signal and are dropped
+    assert [m.role for m in trace.messages] == ["user", "assistant", "assistant"]
+    assert trace.messages[0].content == "fix the bug"
+
+    asst = trace.messages[1]
+    assert asst.content == "running tests"
+    assert len(asst.tool_calls) == 1
+    call = asst.tool_calls[0]
+    assert call.name == "Bash"
+    assert call.arguments == {"command": "pytest"}
+    assert call.result == "2 passed"
+    assert call.status == "success"
+    assert call.success is True
+    assert call.duration_ms == 2000  # 10:00:01 -> 10:00:03
+
+    assert trace.messages[2].content == "done, tests pass"
+
+
+def test_claude_code_adapter_rejects_empty_session(tmp_path):
+    session = tmp_path / "empty.jsonl"
+    _write_session(session, [{"type": "summary", "summary": "only meta"}])
+    try:
+        load_claude_code_session(session)
+    except ValueError:
+        return
+    raise AssertionError("expected ValueError for a session with no usable messages")
+
+
+def test_latest_claude_code_session_picks_newest(tmp_path):
+    proj = tmp_path / "proj-a"
+    proj.mkdir()
+    older, newer = proj / "older.jsonl", proj / "newer.jsonl"
+    msg = {"type": "user", "message": {"role": "user", "content": [{"type": "text", "text": "hi"}]}}
+    _write_session(older, [msg])
+    _write_session(newer, [msg])
+    os.utime(older, (1, 1))
+    os.utime(newer, (10, 10))
+    assert latest_claude_code_session(tmp_path).name == "newer.jsonl"
+
+
+def test_claude_code_preserves_thinking_in_metadata(tmp_path):
+    lines = [
+        {"type": "assistant", "uuid": "a1", "timestamp": "2026-06-13T10:00:00Z",
+         "message": {"role": "assistant", "content": [
+             {"type": "thinking", "thinking": "the user wants X, so I should do Y"},
+             {"type": "text", "text": "here is the answer"},
+         ]}},
+    ]
+    session = tmp_path / "thinking.jsonl"
+    _write_session(session, lines)
+    trace = load_claude_code_session(session)
+
+    assert len(trace.messages) == 1
+    msg = trace.messages[0]
+    assert msg.content == "here is the answer"  # content is not muddied by thinking
+    assert msg.metadata.get("thinking") == "the user wants X, so I should do Y"
+
+
+def test_claude_code_no_sidechains_filter(tmp_path):
+    lines = [
+        {"type": "user", "uuid": "u1",
+         "message": {"role": "user", "content": [{"type": "text", "text": "main turn"}]}},
+        {"type": "assistant", "uuid": "s1", "isSidechain": True,
+         "message": {"role": "assistant", "content": [{"type": "text", "text": "subagent turn"}]}},
+    ]
+    session = tmp_path / "sidechain.jsonl"
+    _write_session(session, lines)
+
+    full = load_claude_code_session(session, include_sidechains=True)
+    assert [m.content for m in full.messages] == ["main turn", "subagent turn"]
+
+    filtered = load_claude_code_session(session, include_sidechains=False)
+    assert [m.content for m in filtered.messages] == ["main turn"]
+
+
+def test_claude_code_empty_thinking_text_is_not_preserved(tmp_path):
+    # Claude Code persists thinking blocks with empty text and only a signature,
+    # so there is no reasoning text to capture.
+    lines = [
+        {"type": "assistant", "uuid": "a1",
+         "message": {"role": "assistant", "content": [
+             {"type": "thinking", "thinking": "", "signature": "sig-abc"},
+             {"type": "text", "text": "answer"},
+         ]}},
+    ]
+    session = tmp_path / "empty_think.jsonl"
+    _write_session(session, lines)
+    msg = load_claude_code_session(session).messages[0]
+    assert msg.content == "answer"
+    assert "thinking" not in msg.metadata