#[cfg(feature = "sqlite-store")]
mod inner {
use std::path::{Path, PathBuf};
#[cfg(test)]
use std::sync::atomic::{AtomicU8, Ordering};
use meerkat_core::lifecycle::{InputId, RunBoundaryReceipt, RunId};
use meerkat_store::json_column::JsonColumnBytes;
use meerkat_store::sqlite_store::{begin_immediate_transaction, open_connection};
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use crate::identifiers::LogicalRuntimeId;
use crate::input_state::{InputStatePersistenceRecord, StoredInputState};
use crate::store::{
AuthOAuthFlowSnapshotUpdate, MachineLifecycleCommit, MachineLifecycleSnapshot,
MachineLifecycleStoreRecord, RuntimeStore, RuntimeStoreError, SessionDelta,
};
const CREATE_RUNTIME_SCHEMA_SQL: &str = r"
CREATE TABLE IF NOT EXISTS runtime_input_states (
runtime_id TEXT NOT NULL,
input_id TEXT NOT NULL,
state_json BLOB NOT NULL,
PRIMARY KEY (runtime_id, input_id)
);
CREATE TABLE IF NOT EXISTS runtime_boundary_receipts (
runtime_id TEXT NOT NULL,
run_id TEXT NOT NULL,
sequence INTEGER NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (runtime_id, run_id, sequence)
);
CREATE TABLE IF NOT EXISTS runtime_session_snapshots (
runtime_id TEXT PRIMARY KEY,
session_snapshot BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS runtime_states (
runtime_id TEXT PRIMARY KEY,
runtime_state_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS runtime_ops_lifecycle (
runtime_id TEXT PRIMARY KEY,
state_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS runtime_retired_ops_epochs (
runtime_id TEXT NOT NULL,
epoch_id TEXT NOT NULL,
PRIMARY KEY (runtime_id, epoch_id)
);
CREATE TABLE IF NOT EXISTS runtime_auth_oauth_flow_state (
id TEXT PRIMARY KEY,
state_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS runtime_projection_quarantine (
runtime_id TEXT PRIMARY KEY
);
CREATE TABLE IF NOT EXISTS runtime_compaction_projection_outbox (
runtime_id TEXT NOT NULL,
session_id TEXT NOT NULL,
parent_revision TEXT NOT NULL,
revision TEXT NOT NULL,
commit_fingerprint TEXT NOT NULL,
intent_json BLOB NOT NULL,
state TEXT NOT NULL CHECK (state IN ('pending', 'finalized')),
PRIMARY KEY (runtime_id, session_id, parent_revision, revision, commit_fingerprint)
)";
fn ensure_runtime_schema(conn: &Connection) -> Result<(), RuntimeStoreError> {
conn.execute_batch(CREATE_RUNTIME_SCHEMA_SQL)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
}
fn open_runtime_connection(path: &Path) -> Result<Connection, RuntimeStoreError> {
let conn =
open_connection(path).map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
ensure_runtime_schema(&conn)?;
Ok(conn)
}
fn begin_runtime_transaction(
conn: &mut Connection,
) -> Result<Transaction<'_>, RuntimeStoreError> {
begin_immediate_transaction(conn)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
}
fn runtime_id_text(runtime_id: &LogicalRuntimeId) -> &str {
&runtime_id.0
}
fn deserialize_persisted_session(
bytes: &[u8],
) -> Result<meerkat_core::Session, RuntimeStoreError> {
serde_json::from_slice(bytes).map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
}
fn deserialize_persisted_input_state(
bytes: &[u8],
) -> Result<StoredInputState, RuntimeStoreError> {
serde_json::from_slice(bytes).map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
}
fn encode_receipt_sequence(sequence: u64) -> i64 {
i64::from_ne_bytes(sequence.to_ne_bytes())
}
#[cfg(test)]
fn decode_receipt_sequence(stored: i64) -> u64 {
u64::from_ne_bytes(stored.to_ne_bytes())
}
fn is_runtime_placeholder_session(session: &meerkat_core::Session) -> bool {
session.transcript_history_state().ok().flatten().is_none()
&& matches!(
session.messages(),
[] | [meerkat_core::types::Message::System(_)]
)
}
const AUTH_OAUTH_FLOW_STATE_ID: &str = "auth_oauth_flow_state";
fn upsert_runtime_snapshot(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
snapshot: &[u8],
) -> Result<(), RuntimeStoreError> {
tx.execute(
r"
INSERT INTO runtime_session_snapshots (runtime_id, session_snapshot)
VALUES (?1, ?2)
ON CONFLICT(runtime_id) DO UPDATE SET session_snapshot = excluded.session_snapshot
",
params![runtime_id_text(runtime_id), snapshot],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
clear_runtime_projection_quarantine(tx, runtime_id)?;
Ok(())
}
fn set_runtime_projection_quarantine(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
tx.execute(
r"
INSERT OR REPLACE INTO runtime_projection_quarantine (runtime_id)
VALUES (?1)
",
params![runtime_id_text(runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
fn clear_runtime_projection_quarantine(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
tx.execute(
"DELETE FROM runtime_projection_quarantine WHERE runtime_id = ?1",
params![runtime_id_text(runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
fn insert_compaction_projection_outbox_intents(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
intents: &[meerkat_core::CompactionProjectionIntent],
) -> Result<(), RuntimeStoreError> {
for intent in intents {
let encoded = serde_json::to_vec(intent)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
tx.execute(
r"
INSERT OR IGNORE INTO runtime_compaction_projection_outbox
(runtime_id, session_id, parent_revision, revision, commit_fingerprint, intent_json, state)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'pending')
",
params![
runtime_id_text(runtime_id),
intent.projection.session_id().to_string(),
intent.projection.parent_revision(),
intent.projection.revision(),
intent.projection.commit_fingerprint(),
encoded,
],
)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
let existing = tx
.query_row(
r"
SELECT intent_json
FROM runtime_compaction_projection_outbox
WHERE runtime_id = ?1 AND session_id = ?2
AND parent_revision = ?3 AND revision = ?4
AND commit_fingerprint = ?5
",
params![
runtime_id_text(runtime_id),
intent.projection.session_id().to_string(),
intent.projection.parent_revision(),
intent.projection.revision(),
intent.projection.commit_fingerprint(),
],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
let existing: meerkat_core::CompactionProjectionIntent =
serde_json::from_slice(&existing)
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
if existing != *intent {
return Err(RuntimeStoreError::WriteFailed(format!(
"conflicting compaction outbox intent for rewrite {}",
intent.projection.revision()
)));
}
}
Ok(())
}
fn ensure_compaction_intents_already_outboxed(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
session: &meerkat_core::Session,
) -> Result<(), RuntimeStoreError> {
for intent in crate::store::validated_compaction_projection_intents(session)? {
let (encoded, state) = tx
.query_row(
r"
SELECT intent_json, state
FROM runtime_compaction_projection_outbox
WHERE runtime_id = ?1 AND session_id = ?2
AND parent_revision = ?3 AND revision = ?4
AND commit_fingerprint = ?5
",
params![
runtime_id_text(runtime_id),
intent.projection.session_id().to_string(),
intent.projection.parent_revision(),
intent.projection.revision(),
intent.projection.commit_fingerprint(),
],
|row| {
Ok((
row.get::<_, JsonColumnBytes>(0)?.into_bytes(),
row.get::<_, String>(1)?,
))
},
)
.optional()
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?
.ok_or_else(|| {
RuntimeStoreError::WriteFailed(format!(
"non-boundary snapshot introduces compaction intent {} without atomic outbox authority",
intent.projection.revision()
))
})?;
if state == "finalized" {
return Err(RuntimeStoreError::WriteFailed(format!(
"non-boundary snapshot replays finalized compaction intent {}",
intent.projection.revision()
)));
}
let existing: meerkat_core::CompactionProjectionIntent =
serde_json::from_slice(&encoded)
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
if existing != intent {
return Err(RuntimeStoreError::WriteFailed(format!(
"non-boundary snapshot conflicts with compaction outbox rewrite {}",
intent.projection.revision()
)));
}
}
Ok(())
}
fn reject_finalized_compaction_projection_replays(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
intents: &[meerkat_core::CompactionProjectionIntent],
) -> Result<(), RuntimeStoreError> {
for intent in intents {
let state = tx
.query_row(
r"
SELECT state
FROM runtime_compaction_projection_outbox
WHERE runtime_id = ?1 AND session_id = ?2
AND parent_revision = ?3 AND revision = ?4
AND commit_fingerprint = ?5
",
params![
runtime_id_text(runtime_id),
intent.projection.session_id().to_string(),
intent.projection.parent_revision(),
intent.projection.revision(),
intent.projection.commit_fingerprint(),
],
|row| row.get::<_, String>(0),
)
.optional()
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
if state.as_deref() == Some("finalized") {
return Err(RuntimeStoreError::WriteFailed(format!(
"atomic session snapshot replays finalized compaction intent {}",
intent.projection.revision()
)));
}
}
Ok(())
}
fn insert_receipt(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
receipt: &RunBoundaryReceipt,
) -> Result<(), RuntimeStoreError> {
let receipt_json = serde_json::to_vec(receipt)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.execute(
r"
INSERT INTO runtime_boundary_receipts (runtime_id, run_id, sequence, receipt_json)
VALUES (?1, ?2, ?3, ?4)
",
params![
runtime_id_text(runtime_id),
receipt.run_id.0.to_string(),
encode_receipt_sequence(receipt.sequence),
receipt_json,
],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
fn upsert_input_states(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
input_states: &[StoredInputState],
) -> Result<(), RuntimeStoreError> {
for bundle in input_states {
let state_json = serde_json::to_vec(bundle)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.execute(
r"
INSERT INTO runtime_input_states (runtime_id, input_id, state_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(runtime_id, input_id) DO UPDATE SET state_json = excluded.state_json
",
params![
runtime_id_text(runtime_id),
bundle.state.input_id.0.to_string(),
state_json
],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
}
Ok(())
}
fn upsert_machine_lifecycle_snapshot(
tx: &Transaction<'_>,
runtime_id: &LogicalRuntimeId,
snapshot: &MachineLifecycleSnapshot,
) -> Result<(), RuntimeStoreError> {
let state_json = MachineLifecycleStoreRecord::from_snapshot(snapshot).encode()?;
tx.execute(
r"
INSERT INTO runtime_states (runtime_id, runtime_state_json)
VALUES (?1, ?2)
ON CONFLICT(runtime_id) DO UPDATE SET runtime_state_json = excluded.runtime_state_json
",
params![runtime_id_text(runtime_id), state_json],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
#[derive(Debug, PartialEq, Eq)]
struct UnregisterFinalizationObservation {
lifecycle_record: Option<Vec<u8>>,
input_state_records: Vec<(String, Option<Vec<u8>>)>,
ops_record: Option<Vec<u8>>,
retired_ops_epoch_present: bool,
}
fn observe_unregister_finalization(
conn: &Connection,
runtime_id: &LogicalRuntimeId,
input_ids: &[String],
retired_ops_epoch: &meerkat_core::RuntimeEpochId,
) -> Result<UnregisterFinalizationObservation, RuntimeStoreError> {
let lifecycle_record = conn
.query_row(
"SELECT runtime_state_json FROM runtime_states WHERE runtime_id = ?1",
params![runtime_id_text(runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
let input_state_records = input_ids
.iter()
.map(|input_id| {
conn.query_row(
r"
SELECT state_json
FROM runtime_input_states
WHERE runtime_id = ?1 AND input_id = ?2
",
params![runtime_id_text(runtime_id), input_id],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map(|record| (input_id.clone(), record))
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.collect::<Result<Vec<_>, _>>()?;
let ops_record = conn
.query_row(
"SELECT state_json FROM runtime_ops_lifecycle WHERE runtime_id = ?1",
params![runtime_id_text(runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
let retired_ops_epoch_present = conn
.query_row(
r"
SELECT 1
FROM runtime_retired_ops_epochs
WHERE runtime_id = ?1 AND epoch_id = ?2
",
params![runtime_id_text(runtime_id), retired_ops_epoch.to_string()],
|_row| Ok(()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.is_some();
Ok(UnregisterFinalizationObservation {
lifecycle_record,
input_state_records,
ops_record,
retired_ops_epoch_present,
})
}
pub struct SqliteRuntimeStore {
path: PathBuf,
#[cfg(test)]
unregister_finalization_fault: AtomicU8,
}
impl SqliteRuntimeStore {
pub fn new(path: impl Into<PathBuf>) -> Result<Self, RuntimeStoreError> {
let path = path.into();
let conn = open_runtime_connection(&path)?;
drop(conn);
Ok(Self {
path,
#[cfg(test)]
unregister_finalization_fault: AtomicU8::new(0),
})
}
pub fn path(&self) -> &Path {
&self.path
}
#[cfg(test)]
fn inject_unregister_finalization_fault(&self, fault: u8) {
self.unregister_finalization_fault
.store(fault, Ordering::SeqCst);
}
}
#[async_trait::async_trait]
impl RuntimeStore for SqliteRuntimeStore {
fn supports_compaction_projection_outbox(&self) -> bool {
true
}
fn auth_authority_key(&self) -> Option<String> {
let path = std::fs::canonicalize(&self.path).unwrap_or_else(|_| self.path.clone());
Some(format!("sqlite:{}", path.display()))
}
fn persist_auth_oauth_flow_snapshot(
&self,
snapshot_json: &[u8],
) -> Result<(), RuntimeStoreError> {
let mut conn = open_runtime_connection(&self.path)?;
let tx = begin_runtime_transaction(&mut conn)?;
tx.execute(
r"
INSERT INTO runtime_auth_oauth_flow_state (id, state_json)
VALUES (?1, ?2)
ON CONFLICT(id) DO UPDATE SET state_json = excluded.state_json
",
params![AUTH_OAUTH_FLOW_STATE_ID, snapshot_json],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
fn load_auth_oauth_flow_snapshot(&self) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
let conn = open_runtime_connection(&self.path)?;
conn.query_row(
r"
SELECT state_json
FROM runtime_auth_oauth_flow_state
WHERE id = ?1
",
params![AUTH_OAUTH_FLOW_STATE_ID],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
}
fn update_auth_oauth_flow_snapshot(
&self,
update: &mut AuthOAuthFlowSnapshotUpdate<'_>,
) -> Result<(), RuntimeStoreError> {
let mut conn = open_runtime_connection(&self.path)?;
let tx = begin_runtime_transaction(&mut conn)?;
let current = tx
.query_row(
r"
SELECT state_json
FROM runtime_auth_oauth_flow_state
WHERE id = ?1
",
params![AUTH_OAUTH_FLOW_STATE_ID],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
let next = update(current.as_deref())?;
tx.execute(
r"
INSERT INTO runtime_auth_oauth_flow_state (id, state_json)
VALUES (?1, ?2)
ON CONFLICT(id) DO UPDATE SET state_json = excluded.state_json
",
params![AUTH_OAUTH_FLOW_STATE_ID, next],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
}
async fn commit_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: SessionDelta,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let incoming =
serde_json::from_slice(&session_delta.session_snapshot)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
ensure_compaction_intents_already_outboxed(&tx, &runtime_id, &incoming)?;
let previous = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| deserialize_persisted_session(&bytes))
.transpose()?;
meerkat_core::session_store::run_boundary_snapshot_save_guard(
&incoming,
previous.as_ref(),
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
upsert_runtime_snapshot(&tx, &runtime_id, &session_delta.session_snapshot)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn commit_session_transcript_rewrite_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: SessionDelta,
commit: &meerkat_core::TranscriptRewriteCommit,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let commit = commit.clone();
tokio::task::spawn_blocking(move || {
let incoming = serde_json::from_slice::<meerkat_core::Session>(
&session_delta.session_snapshot,
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
ensure_compaction_intents_already_outboxed(&tx, &runtime_id, &incoming)?;
let previous = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| deserialize_persisted_session(&bytes))
.transpose()?;
meerkat_core::session_store::transcript_rewrite_save_guard(
&incoming,
previous.as_ref(),
&commit,
)
.map_err(|err| match err {
meerkat_core::SessionStoreError::TranscriptRevisionConflict {
expected, actual, ..
} => RuntimeStoreError::TranscriptRevisionConflict { expected, actual },
other => RuntimeStoreError::WriteFailed(other.to_string()),
})?;
upsert_runtime_snapshot(&tx, &runtime_id, &session_delta.session_snapshot)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn atomic_apply(
&self,
runtime_id: &LogicalRuntimeId,
session_delta: Option<SessionDelta>,
receipt: RunBoundaryReceipt,
input_updates: Vec<InputStatePersistenceRecord>,
session_store_key: Option<meerkat_core::types::SessionId>,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let input_updates = input_updates
.into_iter()
.map(InputStatePersistenceRecord::into_stored)
.collect::<Vec<_>>();
tokio::task::spawn_blocking(move || {
let session_snapshot = session_delta
.as_ref()
.map(|delta| {
serde_json::from_slice::<meerkat_core::Session>(&delta.session_snapshot)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
})
.transpose()?;
let compaction_intents = session_snapshot
.as_ref()
.map(crate::store::validated_compaction_projection_intents)
.transpose()?
.unwrap_or_default();
if let (Some(session), Some(session_store_key)) =
(session_snapshot.as_ref(), session_store_key.as_ref())
&& session.id() != session_store_key
{
return Err(RuntimeStoreError::SessionKeyMismatch {
expected: session_store_key.clone(),
actual: session.id().clone(),
});
}
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
reject_finalized_compaction_projection_replays(
&tx,
&runtime_id,
&compaction_intents,
)?;
let mut session_snapshot_superseded = false;
if let Some(session) = session_snapshot.as_ref() {
let mut persist_session_snapshot = true;
let previous = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| deserialize_persisted_session(&bytes))
.transpose()?;
if let Err(err) = meerkat_core::session_store::run_boundary_snapshot_save_guard(
session,
previous.as_ref(),
) {
if previous.as_ref().is_some_and(is_runtime_placeholder_session) {
persist_session_snapshot = true;
} else if previous.as_ref().is_some_and(|previous| {
meerkat_core::session_store::run_boundary_snapshot_save_guard(
previous,
Some(session),
)
.is_ok()
}) {
persist_session_snapshot = false;
session_snapshot_superseded = true;
} else {
return Err(RuntimeStoreError::WriteFailed(err.to_string()));
}
}
if persist_session_snapshot
&& let Some(delta) = session_delta.as_ref()
{
upsert_runtime_snapshot(&tx, &runtime_id, &delta.session_snapshot)?;
}
}
if session_snapshot_superseded {
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
return Ok(());
}
insert_compaction_projection_outbox_intents(
&tx,
&runtime_id,
&compaction_intents,
)?;
insert_receipt(&tx, &runtime_id, &receipt)?;
upsert_input_states(&tx, &runtime_id, &input_updates)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_pending_compaction_projections(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<meerkat_core::CompactionProjectionIntent>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
let mut statement = conn
.prepare(
r"
SELECT intent_json
FROM runtime_compaction_projection_outbox
WHERE runtime_id = ?1 AND state = 'pending'
ORDER BY session_id, parent_revision, revision, commit_fingerprint
",
)
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
let rows = statement
.query_map(params![runtime_id_text(&runtime_id)], |row| {
Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes())
})
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
rows.map(|row| {
let encoded =
row.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?;
serde_json::from_slice(&encoded)
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))
})
.collect()
})
.await
.map_err(|error| RuntimeStoreError::Internal(format!("Task join failed: {error}")))?
}
async fn mark_compaction_projection_finalized(
&self,
runtime_id: &LogicalRuntimeId,
projection: &meerkat_core::CompactionProjectionId,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let projection = projection.clone();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
let exists = tx
.query_row(
r"
SELECT 1 FROM runtime_compaction_projection_outbox
WHERE runtime_id = ?1 AND session_id = ?2
AND parent_revision = ?3 AND revision = ?4
AND commit_fingerprint = ?5
",
params![
runtime_id_text(&runtime_id),
projection.session_id().to_string(),
projection.parent_revision(),
projection.revision(),
projection.commit_fingerprint(),
],
|_row| Ok(()),
)
.optional()
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?
.is_some();
if !exists {
return Err(RuntimeStoreError::NotFound(format!(
"compaction outbox rewrite {}",
projection.revision()
)));
}
if let Some(snapshot) = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|error| RuntimeStoreError::ReadFailed(error.to_string()))?
{
let mut session = deserialize_persisted_session(&snapshot)?;
session
.complete_compaction_projection_intent(&projection)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
let cleaned = serde_json::to_vec(&session)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
upsert_runtime_snapshot(&tx, &runtime_id, &cleaned)?;
}
tx
.execute(
r"
UPDATE runtime_compaction_projection_outbox
SET state = 'finalized'
WHERE runtime_id = ?1 AND session_id = ?2
AND parent_revision = ?3 AND revision = ?4
AND commit_fingerprint = ?5
",
params![
runtime_id_text(&runtime_id),
projection.session_id().to_string(),
projection.parent_revision(),
projection.revision(),
projection.commit_fingerprint(),
],
)
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
tx.commit()
.map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
Ok(())
})
.await
.map_err(|error| RuntimeStoreError::Internal(format!("Task join failed: {error}")))?
}
async fn load_input_states(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Vec<StoredInputState>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
let mut stmt = conn
.prepare(
r"
SELECT state_json
FROM runtime_input_states
WHERE runtime_id = ?1
ORDER BY input_id ASC
",
)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
let rows = stmt
.query_map(params![runtime_id_text(&runtime_id)], |row| {
row.get::<_, JsonColumnBytes>(0)
.map(JsonColumnBytes::into_bytes)
})
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
rows.map(|row| {
let bytes =
row.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
deserialize_persisted_input_state(&bytes)
})
.collect()
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_boundary_receipt(
&self,
runtime_id: &LogicalRuntimeId,
run_id: &RunId,
sequence: u64,
) -> Result<Option<RunBoundaryReceipt>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let run_id = run_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
r"
SELECT receipt_json
FROM runtime_boundary_receipts
WHERE runtime_id = ?1 AND run_id = ?2 AND sequence = ?3
",
params![
runtime_id_text(&runtime_id),
run_id.0.to_string(),
encode_receipt_sequence(sequence)
],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| {
serde_json::from_slice(&bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.transpose()
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn clear_session_snapshot(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
tx.execute(
"DELETE FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn replace_session_snapshot_if_current(
&self,
runtime_id: &LogicalRuntimeId,
expected_current: &[u8],
replacement: Vec<u8>,
) -> Result<bool, RuntimeStoreError> {
let replacement_session: meerkat_core::Session =
serde_json::from_slice(&replacement)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let expected_current = expected_current.to_vec();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
let current = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
if current.as_deref() != Some(expected_current.as_slice()) {
return Ok(false);
}
ensure_compaction_intents_already_outboxed(
&tx,
&runtime_id,
&replacement_session,
)?;
upsert_runtime_snapshot(&tx, &runtime_id, &replacement)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(true)
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn clear_session_snapshot_if_current(
&self,
runtime_id: &LogicalRuntimeId,
expected_current: &[u8],
) -> Result<bool, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let expected_current = expected_current.to_vec();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
let current = tx
.query_row(
"SELECT session_snapshot FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
if current.as_deref() != Some(expected_current.as_slice()) {
return Ok(false);
}
tx.execute(
"DELETE FROM runtime_session_snapshots WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
set_runtime_projection_quarantine(&tx, &runtime_id)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(true)
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn is_runtime_projection_quarantined(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<bool, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
"SELECT EXISTS(SELECT 1 FROM runtime_projection_quarantine WHERE runtime_id = ?1)",
params![runtime_id_text(&runtime_id)],
|row| row.get::<_, bool>(0),
)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn persist_input_state(
&self,
runtime_id: &LogicalRuntimeId,
state: &InputStatePersistenceRecord,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let state = state.clone_stored();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
upsert_input_states(&tx, &runtime_id, &[state])?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_input_state(
&self,
runtime_id: &LogicalRuntimeId,
input_id: &InputId,
) -> Result<Option<StoredInputState>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let input_id = input_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
r"
SELECT state_json
FROM runtime_input_states
WHERE runtime_id = ?1 AND input_id = ?2
",
params![runtime_id_text(&runtime_id), input_id.0.to_string()],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| deserialize_persisted_input_state(&bytes))
.transpose()
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_machine_lifecycle_record(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
"SELECT runtime_state_json FROM runtime_states WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn commit_machine_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
commit: MachineLifecycleCommit,
input_states: &[InputStatePersistenceRecord],
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let snapshot = commit.into_snapshot();
let input_states = input_states
.iter()
.map(InputStatePersistenceRecord::clone_stored)
.collect::<Vec<_>>();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
upsert_machine_lifecycle_snapshot(&tx, &runtime_id, &snapshot)?;
upsert_input_states(&tx, &runtime_id, &input_states)?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn commit_unregister_finalization(
&self,
runtime_id: &LogicalRuntimeId,
commit: MachineLifecycleCommit,
input_states: &[InputStatePersistenceRecord],
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let retired_ops_epoch = commit.retired_ops_epoch().cloned().ok_or_else(|| {
RuntimeStoreError::WriteFailed(
"unregister finalization missing exact retired ops epoch".into(),
)
})?;
let snapshot = commit.into_snapshot();
let input_states = input_states
.iter()
.map(InputStatePersistenceRecord::clone_stored)
.collect::<Vec<_>>();
#[cfg(test)]
let fault = self.unregister_finalization_fault.swap(0, Ordering::SeqCst);
#[cfg(not(test))]
let fault = 0_u8;
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let final_lifecycle_record =
MachineLifecycleStoreRecord::from_snapshot(&snapshot).encode()?;
let final_input_state_records = input_states
.iter()
.map(|state| {
serde_json::to_vec(state)
.map(|record| {
(state.state.input_id.0.to_string(), Some(record))
})
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
})
.collect::<Result<Vec<_>, _>>()?;
let input_ids = final_input_state_records
.iter()
.map(|(input_id, _)| input_id.clone())
.collect::<Vec<_>>();
let tx = begin_runtime_transaction(&mut conn)?;
let before_observation = observe_unregister_finalization(
&tx,
&runtime_id,
&input_ids,
&retired_ops_epoch,
)?;
let final_ops_record = match before_observation.ops_record.as_ref() {
Some(bytes) => {
let persisted: crate::ops_lifecycle::PersistedOpsSnapshot =
serde_json::from_slice(bytes).map_err(|error| {
RuntimeStoreError::ReadFailed(format!(
"failed to decode ops epoch before unregister finalization: {error}"
))
})?;
if persisted.epoch_id == retired_ops_epoch {
None
} else {
Some(bytes.clone())
}
}
None => None,
};
let final_observation = UnregisterFinalizationObservation {
lifecycle_record: Some(final_lifecycle_record),
input_state_records: final_input_state_records,
ops_record: final_ops_record.clone(),
retired_ops_epoch_present: true,
};
upsert_machine_lifecycle_snapshot(&tx, &runtime_id, &snapshot)?;
upsert_input_states(&tx, &runtime_id, &input_states)?;
if fault == 1 {
return Err(RuntimeStoreError::WriteFailed(
"synthetic power cut after unregister lifecycle write".to_string(),
));
}
tx.execute(
r"
INSERT INTO runtime_retired_ops_epochs (runtime_id, epoch_id)
VALUES (?1, ?2)
ON CONFLICT(runtime_id, epoch_id) DO NOTHING
",
params![runtime_id_text(&runtime_id), retired_ops_epoch.to_string()],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
if before_observation.ops_record.is_some() && final_ops_record.is_none() {
tx.execute(
"DELETE FROM runtime_ops_lifecycle WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
}
if fault == 2 {
return Err(RuntimeStoreError::WriteFailed(
"synthetic power cut after unregister ops deletion".to_string(),
));
}
let commit_error = match tx.commit() {
Ok(()) if fault != 3 => return Ok(()),
Ok(()) => {
"synthetic lost acknowledgement after unregister commit".to_string()
}
Err(err) => err.to_string(),
};
drop(conn);
let observed = open_runtime_connection(&path)
.and_then(|conn| {
observe_unregister_finalization(
&conn,
&runtime_id,
&input_ids,
&retired_ops_epoch,
)
})
.map_err(|observation_error| {
RuntimeStoreError::UnregisterFinalizationOutcomeUnknown(format!(
"commit acknowledgement failed ({commit_error}); durable outcome read failed: {observation_error}"
))
})?;
if observed == final_observation {
return Ok(());
}
if observed == before_observation {
return Err(RuntimeStoreError::WriteFailed(commit_error));
}
Err(RuntimeStoreError::UnregisterFinalizationOutcomeUnknown(
format!(
"commit acknowledgement failed ({commit_error}); reopened lifecycle/input/ops bytes match neither final nor pre-transaction authority"
),
))
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn persist_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
snapshot: &crate::ops_lifecycle::PersistedOpsSnapshot,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
let snapshot = snapshot.clone();
tokio::task::spawn_blocking(move || {
let state_json = serde_json::to_vec(&snapshot)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
let retired = tx
.query_row(
r"
SELECT 1
FROM runtime_retired_ops_epochs
WHERE runtime_id = ?1 AND epoch_id = ?2
",
params![runtime_id_text(&runtime_id), snapshot.epoch_id.to_string()],
|_row| Ok(()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.is_some();
if retired {
return Err(RuntimeStoreError::OpsLifecycleEpochRetired {
runtime_id: runtime_id.0.clone(),
epoch_id: snapshot.epoch_id,
});
}
tx.execute(
r"
INSERT INTO runtime_ops_lifecycle (runtime_id, state_json)
VALUES (?1, ?2)
ON CONFLICT(runtime_id) DO UPDATE SET state_json = excluded.state_json
",
params![runtime_id_text(&runtime_id), state_json],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn load_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<Option<crate::ops_lifecycle::PersistedOpsSnapshot>, RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let conn = open_runtime_connection(&path)?;
conn.query_row(
"SELECT state_json FROM runtime_ops_lifecycle WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
|row| Ok(row.get::<_, JsonColumnBytes>(0)?.into_bytes()),
)
.optional()
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?
.map(|bytes| {
serde_json::from_slice(&bytes)
.map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))
})
.transpose()
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
async fn delete_ops_lifecycle(
&self,
runtime_id: &LogicalRuntimeId,
) -> Result<(), RuntimeStoreError> {
let path = self.path.clone();
let runtime_id = runtime_id.clone();
tokio::task::spawn_blocking(move || {
let mut conn = open_runtime_connection(&path)?;
let tx = begin_runtime_transaction(&mut conn)?;
tx.execute(
"DELETE FROM runtime_ops_lifecycle WHERE runtime_id = ?1",
params![runtime_id_text(&runtime_id)],
)
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
tx.commit()
.map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))?;
Ok(())
})
.await
.map_err(|err| RuntimeStoreError::Internal(format!("Task join failed: {err}")))?
}
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
use tempfile::TempDir;
use super::*;
use crate::identifiers::LogicalRuntimeId;
use crate::runtime_state::RuntimeState;
use meerkat_core::lifecycle::run_primitive::RunApplyBoundary;
use meerkat_core::lifecycle::{InputId, RunBoundaryReceipt, RunId};
use meerkat_core::session_store::SessionStore as _;
use meerkat_core::types::{
AssistantBlock, BlockAssistantMessage, Message, StopReason, UserMessage,
};
use meerkat_core::{Session, TranscriptRewriteReason, TranscriptRewriteSelection};
use meerkat_store::SqliteSessionStore;
fn temp_store() -> (TempDir, SqliteRuntimeStore) {
let dir = TempDir::new().unwrap();
let path = dir.path().join("sessions.sqlite3");
let store = SqliteRuntimeStore::new(path).unwrap();
(dir, store)
}
fn runtime_id() -> LogicalRuntimeId {
LogicalRuntimeId("runtime-1".to_string())
}
fn input_state() -> InputStatePersistenceRecord {
InputStatePersistenceRecord::from_machine_snapshot(StoredInputState::new_accepted(
InputId::new(),
))
.expect("accepted test input state seed must be machine-authorized")
}
fn session_with_one_turn() -> Session {
let mut session = Session::new();
session.push(Message::User(UserMessage::text("hello".to_string())));
session.push(Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "verbose answer".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
}));
session
}
fn session_with_user(content: &str) -> Session {
let mut session = Session::new();
session.push(Message::User(UserMessage::text(content.to_string())));
session
}
fn session_with_compaction_intent() -> (Session, meerkat_core::CompactionProjectionIntent) {
let mut session = session_with_user("verbose context one");
session.push(Message::User(UserMessage::text("verbose context two")));
let parent = session.transcript_revision().unwrap();
session
.commit_transcript_rewrite(
TranscriptRewriteSelection::MessageRange { start: 0, end: 2 },
vec![Message::User(UserMessage::compaction_summary(
"compacted context",
))],
TranscriptRewriteReason::new("compaction"),
Some("sqlite-outbox-test".to_string()),
Some(parent),
)
.unwrap();
let mut encoded = serde_json::to_value(&session).unwrap();
encoded["metadata"][meerkat_core::SESSION_TRANSCRIPT_HISTORY_STATE_KEY]["commits"][0]
["selection"] = serde_json::json!({
"type": "compaction_message_range",
"range": { "start": 0, "end": 2 }
});
let mut session: Session = serde_json::from_value(encoded).unwrap();
let commit = session
.transcript_history_state()
.unwrap()
.unwrap()
.commits
.last()
.unwrap()
.clone();
let intent = meerkat_core::CompactionProjectionIntent {
projection: serde_json::from_value(serde_json::json!({
"session_id": session.id(),
"parent_revision": &commit.parent_revision,
"revision": &commit.revision,
"commit_fingerprint": "sha256:aee1fea2386a630969f33a58068390400ed9c0e5964a1838269ae2eeab2761da",
}))
.unwrap(),
summary_tokens: 5,
messages_before: 2,
messages_after: 1,
};
session
.add_compaction_projection_intent(intent.clone())
.unwrap();
(session, intent)
}
fn snapshot_with_raw_intents(
session: &Session,
intents: &[meerkat_core::CompactionProjectionIntent],
) -> Vec<u8> {
let mut value = serde_json::to_value(session).unwrap();
value["metadata"][meerkat_core::memory::SESSION_COMPACTION_PROJECTION_INTENTS_KEY] =
serde_json::to_value(intents).unwrap();
serde_json::to_vec(&value).unwrap()
}
fn unbacked_intent(
session_id: &meerkat_core::types::SessionId,
) -> meerkat_core::CompactionProjectionIntent {
meerkat_core::CompactionProjectionIntent {
projection: serde_json::from_value(serde_json::json!({
"session_id": session_id,
"parent_revision": "missing-parent",
"revision": "missing-revision",
"commit_fingerprint": "sha256:unbacked-persisted-fixture",
}))
.unwrap(),
summary_tokens: 1,
messages_before: 2,
messages_after: 1,
}
}
#[tokio::test]
async fn compaction_outbox_is_atomic_durable_and_finalize_ack_is_idempotent() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("runtime.sqlite3");
let runtime_id = runtime_id();
let (session, intent) = session_with_compaction_intent();
let snapshot = serde_json::to_vec(&session).unwrap();
{
let store = SqliteRuntimeStore::new(&path).unwrap();
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: snapshot.clone(),
}),
RunBoundaryReceipt {
run_id: RunId::new(),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence: 41,
},
vec![],
Some(session.id().clone()),
)
.await
.unwrap();
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(snapshot)
);
}
let reopened = SqliteRuntimeStore::new(&path).unwrap();
assert_eq!(
reopened
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap(),
vec![intent.clone()]
);
reopened
.mark_compaction_projection_finalized(&runtime_id, &intent.projection)
.await
.unwrap();
reopened
.mark_compaction_projection_finalized(&runtime_id, &intent.projection)
.await
.unwrap();
assert!(
reopened
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
drop(reopened);
let after_ack_reopen = SqliteRuntimeStore::new(&path).unwrap();
let persisted: Session = serde_json::from_slice(
&after_ack_reopen
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.unwrap(),
)
.unwrap();
assert!(
persisted
.compaction_projection_intents()
.unwrap()
.is_empty()
);
assert!(
after_ack_reopen
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn finalized_sqlite_outbox_tombstone_rejects_all_snapshot_replay_paths() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let (session, intent) = session_with_compaction_intent();
let replay_snapshot = serde_json::to_vec(&session).unwrap();
let commit = session
.transcript_history_state()
.unwrap()
.unwrap()
.commits
.last()
.unwrap()
.clone();
let receipt = |run_id, sequence| RunBoundaryReceipt {
run_id,
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence,
};
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: replay_snapshot.clone(),
}),
receipt(RunId::new(), 80),
vec![],
Some(session.id().clone()),
)
.await
.unwrap();
store
.mark_compaction_projection_finalized(&runtime_id, &intent.projection)
.await
.unwrap();
let cleaned_snapshot = store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.unwrap();
let replay_run_id = RunId::new();
let error = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: replay_snapshot.clone(),
}),
receipt(replay_run_id.clone(), 81),
vec![],
Some(session.id().clone()),
)
.await
.unwrap_err();
assert!(error.to_string().contains("finalized compaction intent"));
assert!(
store
.load_boundary_receipt(&runtime_id, &replay_run_id, 81)
.await
.unwrap()
.is_none(),
"finalized replay rejection must roll back the whole SQLite transaction"
);
let error = store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: replay_snapshot.clone(),
},
)
.await
.unwrap_err();
assert!(error.to_string().contains("finalized compaction intent"));
let error = store
.commit_session_transcript_rewrite_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: replay_snapshot.clone(),
},
&commit,
)
.await
.unwrap_err();
assert!(error.to_string().contains("finalized compaction intent"));
let error = store
.replace_session_snapshot_if_current(
&runtime_id,
&cleaned_snapshot,
replay_snapshot,
)
.await
.unwrap_err();
assert!(error.to_string().contains("finalized compaction intent"));
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(cleaned_snapshot)
);
assert!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty(),
"a finalized SQLite tombstone must never be silently revived or left untracked"
);
}
#[tokio::test]
async fn invalid_compaction_outbox_intent_rolls_back_snapshot_and_outbox() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let (session, mut conflicting) = session_with_compaction_intent();
let original = session.compaction_projection_intents().unwrap()[0].clone();
conflicting.summary_tokens += 1;
let error = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: snapshot_with_raw_intents(
&session,
&[original, conflicting],
),
}),
RunBoundaryReceipt {
run_id: RunId::new(),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence: 42,
},
vec![],
Some(session.id().clone()),
)
.await
.unwrap_err();
assert!(matches!(error, RuntimeStoreError::WriteFailed(_)));
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
None
);
assert!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
let foreign = session_with_compaction_intent().1;
for (sequence, invalid) in [foreign, unbacked_intent(session.id())]
.into_iter()
.enumerate()
{
let error = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: snapshot_with_raw_intents(&session, &[invalid]),
}),
RunBoundaryReceipt {
run_id: RunId::new(),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence: 50 + sequence as u64,
},
vec![],
Some(session.id().clone()),
)
.await
.unwrap_err();
assert!(matches!(error, RuntimeStoreError::WriteFailed(_)));
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
None
);
assert!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
}
#[tokio::test]
async fn superseded_snapshot_does_not_advance_sqlite_compaction_outbox() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let (incoming, intent) = session_with_compaction_intent();
let mut current = incoming.clone();
current
.complete_compaction_projection_intent(&intent.projection)
.unwrap();
current.push(Message::User(UserMessage::text("already advanced")));
let current_snapshot = serde_json::to_vec(¤t).unwrap();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: current_snapshot.clone(),
},
)
.await
.unwrap();
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: serde_json::to_vec(&incoming).unwrap(),
}),
RunBoundaryReceipt {
run_id: RunId::new(),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence: 43,
},
vec![],
Some(incoming.id().clone()),
)
.await
.unwrap();
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(current_snapshot)
);
assert!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn existing_sqlite_outbox_rejects_changed_intent_without_advancing_snapshot() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let (session, intent) = session_with_compaction_intent();
let original_snapshot = serde_json::to_vec(&session).unwrap();
let receipt = |sequence| RunBoundaryReceipt {
run_id: RunId::new(),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 1,
sequence,
};
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: original_snapshot.clone(),
}),
receipt(70),
vec![],
Some(session.id().clone()),
)
.await
.unwrap();
let mut advanced = session.clone();
advanced.push(Message::User(UserMessage::text("later turn")));
let mut conflicting = intent.clone();
conflicting.summary_tokens += 1;
let error = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: snapshot_with_raw_intents(&advanced, &[conflicting]),
}),
receipt(71),
vec![],
Some(session.id().clone()),
)
.await
.unwrap_err();
assert!(matches!(error, RuntimeStoreError::WriteFailed(_)));
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(original_snapshot)
);
assert_eq!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap(),
vec![intent]
);
}
#[tokio::test]
async fn sqlite_non_boundary_snapshot_apis_cannot_bypass_compaction_outbox() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let (session, _intent) = session_with_compaction_intent();
let snapshot = serde_json::to_vec(&session).unwrap();
let commit = session
.transcript_history_state()
.unwrap()
.unwrap()
.commits
.last()
.unwrap()
.clone();
assert!(
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: snapshot.clone(),
},
)
.await
.is_err()
);
assert!(
store
.commit_session_transcript_rewrite_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: snapshot.clone(),
},
&commit,
)
.await
.is_err()
);
let clean = Session::with_id(session.id().clone());
let clean_snapshot = serde_json::to_vec(&clean).unwrap();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: clean_snapshot.clone(),
},
)
.await
.unwrap();
assert!(
store
.replace_session_snapshot_if_current(&runtime_id, &clean_snapshot, snapshot,)
.await
.is_err()
);
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(clean_snapshot)
);
assert!(
store
.load_pending_compaction_projections(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
fn receipt_row_count(store: &SqliteRuntimeStore) -> usize {
let conn = open_runtime_connection(store.path()).unwrap();
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM runtime_boundary_receipts",
[],
|row| row.get(0),
)
.unwrap();
usize::try_from(count).unwrap()
}
#[tokio::test]
async fn atomic_apply_roundtrip() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let session = serde_json::to_vec(&meerkat_core::Session::new()).unwrap();
let receipt = RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: Some("machine-owned-digest".to_string()),
message_count: 42,
sequence: 5,
};
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: session.clone(),
}),
receipt.clone(),
vec![input_state()],
None,
)
.await
.unwrap();
assert!(
store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.is_some()
);
assert_eq!(store.load_input_states(&runtime_id).await.unwrap().len(), 1);
}
#[tokio::test]
async fn commit_session_snapshot_does_not_write_boundary_receipt() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let session = serde_json::to_vec(&meerkat_core::Session::new()).unwrap();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: session,
},
)
.await
.unwrap();
assert!(
store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.is_some()
);
assert_eq!(receipt_row_count(&store), 0);
assert!(
store
.load_input_states(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn commit_session_snapshot_does_not_write_session_projection_row() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("sessions.sqlite3");
let store = SqliteRuntimeStore::new(path.clone()).unwrap();
let runtime_id = runtime_id();
let session = meerkat_core::Session::new();
let session_id = session.id().clone();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&session).unwrap(),
},
)
.await
.unwrap();
let session_store = SqliteSessionStore::open(path).unwrap();
assert!(
session_store.load(&session_id).await.unwrap().is_none(),
"runtime snapshot commits must not contaminate the SessionStore projection row before checkpoint continuity validation"
);
assert!(
store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn commit_session_snapshot_rejects_stale_runtime_parent() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let accepted = session_with_user("accepted runtime turn");
let mut stale = Session::with_id(accepted.id().clone());
stale.push(Message::User(UserMessage::text(
"stale runtime turn".to_string(),
)));
let accepted_snapshot = serde_json::to_vec(&accepted).unwrap();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: accepted_snapshot.clone(),
},
)
.await
.unwrap();
let err = store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&stale).unwrap(),
},
)
.await
.expect_err("stale non-continuation must not overwrite runtime snapshot");
assert!(matches!(err, RuntimeStoreError::WriteFailed(_)));
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(accepted_snapshot)
);
}
#[tokio::test]
async fn atomic_apply_keeps_current_snapshot_when_incoming_is_superseded() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let incoming = session_with_user("turn input");
let mut current = incoming.clone();
current.push(Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "peer response already applied".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
}));
let current_snapshot = serde_json::to_vec(¤t).unwrap();
let receipt = RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: Some("machine-owned-digest".to_string()),
message_count: 2,
sequence: 11,
};
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: current_snapshot.clone(),
},
)
.await
.unwrap();
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: serde_json::to_vec(&incoming).unwrap(),
}),
receipt.clone(),
vec![input_state()],
Some(incoming.id().clone()),
)
.await
.unwrap();
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(current_snapshot)
);
assert_eq!(
store
.load_boundary_receipt(&runtime_id, &receipt.run_id, receipt.sequence)
.await
.unwrap(),
None
);
assert!(
store
.load_input_states(&runtime_id)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn atomic_apply_allows_first_generated_snapshot_after_placeholder() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let mut placeholder = Session::new();
placeholder.set_system_prompt("base system".to_string());
let mut incoming = Session::with_id(placeholder.id().clone());
incoming.set_system_prompt("base system".to_string());
incoming.push(Message::User(UserMessage::text(
"verbose first turn".to_string(),
)));
let parent_revision = incoming.transcript_revision().unwrap();
incoming
.commit_transcript_rewrite(
TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
vec![Message::User(UserMessage::compaction_summary(
"[Context compacted] first turn",
))],
TranscriptRewriteReason::new("compaction"),
Some("meerkat-core".to_string()),
Some(parent_revision),
)
.unwrap();
let incoming_snapshot = serde_json::to_vec(&incoming).unwrap();
let receipt = RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: Some("machine-owned-digest".to_string()),
message_count: incoming.messages().len(),
sequence: 12,
};
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&placeholder).unwrap(),
},
)
.await
.unwrap();
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: incoming_snapshot.clone(),
}),
receipt.clone(),
vec![],
Some(incoming.id().clone()),
)
.await
.unwrap();
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(incoming_snapshot)
);
assert_eq!(
store
.load_boundary_receipt(&runtime_id, &receipt.run_id, receipt.sequence)
.await
.unwrap(),
Some(receipt)
);
}
#[tokio::test]
async fn atomic_apply_allows_generated_compaction_before_retained_tail() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let mut previous = Session::new();
previous.set_system_prompt("runtime system before context refresh".to_string());
previous.push(Message::User(UserMessage::text(
"Turn 1 request".to_string(),
)));
previous.push(Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "Turn 1 answer".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
}));
let mut incoming = Session::with_id(previous.id().clone());
incoming.set_system_prompt("runtime system after context refresh".to_string());
incoming.push(Message::User(UserMessage::text(
"Verbose context that will be compacted".to_string(),
)));
for message in previous.messages()[1..].iter().cloned() {
incoming.push(message);
}
incoming.push(Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "Turn 2 generated answer".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
}));
let parent_revision = incoming.transcript_revision().unwrap();
incoming
.commit_transcript_rewrite(
TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
vec![Message::User(UserMessage::compaction_summary(
"[Context compacted] Earlier runtime context".to_string(),
))],
TranscriptRewriteReason::new("compaction"),
Some("meerkat-core".to_string()),
Some(parent_revision),
)
.unwrap();
let incoming_snapshot = serde_json::to_vec(&incoming).unwrap();
let receipt = RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: Some("machine-owned-digest".to_string()),
message_count: incoming.messages().len(),
sequence: 13,
};
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&previous).unwrap(),
},
)
.await
.unwrap();
store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: incoming_snapshot.clone(),
}),
receipt.clone(),
vec![],
Some(incoming.id().clone()),
)
.await
.unwrap();
assert_eq!(
store.load_session_snapshot(&runtime_id).await.unwrap(),
Some(incoming_snapshot)
);
assert_eq!(
store
.load_boundary_receipt(&runtime_id, &receipt.run_id, receipt.sequence)
.await
.unwrap(),
Some(receipt)
);
}
#[tokio::test]
async fn transcript_rewrite_snapshot_rejects_stale_runtime_parent() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let original = session_with_one_turn();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&original).unwrap(),
},
)
.await
.unwrap();
let parent_revision = original.transcript_revision().unwrap();
let mut first_rewrite = original.clone();
let first_commit = first_rewrite
.commit_transcript_rewrite(
TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
vec![Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "first compact answer".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
})],
TranscriptRewriteReason::new("compaction"),
Some("sqlite-test".to_string()),
Some(parent_revision.clone()),
)
.unwrap();
store
.commit_session_transcript_rewrite_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&first_rewrite).unwrap(),
},
&first_commit,
)
.await
.unwrap();
let mut stale_rewrite = original;
let stale_commit = stale_rewrite
.commit_transcript_rewrite(
TranscriptRewriteSelection::MessageRange { start: 1, end: 2 },
vec![Message::BlockAssistant(BlockAssistantMessage {
blocks: vec![AssistantBlock::Text {
text: "stale compact answer".to_string(),
meta: None,
}],
stop_reason: StopReason::EndTurn,
identity: meerkat_core::types::TranscriptMessageIdentity::default(),
created_at: meerkat_core::types::message_timestamp_now(),
})],
TranscriptRewriteReason::new("compaction"),
Some("sqlite-test".to_string()),
Some(parent_revision),
)
.unwrap();
let err = store
.commit_session_transcript_rewrite_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&stale_rewrite).unwrap(),
},
&stale_commit,
)
.await
.expect_err("stale rewrite parent should be rejected atomically");
assert!(matches!(
err,
RuntimeStoreError::TranscriptRevisionConflict { .. }
));
let stored = store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.unwrap();
let stored: Session = serde_json::from_slice(&stored).unwrap();
assert_eq!(stored.transcript_revision().unwrap(), first_commit.revision);
}
#[tokio::test]
async fn atomic_apply_is_atomic_on_receipt_conflict() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let receipt = RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 0,
sequence: 0,
};
store
.atomic_apply(
&runtime_id,
None,
receipt.clone(),
vec![input_state()],
None,
)
.await
.unwrap();
let session = serde_json::to_vec(&meerkat_core::Session::new()).unwrap();
let err = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: session,
}),
receipt,
vec![input_state()],
None,
)
.await
.expect_err("duplicate receipt should fail");
assert!(matches!(err, RuntimeStoreError::WriteFailed(_)));
assert!(
store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.is_none()
);
let states = store.load_input_states(&runtime_id).await.unwrap();
assert_eq!(states.len(), 1);
}
#[tokio::test]
async fn atomic_apply_rejects_mismatched_session_store_key() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let session = meerkat_core::Session::new();
let wrong_session_id = meerkat_core::Session::new().id().clone();
let snapshot = serde_json::to_vec(&session).unwrap();
let err = store
.atomic_apply(
&runtime_id,
Some(SessionDelta {
session_snapshot: snapshot,
}),
RunBoundaryReceipt {
run_id: RunId(uuid::Uuid::new_v4()),
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: None,
message_count: 0,
sequence: 0,
},
vec![input_state()],
Some(wrong_session_id),
)
.await
.expect_err("mismatched session_store_key should fail");
assert!(matches!(err, RuntimeStoreError::SessionKeyMismatch { .. }));
assert!(
store
.load_session_snapshot(&runtime_id)
.await
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn commit_machine_lifecycle_persists_both_parts() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let runtime_state = RuntimeState::Stopped;
let binding = crate::store::MachineLifecycleBindingFacts::new(
Some("rt:session:sqlite".to_string()),
Some(11),
None,
Some("epoch-sqlite".to_string()),
);
store
.commit_machine_lifecycle(
&runtime_id,
MachineLifecycleCommit::new_with_binding(
runtime_state,
binding.clone(),
crate::store::SupervisorAuthoritySnapshot::UnboundNoReceipt,
),
&[input_state()],
)
.await
.unwrap();
assert!(
crate::store::load_runtime_state(&store, &runtime_id)
.await
.unwrap()
.is_some()
);
let lifecycle = crate::store::load_machine_lifecycle(&store, &runtime_id)
.await
.unwrap()
.expect("machine lifecycle snapshot");
assert_eq!(lifecycle.runtime_state(), runtime_state);
assert_eq!(lifecycle.binding(), &binding);
assert_eq!(store.load_input_states(&runtime_id).await.unwrap().len(), 1);
}
#[tokio::test]
async fn unregister_finalization_power_cuts_reopen_without_split_epoch_truth() {
for (fault, commit_was_durable) in [(1_u8, false), (2_u8, false), (3_u8, true)] {
let dir = TempDir::new().unwrap();
let path = dir.path().join(format!("unregister-fault-{fault}.sqlite3"));
let runtime_id = LogicalRuntimeId::new(format!("runtime-fault-{fault}"));
let store = SqliteRuntimeStore::new(path.clone()).unwrap();
store
.commit_machine_lifecycle(
&runtime_id,
MachineLifecycleCommit::new_with_binding(
RuntimeState::Idle,
crate::store::MachineLifecycleBindingFacts::new(
Some(format!("rt:session:fault-{fault}")),
Some(1),
Some(1),
Some(format!("epoch-{fault}")),
),
crate::store::SupervisorAuthoritySnapshot::UnboundNoReceipt,
),
&[],
)
.await
.unwrap();
let stale_ops = crate::ops_lifecycle::RuntimeOpsLifecycleRegistry::new()
.capture_persistence_snapshot(
meerkat_core::RuntimeEpochId::new(),
&meerkat_core::EpochCursorState::new(),
)
.unwrap();
store
.persist_ops_lifecycle(&runtime_id, &stale_ops)
.await
.unwrap();
let retired_ops_epoch = stale_ops.epoch_id.clone();
store.inject_unregister_finalization_fault(fault);
let result = store
.commit_unregister_finalization(
&runtime_id,
MachineLifecycleCommit::new_with_binding(
RuntimeState::Stopped,
crate::store::MachineLifecycleBindingFacts::new(None, None, None, None),
crate::store::SupervisorAuthoritySnapshot::UnboundNoReceipt,
)
.for_unregister_finalization(retired_ops_epoch.clone()),
&[],
)
.await;
if commit_was_durable {
result.expect("post-commit kill window has durable finalization");
} else {
let error = result
.expect_err("pre-commit unregister finalization interruption must surface");
assert!(error.to_string().contains("synthetic"));
}
drop(store);
let reopened = SqliteRuntimeStore::new(path.clone()).unwrap();
let recovered_state = crate::store::load_runtime_state(&reopened, &runtime_id)
.await
.unwrap();
let recovered_ops = reopened.load_ops_lifecycle(&runtime_id).await.unwrap();
if commit_was_durable {
assert_eq!(recovered_state, Some(RuntimeState::Stopped));
assert!(recovered_ops.is_none());
} else {
assert_eq!(recovered_state, Some(RuntimeState::Idle));
assert!(recovered_ops.is_some());
}
assert!(
recovered_state != Some(RuntimeState::Stopped) || recovered_ops.is_none(),
"reopen must never expose terminal lifecycle with the stale ops epoch"
);
reopened
.commit_unregister_finalization(
&runtime_id,
MachineLifecycleCommit::new_with_binding(
RuntimeState::Stopped,
crate::store::MachineLifecycleBindingFacts::new(None, None, None, None),
crate::store::SupervisorAuthoritySnapshot::UnboundNoReceipt,
)
.for_unregister_finalization(retired_ops_epoch.clone()),
&[],
)
.await
.unwrap();
drop(reopened);
let reopened_after_retry = SqliteRuntimeStore::new(path).unwrap();
assert_eq!(
crate::store::load_runtime_state(&reopened_after_retry, &runtime_id)
.await
.unwrap(),
Some(RuntimeState::Stopped)
);
assert!(
reopened_after_retry
.load_ops_lifecycle(&runtime_id)
.await
.unwrap()
.is_none()
);
let late_error = reopened_after_retry
.persist_ops_lifecycle(&runtime_id, &stale_ops)
.await
.expect_err("reopen must retain the exact retired-epoch fence");
assert!(matches!(
late_error,
RuntimeStoreError::OpsLifecycleEpochRetired { epoch_id, .. }
if epoch_id == retired_ops_epoch
));
assert!(
reopened_after_retry
.load_ops_lifecycle(&runtime_id)
.await
.unwrap()
.is_none()
);
}
}
#[tokio::test]
async fn delayed_old_epoch_finalizer_preserves_new_epoch_across_reopen() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("old-finalizer-new-epoch.sqlite3");
let runtime_id = LogicalRuntimeId::new("runtime-old-finalizer-new-epoch");
let store = SqliteRuntimeStore::new(path.clone()).unwrap();
let registry = crate::ops_lifecycle::RuntimeOpsLifecycleRegistry::new();
let old_ops = registry
.capture_persistence_snapshot(
meerkat_core::RuntimeEpochId::new(),
&meerkat_core::EpochCursorState::new(),
)
.unwrap();
let new_ops = registry
.capture_persistence_snapshot(
meerkat_core::RuntimeEpochId::new(),
&meerkat_core::EpochCursorState::new(),
)
.unwrap();
store
.persist_ops_lifecycle(&runtime_id, &old_ops)
.await
.unwrap();
store
.persist_ops_lifecycle(&runtime_id, &new_ops)
.await
.unwrap();
store.inject_unregister_finalization_fault(3);
store
.commit_unregister_finalization(
&runtime_id,
MachineLifecycleCommit::new_with_binding(
RuntimeState::Stopped,
crate::store::MachineLifecycleBindingFacts::new(None, None, None, None),
crate::store::SupervisorAuthoritySnapshot::UnboundNoReceipt,
)
.for_unregister_finalization(old_ops.epoch_id.clone()),
&[],
)
.await
.unwrap();
drop(store);
let reopened = SqliteRuntimeStore::new(path).unwrap();
assert_eq!(
reopened
.load_ops_lifecycle(&runtime_id)
.await
.unwrap()
.expect("new epoch row must survive delayed old finalization")
.epoch_id,
new_ops.epoch_id
);
assert!(matches!(
reopened
.persist_ops_lifecycle(&runtime_id, &old_ops)
.await
.expect_err("old epoch tombstone must survive reopen"),
RuntimeStoreError::OpsLifecycleEpochRetired { .. }
));
reopened
.persist_ops_lifecycle(&runtime_id, &new_ops)
.await
.unwrap();
}
#[tokio::test]
async fn legacy_runtime_state_row_is_not_lifecycle_authority() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let legacy_state_json = serde_json::to_vec(&RuntimeState::Retired).unwrap();
let conn = open_runtime_connection(&store.path).unwrap();
conn.execute(
r"
INSERT INTO runtime_states (runtime_id, runtime_state_json)
VALUES (?1, ?2)
",
params![runtime_id_text(&runtime_id), legacy_state_json],
)
.unwrap();
assert!(matches!(
crate::store::load_runtime_state(&store, &runtime_id).await,
Err(RuntimeStoreError::ReadFailed(_))
));
assert!(matches!(
crate::store::load_machine_lifecycle(&store, &runtime_id).await,
Err(RuntimeStoreError::ReadFailed(_))
));
}
#[tokio::test]
async fn legacy_machine_lifecycle_snapshot_row_is_not_lifecycle_authority() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let runtime_state = RuntimeState::Retired;
let legacy_snapshot_json = serde_json::to_vec(&serde_json::json!({
"runtime_state": runtime_state,
"binding": {
"agent_runtime_id": "rt:session:legacy",
"fence_token": 23,
"runtime_generation": 7,
"runtime_epoch_id": "epoch-legacy"
}
}))
.unwrap();
let conn = open_runtime_connection(&store.path).unwrap();
conn.execute(
r"
INSERT INTO runtime_states (runtime_id, runtime_state_json)
VALUES (?1, ?2)
",
params![runtime_id_text(&runtime_id), legacy_snapshot_json],
)
.unwrap();
assert!(matches!(
crate::store::load_runtime_state(&store, &runtime_id).await,
Err(RuntimeStoreError::ReadFailed(_))
));
assert!(matches!(
crate::store::load_machine_lifecycle(&store, &runtime_id).await,
Err(RuntimeStoreError::ReadFailed(_))
));
}
#[test]
fn receipt_sequence_encoding_is_injective_across_i64_boundary() {
let probes: [u64; 6] = [
0,
1,
i64::MAX as u64 - 1,
i64::MAX as u64,
i64::MAX as u64 + 1,
u64::MAX,
];
let mut seen = std::collections::HashSet::new();
for sequence in probes {
let encoded = encode_receipt_sequence(sequence);
assert!(
seen.insert(encoded),
"sequence {sequence} aliased an already-stored key {encoded}"
);
assert_eq!(
decode_receipt_sequence(encoded),
sequence,
"round-trip failed for sequence {sequence}"
);
}
}
fn receipt_with_sequence(run_id: RunId, sequence: u64) -> RunBoundaryReceipt {
RunBoundaryReceipt {
run_id,
boundary: RunApplyBoundary::RunStart,
contributing_input_ids: vec![],
conversation_digest: Some("machine-owned-digest".to_string()),
message_count: 1,
sequence,
}
}
#[tokio::test]
async fn boundary_receipts_straddling_i64_max_persist_and_read_distinctly() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let run_id = RunId(uuid::Uuid::new_v4());
let low = receipt_with_sequence(run_id.clone(), i64::MAX as u64);
let high = receipt_with_sequence(run_id.clone(), i64::MAX as u64 + 1);
store
.atomic_apply(&runtime_id, None, low.clone(), vec![], None)
.await
.unwrap();
store
.atomic_apply(&runtime_id, None, high.clone(), vec![], None)
.await
.unwrap();
assert_eq!(receipt_row_count(&store), 2);
assert_eq!(
store
.load_boundary_receipt(&runtime_id, &run_id, low.sequence)
.await
.unwrap(),
Some(low)
);
assert_eq!(
store
.load_boundary_receipt(&runtime_id, &run_id, high.sequence)
.await
.unwrap(),
Some(high)
);
}
#[test]
fn deserialize_persisted_session_rejects_missing_version_row() {
let v0_blob = serde_json::json!({
"id": "00000000-0000-0000-0000-000000000012",
"messages": [],
"created_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"updated_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"metadata": {}
});
let bytes = serde_json::to_vec(&v0_blob).unwrap();
let err = deserialize_persisted_session(&bytes)
.expect_err("missing-version session row must fail closed");
assert!(
err.to_string().contains("version"),
"unexpected error: {err}"
);
}
#[test]
fn deserialize_persisted_session_rejects_legacy_v1_version_row() {
let v1_blob = serde_json::json!({
"version": 1,
"id": "00000000-0000-0000-0000-000000000012",
"messages": [],
"created_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"updated_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"metadata": {}
});
let bytes = serde_json::to_vec(&v1_blob).unwrap();
let err = deserialize_persisted_session(&bytes)
.expect_err("legacy v1 session row must fail closed");
assert!(
err.to_string()
.contains("generated session persistence version authority rejected"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn runtime_store_read_path_rejects_v0_session_row() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let v0_blob = serde_json::json!({
"id": "00000000-0000-0000-0000-000000000012",
"messages": [],
"created_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"updated_at": { "secs_since_epoch": 1727784000, "nanos_since_epoch": 0 },
"metadata": {}
});
let v0_bytes = serde_json::to_vec(&v0_blob).unwrap();
{
let mut conn = open_runtime_connection(store.path()).unwrap();
let tx = begin_runtime_transaction(&mut conn).unwrap();
upsert_runtime_snapshot(&tx, &runtime_id, &v0_bytes).unwrap();
tx.commit().unwrap();
}
let mut incoming = Session::new();
incoming.push(Message::User(UserMessage::text("hello".to_string())));
let err = store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&incoming).unwrap(),
},
)
.await
.expect_err("v0 previous row must fail the read path closed");
assert!(
err.to_string().contains("version"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn runtime_store_read_path_rejects_v0_input_state_row() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let input_id = InputId::new();
let bundle = StoredInputState::new_accepted(input_id.clone());
let mut row = serde_json::to_value(&bundle).unwrap();
row.as_object_mut()
.unwrap()
.remove("stored_input_state_version");
let row_bytes = serde_json::to_vec(&row).unwrap();
{
let mut conn = open_runtime_connection(store.path()).unwrap();
let tx = begin_runtime_transaction(&mut conn).unwrap();
tx.execute(
r"
INSERT INTO runtime_input_states (runtime_id, input_id, state_json)
VALUES (?1, ?2, ?3)
",
params![
runtime_id_text(&runtime_id),
input_id.0.to_string(),
row_bytes
],
)
.unwrap();
tx.commit().unwrap();
}
let err = store
.load_input_state(&runtime_id, &input_id)
.await
.expect_err("v0 input-state row must fail the read path closed");
assert!(
err.to_string().contains("stored_input_state_version"),
"unexpected error: {err}"
);
let err = store
.load_input_states(&runtime_id)
.await
.expect_err("v0 input-state row must fail the bulk read path closed");
assert!(
err.to_string().contains("stored_input_state_version"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn projection_quarantine_marker_survives_restart_and_clears_on_write() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("sessions.sqlite3");
let runtime_id = runtime_id();
let rejected = session_with_user("rejected runtime turn");
let rejected_snapshot = serde_json::to_vec(&rejected).unwrap();
{
let store = SqliteRuntimeStore::new(path.clone()).unwrap();
assert!(
!store
.is_runtime_projection_quarantined(&runtime_id)
.await
.unwrap(),
"a fresh runtime must not start quarantined"
);
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: rejected_snapshot.clone(),
},
)
.await
.unwrap();
assert!(
store
.clear_session_snapshot_if_current(&runtime_id, &rejected_snapshot)
.await
.unwrap(),
"matching snapshot must be cleared"
);
assert!(
store
.is_runtime_projection_quarantined(&runtime_id)
.await
.unwrap(),
"clearing the rejected snapshot must record the quarantine marker"
);
}
{
let restarted = SqliteRuntimeStore::new(path.clone()).unwrap();
assert!(
restarted
.is_runtime_projection_quarantined(&runtime_id)
.await
.unwrap(),
"quarantine marker must survive a simulated process restart"
);
let revived = session_with_user("revived runtime turn");
restarted
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: serde_json::to_vec(&revived).unwrap(),
},
)
.await
.unwrap();
assert!(
!restarted
.is_runtime_projection_quarantined(&runtime_id)
.await
.unwrap(),
"a live snapshot write must clear the quarantine marker"
);
}
{
let restarted_again = SqliteRuntimeStore::new(path).unwrap();
assert!(
!restarted_again
.is_runtime_projection_quarantined(&runtime_id)
.await
.unwrap(),
"cleared quarantine marker must stay cleared across restart"
);
}
}
#[tokio::test]
async fn legacy_text_json_columns_still_read() {
let (_dir, store) = temp_store();
let runtime_id = runtime_id();
let session = session_with_one_turn();
let snapshot = serde_json::to_vec(&session).unwrap();
store
.commit_session_snapshot(
&runtime_id,
SessionDelta {
session_snapshot: snapshot.clone(),
},
)
.await
.unwrap();
{
let conn = open_runtime_connection(store.path()).unwrap();
let changed = conn
.execute(
"UPDATE runtime_session_snapshots SET session_snapshot = CAST(session_snapshot AS TEXT)",
[],
)
.unwrap();
assert!(changed > 0, "expected snapshot rows to degrade");
}
let carried = store
.load_session_snapshot(&runtime_id)
.await
.expect("load over TEXT snapshot must not fail")
.expect("snapshot present");
assert_eq!(carried, snapshot, "TEXT snapshot bytes must round-trip");
}
}
}
#[cfg(feature = "sqlite-store")]
pub use inner::SqliteRuntimeStore;