nisse/nisse-dev/tests/test_canonical.py

import sqlite3
from pathlib import Path

import pytest

from nisse_dev.canonical import (
    AssistantText,
    Compaction,
    DelegationCompleted,
    DelegationStarted,
    GateOpened,
    PlanSubmitted,
    ToolCall,
    ToolResult,
    UserMessage,
    from_payload,
    mint_call_id,
)
from nisse_dev.state import CURRENT_SCHEMA_VERSION, StateStore

ALL_EVENTS = [
    UserMessage(text="use demo", source="matrix:$abc"),
    AssistantText(text="starting", model="local/qwen"),
    ToolCall(call_id="call-1", tool="run_tests", arguments={"path": "tests/"}),
    ToolResult(call_id="call-1", result="3 passed", success=True),
    PlanSubmitted(plan="the plan", model="claude-opus-5"),
    GateOpened(gate="arch", reason="plan submitted (rev-1)"),
    DelegationStarted(
        brief="fix the flaky test",
        role="delegate",
        model="claude-opus-5",
        caps={"turns": 20, "tokens": 100000},
    ),
    DelegationCompleted(
        handback="test fixed; root cause was a shared tmpdir",
        artifacts=["nisse-dev/tests/test_x.py"],
        tokens=51234,
        duration_s=61.5,
        exit_status="success",
    ),
    Compaction(summary="earlier work summarized", model="local/qwen", replaces=[1, 4]),
]


def test_all_event_types_round_trip(tmp_path: Path) -> None:
    with StateStore(tmp_path / "state.db") as store:
        for event in ALL_EVENTS:
            store.append_event(event)
    with StateStore(tmp_path / "state.db") as store:
        stored = store.read_events()
    assert [s.event for s in stored] == ALL_EVENTS
    assert [s.id for s in stored] == sorted(s.id for s in stored)


def test_read_events_after_id(tmp_path: Path) -> None:
    with StateStore(tmp_path / "state.db") as store:
        first = store.append_event(UserMessage(text="one", source="m:1"))
        store.append_event(UserMessage(text="two", source="m:2"))
        tail = store.read_events(after=first)
    assert [s.event.text for s in tail] == ["two"]


def test_mint_call_id_is_unique_and_prefixed() -> None:
    ids = {mint_call_id() for _ in range(100)}
    assert len(ids) == 100
    assert all(i.startswith("call-") for i in ids)


def test_unknown_event_type_rejected() -> None:
    with pytest.raises(ValueError, match="unknown canonical event type"):
        from_payload("mystery", {})


def test_unknown_fields_rejected() -> None:
    with pytest.raises(ValueError, match="unknown fields"):
        from_payload("user_message", {"text": "hi", "source": "m", "extra": 1})


# The v1 schema as it shipped (PR #2), frozen here to prove a newer image
# opens and upgrades a database written by the older one (L8).
V1_SCHEMA = [
    "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)",
    """
    CREATE TABLE agent_state (
        id INTEGER PRIMARY KEY CHECK (id = 1),
        state TEXT NOT NULL,
        project TEXT,
        data TEXT NOT NULL DEFAULT '{}',
        updated_at TEXT NOT NULL
    )
    """,
    """
    CREATE TABLE intent_log (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        kind TEXT NOT NULL,
        payload TEXT NOT NULL,
        status TEXT NOT NULL DEFAULT 'pending',
        created_at TEXT NOT NULL,
        resolved_at TEXT
    )
    """,
]


def test_default_migration_upgrades_a_real_v1_store(tmp_path: Path) -> None:
    db = tmp_path / "state.db"
    conn = sqlite3.connect(db)
    for statement in V1_SCHEMA:
        conn.execute(statement)
    conn.execute("INSERT INTO meta (key, value) VALUES ('schema_version', '1')")
    conn.execute(
        "INSERT INTO agent_state (id, state, project, data, updated_at) "
        "VALUES (1, 'IDLE', NULL, '{}', 't')"
    )
    conn.commit()
    conn.close()

    with StateStore(db) as store:
        assert store.schema_version == CURRENT_SCHEMA_VERSION
        state = store.get_state()
        assert state is not None
        assert state.state == "IDLE"  # v1 data intact
        store.append_event(UserMessage(text="hello", source="m:1"))
        assert len(store.read_events()) == 1
        # Two migrations were traversed, so both later tables must exist.
        store.mark_outcome(1, "good", "arrived via v1 -> v3")
        assert store.outcomes()[1] == ("good", "arrived via v1 -> v3")