spikes/007-real-trace-ingestion/tests/test_ingester.py

"""Spike 007 ingestion tests — Claude Code JSONL → TraceState.

Uses fixtures/synthetic_session.jsonl which conforms to the Claude Code 2.1.x
schema. Real-session test (skipped if no local sessions) is included as a
sanity check; CI users can ignore it.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

HERE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(HERE))

from claude_code_ingester import (  # noqa: E402
    ClaudeCodeIngester,
    IngestionStats,
    SYSTEM_PROMPT,
)


FIXTURE = HERE / "fixtures" / "synthetic_session.jsonl"


# ---------------------------------------------------------------------
# Synthetic-fixture tests (always run, deterministic)
# ---------------------------------------------------------------------

def test_fixture_exists():
    assert FIXTURE.exists(), f"missing test fixture: {FIXTURE}"


def test_ingest_emits_three_states():
    """Synthetic session has 3 assistant turns → 3 TraceState records."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    assert len(states) == 3, (
        f"expected 3 states (3 assistant turns), got {len(states)}"
    )


def test_state_id_uniqueness():
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    ids = [s["state_id"] for s in states]
    assert len(ids) == len(set(ids)), f"non-unique state_ids: {ids}"


def test_messages_history_grows():
    """Each subsequent state's messages list should be longer than the previous."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    lengths = [len(s["messages"]) for s in states]
    for i in range(1, len(lengths)):
        assert lengths[i] > lengths[i - 1], (
            f"history did not grow: {lengths}"
        )


def test_first_state_has_system_prompt_and_user_message():
    """State 0 has [system, user] in messages (history before first asst turn)."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    assert states[0]["messages"][0]["role"] == "system"
    assert states[0]["messages"][0]["content"] == SYSTEM_PROMPT
    assert states[0]["messages"][1]["role"] == "user"
    assert "1MB" in states[0]["messages"][1]["content"]


def test_thinking_stripped_from_teacher_history():
    """The thinking block in turn 1 should not appear in turn 2's messages history."""
    ingester = ClaudeCodeIngester(strip_thinking=True)
    states = list(ingester.ingest(FIXTURE))
    # State 1's history includes the assistant's first turn (which had a thinking block)
    history_2 = states[1]["messages"]
    asst_msgs_2 = [m for m in history_2 if m["role"] == "assistant"]
    assert len(asst_msgs_2) == 1, "state 1 should have 1 prior assistant turn"
    assert "[THINKING]" not in asst_msgs_2[0]["content"], (
        f"thinking leaked into teacher history: {asst_msgs_2[0]['content']!r}"
    )


def test_thinking_kept_in_student_action():
    """State 0 first assistant turn HAD a thinking block — must appear in student_action."""
    ingester = ClaudeCodeIngester(strip_thinking=True)
    states = list(ingester.ingest(FIXTURE))
    assert "[THINKING]" in states[0]["student_action"], (
        f"thinking missing from student_action: {states[0]['student_action']!r}"
    )


def test_tool_use_serialization():
    """Tool use blocks should be serialized as [TOOL_USE] name=... input=..."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    assert "[TOOL_USE]" in states[0]["student_action"]
    assert "name=Bash" in states[0]["student_action"]
    # Input should be JSON
    assert "find" in states[0]["student_action"]


def test_tool_result_in_user_history():
    """The tool_result observation should be in state 1's history as a user msg."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(FIXTURE))
    history_1 = states[1]["messages"]
    user_msgs = [m for m in history_1 if m["role"] == "user"]
    assert any("[TOOL_RESULT]" in m["content"] for m in user_msgs), (
        f"tool_result missing from history: {user_msgs}"
    )


def test_summary_records_skipped():
    ingester = ClaudeCodeIngester()
    list(ingester.ingest(FIXTURE))
    assert ingester.last_stats.skipped_summary >= 1


def test_stats_populated():
    ingester = ClaudeCodeIngester()
    list(ingester.ingest(FIXTURE))
    s = ingester.last_stats
    assert s.n_assistant_turns == 3
    assert s.n_tool_use_blocks == 2
    assert s.n_text_blocks >= 2  # 2 turns have text blocks
    assert s.n_states_emitted == 3


# ---------------------------------------------------------------------
# Subagent skip
# ---------------------------------------------------------------------

def test_subagent_filename_skipped(tmp_path):
    """Files starting with `agent-` should be entirely skipped."""
    fake = tmp_path / "agent-12345.jsonl"
    fake.write_text(FIXTURE.read_text())
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(fake))
    assert states == [], "subagent file should yield nothing"


def test_sidechain_records_skipped(tmp_path):
    """isSidechain=true records should be skipped."""
    fake = tmp_path / "with_sidechain.jsonl"
    raw = FIXTURE.read_text().splitlines()
    # Add a sidechain assistant record
    sidechain = {
        "type": "assistant",
        "uuid": "side1",
        "parentUuid": "a6",
        "sessionId": "test-session",
        "timestamp": "2026-05-26T10:00:20Z",
        "cwd": "/tmp/test",
        "version": "2.1.143",
        "isSidechain": True,
        "message": {
            "role": "assistant",
            "model": "claude-opus-4-7",
            "content": [{"type": "text", "text": "subagent talking"}],
        },
    }
    raw.append(json.dumps(sidechain))
    fake.write_text("\n".join(raw) + "\n")

    ingester = ClaudeCodeIngester(skip_sidechain=True)
    list(ingester.ingest(fake))
    assert ingester.last_stats.skipped_subagent >= 1


# ---------------------------------------------------------------------
# Error tolerance
# ---------------------------------------------------------------------

def test_truncated_line_tolerated(tmp_path):
    """A truncated/malformed JSON line should be skipped, not crash the ingester."""
    fake = tmp_path / "broken.jsonl"
    raw = FIXTURE.read_text().splitlines()
    raw.insert(2, '{"type": "assistant", "message": {bad json')
    fake.write_text("\n".join(raw) + "\n")

    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(fake))
    assert ingester.last_stats.skipped_truncated_lines == 1
    assert len(states) == 3, "valid records should still parse"


# ---------------------------------------------------------------------
# Real session smoke (skipped if not present)
# ---------------------------------------------------------------------

REAL_SESSION = Path(
    "/home/codeseys/.claude/projects/-mnt-e-CS-github-VIGOR--overstory-worktrees-builder-iteration-checkpoint/e4a34e2b-40c6-49ce-b253-912a43224aae.jsonl"
)


@pytest.mark.skipif(not REAL_SESSION.exists(), reason="real Claude Code session not on this machine")
def test_real_session_ingest_smoke():
    """Sanity-check the ingester on a real session — should yield ≥10 states with no exceptions."""
    ingester = ClaudeCodeIngester()
    states = list(ingester.ingest(REAL_SESSION))
    assert len(states) >= 10, f"expected ≥10 states from real session, got {len(states)}"
    # Spot-check: every state should have a non-empty student_action
    for i, s in enumerate(states):
        assert s["student_action"], f"empty student_action at state {i}"
        assert s["messages"], f"empty messages at state {i}"
    # No version warnings on a known-good session
    assert not ingester.last_stats.version_warnings, (
        f"unexpected version warnings: {ingester.last_stats.version_warnings}"
    )