nisse/nisse-core/tests/test_canonical.py

import sqlite3
from pathlib import Path

import pytest

from nisse_core.log.canonical import (
    AssistantText,
    Compaction,
    DelegationCompleted,
    DelegationStarted,
    GateOpened,
    PlanSubmitted,
    ToolCall,
    ToolResult,
    UserMessage,
    from_payload,
    mint_call_id,
)
from nisse_core.state.store import StateSchemaError, 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="helper",
        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.transcript("t", "driver").append(event)
    with StateStore(tmp_path / "state.db") as store:
        stored = store.transcript("t", "driver").read()
    assert [s.event for s in stored] == ALL_EVENTS
    assert [s.id for s in stored] == sorted(s.id for s in stored)
    assert {(s.work_item, s.actor) for s in stored} == {("t", "driver")}


def test_read_events_after_id(tmp_path: Path) -> None:
    with StateStore(tmp_path / "state.db") as store:
        log = store.transcript("t", "operator")
        first = log.append(UserMessage(text="one", source="m:1"))
        log.append(UserMessage(text="two", source="m:2"))
        tail = log.read(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_a_store_from_before_work_items_is_refused_clearly(tmp_path: Path) -> None:
    # No production data predates work items, so there is no migration from
    # these versions (Development Plan M1.1): the store must say so rather
    # than open half-understood.
    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', '3')")
    conn.commit()
    conn.close()

    with pytest.raises(StateSchemaError, match="predates work items"):
        StateStore(db)