tests/agent/test_transcript_store.py

from datetime import UTC, datetime

from TerraFin.agent.loop import TerraFinConversationMessage
from TerraFin.agent.transcript_store import HostedTranscriptStore


def _ts(hour: int) -> datetime:
    return datetime(2026, 4, 16, hour, 0, tzinfo=UTC)


def test_transcript_store_derives_summary_and_conversation_from_events(tmp_path) -> None:
    store = HostedTranscriptStore(root_dir=tmp_path / "agent")
    store.create_session(
        session_id="session:alpha",
        agent_name="terrafin-assistant",
        created_at=_ts(9),
        runtime_model={
            "modelRef": "github-copilot/gpt-4o",
            "providerId": "github-copilot",
            "providerLabel": "GitHub Copilot",
            "modelId": "gpt-4o",
        },
        system_message=TerraFinConversationMessage(
            role="system",
            content="You are TerraFin Agent.",
            created_at=_ts(9),
        ),
    )
    store.append_message(
        "session:alpha",
        TerraFinConversationMessage(role="user", content="Check AAPL.", created_at=_ts(10)),
    )
    store.append_message(
        "session:alpha",
        TerraFinConversationMessage(role="assistant", content="AAPL looks stable.", created_at=_ts(11)),
    )
    store.append_message(
        "session:alpha",
        TerraFinConversationMessage(
            role="tool",
            content='{"ticker":"AAPL"}',
            created_at=_ts(12),
            name="market_snapshot",
        ),
    )

    summary = store.build_summary("session:alpha")
    conversation = store.load_conversation("session:alpha")

    assert summary.title == "Check AAPL."
    assert summary.last_message_preview == "AAPL looks stable."
    assert summary.message_count == 2
    assert summary.runtime_model is not None
    assert summary.runtime_model["modelRef"] == "github-copilot/gpt-4o"
    assert [message.role for message in conversation.snapshot()] == [
        "system",
        "user",
        "assistant",
        "tool",
    ]


def test_transcript_store_archives_deleted_sessions(tmp_path) -> None:
    store = HostedTranscriptStore(root_dir=tmp_path / "agent")
    store.create_session(
        session_id="session:delete-me",
        agent_name="terrafin-assistant",
        created_at=_ts(9),
    )

    archived = store.archive_session("session:delete-me", deleted_at=_ts(10))

    assert archived.deleted_at == _ts(10)
    assert store.session_exists("session:delete-me") is False
    assert store.list_sessions() == ()
    assert len(store.list_sessions(include_deleted=True)) == 1
    assert list((tmp_path / "agent" / "sessions").glob("session:delete-me.deleted.*.jsonl"))