Persistence and Caching

Experiment store

store

SQLite-backed study, trial, attempt, observation, and event persistence.

StudyStatus

Bases: str, Enum

Durable lifecycle states for a study.

CREATED class-attribute instance-attribute

CREATED = 'created'

RUNNING class-attribute instance-attribute

RUNNING = 'running'

COMPLETED class-attribute instance-attribute

COMPLETED = 'completed'

FAILED class-attribute instance-attribute

FAILED = 'failed'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

TrialStatus

Bases: str, Enum

Durable lifecycle states for a trial.

PENDING class-attribute instance-attribute

PENDING = 'pending'

RUNNING class-attribute instance-attribute

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

FAILED = 'failed'

ABANDONED class-attribute instance-attribute

ABANDONED = 'abandoned'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

AttemptStatus

Bases: str, Enum

Durable lifecycle states for a trial attempt.

RUNNING class-attribute instance-attribute

RUNNING = 'running'

SUCCEEDED class-attribute instance-attribute

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute

FAILED = 'failed'

ABANDONED class-attribute instance-attribute

ABANDONED = 'abandoned'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

StudyRecord dataclass

StudyRecord(id: str, name: str, config_hash: str, status: StudyStatus, created_at: str, updated_at: str)

Persisted study identity and lifecycle state.

id instance-attribute

id: str

name instance-attribute

name: str

config_hash instance-attribute

config_hash: str

status instance-attribute

status: StudyStatus

created_at instance-attribute

created_at: str

updated_at instance-attribute

updated_at: str

TrialRecord dataclass

TrialRecord(id: str, study_id: str, sequence: int, candidate_hash: str, candidate_json: str, status: TrialStatus, error: str | None, created_at: str, updated_at: str)

Persisted trial identity and lifecycle state.

id instance-attribute

id: str

study_id instance-attribute

study_id: str

sequence instance-attribute

sequence: int

candidate_hash instance-attribute

candidate_hash: str

candidate_json instance-attribute

candidate_json: str

status instance-attribute

status: TrialStatus

error instance-attribute

error: str | None

created_at instance-attribute

created_at: str

updated_at instance-attribute

updated_at: str

AttemptRecord dataclass

AttemptRecord(id: str, trial_id: str, number: int, status: AttemptStatus, started_at: float, lease_expires_at: float, finished_at: float | None, error: str | None)

Persisted execution attempt with a recovery lease.

id instance-attribute

id: str

trial_id instance-attribute

trial_id: str

number instance-attribute

number: int

status instance-attribute

status: AttemptStatus

started_at instance-attribute

started_at: float

lease_expires_at instance-attribute

lease_expires_at: float

finished_at instance-attribute

finished_at: float | None

error instance-attribute

error: str | None

PersistedUsageRecord dataclass

PersistedUsageRecord(id: str, trial_id: str, attempt_id: str | None, usage: UsageRecord, created_at: str)

Append-only usage observation associated with a trial and optional attempt.

id instance-attribute

id: str

trial_id instance-attribute

trial_id: str

attempt_id instance-attribute

attempt_id: str | None

usage instance-attribute

usage: UsageRecord

created_at instance-attribute

created_at: str

StudyEvent dataclass

StudyEvent(study_id: str, sequence: int, type: str, payload: Mapping[str, Any], created_at: str)

Ordered durable event for replay and observability.

study_id instance-attribute

study_id: str

sequence instance-attribute

sequence: int

type instance-attribute

type: str

payload instance-attribute

payload: Mapping[str, Any]

created_at instance-attribute

created_at: str

SQLiteStore

SQLiteStore(path: str | Path = ':memory:', *, clock: Callable[[], float] = time)

Transactional SQLite authority for optimization lifecycle state.

Source code in src/tunarag/store.py

def __init__(
    self,
    path: str | Path = ":memory:",
    *,
    clock: Callable[[], float] = time.time,
) -> None:
    self._connection = sqlite3.connect(str(path), timeout=30.0)
    self._connection.row_factory = sqlite3.Row
    self._clock = clock
    self.initialize()

schema_version class-attribute instance-attribute

schema_version = 2

initialize

initialize() -> None

Create or migrate the experiment schema to the supported version.

Source code in src/tunarag/store.py

def initialize(self) -> None:
    """Create or migrate the experiment schema to the supported version."""

    self._connection.execute("PRAGMA foreign_keys = ON")
    self._connection.execute(
        "CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)"
    )
    self._connection.commit()
    versions = self._schema_versions()
    if not versions:
        self._create_schema_version_1()
        versions = [1]
    if versions == [1]:
        self._migrate_version_1_to_2()
        versions = [1, 2]
    if versions != [1, 2]:
        self._connection.close()
        raise StoreError(
            "unsupported experiment schema versions",
            stage="store.initialize",
            details={"versions": versions, "supported": self.schema_version},
        )

create_study

create_study(name: str, configuration: object) -> StudyRecord

Create a study in the created state.

Source code in src/tunarag/store.py

def create_study(self, name: str, configuration: object) -> StudyRecord:
    """Create a study in the created state."""

    if not name.strip():
        raise ValueError("study name must not be empty")
    study_id = str(uuid.uuid4())
    config_hash = content_hash(configuration, namespace="study:v1")
    with self._connection:
        self._connection.execute(
            """INSERT INTO studies(id, name, config_hash, status, updated_at)
            VALUES (?, ?, ?, 'created', CURRENT_TIMESTAMP)""",
            (study_id, name, config_hash),
        )
        self._append_event_locked(study_id, "study.created", {"name": name})
    return self.get_study(study_id)

get_study

get_study(study_id: str) -> StudyRecord

Load one study or raise KeyError.

Source code in src/tunarag/store.py

def get_study(self, study_id: str) -> StudyRecord:
    """Load one study or raise `KeyError`."""

    row = self._connection.execute(
        """SELECT id, name, config_hash, status, created_at, updated_at
        FROM studies WHERE id = ?""",
        (study_id,),
    ).fetchone()
    if row is None:
        raise KeyError(f"unknown study: {study_id}")
    return _study_from_row(cast(sqlite3.Row, row))

transition_study

transition_study(study_id: str, status: StudyStatus) -> StudyRecord

Apply a guarded study lifecycle transition.

Source code in src/tunarag/store.py

def transition_study(self, study_id: str, status: StudyStatus) -> StudyRecord:
    """Apply a guarded study lifecycle transition."""

    self._begin_immediate()
    try:
        current = self._get_study_status_locked(study_id)
        if status not in _STUDY_TRANSITIONS[current]:
            raise ValueError(f"invalid study transition: {current.value} -> {status.value}")
        if status is StudyStatus.COMPLETED:
            active = self._connection.execute(
                """SELECT 1 FROM trials
                WHERE study_id = ? AND status IN ('pending', 'running') LIMIT 1""",
                (study_id,),
            ).fetchone()
            if active is not None:
                raise ValueError("study cannot complete while trials are active")
        self._connection.execute(
            """UPDATE studies SET status = ?, updated_at = CURRENT_TIMESTAMP
            WHERE id = ?""",
            (status.value, study_id),
        )
        self._append_event_locked(
            study_id, f"study.{status.value}", {"previous_status": current.value}
        )
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return self.get_study(study_id)

create_trial

create_trial(study_id: str, sequence: int, candidate: Candidate) -> TrialRecord

Create a trial with an explicit nonnegative sequence.

Source code in src/tunarag/store.py

def create_trial(self, study_id: str, sequence: int, candidate: Candidate) -> TrialRecord:
    """Create a trial with an explicit nonnegative sequence."""

    if sequence < 0:
        raise ValueError("trial sequence must not be negative")
    self._begin_immediate()
    try:
        self._require_running_study_locked(study_id)
        record = self._insert_trial_locked(study_id, sequence, candidate)
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return record

reserve_trial

reserve_trial(study_id: str, candidate: Candidate) -> TrialRecord

Atomically allocate the next sequence and reserve a unique candidate.

Source code in src/tunarag/store.py

def reserve_trial(self, study_id: str, candidate: Candidate) -> TrialRecord:
    """Atomically allocate the next sequence and reserve a unique candidate."""

    self._begin_immediate()
    try:
        self._require_running_study_locked(study_id)
        row = self._connection.execute(
            "SELECT COALESCE(MAX(sequence), -1) + 1 FROM trials WHERE study_id = ?",
            (study_id,),
        ).fetchone()
        sequence = int(cast(sqlite3.Row, row)[0])
        record = self._insert_trial_locked(study_id, sequence, candidate)
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return record

get_trial

get_trial(trial_id: str) -> TrialRecord

Load one trial or raise KeyError.

Source code in src/tunarag/store.py

def get_trial(self, trial_id: str) -> TrialRecord:
    """Load one trial or raise `KeyError`."""

    row = self._connection.execute(
        """SELECT id, study_id, sequence, candidate_hash, candidate_json, status,
        error, created_at, updated_at FROM trials WHERE id = ?""",
        (trial_id,),
    ).fetchone()
    if row is None:
        raise KeyError(f"unknown trial: {trial_id}")
    return _trial_from_row(cast(sqlite3.Row, row))

list_trials

list_trials(study_id: str) -> list[TrialRecord]

Return trials in durable sequence order.

Source code in src/tunarag/store.py

def list_trials(self, study_id: str) -> list[TrialRecord]:
    """Return trials in durable sequence order."""

    rows = self._connection.execute(
        """SELECT id, study_id, sequence, candidate_hash, candidate_json, status,
        error, created_at, updated_at FROM trials
        WHERE study_id = ? ORDER BY sequence""",
        (study_id,),
    ).fetchall()
    return [_trial_from_row(row) for row in rows]

transition_trial

transition_trial(trial_id: str, status: TrialStatus, *, error: str | None = None) -> TrialRecord

Apply a guarded trial lifecycle transition.

Source code in src/tunarag/store.py

def transition_trial(
    self, trial_id: str, status: TrialStatus, *, error: str | None = None
) -> TrialRecord:
    """Apply a guarded trial lifecycle transition."""

    self._begin_immediate()
    try:
        row = self._get_trial_row_locked(trial_id)
        current = TrialStatus(row["status"])
        if status not in _TRIAL_TRANSITIONS[current]:
            raise ValueError(f"invalid trial transition: {current.value} -> {status.value}")
        if status in _TERMINAL_TRIAL_STATUSES:
            active = self._connection.execute(
                "SELECT 1 FROM trial_attempts WHERE trial_id = ? AND status = 'running'",
                (trial_id,),
            ).fetchone()
            if active is not None:
                raise ValueError("active attempt must be completed through commit_trial_result")
        self._connection.execute(
            """UPDATE trials SET status = ?, error = ?, updated_at = CURRENT_TIMESTAMP
            WHERE id = ?""",
            (status.value, error, trial_id),
        )
        self._append_event_locked(
            row["study_id"],
            f"trial.{status.value}",
            {"trial_id": trial_id, "previous_status": current.value},
        )
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return self.get_trial(trial_id)

start_attempt

start_attempt(trial_id: str, *, lease_seconds: float, now: float | None = None) -> AttemptRecord

Start the next attempt and acquire its recovery lease.

Source code in src/tunarag/store.py

def start_attempt(
    self,
    trial_id: str,
    *,
    lease_seconds: float,
    now: float | None = None,
) -> AttemptRecord:
    """Start the next attempt and acquire its recovery lease."""

    if not math.isfinite(lease_seconds) or lease_seconds <= 0:
        raise ValueError("attempt lease must be a positive finite number")
    started_at = self._timestamp(now)
    self._begin_immediate()
    try:
        trial = self._get_trial_row_locked(trial_id)
        trial_status = TrialStatus(trial["status"])
        if trial_status not in {TrialStatus.PENDING, TrialStatus.RUNNING}:
            raise ValueError(f"cannot start attempt for {trial_status.value} trial")
        active = self._connection.execute(
            "SELECT 1 FROM trial_attempts WHERE trial_id = ? AND status = 'running'",
            (trial_id,),
        ).fetchone()
        if active is not None:
            raise ValueError("trial already has a running attempt")
        row = self._connection.execute(
            """SELECT COALESCE(MAX(attempt_number), -1) + 1
            FROM trial_attempts WHERE trial_id = ?""",
            (trial_id,),
        ).fetchone()
        number = int(cast(sqlite3.Row, row)[0])
        attempt_id = str(uuid.uuid4())
        lease_expires_at = started_at + lease_seconds
        self._connection.execute(
            """INSERT INTO trial_attempts(
                id, trial_id, attempt_number, status, started_at, lease_expires_at
            ) VALUES (?, ?, ?, 'running', ?, ?)""",
            (attempt_id, trial_id, number, started_at, lease_expires_at),
        )
        if trial_status is TrialStatus.PENDING:
            self._connection.execute(
                """UPDATE trials SET status = 'running', updated_at = CURRENT_TIMESTAMP
                WHERE id = ?""",
                (trial_id,),
            )
        self._append_event_locked(
            trial["study_id"],
            "trial.attempt.started",
            {"trial_id": trial_id, "attempt_id": attempt_id, "attempt": number},
        )
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return self.get_attempt(attempt_id)

get_attempt

get_attempt(attempt_id: str) -> AttemptRecord

Load one attempt or raise KeyError.

Source code in src/tunarag/store.py

def get_attempt(self, attempt_id: str) -> AttemptRecord:
    """Load one attempt or raise `KeyError`."""

    row = self._connection.execute(
        """SELECT id, trial_id, attempt_number, status, started_at,
        lease_expires_at, finished_at, error FROM trial_attempts WHERE id = ?""",
        (attempt_id,),
    ).fetchone()
    if row is None:
        raise KeyError(f"unknown attempt: {attempt_id}")
    return _attempt_from_row(cast(sqlite3.Row, row))

list_attempts

list_attempts(trial_id: str) -> list[AttemptRecord]

Return attempts in monotonic attempt-number order.

Source code in src/tunarag/store.py

def list_attempts(self, trial_id: str) -> list[AttemptRecord]:
    """Return attempts in monotonic attempt-number order."""

    rows = self._connection.execute(
        """SELECT id, trial_id, attempt_number, status, started_at,
        lease_expires_at, finished_at, error FROM trial_attempts
        WHERE trial_id = ? ORDER BY attempt_number""",
        (trial_id,),
    ).fetchall()
    return [_attempt_from_row(row) for row in rows]

finish_attempt

finish_attempt(attempt_id: str, status: AttemptStatus, *, usage: Sequence[UsageRecord] = (), error: str | None = None, now: float | None = None) -> AttemptRecord

Finish a running attempt while leaving retryable trial state intact.

Source code in src/tunarag/store.py

def finish_attempt(
    self,
    attempt_id: str,
    status: AttemptStatus,
    *,
    usage: Sequence[UsageRecord] = (),
    error: str | None = None,
    now: float | None = None,
) -> AttemptRecord:
    """Finish a running attempt while leaving retryable trial state intact."""

    if status is AttemptStatus.RUNNING:
        raise ValueError("attempt terminal status must not be running")
    finished_at = self._timestamp(now)
    self._begin_immediate()
    try:
        row = self._get_attempt_row_locked(attempt_id)
        if AttemptStatus(row["status"]) is not AttemptStatus.RUNNING:
            raise ValueError("only a running attempt can be finished")
        self._connection.execute(
            """UPDATE trial_attempts SET status = ?, finished_at = ?, error = ?
            WHERE id = ?""",
            (status.value, finished_at, error, attempt_id),
        )
        self._insert_usage_locked(row["trial_id"], attempt_id, usage)
        trial = self._get_trial_row_locked(row["trial_id"])
        self._append_event_locked(
            trial["study_id"],
            f"trial.attempt.{status.value}",
            {"trial_id": row["trial_id"], "attempt_id": attempt_id},
        )
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return self.get_attempt(attempt_id)

abandon_expired_attempts

abandon_expired_attempts(*, study_id: str | None = None, now: float | None = None) -> list[AttemptRecord]

Persist abandonment for expired attempts, optionally scoped to one study.

Source code in src/tunarag/store.py

def abandon_expired_attempts(
    self, *, study_id: str | None = None, now: float | None = None
) -> list[AttemptRecord]:
    """Persist abandonment for expired attempts, optionally scoped to one study."""

    timestamp = self._timestamp(now)
    abandoned_ids: list[str] = []
    self._begin_immediate()
    try:
        if study_id is None:
            rows = self._connection.execute(
                """SELECT id, trial_id FROM trial_attempts
                WHERE status = 'running' AND lease_expires_at <= ?
                ORDER BY started_at, id""",
                (timestamp,),
            ).fetchall()
        else:
            self._get_study_status_locked(study_id)
            rows = self._connection.execute(
                """SELECT a.id, a.trial_id FROM trial_attempts AS a
                JOIN trials AS t ON t.id = a.trial_id
                WHERE t.study_id = ? AND a.status = 'running'
                  AND a.lease_expires_at <= ?
                ORDER BY a.started_at, a.id""",
                (study_id, timestamp),
            ).fetchall()
        for row in rows:
            self._connection.execute(
                """UPDATE trial_attempts
                SET status = 'abandoned', finished_at = ?, error = 'lease expired'
                WHERE id = ?""",
                (timestamp, row["id"]),
            )
            trial = self._get_trial_row_locked(row["trial_id"])
            self._append_event_locked(
                trial["study_id"],
                "trial.attempt.abandoned",
                {"trial_id": row["trial_id"], "attempt_id": row["id"]},
            )
            abandoned_ids.append(row["id"])
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return [self.get_attempt(attempt_id) for attempt_id in abandoned_ids]

list_recoverable_trials

list_recoverable_trials(study_id: str) -> list[TrialRecord]

Return running trials that have no active attempt.

Source code in src/tunarag/store.py

def list_recoverable_trials(self, study_id: str) -> list[TrialRecord]:
    """Return running trials that have no active attempt."""

    rows = self._connection.execute(
        """SELECT t.id, t.study_id, t.sequence, t.candidate_hash, t.candidate_json,
        t.status, t.error, t.created_at, t.updated_at
        FROM trials AS t
        WHERE t.study_id = ? AND t.status = 'running'
          AND NOT EXISTS (
              SELECT 1 FROM trial_attempts AS a
              WHERE a.trial_id = t.id AND a.status = 'running'
          )
        ORDER BY t.sequence""",
        (study_id,),
    ).fetchall()
    return [_trial_from_row(row) for row in rows]

commit_trial_result

commit_trial_result(trial_id: str, *, status: TrialStatus, metrics: Sequence[MetricValue] = (), usage: Sequence[UsageRecord] = (), attempt_id: str | None = None, error: str | None = None, now: float | None = None) -> TrialRecord

Atomically persist observations, attempt completion, and terminal trial state.

Source code in src/tunarag/store.py

def commit_trial_result(
    self,
    trial_id: str,
    *,
    status: TrialStatus,
    metrics: Sequence[MetricValue] = (),
    usage: Sequence[UsageRecord] = (),
    attempt_id: str | None = None,
    error: str | None = None,
    now: float | None = None,
) -> TrialRecord:
    """Atomically persist observations, attempt completion, and terminal trial state."""

    if status not in _TERMINAL_TRIAL_STATUSES:
        raise ValueError("trial result status must be terminal")
    metric_names = [metric.name for metric in metrics]
    if len(metric_names) != len(set(metric_names)):
        raise ValueError("trial result contains duplicate metric names")
    finished_at = self._timestamp(now)
    self._begin_immediate()
    try:
        trial = self._get_trial_row_locked(trial_id)
        if TrialStatus(trial["status"]) is not TrialStatus.RUNNING:
            raise ValueError("only a running trial can commit a result")
        if attempt_id is not None:
            attempt = self._get_attempt_row_locked(attempt_id)
            if attempt["trial_id"] != trial_id:
                raise ValueError("attempt does not belong to trial")
            if AttemptStatus(attempt["status"]) is not AttemptStatus.RUNNING:
                raise ValueError("only a running attempt can commit a trial result")
            attempt_status = _attempt_status_for_trial(status)
            self._connection.execute(
                """UPDATE trial_attempts SET status = ?, finished_at = ?, error = ?
                WHERE id = ?""",
                (attempt_status.value, finished_at, error, attempt_id),
            )
        for metric in metrics:
            self._connection.execute(
                """INSERT INTO metric_values(
                    trial_id, name, value, status, coverage
                ) VALUES (?, ?, ?, ?, ?)""",
                (trial_id, metric.name, metric.value, metric.status.value, metric.coverage),
            )
        self._insert_usage_locked(trial_id, attempt_id, usage)
        self._connection.execute(
            """UPDATE trials SET status = ?, error = ?, updated_at = CURRENT_TIMESTAMP
            WHERE id = ?""",
            (status.value, error, trial_id),
        )
        self._append_event_locked(
            trial["study_id"],
            f"trial.{status.value}",
            {
                "trial_id": trial_id,
                "attempt_id": attempt_id,
                "metric_count": len(metrics),
                "usage_count": len(usage),
            },
        )
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return self.get_trial(trial_id)

list_metrics

list_metrics(trial_id: str) -> list[MetricValue]

Return committed metrics ordered by name.

Source code in src/tunarag/store.py

def list_metrics(self, trial_id: str) -> list[MetricValue]:
    """Return committed metrics ordered by name."""

    rows = self._connection.execute(
        """SELECT name, value, status, coverage FROM metric_values
        WHERE trial_id = ? ORDER BY name""",
        (trial_id,),
    ).fetchall()
    return [
        MetricValue(row["name"], row["value"], ValueStatus(row["status"]), row["coverage"])
        for row in rows
    ]

list_usage

list_usage(trial_id: str) -> list[PersistedUsageRecord]

Return append-only usage in insertion order.

Source code in src/tunarag/store.py

def list_usage(self, trial_id: str) -> list[PersistedUsageRecord]:
    """Return append-only usage in insertion order."""

    rows = self._connection.execute(
        """SELECT id, trial_id, attempt_id, component, input_tokens, output_tokens,
        cost_text, latency_seconds, status, pricing_version, created_at
        FROM usage_records WHERE trial_id = ? ORDER BY rowid""",
        (trial_id,),
    ).fetchall()
    records: list[PersistedUsageRecord] = []
    for row in rows:
        cost = None if row["cost_text"] is None else float(row["cost_text"])
        records.append(
            PersistedUsageRecord(
                id=row["id"],
                trial_id=row["trial_id"],
                attempt_id=row["attempt_id"],
                usage=UsageRecord(
                    component=row["component"],
                    input_tokens=row["input_tokens"],
                    output_tokens=row["output_tokens"],
                    cost=cost,
                    latency_seconds=row["latency_seconds"],
                    status=ValueStatus(row["status"]),
                    pricing_version=row["pricing_version"],
                ),
                created_at=row["created_at"],
            )
        )
    return records

append_event

append_event(study_id: str, event_type: str, payload: Mapping[str, Any]) -> StudyEvent

Append a canonical event with an atomic per-study sequence.

Source code in src/tunarag/store.py

def append_event(
    self, study_id: str, event_type: str, payload: Mapping[str, Any]
) -> StudyEvent:
    """Append a canonical event with an atomic per-study sequence."""

    self._validate_event_type(event_type)
    self._begin_immediate()
    try:
        self._get_study_status_locked(study_id)
        event = self._append_event_locked(study_id, event_type, payload)
        self._connection.commit()
    except BaseException:
        self._connection.rollback()
        raise
    return event

list_events

list_events(study_id: str, *, after_sequence: int = -1) -> list[StudyEvent]

Return durable events after a sequence in replay order.

Source code in src/tunarag/store.py

def list_events(self, study_id: str, *, after_sequence: int = -1) -> list[StudyEvent]:
    """Return durable events after a sequence in replay order."""

    rows = self._connection.execute(
        """SELECT study_id, sequence, type, payload_json, created_at
        FROM study_events WHERE study_id = ? AND sequence > ? ORDER BY sequence""",
        (study_id, after_sequence),
    ).fetchall()
    return [_event_from_row(row) for row in rows]

close

close() -> None

Close the underlying SQLite connection.

Source code in src/tunarag/store.py

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

    self._connection.close()

Evaluation cache

cache

Versioned, checksum-verified local cache persistence.

CachePrivacy

Bases: str, Enum

Privacy classification retained with a cache payload.

STANDARD class-attribute instance-attribute

STANDARD = 'standard'

SENSITIVE class-attribute instance-attribute

SENSITIVE = 'sensitive'

RESTRICTED class-attribute instance-attribute

RESTRICTED = 'restricted'

CacheLookupStatus

Bases: str, Enum

Outcome of a cache lookup.

HIT class-attribute instance-attribute

HIT = 'hit'

MISS class-attribute instance-attribute

MISS = 'miss'

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

CORRUPT class-attribute instance-attribute

CORRUPT = 'corrupt'

QUARANTINED class-attribute instance-attribute

QUARANTINED = 'quarantined'

CacheEntryState

Bases: str, Enum

Persisted cache-entry lifecycle state.

ACTIVE class-attribute instance-attribute

ACTIVE = 'active'

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

QUARANTINED class-attribute instance-attribute

QUARANTINED = 'quarantined'

CacheKey dataclass

CacheKey(namespace: str, digest: str, format_version: int = 1)

Namespaced identity for one semantic cache value.

namespace instance-attribute

namespace: str

digest instance-attribute

digest: str

format_version class-attribute instance-attribute

format_version: int = 1

from_parts classmethod

from_parts(namespace: str, *parts: Any, format_version: int = 1) -> CacheKey

Build a key from canonical semantic inputs.

Source code in src/tunarag/cache.py

@classmethod
def from_parts(cls, namespace: str, *parts: Any, format_version: int = 1) -> "CacheKey":
    """Build a key from canonical semantic inputs."""

    if not parts:
        raise ValueError("cache key requires at least one semantic part")
    digest = content_hash(
        {"format_version": format_version, "parts": parts},
        namespace=f"tunarag:cache-key:{namespace}:v1",
    )
    return cls(namespace=namespace, digest=digest, format_version=format_version)

CacheEntry dataclass

CacheEntry(key: CacheKey, checksum: str, privacy: CachePrivacy, created_at: float, expires_at: float | None)

Metadata returned after an atomic cache write.

key instance-attribute

key: CacheKey

checksum instance-attribute

checksum: str

privacy instance-attribute

privacy: CachePrivacy

created_at instance-attribute

created_at: float

expires_at instance-attribute

expires_at: float | None

CacheLookup dataclass

CacheLookup(key: CacheKey, status: CacheLookupStatus, payload: bytes | None = None, checksum: str | None = None, privacy: CachePrivacy | None = None, created_at: float | None = None, expires_at: float | None = None, reason: str | None = None)

Payload and metadata returned by a cache lookup.

key instance-attribute

key: CacheKey

status instance-attribute

status: CacheLookupStatus

payload class-attribute instance-attribute

payload: bytes | None = None

checksum class-attribute instance-attribute

checksum: str | None = None

privacy class-attribute instance-attribute

privacy: CachePrivacy | None = None

created_at class-attribute instance-attribute

created_at: float | None = None

expires_at class-attribute instance-attribute

expires_at: float | None = None

reason class-attribute instance-attribute

reason: str | None = None

is_hit property

is_hit: bool

Return whether this lookup contains a verified payload.

SQLiteCache

SQLiteCache(path: str | Path = ':memory:', *, clock: Callable[[], float] = time)

Transactional SQLite cache with expiry and corruption quarantine.

Source code in src/tunarag/cache.py

def __init__(
    self,
    path: str | Path = ":memory:",
    *,
    clock: Callable[[], float] = time.time,
) -> None:
    self._connection = sqlite3.connect(str(path))
    self._connection.row_factory = sqlite3.Row
    self._clock = clock
    self.initialize()

initialize

initialize() -> None

Create the independently versioned cache schema.

Source code in src/tunarag/cache.py

def initialize(self) -> None:
    """Create the independently versioned cache schema."""

    self._connection.executescript(
        """
        CREATE TABLE IF NOT EXISTS cache_schema_version (
            version INTEGER PRIMARY KEY
        );
        INSERT OR IGNORE INTO cache_schema_version(version) VALUES (1);
        CREATE TABLE IF NOT EXISTS cache_entries (
            namespace TEXT NOT NULL,
            format_version INTEGER NOT NULL,
            key_digest TEXT NOT NULL,
            payload BLOB NOT NULL,
            checksum TEXT NOT NULL,
            privacy TEXT NOT NULL,
            state TEXT NOT NULL,
            created_at REAL NOT NULL,
            expires_at REAL,
            last_accessed_at REAL,
            hit_count INTEGER NOT NULL DEFAULT 0,
            quarantine_reason TEXT,
            PRIMARY KEY(namespace, format_version, key_digest),
            CHECK(format_version > 0),
            CHECK(hit_count >= 0),
            CHECK(privacy IN ('standard', 'sensitive', 'restricted')),
            CHECK(state IN ('active', 'expired', 'quarantined'))
        );
        CREATE INDEX IF NOT EXISTS idx_cache_entries_retention
            ON cache_entries(state, expires_at);
        """
    )
    self._connection.commit()
    versions = [
        int(row[0])
        for row in self._connection.execute(
            "SELECT version FROM cache_schema_version ORDER BY version"
        )
    ]
    if versions != [1]:
        self._connection.close()
        raise RuntimeError(f"unsupported cache schema versions: {versions}")

put

put(key: CacheKey, payload: bytes | bytearray | memoryview, *, privacy: CachePrivacy = STANDARD, ttl_seconds: float | None = None) -> CacheEntry

Atomically insert or replace a cache payload.

Source code in src/tunarag/cache.py

def put(
    self,
    key: CacheKey,
    payload: bytes | bytearray | memoryview,
    *,
    privacy: CachePrivacy = CachePrivacy.STANDARD,
    ttl_seconds: float | None = None,
) -> CacheEntry:
    """Atomically insert or replace a cache payload."""

    payload_bytes = bytes(payload)
    if ttl_seconds is not None:
        if not math.isfinite(ttl_seconds) or ttl_seconds <= 0:
            raise ValueError("cache TTL must be a positive finite number")
    now = self._now()
    expires_at = None if ttl_seconds is None else now + ttl_seconds
    checksum = _payload_checksum(payload_bytes)
    with self._connection:
        self._connection.execute(
            """
            INSERT INTO cache_entries(
                namespace, format_version, key_digest, payload, checksum, privacy,
                state, created_at, expires_at, last_accessed_at, hit_count,
                quarantine_reason
            ) VALUES (?, ?, ?, ?, ?, ?, 'active', ?, ?, NULL, 0, NULL)
            ON CONFLICT(namespace, format_version, key_digest) DO UPDATE SET
                payload = excluded.payload,
                checksum = excluded.checksum,
                privacy = excluded.privacy,
                state = 'active',
                created_at = excluded.created_at,
                expires_at = excluded.expires_at,
                last_accessed_at = NULL,
                hit_count = 0,
                quarantine_reason = NULL
            """,
            (
                key.namespace,
                key.format_version,
                key.digest,
                payload_bytes,
                checksum,
                privacy.value,
                now,
                expires_at,
            ),
        )
    return CacheEntry(key, checksum, privacy, now, expires_at)

get

get(key: CacheKey) -> CacheLookup

Return a verified payload or an explicit miss reason.

Source code in src/tunarag/cache.py

def get(self, key: CacheKey) -> CacheLookup:
    """Return a verified payload or an explicit miss reason."""

    row = self._select(key)
    if row is None:
        return CacheLookup(key, CacheLookupStatus.MISS)

    privacy = CachePrivacy(row["privacy"])
    created_at = float(row["created_at"])
    expires_at = row["expires_at"]
    metadata = {
        "checksum": row["checksum"],
        "privacy": privacy,
        "created_at": created_at,
        "expires_at": expires_at,
    }
    state = CacheEntryState(row["state"])
    if state is CacheEntryState.QUARANTINED:
        return CacheLookup(
            key,
            CacheLookupStatus.QUARANTINED,
            reason=row["quarantine_reason"],
            **metadata,
        )

    now = self._now()
    if state is CacheEntryState.EXPIRED or (expires_at is not None and expires_at <= now):
        if state is CacheEntryState.ACTIVE:
            with self._connection:
                self._connection.execute(
                    """UPDATE cache_entries SET state = 'expired'
                    WHERE namespace = ? AND format_version = ? AND key_digest = ?""",
                    (key.namespace, key.format_version, key.digest),
                )
        return CacheLookup(key, CacheLookupStatus.EXPIRED, **metadata)

    payload = bytes(row["payload"])
    if _payload_checksum(payload) != row["checksum"]:
        reason = "payload checksum mismatch"
        with self._connection:
            self._connection.execute(
                """UPDATE cache_entries
                SET state = 'quarantined', quarantine_reason = ?
                WHERE namespace = ? AND format_version = ? AND key_digest = ?""",
                (reason, key.namespace, key.format_version, key.digest),
            )
        return CacheLookup(key, CacheLookupStatus.CORRUPT, reason=reason, **metadata)

    with self._connection:
        self._connection.execute(
            """UPDATE cache_entries
            SET last_accessed_at = ?, hit_count = hit_count + 1
            WHERE namespace = ? AND format_version = ? AND key_digest = ?""",
            (now, key.namespace, key.format_version, key.digest),
        )
    return CacheLookup(key, CacheLookupStatus.HIT, payload=payload, **metadata)

delete

delete(key: CacheKey) -> bool

Delete one cache entry and report whether it existed.

Source code in src/tunarag/cache.py

def delete(self, key: CacheKey) -> bool:
    """Delete one cache entry and report whether it existed."""

    with self._connection:
        cursor = self._connection.execute(
            """DELETE FROM cache_entries
            WHERE namespace = ? AND format_version = ? AND key_digest = ?""",
            (key.namespace, key.format_version, key.digest),
        )
    return cursor.rowcount == 1

prune

prune(*, include_quarantined: bool = False) -> int

Remove expired entries and optionally quarantined entries.

Source code in src/tunarag/cache.py

def prune(self, *, include_quarantined: bool = False) -> int:
    """Remove expired entries and optionally quarantined entries."""

    now = self._now()
    conditions: Sequence[str] = (
        "state = 'expired'",
        "(expires_at IS NOT NULL AND expires_at <= ?)",
    )
    where = " OR ".join(conditions)
    parameters: list[float] = [now]
    if include_quarantined:
        where += " OR state = 'quarantined'"
    with self._connection:
        cursor = self._connection.execute(
            f"DELETE FROM cache_entries WHERE {where}", parameters
        )
    return cursor.rowcount

close

close() -> None

Close the underlying SQLite connection.

Source code in src/tunarag/cache.py

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

    self._connection.close()