Storage and History

Copydef add_history(
    self,
    memory_id: str,
    old_memory: Optional[str],
    new_memory: Optional[str],
    event: str,
    *,
    created_at: Optional[str] = None,
    updated_at: Optional[str] = None,
    is_deleted: int = 0,
    actor_id: Optional[str] = None,
    role: Optional[str] = None,
) -> None:
    with self._lock:
        try:
            self.connection.execute("BEGIN")
            self.connection.execute(
                """
                INSERT INTO history (
                    id, memory_id, old_memory, new_memory, event,
                    created_at, updated_at, is_deleted, actor_id, role
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
                (
                    str(uuid.uuid4()),
                    memory_id,
                    old_memory,
                    new_memory,
                    event,
                    created_at,
                    updated_at,
                    is_deleted,
                    actor_id,
                    role,
                ),
            )
            self.connection.execute("COMMIT")
        except Exception as e:
            self.connection.execute("ROLLBACK")
            logger.error(f"Failed to add history record: {e}")
            raise
 
def batch_add_history(self, records: List[Dict[str, Any]]) -> None:
    with self._lock:
        try:
            self.connection.execute("BEGIN")
            self.connection.executemany(
                """
                INSERT INTO history (
                    id, memory_id, old_memory, new_memory, event,
                    created_at, updated_at, is_deleted, actor_id, role
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
                [
                    (
                        str(uuid.uuid4()),
                        record.get("memory_id"),
                        record.get("old_memory"),
                        record.get("new_memory"),
                        record.get("event"),
                        record.get("created_at"),
                        record.get("updated_at"),
                        record.get("is_deleted", 0),
                        record.get("actor_id"),
                        record.get("role"),
                    )
                    for record in records
                ],
            )
            self.connection.execute("COMMIT")
        except Exception as e:
            self.connection.execute("ROLLBACK")
            logger.error(f"Failed to batch add history records: {e}")
            raise
 

Copydef get_history(self, memory_id: str) -> List[Dict[str, Any]]:
    with self._lock:
        cur = self.connection.execute(
            """
            SELECT id, memory_id, old_memory, new_memory, event,
                   created_at, updated_at, is_deleted, actor_id, role
            FROM history
            WHERE memory_id = ?
            ORDER BY created_at ASC, DATETIME(updated_at) ASC
        """,
            (memory_id,),
        )
        rows = cur.fetchall()
 
    return [
        {
            "id": r[0],
            "memory_id": r[1],
            "old_memory": r[2],
            "new_memory": r[3],
            "event": r[4],
            "created_at": r[5],
            "updated_at": r[6],
            "is_deleted": bool(r[7]),
            "actor_id": r[8],
            "role": r[9],
        }
        for r in rows
    ]

Copydef save_messages(self, messages: List[Dict[str, Any]], session_scope: str) -> None:
    if not messages:
        return
    with self._lock:
        try:
            self.connection.execute("BEGIN")
            now = datetime.now(timezone.utc).isoformat()
            for message in messages:
                self.connection.execute(
                    """
                    INSERT INTO messages (id, session_scope, role, content, name, created_at)
                    VALUES (?, ?, ?, ?, ?, ?)
                """,
                    (
                        str(uuid.uuid4()),
                        session_scope,
                        message.get("role"),
                        message.get("content"),
                        message.get("name"),
                        now,
                    ),
                )

Copydef get_last_messages(self, session_scope: str, limit: int = 10) -> List[Dict[str, Any]]:
    with self._lock:
        # Subquery picks the latest N rows (DESC + LIMIT), outer query
        # re-sorts them chronologically (ASC) for the caller.
        cur = self.connection.execute(
            """
            SELECT role, content, name, created_at FROM (
                SELECT role, content, name, created_at
                FROM messages
                WHERE session_scope = ?
                ORDER BY created_at DESC
                LIMIT ?
            ) ORDER BY created_at ASC
        """,
            (session_scope, limit),
        )
        rows = cur.fetchall()
 
    return [
        {
            "role": r[0],
            "content": r[1],
            "name": r[2],
            "created_at": r[3],
        }
        for r in rows
    ]

Copycur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='history'")
if cur.fetchone() is None:
    self.connection.execute("COMMIT")
    return  # nothing to migrate
 
cur.execute("PRAGMA table_info(history)")
old_cols = {row[1] for row in cur.fetchall()}
 
expected_cols = {
    "id",
    "memory_id",
    "old_memory",
    "new_memory",
    "event",
    "created_at",
    "updated_at",
    "is_deleted",
    "actor_id",
    "role",
}
 
if old_cols == expected_cols:
    self.connection.execute("COMMIT")
    return

Copylogger.info("Migrating history table to new schema (no convo columns).")
 
# Clean up any existing history_old table from previous failed migration
cur.execute("DROP TABLE IF EXISTS history_old")
 
# Rename the current history table
cur.execute("ALTER TABLE history RENAME TO history_old")
 
# Create the new history table with updated schema
cur.execute(
    """
    CREATE TABLE history (
        id           TEXT PRIMARY KEY,
        memory_id    TEXT,
        old_memory   TEXT,
        new_memory   TEXT,
        event        TEXT,
        created_at   DATETIME,
        updated_at   DATETIME,
        is_deleted   INTEGER,
        actor_id     TEXT,
        role         TEXT
    )
"""
)
 
# Copy data from old table to new table
intersecting = list(expected_cols & old_cols)
if intersecting:
    cols_csv = ", ".join(intersecting)
    cur.execute(f"INSERT INTO history ({cols_csv}) SELECT {cols_csv} FROM history_old")
 
# Drop the old table
cur.execute("DROP TABLE history_old")
 
# Commit the transaction
self.connection.execute("COMMIT")
logger.info("History table migration completed successfully.")

CopyDROP TABLE IF EXISTS sidekick_calls;
DROP TABLE IF EXISTS sidekick_snapshots;
DROP TABLE IF EXISTS sidekick_state;
DROP TABLE IF EXISTS sidekick_packets;

Copydef prepare_flush(
    self, repo: RepoContext, session_id: str, reason: str
) -> tuple[str, list[dict[str, Any]]] | None:
    with self.conn:
        self.conn.execute("BEGIN IMMEDIATE")
        existing = self.conn.execute(
            """SELECT * FROM flushes
               WHERE repo_id = ? AND session_id = ?
                 AND status NOT IN ('semantic-succeeded', 'explicitly-stored', 'gave-up')
               ORDER BY created_at LIMIT 1""",
            (repo.identity, session_id),
        ).fetchone()

Copydef stop_subagent(
    self,
    repo: RepoContext,
    session_id: str,
    agent_id: str,
    agent_type: str,
    transcript_path: str,
    final_message: str,
) -> str:
    now = utc_now()
    self.conn.execute("BEGIN IMMEDIATE")

Copyself.conn.execute(
    """INSERT INTO sidekick_runs
       (repo_id, session_id, agent_id, agent_type, started_at,
        stopped_at, transcript_path, final_message)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?)
       ON CONFLICT(repo_id, session_id, agent_id) DO UPDATE SET
         stopped_at = excluded.stopped_at,
         transcript_path = excluded.transcript_path,
         final_message = excluded.final_message""",
    (
        repo.identity,
        session_id,
        agent_id,
        agent_type,
        now,
        now,
        bounded(transcript_path, 2000),
        redact(final_message).strip(),
    ),
)