nixos-config/hosts/blunix/localServices/agent-codex/db.py

"""SQLite persistence for the XMPP Codex roles.

The job queues and approval workflows remain file based because systemd path
units consume them. This first migration phase only stores conversational
state.
"""
import contextlib
import os
import sqlite3
import time


SCHEMA = """
CREATE TABLE IF NOT EXISTS sessions (
  role TEXT NOT NULL,
  sender TEXT NOT NULL,
  session_id TEXT NOT NULL,
  updated_at REAL NOT NULL,
  PRIMARY KEY (role, sender)
);

CREATE TABLE IF NOT EXISTS introductions (
  role TEXT NOT NULL,
  sender TEXT NOT NULL,
  introduced_at REAL NOT NULL,
  updated_at REAL NOT NULL,
  PRIMARY KEY (role, sender)
);

-- Carries the previous reply forward as plain-text context for a
-- sender's next job, for roles that run atomic per-job Codex sessions
-- instead of a resumed one (see docs/intent/builder-iterationslimit.md).
-- Deliberately separate from `sessions` above, which stays session-id
-- based for roles that do resume (the Agent role in worker.py).
--
-- `scope` defaults to "" (plain per-sender conversational continuity,
-- e.g. direct XMPP chat with the Builder) but board-triggered jobs pass
-- their board_slug instead -- every board job shares one fixed owner
-- sender identity (BOARD_BUILDER_OWNER_JID), so a bare (role, sender)
-- key let two genuinely unrelated jobs for different cards (or a card
-- and an unrelated direct chat) bleed context into each other once two
-- agent-codex-builder instances could run concurrently for the same
-- sender (found live 2026-09-09, board-worktree-parallelism-smoke-test-a:
-- an unrelated Prosody/XMPP conversation's reply got threaded into the
-- smoke test's prompt). Scoping by board_slug keeps the intentional
-- retry-carries-forward-its-own-interim-report behavior
-- (docs/intent/builder-iterationslimit.md) isolated per card.
CREATE TABLE IF NOT EXISTS context (
  role TEXT NOT NULL,
  sender TEXT NOT NULL,
  scope TEXT NOT NULL DEFAULT '',
  context_text TEXT NOT NULL,
  updated_at REAL NOT NULL,
  PRIMARY KEY (role, sender, scope)
);

"""


def _migrate_context_table(connection):
    """One-time, in-place migration for a `context` table created before
    `scope` existed -- see the schema comment above. Conversational cache,
    not durable data: copying forward under scope='' (today's only
    existing rows) rather than discarding it is a nicety, not a
    correctness requirement."""
    columns = [row[1] for row in connection.execute("PRAGMA table_info(context)")]
    if not columns or "scope" in columns:
        return
    connection.executescript(
        """
        ALTER TABLE context RENAME TO context_pre_scope;
        CREATE TABLE context (
          role TEXT NOT NULL,
          sender TEXT NOT NULL,
          scope TEXT NOT NULL DEFAULT '',
          context_text TEXT NOT NULL,
          updated_at REAL NOT NULL,
          PRIMARY KEY (role, sender, scope)
        );
        INSERT INTO context (role, sender, scope, context_text, updated_at)
          SELECT role, sender, '', context_text, updated_at FROM context_pre_scope;
        DROP TABLE context_pre_scope;
        """
    )


@contextlib.contextmanager
def connect():
    """Open the configured database and initialize its small local schema."""
    path = os.environ["AGENT_DB_PATH"]
    connection = sqlite3.connect(path, timeout=30)
    try:
        connection.execute("PRAGMA foreign_keys = ON")
        connection.executescript(SCHEMA)
        _migrate_context_table(connection)
        yield connection
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()


def load_session(role, sender):
    with connect() as connection:
        return connection.execute(
            "SELECT session_id, updated_at FROM sessions WHERE role = ? AND sender = ?",
            (role, sender),
        ).fetchone()


def save_session(role, sender, session_id, updated_at=None):
    if updated_at is None:
        updated_at = time.time()
    with connect() as connection:
        connection.execute(
            """
            INSERT INTO sessions (role, sender, session_id, updated_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(role, sender) DO UPDATE SET
              session_id = excluded.session_id,
              updated_at = excluded.updated_at
            """,
            (role, sender, session_id, updated_at),
        )


def reset_session(role, sender):
    with connect() as connection:
        connection.execute(
            "DELETE FROM sessions WHERE role = ? AND sender = ?",
            (role, sender),
        )


def load_context(role, sender, scope=""):
    with connect() as connection:
        return connection.execute(
            "SELECT context_text, updated_at FROM context WHERE role = ? AND sender = ? AND scope = ?",
            (role, sender, scope),
        ).fetchone()


def save_context(role, sender, context_text, scope="", updated_at=None):
    if updated_at is None:
        updated_at = time.time()
    with connect() as connection:
        connection.execute(
            """
            INSERT INTO context (role, sender, scope, context_text, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(role, sender, scope) DO UPDATE SET
              context_text = excluded.context_text,
              updated_at = excluded.updated_at
            """,
            (role, sender, scope, context_text, updated_at),
        )


def reset_context(role, sender, scope=""):
    with connect() as connection:
        connection.execute(
            "DELETE FROM context WHERE role = ? AND sender = ? AND scope = ?",
            (role, sender, scope),
        )


def is_introduced(role, sender):
    with connect() as connection:
        return connection.execute(
            "SELECT 1 FROM introductions WHERE role = ? AND sender = ?",
            (role, sender),
        ).fetchone() is not None


def mark_introduced(role, sender, updated_at=None):
    if updated_at is None:
        updated_at = time.time()
    with connect() as connection:
        connection.execute(
            """
            INSERT INTO introductions (role, sender, introduced_at, updated_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(role, sender) DO UPDATE SET updated_at = excluded.updated_at
            """,
            (role, sender, updated_at, updated_at),
        )