session_store

session_store

SQLite-backed storage for spec-search LearningSession records.

Mirrors the style of diapason.learning.optimize.store.OptimizationStore:

• stdlib sqlite3 in WAL mode
• inline DDL as module-level constants
• persistent connection stored as self._conn
• _migrate() runs additive ALTER TABLEs that swallow OperationalError

The store does NOT share its database file with OptimizationStore (see spec §8.1 / brainstorming Q8). The two SQLite files live side-by-side in ~/.diapason/learning/ but are independent.

Classes

SessionStore

SessionStore(db_path: Union[str, Path])

SQLite-backed storage for LearningSession and EditOutcome records.

The full LearningSession is also serialized to disk as <session_dir>/session.json — that file is the authoritative source if SQLite is ever lost. This store is the index used for fast queries.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def __init__(self, db_path: Union[str, Path]) -> None:
    self._db_path = str(db_path)
    self._conn = sqlite3.connect(self._db_path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute("PRAGMA foreign_keys=ON")
    self._conn.execute(_CREATE_SESSIONS)
    self._conn.execute(_CREATE_OUTCOMES)
    for index in _CREATE_INDEXES:
        self._conn.execute(index)
    self._conn.commit()
    self._migrate()

Methods:

close

close() -> None

Close the underlying SQLite connection.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def close(self) -> None:
    """Close the underlying SQLite connection."""
    self._conn.close()

save_session

save_session(session: LearningSession) -> None

Insert or update a LearningSession (idempotent on session.id).

Source code in src/diapason/learning/spec_search/storage/session_store.py

def save_session(self, session: LearningSession) -> None:
    """Insert or update a LearningSession (idempotent on session.id)."""
    self._conn.execute(
        _INSERT_SESSION,
        (
            session.id,
            session.parent_session_id,
            session.trigger.value,
            json.dumps(session.trigger_metadata),
            session.status.value,
            session.autonomy_mode.value,
            session.started_at.isoformat(),
            _dt_to_iso(session.ended_at),
            str(session.diagnosis_path),
            str(session.plan_path),
            session.benchmark_before.model_dump_json(),
            (
                session.benchmark_after.model_dump_json()
                if session.benchmark_after is not None
                else None
            ),
            session.git_checkpoint_pre,
            session.git_checkpoint_post,
            session.teacher_cost_usd,
            session.error,
        ),
    )
    self._conn.commit()

get_session

get_session(session_id: str) -> Optional[LearningSession]

Return the LearningSession with the given id, or None if missing.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def get_session(self, session_id: str) -> Optional[LearningSession]:
    """Return the LearningSession with the given id, or None if missing."""
    row = self._conn.execute(
        "SELECT id, parent_session_id, trigger, trigger_metadata, status, "
        "autonomy_mode, started_at, ended_at, diagnosis_path, plan_path, "
        "benchmark_before, benchmark_after, git_checkpoint_pre, "
        "git_checkpoint_post, teacher_cost_usd, error "
        "FROM learning_sessions WHERE id = ?",
        (session_id,),
    ).fetchone()
    if row is None:
        return None
    return self._row_to_session(row, with_outcomes=True)

list_sessions

list_sessions(
    status: SessionStatus | None = None,
    limit: int | None = None,
) -> list[LearningSession]

List sessions ordered by started_at DESC.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def list_sessions(
    self,
    status: SessionStatus | None = None,
    limit: int | None = None,
) -> list[LearningSession]:
    """List sessions ordered by ``started_at DESC``."""
    sql = (
        "SELECT id, parent_session_id, trigger, trigger_metadata, status, "
        "autonomy_mode, started_at, ended_at, diagnosis_path, plan_path, "
        "benchmark_before, benchmark_after, git_checkpoint_pre, "
        "git_checkpoint_post, teacher_cost_usd, error "
        "FROM learning_sessions"
    )
    params: list[object] = []
    if status is not None:
        sql += " WHERE status = ?"
        params.append(status.value)
    sql += " ORDER BY started_at DESC"
    if limit is not None:
        sql += " LIMIT ?"
        params.append(limit)
    rows = self._conn.execute(sql, params).fetchall()
    return [self._row_to_session(r, with_outcomes=True) for r in rows]

save_outcome

save_outcome(
    session_id: str,
    outcome: EditOutcome,
    *,
    pillar: str,
    op: str,
    target: str,
    risk_tier: str,
    rationale: str = "",
) -> None

Insert an EditOutcome row.

pillar, op, target, risk_tier, and rationale come from the parent Edit (which is not stored on the EditOutcome model). They are kept as columns to make WHERE op = ? queries possible without joining against the on-disk plan.json.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def save_outcome(
    self,
    session_id: str,
    outcome: EditOutcome,
    *,
    pillar: str,
    op: str,
    target: str,
    risk_tier: str,
    rationale: str = "",
) -> None:
    """Insert an EditOutcome row.

    ``pillar``, ``op``, ``target``, ``risk_tier``, and ``rationale`` come
    from the parent ``Edit`` (which is not stored on the EditOutcome
    model). They are kept as columns to make ``WHERE op = ?`` queries
    possible without joining against the on-disk plan.json.
    """
    self._conn.execute(
        _INSERT_OUTCOME,
        (
            session_id,
            outcome.edit_id,
            pillar,
            op,
            target,
            risk_tier,
            outcome.status,
            outcome.benchmark_delta,
            json.dumps(outcome.cluster_deltas),
            rationale,
            outcome.error,
            _dt_to_iso(outcome.applied_at),
        ),
    )
    self._conn.commit()

list_outcomes

list_outcomes(session_id: str) -> list[EditOutcome]

Return all EditOutcomes for a session, ordered by insertion id.

Source code in src/diapason/learning/spec_search/storage/session_store.py

def list_outcomes(self, session_id: str) -> list[EditOutcome]:
    """Return all EditOutcomes for a session, ordered by insertion id."""
    rows = self._conn.execute(
        "SELECT edit_id, status, benchmark_delta, cluster_deltas, error, "
        "applied_at FROM edit_outcomes WHERE session_id = ? ORDER BY id",
        (session_id,),
    ).fetchall()
    return [
        EditOutcome(
            edit_id=row[0],
            status=row[1],
            benchmark_delta=row[2],
            cluster_deltas=json.loads(row[3]),
            error=row[4],
            applied_at=_iso_to_dt(row[5]),
        )
        for row in rows
    ]