chemenu/tools/chemenu/tests/test_import_chronicle.py

"""Reconstructing a trace from a chronicle store.

The fixture builds a store with the schema both Copilot CLI and VS Code Chat
use, so these tests pin the reconstruction without depending on a populated
store on the developer's machine.
"""
import json
import sqlite3

import pytest

import import_chronicle
from chemenu.telemetry import schema

SCHEMA = """
CREATE TABLE sessions (
    id TEXT PRIMARY KEY, cwd TEXT, repository TEXT, host_type TEXT, branch TEXT,
    summary TEXT, agent_name TEXT, agent_description TEXT,
    created_at TEXT, updated_at TEXT
);
CREATE TABLE turns (
    id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, turn_index INTEGER,
    user_message TEXT, assistant_response TEXT, timestamp TEXT
);
CREATE TABLE session_files (
    id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, file_path TEXT,
    tool_name TEXT, turn_index INTEGER, first_seen_at TEXT
);
"""


@pytest.fixture
def store(tmp_path):
    db = tmp_path / "session-store.db"
    connection = sqlite3.connect(db)
    connection.executescript(SCHEMA)
    connection.execute(
        "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?,?,?)",
        ("sess-0001", "/home/u/src/chemenu", "chemenu", "vscode",
         "main", "Built the telemetry layer", "Copilot", "default",
         "2026-08-23T10:00:00.000Z", "2026-08-23T11:30:00.000Z"),
    )
    connection.execute(
        "INSERT INTO sessions VALUES (?,?,?,?,?,?,?,?,?,?)",
        ("sess-0002", "/home/u/src/other-repo", "other-repo", "vscode",
         "main", "Unrelated work", "Copilot", "default",
         "2026-08-22T10:00:00.000Z", "2026-08-22T10:30:00.000Z"),
    )
    connection.executemany(
        "INSERT INTO turns (session_id, turn_index, user_message, assistant_response, timestamp) "
        "VALUES (?,?,?,?,?)",
        [
            ("sess-0001", 0, "build the emitter", "Done.", "2026-08-23T10:05:00.000Z"),
            ("sess-0001", 1, "now the tests", "Green.", "2026-08-23T10:40:00.000Z"),
        ],
    )
    connection.executemany(
        "INSERT INTO session_files (session_id, file_path, tool_name, turn_index, first_seen_at) "
        "VALUES (?,?,?,?,?)",
        [
            ("sess-0001", "tools/chemenu/telemetry/writer.py", "create_file", 0,
             "2026-08-23T10:06:00.000Z"),
            ("sess-0001", "tools/chemenu/tests/test_telemetry_emit.py", "create_file", 1,
             "2026-08-23T10:41:00.000Z"),
        ],
    )
    connection.commit()
    connection.close()
    return db


def read_trace(path):
    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]


def run(store, isolated_trace_dir, **kwargs):
    connection = import_chronicle.connect(store)
    session = connection.execute("SELECT * FROM sessions WHERE id = 'sess-0001'").fetchone()
    outcome, count = import_chronicle.import_session(
        connection, session, kwargs.get("force", False), kwargs.get("dry_run", False)
    )
    return outcome, count, isolated_trace_dir / "sess-0001" / "trace.jsonl"


def test_a_session_becomes_an_ordered_trace(store, isolated_trace_dir):
    outcome, count, path = run(store, isolated_trace_dir)
    assert outcome == "imported"
    records = read_trace(path)
    assert len(records) == count

    # A turn's prompt and reply share the store's single turn timestamp, while a
    # touched file carries its own, later one - so the file lands after the
    # reply. That is what the store knows; ordering it any other way would be
    # inventing a sequence nobody recorded.
    assert [r["event"] for r in records] == [
        "session.start",
        "prompt.submitted",
        "assistant.message",
        "tool.post",
        "prompt.submitted",
        "assistant.message",
        "tool.post",
        "session.end",
    ]
    assert [r["ts"] for r in records] == sorted(r["ts"] for r in records)


def test_the_trace_says_what_it_could_not_observe(store, isolated_trace_dir):
    _, _, path = run(store, isolated_trace_dir)
    start = read_trace(path)[0]
    assert start["attrs"]["reconstructed"] is True
    assert start["attrs"]["completeness"] == list(schema.HARNESS_CAPABILITIES["vscode-chat"])
    # The store records that a file was touched, not that a tool was about to
    # run - a scorer must be able to see that gap.
    assert "tool.pre" not in start["attrs"]["completeness"]


def test_original_timestamps_survive_the_import(store, isolated_trace_dir):
    _, _, path = run(store, isolated_trace_dir)
    records = read_trace(path)
    assert records[0]["ts"].startswith("2026-08-23T10:00:00")
    assert records[-1]["ts"].startswith("2026-08-23T11:30:00")
    # Normalised, not passed through: 'Z' and '+00:00' sort differently.
    assert records[0]["ts"].endswith("+00:00")


def test_prompts_and_replies_are_carried_over(store, isolated_trace_dir):
    _, _, path = run(store, isolated_trace_dir)
    records = read_trace(path)
    prompts = [r["attrs"]["prompt"] for r in records if r["event"] == "prompt.submitted"]
    replies = [r["attrs"]["message"] for r in records if r["event"] == "assistant.message"]
    assert prompts == ["build the emitter", "now the tests"]
    assert replies == ["Done.", "Green."]


def test_touched_files_become_tool_events(store, isolated_trace_dir):
    _, _, path = run(store, isolated_trace_dir)
    touched = [r["attrs"] for r in read_trace(path) if r["event"] == "tool.post"]
    assert [t["file_path"] for t in touched] == [
        "tools/chemenu/telemetry/writer.py",
        "tools/chemenu/tests/test_telemetry_emit.py",
    ]
    assert {t["tool_name"] for t in touched} == {"create_file"}


def test_import_is_idempotent_unless_forced(store, isolated_trace_dir):
    run(store, isolated_trace_dir)
    outcome, count, path = run(store, isolated_trace_dir)
    assert (outcome, count) == ("skipped", 0)
    assert len(read_trace(path)) == 8

    outcome, _, path = run(store, isolated_trace_dir, force=True)
    assert outcome == "imported"
    assert len(read_trace(path)) == 8  # replaced, not appended to


def test_dry_run_writes_nothing(store, isolated_trace_dir):
    outcome, count, path = run(store, isolated_trace_dir, dry_run=True)
    assert outcome == "would import"
    assert count == 8
    assert not path.exists()


def test_the_store_is_opened_read_only(store):
    connection = import_chronicle.connect(store)
    with pytest.raises(sqlite3.OperationalError):
        connection.execute("DELETE FROM sessions")