sqlrite-engine 0.10.1

Light version of SQLite developed with Rust. Published as `sqlrite-engine` on crates.io; import as `use sqlrite::…`.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

"""Top-level agent tests: turn loop, prompt assembly, summarization."""

from __future__ import annotations

from sqlrite_agent.agent import ChatAgent
from sqlrite_agent.chat import ChatProvider


class CapturingChat:
    """A test double that records what the agent asked it."""

    def __init__(self, reply: str = "ok") -> None:
        self.reply = reply
        self.last_system: str = ""
        self.last_messages: list[dict[str, str]] = []

    def complete(self, *, system: str, messages: list[dict[str, str]]) -> str:
        self.last_system = system
        self.last_messages = list(messages)
        return self.reply


def test_turn_writes_user_and_assistant_messages(memory):
    chat: ChatProvider = CapturingChat(reply="hi there")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")

    turn = agent.turn("Hello!")
    assert turn.assistant_reply == "hi there"

    s = memory.stats()
    assert s["messages"] == 2  # one user, one assistant


def test_turn_assembles_system_prompt_with_recalled_facts(memory):
    # Seed a known fact, then issue a related query in a NEW turn.
    memory.log_message(
        conversation_id="c1",
        role="user",
        content="My dog's name is Mochi.",
    )

    chat = CapturingChat(reply="cool")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")
    agent.turn("Tell me about Mochi")

    assert "user.dog.name = Mochi" in chat.last_system


def test_turn_includes_recent_chat_history(memory):
    chat = CapturingChat(reply="ok")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")

    agent.turn("first message")
    agent.turn("second message")

    # Third turn should see both prior turns in the message list.
    agent.turn("third message")
    user_contents = [m["content"] for m in chat.last_messages if m["role"] == "user"]
    assert "first message" in user_contents
    assert "second message" in user_contents
    assert "third message" in user_contents


def test_summarize_window_writes_a_summary(memory):
    chat = CapturingChat(reply="The user talked about their dog and the weather.")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")
    agent.turn("My dog's name is Mochi.")
    agent.turn("The weather in Lisbon is sunny.")

    summary = agent.summarize_window(last_n=10)
    assert summary is not None
    assert memory.stats()["summaries"] == 1


def test_recall_survives_db_reopen(tmp_path, embedder):
    """The headline demo: memory across process restarts."""
    from sqlrite_agent.db import AgentDB
    from sqlrite_agent.memory import Memory

    path = str(tmp_path / "agent.sqlrite")

    # --- session 1 ---
    db = AgentDB(path)
    memory = Memory(db, embedder)
    chat = CapturingChat(reply="ok")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")
    agent.turn("My dog's name is Mochi.")
    agent.turn("Mochi loves carrots.")
    db.close()

    # --- session 2: fresh process, same DB ---
    db = AgentDB(path)
    memory = Memory(db, embedder)
    chat = CapturingChat(reply="great")
    agent = ChatAgent(memory=memory, chat=chat, conversation_id="c1")

    agent.turn("What does Mochi eat?")
    # The system prompt for the new turn should contain the fact we
    # stored in session 1. That's the entire point of the demo.
    assert "Mochi" in chat.last_system