use crate::evidence::{
digest, hmac_sha256, validate_witness_capture_with_key, verify_witness_record_with_key,
WitnessCapture, WitnessError, WitnessRecord,
};
use chrono::{DateTime, SecondsFormat, Utc};
use rusqlite::{params, Connection, OptionalExtension};
use serde_json::Value;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
pub struct PersistentStore {
conn: std::sync::Arc<Mutex<Connection>>,
integrity_key: Option<std::sync::Arc<[u8]>>,
#[cfg(test)]
terminal_projection_fault: std::sync::Arc<std::sync::atomic::AtomicBool>,
#[cfg(test)]
checkpoint_persistence_fault: std::sync::Arc<std::sync::atomic::AtomicBool>,
#[cfg(test)]
graph_delete_fault: std::sync::Arc<std::sync::atomic::AtomicBool>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct CheckpointRecord {
pub checkpoint_id: String,
pub run_id: String,
pub graph_id: String,
pub graph_version: String,
pub next_node_cursor: String,
pub state: Value,
pub state_digest: String,
pub budgets: Value,
pub budget_counters: Value,
pub dependency_summary: Value,
pub dependency_digest: String,
pub terminal_cursor: u64,
pub event_cursor: u64,
pub checkpoint_digest: String,
pub created_at: String,
pub consumed_at: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CheckpointError {
NotFound,
Consumed,
Integrity,
Persistence,
IntegrityKeyRequired,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApprovalError {
NotFound,
Conflict,
AlreadyDecided,
Expired,
DecisionNotAllowed,
Integrity,
Checkpoint(CheckpointError),
Persistence,
IntegrityKeyRequired,
}
impl ApprovalError {
pub fn code(&self) -> &'static str {
match self {
Self::NotFound => "APPROVAL_NOT_FOUND",
Self::Conflict => "APPROVAL_REQUEST_CONFLICT",
Self::AlreadyDecided => "APPROVAL_ALREADY_DECIDED",
Self::Expired => "APPROVAL_EXPIRED",
Self::DecisionNotAllowed => "APPROVAL_DECISION_NOT_ALLOWED",
Self::Integrity => "APPROVAL_INTEGRITY_FAILURE",
Self::Checkpoint(error) => error.code(),
Self::Persistence => "APPROVAL_PERSISTENCE_FAILURE",
Self::IntegrityKeyRequired => "INTEGRITY_KEY_REQUIRED",
}
}
pub fn message(&self) -> String {
match self {
Self::NotFound => "approval was not found".into(),
Self::Conflict => {
"a conflicting pending approval already exists for this checkpoint and audience"
.into()
}
Self::AlreadyDecided => "approval has already been decided".into(),
Self::Expired => "approval has expired".into(),
Self::DecisionNotAllowed => "decision is not allowed by this approval".into(),
Self::Integrity => "approval integrity validation failed".into(),
Self::Checkpoint(error) => error.message().into(),
Self::Persistence => "approval persistence failed".into(),
Self::IntegrityKeyRequired => {
"integrity key is required for durable approval operations".into()
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalRecord {
pub approval_id: String,
pub checkpoint_id: String,
pub run_id: String,
pub graph_id: String,
pub graph_version: String,
pub checkpoint_digest: String,
pub audience: String,
pub prompt_digest: String,
pub allowed_decisions: Vec<String>,
pub approval_digest: String,
pub status: String,
pub decision: Option<String>,
pub decided_by: Option<String>,
pub decided_at: Option<String>,
pub expires_at: String,
pub created_at: String,
}
#[derive(Debug, Clone)]
pub struct ApprovedCheckpoint {
pub approval: ApprovalRecord,
pub checkpoint: CheckpointRecord,
}
impl CheckpointError {
pub fn code(&self) -> &'static str {
match self {
Self::NotFound => "CHECKPOINT_NOT_FOUND",
Self::Consumed => "CHECKPOINT_CONSUMED",
Self::Integrity => "CHECKPOINT_INTEGRITY_FAILURE",
Self::Persistence => "CHECKPOINT_PERSISTENCE_FAILURE",
Self::IntegrityKeyRequired => "INTEGRITY_KEY_REQUIRED",
}
}
pub fn message(&self) -> &'static str {
match self {
Self::NotFound => "checkpoint was not found",
Self::Consumed => "checkpoint has already been consumed",
Self::Integrity => "checkpoint integrity validation failed",
Self::Persistence => "checkpoint persistence failed; resumability was not advertised",
Self::IntegrityKeyRequired => {
"integrity key is required for durable checkpoint operations"
}
}
}
}
#[derive(Debug, Clone)]
pub struct ExecutionContract {
pub graph_id: String,
pub graph_version: String,
pub input: Value,
pub budgets: Value,
}
type CheckpointParts = (
String,
String,
String,
Option<String>,
Option<String>,
Option<String>,
Option<String>,
Option<String>,
Option<String>,
Option<String>,
Option<i64>,
Option<i64>,
Option<String>,
Option<String>,
Option<String>,
Option<String>,
);
fn checkpoint_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<CheckpointParts> {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
row.get(6)?,
row.get(7)?,
row.get(8)?,
row.get(9)?,
row.get(10)?,
row.get(11)?,
row.get(12)?,
row.get(13)?,
row.get(14)?,
row.get(15)?,
))
}
fn checkpoint_from_parts(parts: CheckpointParts) -> Option<CheckpointRecord> {
let (
run_id,
graph_id,
graph_version,
next_node_cursor,
state_json,
state_digest,
budgets_json,
counters_json,
dependency_json,
dependency_digest,
terminal_cursor,
event_cursor,
checkpoint_id,
checkpoint_digest,
created_at,
consumed_at,
) = parts;
Some(CheckpointRecord {
checkpoint_id: checkpoint_id?,
run_id,
graph_id,
graph_version,
next_node_cursor: next_node_cursor?,
state: serde_json::from_str(&state_json?).ok()?,
state_digest: state_digest?,
budgets: serde_json::from_str(&budgets_json?).ok()?,
budget_counters: serde_json::from_str(&counters_json?).ok()?,
dependency_summary: serde_json::from_str(&dependency_json?).ok()?,
dependency_digest: dependency_digest?,
terminal_cursor: u64::try_from(terminal_cursor?).ok()?,
event_cursor: u64::try_from(event_cursor?).ok()?,
checkpoint_digest: checkpoint_digest?,
created_at: created_at?,
consumed_at,
})
}
fn checkpoint_digest(record: &CheckpointRecord, key: &[u8]) -> String {
hmac_sha256(
&serde_json::json!({
"checkpoint_id": record.checkpoint_id,
"run_id": record.run_id,
"graph_id": record.graph_id,
"graph_version": record.graph_version,
"next_node_cursor": record.next_node_cursor,
"state": record.state,
"state_digest": record.state_digest,
"budgets": record.budgets,
"budget_counters": record.budget_counters,
"dependency_summary": record.dependency_summary,
"dependency_digest": record.dependency_digest,
"terminal_cursor": record.terminal_cursor,
"event_cursor": record.event_cursor,
"created_at": record.created_at,
}),
key,
)
}
fn validate_checkpoint_record(
record: &CheckpointRecord,
key: &[u8],
) -> Result<(), CheckpointError> {
if record.checkpoint_id != format!("checkpoint-{}-{}", record.run_id, record.next_node_cursor)
|| record.state_digest != digest(&record.state)
|| record.dependency_digest != digest(&record.dependency_summary)
|| record.checkpoint_digest != checkpoint_digest(record, key)
{
return Err(CheckpointError::Integrity);
}
Ok(())
}
type ApprovalParts = (
String,
Option<String>,
String,
Option<String>,
Option<String>,
Option<String>,
String,
String,
Option<String>,
Option<String>,
String,
Option<String>,
Option<String>,
Option<String>,
String,
String,
String,
);
fn approval_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ApprovalParts> {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
row.get(6)?,
row.get(7)?,
row.get(8)?,
row.get(9)?,
row.get(10)?,
row.get(11)?,
row.get(12)?,
row.get(13)?,
row.get(14)?,
row.get(15)?,
row.get(16)?,
))
}
fn approval_from_parts(parts: ApprovalParts) -> Option<ApprovalRecord> {
let (
approval_id,
checkpoint_id,
run_id,
graph_id,
graph_version,
checkpoint_digest,
audience,
prompt_digest,
allowed_decisions,
approval_digest,
status,
decision,
decided_by,
decided_at,
expires_at,
created_at,
_legacy,
) = parts;
let allowed_decisions = serde_json::from_str(&allowed_decisions?).ok()?;
Some(ApprovalRecord {
approval_id,
checkpoint_id: checkpoint_id?,
run_id,
graph_id: graph_id?,
graph_version: graph_version?,
checkpoint_digest: checkpoint_digest?,
audience,
prompt_digest,
allowed_decisions,
approval_digest: approval_digest?,
status,
decision,
decided_by,
decided_at,
expires_at,
created_at,
})
}
const APPROVAL_COLUMNS: &str = "approval_id, checkpoint_id, run_id, graph_id,
graph_version, checkpoint_digest, audience, prompt_digest,
allowed_decisions, approval_digest, status, decision, decided_by,
decided_at, expires_at, created_at, prompt";
fn approval_digest(record: &ApprovalRecord, key: &[u8]) -> String {
hmac_sha256(
&serde_json::json!({
"approval_id": record.approval_id,
"checkpoint_id": record.checkpoint_id,
"run_id": record.run_id,
"graph_id": record.graph_id,
"graph_version": record.graph_version,
"checkpoint_digest": record.checkpoint_digest,
"audience": record.audience,
"prompt_digest": record.prompt_digest,
"allowed_decisions": record.allowed_decisions,
"status": record.status,
"decision": record.decision,
"decided_by": record.decided_by,
"decided_at": record.decided_at,
"expires_at": record.expires_at,
"created_at": record.created_at,
}),
key,
)
}
fn parse_approval_parts(parts: ApprovalParts, key: &[u8]) -> Result<ApprovalRecord, ApprovalError> {
let record = approval_from_parts(parts).ok_or(ApprovalError::Integrity)?;
if record.approval_digest != approval_digest(&record, key) {
return Err(ApprovalError::Integrity);
}
Ok(record)
}
fn uuid_like() -> String {
digest(&Value::String(
Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
))
.trim_start_matches("sha256:")
.to_owned()
}
fn load_checkpoint_from_tx(
tx: &rusqlite::Transaction<'_>,
checkpoint_id: &str,
key: &[u8],
) -> Result<CheckpointRecord, CheckpointError> {
let row = tx.query_row(
"SELECT run_id, graph_id, graph_version, next_cursor, state_json,
state_digest, budgets_json, budget_counters_json,
dependency_json, dependency_digest, terminal_cursor,
event_cursor, checkpoint_id, checkpoint_digest, created_at,
consumed_at
FROM checkpoints WHERE checkpoint_id = ?1",
params![checkpoint_id],
checkpoint_row,
);
let parts = match row {
Ok(parts) => parts,
Err(rusqlite::Error::QueryReturnedNoRows) => return Err(CheckpointError::NotFound),
Err(_) => return Err(CheckpointError::Persistence),
};
let record = checkpoint_from_parts(parts).ok_or(CheckpointError::Integrity)?;
validate_checkpoint_record(&record, key)?;
Ok(record)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GraphDeleteResult {
Deleted,
NotFound,
Referenced,
}
impl Clone for PersistentStore {
fn clone(&self) -> Self {
Self {
conn: self.conn.clone(),
integrity_key: self.integrity_key.clone(),
#[cfg(test)]
terminal_projection_fault: self.terminal_projection_fault.clone(),
#[cfg(test)]
checkpoint_persistence_fault: self.checkpoint_persistence_fault.clone(),
#[cfg(test)]
graph_delete_fault: self.graph_delete_fault.clone(),
}
}
}
impl PersistentStore {
pub fn open(data_dir: &Path) -> Result<Self, String> {
Self::open_with_integrity_key(data_dir, None)
}
pub fn open_with_integrity_key(
data_dir: &Path,
integrity_key_path: Option<&Path>,
) -> Result<Self, String> {
crate::fs_security::validate_data_store(data_dir, integrity_key_path)
.map_err(|e| format!("filesystem security check failed: {e}"))?;
std::fs::create_dir_all(data_dir).map_err(|e| format!("failed to create data dir: {e}"))?;
let db_path = data_dir.join("agent-graph.db");
let conn =
Connection::open(&db_path).map_err(|e| format!("failed to open database: {e}"))?;
conn.execute_batch(
"PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = 5000;",
)
.map_err(|e| format!("pragma error: {e}"))?;
crate::fs_security::validate_data_store(data_dir, integrity_key_path)
.map_err(|e| format!("filesystem security check failed after WAL init: {e}"))?;
let store = Self {
conn: std::sync::Arc::new(Mutex::new(conn)),
integrity_key: Self::load_integrity_key_from(integrity_key_path),
#[cfg(test)]
terminal_projection_fault: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
false,
)),
#[cfg(test)]
checkpoint_persistence_fault: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
false,
)),
#[cfg(test)]
graph_delete_fault: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
};
store.migrate()?;
Ok(store)
}
#[allow(dead_code)]
fn load_integrity_key() -> Option<std::sync::Arc<[u8]>> {
Self::load_integrity_key_from(None)
}
fn load_integrity_key_from(explicit: Option<&Path>) -> Option<std::sync::Arc<[u8]>> {
let path = if let Some(p) = explicit {
std::ffi::OsStr::new(p).to_owned()
} else {
std::env::var_os("AGENT_GRAPH_INTEGRITY_KEY_PATH")?
};
let key = std::fs::read(path).ok()?;
(key.len() >= 32).then(|| std::sync::Arc::from(key))
}
fn require_integrity_key(&self) -> Result<&[u8], ()> {
self.integrity_key.as_deref().ok_or(())
}
pub fn has_integrity_key(&self) -> bool {
self.integrity_key.is_some()
}
fn migrate(&self) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS graphs (
name TEXT PRIMARY KEY,
spec_json TEXT NOT NULL,
spec_version TEXT NOT NULL DEFAULT '2',
topology_hash TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS graph_versions (
graph_name TEXT NOT NULL,
topology_hash TEXT NOT NULL,
spec_json TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (graph_name, topology_hash)
);
CREATE TABLE IF NOT EXISTS executions (
run_id TEXT PRIMARY KEY,
graph_name TEXT NOT NULL,
graph_hash TEXT NOT NULL,
thread_id TEXT,
status TEXT NOT NULL,
input_json TEXT,
budgets_json TEXT,
final_state_json TEXT,
started_at TEXT NOT NULL,
finished_at TEXT,
total_nodes INTEGER DEFAULT 0,
failed_attempts INTEGER DEFAULT 0,
idempotency_key TEXT UNIQUE,
FOREIGN KEY (graph_name) REFERENCES graphs(name)
);
CREATE TABLE IF NOT EXISTS checkpoints (
run_id TEXT NOT NULL,
node_id TEXT NOT NULL,
attempt INTEGER NOT NULL,
input_json TEXT,
output_json TEXT,
status TEXT NOT NULL,
error TEXT,
recorded_at TEXT NOT NULL DEFAULT (datetime('now')),
checkpoint_id TEXT,
graph_id TEXT,
graph_version TEXT,
next_cursor TEXT,
state_json TEXT,
state_digest TEXT,
budgets_json TEXT,
budget_counters_json TEXT,
dependency_json TEXT,
dependency_digest TEXT,
terminal_cursor INTEGER,
event_cursor INTEGER,
checkpoint_digest TEXT,
created_at TEXT,
consumed_at TEXT,
PRIMARY KEY (run_id, node_id, attempt),
FOREIGN KEY (run_id) REFERENCES executions(run_id)
);
CREATE TABLE IF NOT EXISTS events (
run_id TEXT NOT NULL,
seq INTEGER NOT NULL,
event_type TEXT NOT NULL,
event_json TEXT NOT NULL,
emitted_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (run_id, seq),
FOREIGN KEY (run_id) REFERENCES executions(run_id)
);
CREATE TABLE IF NOT EXISTS terminal_receipts (
run_id TEXT PRIMARY KEY,
receipt_json TEXT NOT NULL,
bundle_json TEXT NOT NULL,
receipt_digest TEXT NOT NULL,
persisted_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (run_id) REFERENCES executions(run_id)
);
CREATE TABLE IF NOT EXISTS template_candidates (
template_id TEXT PRIMARY KEY, spec_digest TEXT NOT NULL,
graph_id TEXT NOT NULL, graph_version TEXT NOT NULL,
source_ref TEXT NOT NULL, state TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS template_outcome_links (
template_id TEXT NOT NULL, run_id TEXT NOT NULL,
terminal_receipt_id TEXT NOT NULL, receipt_digest TEXT NOT NULL,
disposition TEXT NOT NULL, evidence_digest TEXT NOT NULL,
recorded_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (template_id, run_id),
FOREIGN KEY (template_id) REFERENCES template_candidates(template_id),
FOREIGN KEY (run_id) REFERENCES executions(run_id),
FOREIGN KEY (run_id) REFERENCES terminal_receipts(run_id)
);
CREATE TABLE IF NOT EXISTS template_promotion_decisions (
template_id TEXT NOT NULL, from_state TEXT NOT NULL,
to_state TEXT NOT NULL, evidence_set_digest TEXT NOT NULL,
operator_receipt_id TEXT NOT NULL, decision_digest TEXT NOT NULL,
decided_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (template_id) REFERENCES template_candidates(template_id)
);
CREATE TABLE IF NOT EXISTS approval_requests (
approval_id TEXT PRIMARY KEY,
run_id TEXT NOT NULL,
node_id TEXT NOT NULL,
checkpoint_id TEXT,
graph_id TEXT,
graph_version TEXT,
checkpoint_digest TEXT,
audience TEXT NOT NULL,
prompt TEXT NOT NULL,
prompt_digest TEXT,
allowed_decisions TEXT NOT NULL,
approval_digest TEXT,
status TEXT NOT NULL DEFAULT 'pending',
decision TEXT,
decided_by TEXT,
decided_at TEXT,
expires_at TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
FOREIGN KEY (run_id) REFERENCES executions(run_id)
);
CREATE TABLE IF NOT EXISTS idempotency_keys (
key TEXT PRIMARY KEY,
request_digest TEXT NOT NULL,
result_json TEXT NOT NULL,
valid INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS source_witnesses (
witness_id TEXT PRIMARY KEY,
locator TEXT NOT NULL,
content TEXT NOT NULL,
media_type TEXT NOT NULL,
authority_class TEXT NOT NULL,
retrieved_at TEXT NOT NULL,
digest TEXT NOT NULL UNIQUE,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);",
)
.map_err(|e| format!("migration error: {e}"))?;
let has_request_digest = conn
.prepare("PRAGMA table_info(idempotency_keys)")
.map_err(|e| format!("idempotency schema inspection error: {e}"))?
.query_map([], |row| row.get::<_, String>(1))
.map_err(|e| format!("idempotency schema inspection error: {e}"))?
.filter_map(Result::ok)
.any(|column| column == "request_digest");
if !has_request_digest {
conn.execute(
"ALTER TABLE idempotency_keys ADD COLUMN request_digest TEXT",
[],
)
.map_err(|e| format!("idempotency migration error: {e}"))?;
}
let has_idempotency_valid = conn
.prepare("PRAGMA table_info(idempotency_keys)")
.map_err(|e| format!("idempotency schema inspection error: {e}"))?
.query_map([], |row| row.get::<_, String>(1))
.map_err(|e| format!("idempotency schema inspection error: {e}"))?
.filter_map(Result::ok)
.any(|column| column == "valid");
if !has_idempotency_valid {
conn.execute(
"ALTER TABLE idempotency_keys ADD COLUMN valid INTEGER NOT NULL DEFAULT 1",
[],
)
.map_err(|e| format!("idempotency migration error: {e}"))?;
}
conn.execute(
"UPDATE idempotency_keys SET valid = 0 WHERE request_digest IS NULL",
[],
)
.map_err(|e| format!("idempotency quarantine error: {e}"))?;
for (table, column, definition) in [
("executions", "budgets_json", "TEXT"),
("checkpoints", "checkpoint_id", "TEXT"),
("checkpoints", "graph_id", "TEXT"),
("checkpoints", "graph_version", "TEXT"),
("checkpoints", "next_cursor", "TEXT"),
("checkpoints", "state_json", "TEXT"),
("checkpoints", "state_digest", "TEXT"),
("checkpoints", "budgets_json", "TEXT"),
("checkpoints", "budget_counters_json", "TEXT"),
("checkpoints", "dependency_json", "TEXT"),
("checkpoints", "dependency_digest", "TEXT"),
("checkpoints", "terminal_cursor", "INTEGER"),
("checkpoints", "event_cursor", "INTEGER"),
("checkpoints", "checkpoint_digest", "TEXT"),
("checkpoints", "created_at", "TEXT"),
("checkpoints", "consumed_at", "TEXT"),
("approval_requests", "checkpoint_id", "TEXT"),
("approval_requests", "graph_id", "TEXT"),
("approval_requests", "graph_version", "TEXT"),
("approval_requests", "checkpoint_digest", "TEXT"),
("approval_requests", "prompt_digest", "TEXT"),
("approval_requests", "approval_digest", "TEXT"),
] {
let has_column = conn
.prepare(&format!("PRAGMA table_info({table})"))
.map_err(|e| format!("{table} schema inspection error: {e}"))?
.query_map([], |row| row.get::<_, String>(1))
.map_err(|e| format!("{table} schema inspection error: {e}"))?
.filter_map(Result::ok)
.any(|name| name == column);
if !has_column {
conn.execute(
&format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
[],
)
.map_err(|e| format!("{table} migration error: {e}"))?;
}
}
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS checkpoints_checkpoint_id_idx
ON checkpoints(checkpoint_id) WHERE checkpoint_id IS NOT NULL",
[],
)
.map_err(|e| format!("checkpoint index migration error: {e}"))?;
Ok(())
}
pub fn capture_witness(&self, capture: WitnessCapture) -> Result<WitnessRecord, WitnessError> {
let key = self.require_integrity_key().map_err(|_| {
WitnessError::new(
"INTEGRITY_KEY_REQUIRED",
"an external integrity key is required for durable witness capture",
)
})?;
let expected = validate_witness_capture_with_key(capture, Some(key))
.map_err(|error| WitnessError::new(error.code, error.message))?;
let conn = self
.conn
.lock()
.map_err(|_| WitnessError::new("WITNESS_STORE_ERROR", "witness SQLite lock failed"))?;
conn.execute(
"INSERT OR IGNORE INTO source_witnesses
(witness_id, locator, content, media_type, authority_class, retrieved_at, digest)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
expected.witness_id,
expected.locator,
expected.content,
expected.media_type,
expected.authority_class,
expected.retrieved_at,
expected.digest
],
)
.map_err(|_| WitnessError::new("WITNESS_STORE_ERROR", "witness SQLite write failed"))?;
drop(conn);
self.get_witness(&expected.witness_id)?.ok_or_else(|| {
WitnessError::new(
"WITNESS_STORE_ERROR",
"witness SQLite write did not produce a row",
)
})
}
pub fn get_witness(&self, witness_id: &str) -> Result<Option<WitnessRecord>, WitnessError> {
let key = self.require_integrity_key().map_err(|_| {
WitnessError::new(
"INTEGRITY_KEY_REQUIRED",
"an external integrity key is required for durable witness reads",
)
})?;
let conn = self
.conn
.lock()
.map_err(|_| WitnessError::new("WITNESS_STORE_ERROR", "witness SQLite lock failed"))?;
let row = conn.query_row(
"SELECT witness_id, locator, content, media_type, authority_class, retrieved_at, digest
FROM source_witnesses WHERE witness_id = ?1",
params![witness_id],
|row| {
Ok(WitnessRecord {
witness_id: row.get(0)?,
locator: row.get(1)?,
content: row.get(2)?,
media_type: row.get(3)?,
authority_class: row.get(4)?,
retrieved_at: row.get(5)?,
digest: row.get(6)?,
})
},
);
match row {
Ok(record) => {
verify_witness_record_with_key(&record, Some(key))?;
Ok(Some(record))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(rusqlite::Error::FromSqlConversionFailure(..))
| Err(rusqlite::Error::InvalidColumnType(..))
| Err(rusqlite::Error::InvalidColumnName(_)) => Err(WitnessError::new(
"WITNESS_INTEGRITY_FAILURE",
"stored witness integrity validation failed",
)),
Err(_) => Err(WitnessError::new(
"WITNESS_STORE_ERROR",
"witness SQLite read failed",
)),
}
}
pub fn save_graph(
&self,
name: &str,
spec_json: &str,
topology_hash: &str,
overwrite: bool,
) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let current: Option<String> = conn
.query_row(
"SELECT topology_hash FROM graphs WHERE name = ?1",
params![name],
|row| row.get(0),
)
.ok();
if let Some(current) = current
.as_deref()
.filter(|current| *current != topology_hash)
{
let referenced: bool = conn
.query_row(
"SELECT EXISTS(SELECT 1 FROM executions WHERE graph_name = ?1 AND graph_hash = ?2)",
params![name, current],
|row| row.get(0),
)
.map_err(|e| format!("graph version reference check error: {e}"))?;
if referenced && !overwrite {
return Err(format!("graph '{name}' current version {current} is referenced by a durable execution; explicit overwrite is required"));
}
}
conn.execute(
"INSERT OR IGNORE INTO graph_versions (graph_name, topology_hash, spec_json) VALUES (?1, ?2, ?3)",
params![name, topology_hash, spec_json],
)
.map_err(|e| format!("save graph version error: {e}"))?;
conn.execute(
"INSERT INTO graphs (name, spec_json, topology_hash)
VALUES (?1, ?2, ?3)
ON CONFLICT(name) DO UPDATE SET
spec_json = excluded.spec_json,
topology_hash = excluded.topology_hash,
updated_at = datetime('now')",
params![name, spec_json, topology_hash],
)
.map_err(|e| format!("save_graph error: {e}"))?;
Ok(())
}
pub fn load_graph(&self, name: &str) -> Result<Option<(String, String)>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare("SELECT spec_json, topology_hash FROM graphs WHERE name = ?1")
.map_err(|e| format!("load_graph error: {e}"))?;
let result = stmt
.query_row(params![name], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.ok();
Ok(result)
}
pub fn list_graphs(&self) -> Result<Vec<(String, String, String)>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare("SELECT name, topology_hash, created_at FROM graphs ORDER BY name")
.map_err(|e| format!("list_graphs error: {e}"))?;
let rows = stmt
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})
.map_err(|e| format!("list_graphs error: {e}"))?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
}
pub fn delete_graph(&self, name: &str) -> Result<GraphDeleteResult, String> {
let mut conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let tx = conn
.transaction()
.map_err(|e| format!("delete_graph transaction begin error: {e}"))?;
let referenced: bool = tx
.query_row(
"SELECT EXISTS(SELECT 1 FROM executions WHERE graph_name = ?1)",
params![name],
|row| row.get(0),
)
.map_err(|e| format!("delete_graph reference check error: {e}"))?;
if referenced {
return Ok(GraphDeleteResult::Referenced);
}
let affected = tx
.execute("DELETE FROM graphs WHERE name = ?1", params![name])
.map_err(|e| format!("delete_graph error: {e}"))?;
if affected == 0 {
return Ok(GraphDeleteResult::NotFound);
}
#[cfg(test)]
if self
.graph_delete_fault
.swap(false, std::sync::atomic::Ordering::SeqCst)
{
return Err("injected graph deletion failure after graph row removal".into());
}
tx.execute(
"DELETE FROM graph_versions WHERE graph_name = ?1",
params![name],
)
.map_err(|e| format!("delete_graph versions error: {e}"))?;
tx.commit()
.map_err(|e| format!("delete_graph transaction commit error: {e}"))?;
Ok(GraphDeleteResult::Deleted)
}
#[cfg(test)]
pub(crate) fn fail_graph_delete_after_graph_row(&self) {
self.graph_delete_fault
.store(true, std::sync::atomic::Ordering::SeqCst);
}
#[cfg(test)]
pub(crate) fn fail_terminal_projection_after_events(&self) {
self.terminal_projection_fault
.store(true, std::sync::atomic::Ordering::SeqCst);
}
pub fn load_graph_version(
&self,
name: &str,
topology_hash: &str,
) -> Result<Option<String>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let result = conn
.query_row(
"SELECT spec_json FROM graph_versions WHERE graph_name = ?1 AND topology_hash = ?2",
params![name, topology_hash],
|row| row.get::<_, String>(0),
)
.ok();
Ok(result)
}
pub fn list_graph_versions(&self, name: &str) -> Result<Vec<String>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare("SELECT topology_hash FROM graph_versions WHERE graph_name = ?1 ORDER BY created_at, topology_hash")
.map_err(|e| format!("list graph versions error: {e}"))?;
let versions = stmt
.query_map(params![name], |row| row.get::<_, String>(0))
.map_err(|e| format!("list graph versions error: {e}"))?
.filter_map(Result::ok)
.collect();
Ok(versions)
}
pub fn save_execution(
&self,
run_id: &str,
graph_name: &str,
graph_hash: &str,
status: &str,
input_json: &str,
) -> Result<(), String> {
self.save_execution_with_budgets(run_id, graph_name, graph_hash, status, input_json, None)
}
pub fn save_execution_with_budgets(
&self,
run_id: &str,
graph_name: &str,
graph_hash: &str,
status: &str,
input_json: &str,
budgets_json: Option<&str>,
) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
conn.execute(
"INSERT INTO executions (run_id, graph_name, graph_hash, status, input_json, budgets_json, started_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, datetime('now'))
ON CONFLICT(run_id) DO UPDATE SET
status = excluded.status,
budgets_json = COALESCE(excluded.budgets_json, executions.budgets_json),
finished_at = CASE WHEN excluded.status IN ('completed','failed','cancelled') THEN datetime('now') ELSE finished_at END",
params![run_id, graph_name, graph_hash, status, input_json, budgets_json],
)
.map_err(|e| format!("save_execution error: {e}"))?;
Ok(())
}
pub fn update_execution_status(
&self,
run_id: &str,
status: &str,
final_state_json: Option<&str>,
total_nodes: Option<usize>,
failed_attempts: Option<usize>,
) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let changed = conn.execute(
"UPDATE executions SET
status = ?2,
final_state_json = COALESCE(?3, final_state_json),
total_nodes = COALESCE(?4, total_nodes),
failed_attempts = COALESCE(?5, failed_attempts),
finished_at = CASE WHEN ?2 IN ('completed','failed','cancelled') THEN datetime('now') ELSE finished_at END
WHERE run_id = ?1",
params![
run_id,
status,
final_state_json,
total_nodes.map(|v| v as i64),
failed_attempts.map(|v| v as i64)
],
).map_err(|e| format!("update_execution error: {e}"))?;
if changed != 1 {
return Err(format!(
"update_execution error: run '{run_id}' was not found"
));
}
Ok(())
}
pub fn persist_terminal_projection(
&self,
run_id: &str,
status: &str,
final_state_json: &str,
total_nodes: usize,
events: &[(u64, String, String)],
receipt_json: &str,
bundle_json: &str,
) -> Result<String, String> {
let key = self
.require_integrity_key()
.map_err(|_| "INTEGRITY_KEY_REQUIRED".to_owned())?;
let receipt: Value = serde_json::from_str(receipt_json)
.map_err(|e| format!("terminal receipt JSON error: {e}"))?;
let receipt_digest = hmac_sha256(&receipt, key);
let mut conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let tx = conn
.transaction()
.map_err(|e| format!("terminal transaction begin error: {e}"))?;
let changed = tx.execute(
"UPDATE executions SET status = ?2, final_state_json = ?3, total_nodes = ?4,
finished_at = CASE WHEN ?2 IN ('completed','failed','cancelled') THEN datetime('now') ELSE finished_at END
WHERE run_id = ?1",
params![run_id, status, final_state_json, total_nodes as i64],
).map_err(|e| format!("terminal execution update error: {e}"))?;
if changed != 1 {
return Err(format!("terminal projection run '{run_id}' was not found"));
}
tx.execute("DELETE FROM events WHERE run_id = ?1", params![run_id])
.map_err(|e| format!("terminal event reset error: {e}"))?;
for (seq, event_type, event_json) in events {
tx.execute(
"INSERT INTO events (run_id, seq, event_type, event_json) VALUES (?1, ?2, ?3, ?4)",
params![run_id, *seq, event_type, event_json],
)
.map_err(|e| format!("terminal event insert error: {e}"))?;
}
#[cfg(test)]
if self
.terminal_projection_fault
.swap(false, std::sync::atomic::Ordering::SeqCst)
{
return Err("injected terminal projection failure after events".into());
}
tx.execute(
"INSERT INTO terminal_receipts (run_id, receipt_json, bundle_json, receipt_digest) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(run_id) DO UPDATE SET receipt_json=excluded.receipt_json, bundle_json=excluded.bundle_json, receipt_digest=excluded.receipt_digest, persisted_at=datetime('now')",
params![run_id, receipt_json, bundle_json, receipt_digest],
).map_err(|e| format!("terminal receipt insert error: {e}"))?;
tx.commit()
.map_err(|e| format!("terminal transaction commit error: {e}"))?;
Ok(receipt_digest)
}
pub fn load_terminal_receipt(&self, run_id: &str) -> Result<Option<Value>, String> {
let key = self
.require_integrity_key()
.map_err(|_| "INTEGRITY_KEY_REQUIRED".to_owned())?;
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let row: Result<(String, String), _> = conn.query_row(
"SELECT receipt_json, receipt_digest FROM terminal_receipts WHERE run_id = ?1",
params![run_id],
|row| Ok((row.get(0)?, row.get(1)?)),
);
match row {
Ok((receipt_json, receipt_digest)) => {
let receipt: Value = serde_json::from_str(&receipt_json)
.map_err(|e| format!("stored receipt JSON error: {e}"))?;
if hmac_sha256(&receipt, key) != receipt_digest {
return Err("RECEIPT_INTEGRITY_FAILURE".into());
}
Ok(Some(
serde_json::json!({"receipt":receipt,"receipt_digest":receipt_digest,"storage_class":"sqlite_terminal_receipt","replay_capability":"integrity_only"}),
))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(format!("load terminal receipt error: {e}")),
}
}
pub fn recover_incomplete_executions(&self) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
conn.execute(
"UPDATE executions
SET status = 'interrupted_non_resumable', finished_at = datetime('now')
WHERE status IN ('accepted', 'running')",
[],
)
.map_err(|e| format!("recover executions error: {e}"))?;
Ok(())
}
pub fn load_execution(&self, run_id: &str) -> Result<Option<Value>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare(
"SELECT graph_name, graph_hash, status, final_state_json, started_at, finished_at
FROM executions WHERE run_id = ?1",
)
.map_err(|e| format!("load execution error: {e}"))?;
let row = stmt.query_row(params![run_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, Option<String>>(5)?,
))
});
match row {
Ok((graph_id, graph_version, status, final_state_json, started_at, finished_at)) => {
let final_state = final_state_json
.as_deref()
.map(serde_json::from_str)
.transpose()
.map_err(|e| format!("stored final state JSON error: {e}"))?
.unwrap_or(Value::Null);
Ok(Some(serde_json::json!({
"run_id": run_id,
"graph_id": graph_id,
"graph_version": graph_version,
"status": status,
"success": status == "completed",
"final_state": final_state,
"started_at": started_at,
"finished_at": finished_at,
"storage_class": "sqlite_terminal_record",
"durable_resume": false,
"replay_capability": "integrity_only"
})))
}
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(format!("load execution error: {e}")),
}
}
pub fn load_execution_contract(
&self,
run_id: &str,
) -> Result<Option<ExecutionContract>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let result = conn.query_row(
"SELECT graph_name, graph_hash, input_json, budgets_json
FROM executions WHERE run_id = ?1",
params![run_id],
|row| {
let input_json: Option<String> = row.get(2)?;
let budgets_json: Option<String> = row.get(3)?;
Ok(ExecutionContract {
graph_id: row.get(0)?,
graph_version: row.get(1)?,
input: input_json
.as_deref()
.map(serde_json::from_str)
.transpose()
.map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
2,
rusqlite::types::Type::Text,
Box::new(error),
)
})?
.unwrap_or(Value::Null),
budgets: budgets_json
.as_deref()
.map(serde_json::from_str)
.transpose()
.map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
3,
rusqlite::types::Type::Text,
Box::new(error),
)
})?
.unwrap_or(Value::Null),
})
},
);
match result {
Ok(contract) => Ok(Some(contract)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(error) => Err(format!("load execution contract error: {error}")),
}
}
pub fn create_resume_checkpoint(
&self,
run_id: &str,
graph_id: &str,
graph_version: &str,
next_node_cursor: &str,
state: &Value,
budgets: &Value,
budget_counters: &Value,
dependency_summary: &Value,
terminal_cursor: u64,
event_cursor: u64,
) -> Result<CheckpointRecord, CheckpointError> {
let key = self
.require_integrity_key()
.map_err(|_| CheckpointError::IntegrityKeyRequired)?;
let state_json = serde_json::to_string(state).map_err(|_| CheckpointError::Persistence)?;
let budgets_json =
serde_json::to_string(budgets).map_err(|_| CheckpointError::Persistence)?;
let counters_json =
serde_json::to_string(budget_counters).map_err(|_| CheckpointError::Persistence)?;
let dependency_json =
serde_json::to_string(dependency_summary).map_err(|_| CheckpointError::Persistence)?;
let state_digest = digest(state);
let dependency_digest = digest(dependency_summary);
let created_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
let checkpoint_id = format!("checkpoint-{run_id}-{next_node_cursor}");
let mut record = CheckpointRecord {
checkpoint_id,
run_id: run_id.to_owned(),
graph_id: graph_id.to_owned(),
graph_version: graph_version.to_owned(),
next_node_cursor: next_node_cursor.to_owned(),
state: state.clone(),
state_digest,
budgets: budgets.clone(),
budget_counters: budget_counters.clone(),
dependency_summary: dependency_summary.clone(),
dependency_digest,
terminal_cursor,
event_cursor,
checkpoint_digest: String::new(),
created_at,
consumed_at: None,
};
record.checkpoint_digest = checkpoint_digest(&record, key);
let mut conn = self.conn.lock().map_err(|_| CheckpointError::Persistence)?;
let tx = conn
.transaction()
.map_err(|_| CheckpointError::Persistence)?;
#[cfg(test)]
if self
.checkpoint_persistence_fault
.swap(false, std::sync::atomic::Ordering::SeqCst)
{
return Err(CheckpointError::Persistence);
}
tx.execute(
"INSERT INTO checkpoints
(run_id, node_id, attempt, input_json, status, checkpoint_id,
graph_id, graph_version, next_cursor, state_json, state_digest,
budgets_json, budget_counters_json, dependency_json,
dependency_digest, terminal_cursor, event_cursor, checkpoint_digest,
created_at, consumed_at)
VALUES (?1, ?2, 0, ?3, 'available', ?4, ?5, ?6, ?7, ?3, ?8,
?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, NULL)",
params![
record.run_id,
record.next_node_cursor,
state_json,
record.checkpoint_id,
record.graph_id,
record.graph_version,
record.next_node_cursor,
record.state_digest,
budgets_json,
counters_json,
dependency_json,
record.dependency_digest,
record.terminal_cursor as i64,
record.event_cursor as i64,
record.checkpoint_digest,
record.created_at,
],
)
.map_err(|_| CheckpointError::Persistence)?;
tx.commit().map_err(|_| CheckpointError::Persistence)?;
Ok(record)
}
pub fn load_resume_checkpoint(
&self,
checkpoint_id: Option<&str>,
run_id: Option<&str>,
) -> Result<Option<CheckpointRecord>, CheckpointError> {
let key = self
.require_integrity_key()
.map_err(|_| CheckpointError::IntegrityKeyRequired)?;
let conn = self.conn.lock().map_err(|_| CheckpointError::Persistence)?;
let query = if checkpoint_id.is_some() {
"SELECT run_id, graph_id, graph_version, next_cursor, state_json,
state_digest, budgets_json, budget_counters_json,
dependency_json, dependency_digest, terminal_cursor,
event_cursor, checkpoint_id, checkpoint_digest, created_at,
consumed_at
FROM checkpoints WHERE checkpoint_id = ?1"
} else {
"SELECT run_id, graph_id, graph_version, next_cursor, state_json,
state_digest, budgets_json, budget_counters_json,
dependency_json, dependency_digest, terminal_cursor,
event_cursor, checkpoint_id, checkpoint_digest, created_at,
consumed_at
FROM checkpoints WHERE run_id = ?1 AND checkpoint_id IS NOT NULL
ORDER BY created_at DESC, checkpoint_id DESC LIMIT 1"
};
let selector = checkpoint_id.or(run_id).unwrap_or("");
let row = conn.query_row(query, params![selector], checkpoint_row);
let parts = match row {
Ok(parts) => parts,
Err(rusqlite::Error::QueryReturnedNoRows) => return Ok(None),
Err(_) => return Err(CheckpointError::Persistence),
};
let record = checkpoint_from_parts(parts).ok_or(CheckpointError::Integrity)?;
validate_checkpoint_record(&record, key)?;
Ok(Some(record))
}
pub fn consume_resume_checkpoint(
&self,
checkpoint_id: &str,
) -> Result<CheckpointRecord, CheckpointError> {
let key = self
.require_integrity_key()
.map_err(|_| CheckpointError::IntegrityKeyRequired)?;
let mut conn = self.conn.lock().map_err(|_| CheckpointError::Persistence)?;
let tx = conn
.transaction()
.map_err(|_| CheckpointError::Persistence)?;
let row = tx.query_row(
"SELECT run_id, graph_id, graph_version, next_cursor, state_json,
state_digest, budgets_json, budget_counters_json,
dependency_json, dependency_digest, terminal_cursor,
event_cursor, checkpoint_id, checkpoint_digest, created_at,
consumed_at
FROM checkpoints WHERE checkpoint_id = ?1",
params![checkpoint_id],
checkpoint_row,
);
let parts = match row {
Ok(parts) => parts,
Err(rusqlite::Error::QueryReturnedNoRows) => return Err(CheckpointError::NotFound),
Err(_) => return Err(CheckpointError::Persistence),
};
let record = checkpoint_from_parts(parts).ok_or(CheckpointError::Integrity)?;
validate_checkpoint_record(&record, key)?;
if record.consumed_at.is_some() {
return Err(CheckpointError::Consumed);
}
let consumed_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
let changed = tx
.execute(
"UPDATE checkpoints SET consumed_at = ?2, status = 'consumed'
WHERE checkpoint_id = ?1 AND consumed_at IS NULL",
params![checkpoint_id, consumed_at],
)
.map_err(|_| CheckpointError::Persistence)?;
if changed != 1 {
return Err(CheckpointError::Consumed);
}
tx.commit().map_err(|_| CheckpointError::Persistence)?;
let mut consumed = record;
consumed.consumed_at = Some(consumed_at);
Ok(consumed)
}
pub fn create_checkpoint_approval(
&self,
checkpoint_id: &str,
graph_id: &str,
graph_version: &str,
next_node_cursor: &str,
expected_state: &Value,
expected_budgets: &Value,
expected_budget_counters: &Value,
dependency_summary: &Value,
audience: &str,
prompt_digest: &str,
allowed_decisions: &[String],
expires_at: &str,
) -> Result<ApprovalRecord, ApprovalError> {
let key = self
.require_integrity_key()
.map_err(|_| ApprovalError::IntegrityKeyRequired)?;
let mut conn = self.conn.lock().map_err(|_| ApprovalError::Persistence)?;
let tx = conn.transaction().map_err(|_| ApprovalError::Persistence)?;
let checkpoint =
load_checkpoint_from_tx(&tx, checkpoint_id, key).map_err(ApprovalError::Checkpoint)?;
if checkpoint.consumed_at.is_some() {
return Err(ApprovalError::Checkpoint(CheckpointError::Consumed));
}
if checkpoint.graph_id != graph_id
|| checkpoint.graph_version != graph_version
|| checkpoint.next_node_cursor != next_node_cursor
|| checkpoint.state != *expected_state
|| checkpoint.budgets != *expected_budgets
|| checkpoint.budget_counters != *expected_budget_counters
|| checkpoint.dependency_summary != *dependency_summary
|| checkpoint.terminal_cursor != 0
|| checkpoint.event_cursor != 0
{
return Err(ApprovalError::Checkpoint(CheckpointError::Integrity));
}
let allowed_json =
serde_json::to_string(allowed_decisions).map_err(|_| ApprovalError::Persistence)?;
let existing = tx
.query_row(
&format!(
"SELECT {APPROVAL_COLUMNS} FROM approval_requests
WHERE checkpoint_id = ?1 AND audience = ?2 AND status = 'pending'"
),
params![checkpoint_id, audience],
approval_row,
)
.optional()
.map_err(|_| ApprovalError::Persistence)?;
if let Some(parts) = existing {
let current = parse_approval_parts(parts, key)?;
if current.graph_id == graph_id
&& current.graph_version == graph_version
&& current.checkpoint_digest == checkpoint.checkpoint_digest
&& current.prompt_digest == prompt_digest
&& current.allowed_decisions.as_slice() == allowed_decisions
&& current.expires_at == expires_at
{
tx.commit().map_err(|_| ApprovalError::Persistence)?;
return Ok(current);
}
return Err(ApprovalError::Conflict);
}
let created_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
let mut record = ApprovalRecord {
approval_id: format!("approval-{}", uuid_like()),
checkpoint_id: checkpoint.checkpoint_id.clone(),
run_id: checkpoint.run_id.clone(),
graph_id: graph_id.to_owned(),
graph_version: graph_version.to_owned(),
checkpoint_digest: checkpoint.checkpoint_digest.clone(),
audience: audience.to_owned(),
prompt_digest: prompt_digest.to_owned(),
allowed_decisions: allowed_decisions.to_vec(),
approval_digest: String::new(),
status: "pending".into(),
decision: None,
decided_by: None,
decided_at: None,
expires_at: expires_at.to_owned(),
created_at,
};
record.approval_digest = approval_digest(&record, key);
tx.execute(
"INSERT INTO approval_requests
(approval_id, run_id, node_id, checkpoint_id, graph_id, graph_version,
checkpoint_digest, audience, prompt, prompt_digest, allowed_decisions,
approval_digest, status, expires_at, created_at)
VALUES (?1, ?2, 'durable_checkpoint', ?3, ?4, ?5, ?6, ?7, '', ?8, ?9, ?10,
'pending', ?11, ?12)",
params![
record.approval_id,
record.run_id,
record.checkpoint_id,
record.graph_id,
record.graph_version,
record.checkpoint_digest,
record.audience,
record.prompt_digest,
allowed_json,
record.approval_digest,
record.expires_at,
record.created_at,
],
)
.map_err(|_| ApprovalError::Persistence)?;
tx.commit().map_err(|_| ApprovalError::Persistence)?;
Ok(record)
}
pub fn get_checkpoint_approval(
&self,
approval_id: &str,
) -> Result<Option<ApprovalRecord>, ApprovalError> {
let key = self
.require_integrity_key()
.map_err(|_| ApprovalError::IntegrityKeyRequired)?;
let conn = self.conn.lock().map_err(|_| ApprovalError::Persistence)?;
let result = conn.query_row(
&format!(
"SELECT {APPROVAL_COLUMNS} FROM approval_requests
WHERE approval_id = ?1 AND checkpoint_id IS NOT NULL"
),
params![approval_id],
approval_row,
);
match result {
Ok(parts) => Ok(Some(parse_approval_parts(parts, key)?)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(_) => Err(ApprovalError::Persistence),
}
}
pub fn list_checkpoint_approvals(
&self,
run_id: Option<&str>,
status: Option<&str>,
limit: usize,
) -> Result<Vec<ApprovalRecord>, ApprovalError> {
let key = self
.require_integrity_key()
.map_err(|_| ApprovalError::IntegrityKeyRequired)?;
let conn = self.conn.lock().map_err(|_| ApprovalError::Persistence)?;
let mut sql = format!(
"SELECT {APPROVAL_COLUMNS} FROM approval_requests
WHERE checkpoint_id IS NOT NULL"
);
if run_id.is_some() {
sql.push_str(" AND run_id = ?1");
}
if status.is_some() {
sql.push_str(if run_id.is_some() {
" AND status = ?2"
} else {
" AND status = ?1"
});
}
sql.push_str(" ORDER BY created_at DESC, approval_id DESC LIMIT ?3");
if run_id.is_none() && status.is_none() {
sql = format!("SELECT {APPROVAL_COLUMNS} FROM approval_requests
WHERE checkpoint_id IS NOT NULL ORDER BY created_at DESC, approval_id DESC LIMIT ?1");
} else if run_id.is_none() || status.is_none() {
sql = sql.replace("LIMIT ?3", "LIMIT ?2");
}
let mut statement = conn.prepare(&sql).map_err(|_| ApprovalError::Persistence)?;
let mut rows = if run_id.is_some() && status.is_some() {
statement.query(params![run_id, status, limit.min(200) as i64])
} else if run_id.is_some() {
statement.query(params![run_id, limit.min(200) as i64])
} else if status.is_some() {
statement.query(params![status, limit.min(200) as i64])
} else {
statement.query(params![limit.min(200) as i64])
}
.map_err(|_| ApprovalError::Persistence)?;
let mut approvals = Vec::new();
while let Some(row) = rows.next().map_err(|_| ApprovalError::Persistence)? {
approvals.push(parse_approval_parts(
approval_row(row).map_err(|_| ApprovalError::Persistence)?,
key,
)?);
}
Ok(approvals)
}
pub fn checkpoint_approval_status(
&self,
checkpoint_id: &str,
) -> Result<Option<String>, ApprovalError> {
let conn = self.conn.lock().map_err(|_| ApprovalError::Persistence)?;
conn.query_row(
"SELECT status FROM approval_requests
WHERE checkpoint_id = ?1 AND checkpoint_id IS NOT NULL
ORDER BY created_at DESC LIMIT 1",
params![checkpoint_id],
|row| row.get(0),
)
.optional()
.map_err(|_| ApprovalError::Persistence)
}
pub fn decide_checkpoint_approval(
&self,
approval_id: &str,
decision: &str,
actor: &str,
now: DateTime<Utc>,
) -> Result<ApprovedCheckpoint, ApprovalError> {
let key = self
.require_integrity_key()
.map_err(|_| ApprovalError::IntegrityKeyRequired)?;
let mut conn = self.conn.lock().map_err(|_| ApprovalError::Persistence)?;
let tx = conn.transaction().map_err(|_| ApprovalError::Persistence)?;
let parts = tx
.query_row(
&format!(
"SELECT {APPROVAL_COLUMNS} FROM approval_requests
WHERE approval_id = ?1 AND checkpoint_id IS NOT NULL"
),
params![approval_id],
approval_row,
)
.optional()
.map_err(|_| ApprovalError::Persistence)?
.ok_or(ApprovalError::NotFound)?;
let mut approval = parse_approval_parts(parts, key)?;
if approval.status == "expired" {
return Err(ApprovalError::Expired);
}
if approval.status != "pending" {
return Err(ApprovalError::AlreadyDecided);
}
let expired = DateTime::parse_from_rfc3339(&approval.expires_at)
.map_err(|_| ApprovalError::Integrity)?
.with_timezone(&Utc)
<= now;
if expired {
let checkpoint = load_checkpoint_from_tx(&tx, &approval.checkpoint_id, key)
.map_err(ApprovalError::Checkpoint)?;
if checkpoint.run_id != approval.run_id
|| checkpoint.graph_id != approval.graph_id
|| checkpoint.graph_version != approval.graph_version
|| checkpoint.checkpoint_digest != approval.checkpoint_digest
{
return Err(ApprovalError::Integrity);
}
approval.status = "expired".into();
approval.decision = None;
approval.decided_by = None;
approval.decided_at = Some(now.to_rfc3339_opts(SecondsFormat::Nanos, true));
approval.approval_digest = approval_digest(&approval, key);
let changed = tx
.execute(
"UPDATE approval_requests SET status = 'expired', decision = NULL,
decided_by = NULL, decided_at = ?2, approval_digest = ?3
WHERE approval_id = ?1 AND status = 'pending'",
params![approval_id, approval.decided_at, approval.approval_digest],
)
.map_err(|_| ApprovalError::Persistence)?;
if changed != 1 {
return Err(ApprovalError::AlreadyDecided);
}
if checkpoint.consumed_at.is_none() {
tx.execute(
"UPDATE checkpoints SET consumed_at = ?2, status = 'consumed'
WHERE checkpoint_id = ?1 AND consumed_at IS NULL",
params![
approval.checkpoint_id,
now.to_rfc3339_opts(SecondsFormat::Nanos, true)
],
)
.map_err(|_| ApprovalError::Persistence)?;
}
tx.commit().map_err(|_| ApprovalError::Persistence)?;
return Err(ApprovalError::Expired);
}
if !approval
.allowed_decisions
.iter()
.any(|allowed| allowed == decision)
{
return Err(ApprovalError::DecisionNotAllowed);
}
let checkpoint = load_checkpoint_from_tx(&tx, &approval.checkpoint_id, key)
.map_err(ApprovalError::Checkpoint)?;
if checkpoint.run_id != approval.run_id
|| checkpoint.graph_id != approval.graph_id
|| checkpoint.graph_version != approval.graph_version
|| checkpoint.checkpoint_digest != approval.checkpoint_digest
{
return Err(ApprovalError::Integrity);
}
if checkpoint.consumed_at.is_some() {
return Err(ApprovalError::Checkpoint(CheckpointError::Consumed));
}
let decided_at = now.to_rfc3339_opts(SecondsFormat::Nanos, true);
let status = if decision == "approve" {
"approved"
} else {
"rejected"
};
approval.status = status.into();
approval.decision = Some(decision.into());
approval.decided_by = Some(actor.into());
approval.decided_at = Some(decided_at.clone());
approval.approval_digest = approval_digest(&approval, key);
let changed = tx
.execute(
"UPDATE approval_requests SET status = ?2, decision = ?3,
decided_by = ?4, decided_at = ?5, approval_digest = ?6
WHERE approval_id = ?1 AND status = 'pending'",
params![
approval_id,
status,
decision,
actor,
decided_at,
approval.approval_digest
],
)
.map_err(|_| ApprovalError::Persistence)?;
if changed != 1 {
return Err(ApprovalError::AlreadyDecided);
}
let consumed_at = now.to_rfc3339_opts(SecondsFormat::Nanos, true);
let checkpoint_changed = tx
.execute(
"UPDATE checkpoints SET consumed_at = ?2, status = 'consumed'
WHERE checkpoint_id = ?1 AND consumed_at IS NULL",
params![approval.checkpoint_id, consumed_at],
)
.map_err(|_| ApprovalError::Persistence)?;
if checkpoint_changed != 1 {
return Err(ApprovalError::Checkpoint(CheckpointError::Consumed));
}
tx.commit().map_err(|_| ApprovalError::Persistence)?;
Ok(ApprovedCheckpoint {
approval,
checkpoint,
})
}
pub fn approval_receipt_value(approval: &ApprovalRecord) -> Value {
let decision = approval.decision.clone().unwrap_or_default();
let decided_by_digest = approval
.decided_by
.as_ref()
.map(|actor| digest(&Value::String(actor.clone())));
let decision_digest = digest(&serde_json::json!({
"approval_digest": approval.approval_digest,
"decision": decision,
"decided_by_digest": decided_by_digest,
"decided_at": approval.decided_at,
}));
serde_json::json!({
"approval_id": approval.approval_id,
"approval_digest": approval.approval_digest,
"checkpoint_id": approval.checkpoint_id,
"checkpoint_digest": approval.checkpoint_digest,
"decision": approval.decision,
"decided_by_digest": decided_by_digest,
"decided_at": approval.decided_at,
"allowed_decisions_digest": digest(&serde_json::json!(approval.allowed_decisions)),
"audience_digest": digest(&Value::String(approval.audience.clone())),
"prompt_digest": approval.prompt_digest,
"decision_digest": decision_digest,
})
}
#[cfg(test)]
pub(crate) fn fail_checkpoint_persistence(&self) {
self.checkpoint_persistence_fault
.store(true, std::sync::atomic::Ordering::SeqCst);
}
pub fn save_checkpoint(
&self,
run_id: &str,
node_id: &str,
attempt: u32,
input_json: &str,
output_json: Option<&str>,
status: &str,
error: Option<&str>,
) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
conn.execute(
"INSERT INTO checkpoints (run_id, node_id, attempt, input_json, output_json, status, error)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
ON CONFLICT(run_id, node_id, attempt) DO UPDATE SET
output_json = COALESCE(excluded.output_json, output_json),
status = excluded.status,
error = COALESCE(excluded.error, error)",
params![run_id, node_id, attempt, input_json, output_json, status, error],
)
.map_err(|e| format!("save_checkpoint error: {e}"))?;
Ok(())
}
pub fn save_event(
&self,
run_id: &str,
seq: u64,
event_type: &str,
event_json: &str,
) -> Result<(), String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
conn.execute(
"INSERT OR IGNORE INTO events (run_id, seq, event_type, event_json)
VALUES (?1, ?2, ?3, ?4)",
params![run_id, seq, event_type, event_json],
)
.map_err(|e| format!("save_event error: {e}"))?;
Ok(())
}
pub fn load_events(
&self,
run_id: &str,
cursor: u64,
limit: usize,
) -> Result<Option<Value>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare("SELECT seq, event_json FROM events WHERE run_id = ?1 AND seq >= ?2 ORDER BY seq LIMIT ?3")
.map_err(|e| format!("load events error: {e}"))?;
let events: Vec<Value> = stmt
.query_map(params![run_id, cursor, limit.min(200) as i64], |row| {
let seq: u64 = row.get(0)?;
let event_json: String = row.get(1)?;
let event = serde_json::from_str::<Value>(&event_json)
.unwrap_or_else(|_| serde_json::json!({"receipt":"terminal event persisted with reduced fidelity"}));
Ok(serde_json::json!({"cursor": seq, "event": event}))
})
.map_err(|e| format!("load events error: {e}"))?
.filter_map(Result::ok)
.collect();
let exists: bool = conn
.query_row(
"SELECT EXISTS(SELECT 1 FROM events WHERE run_id = ?1)",
params![run_id],
|row| row.get(0),
)
.map_err(|e| format!("load events existence error: {e}"))?;
if !exists {
return Ok(None);
}
let first: u64 = conn
.query_row(
"SELECT MIN(seq) FROM events WHERE run_id = ?1",
params![run_id],
|row| row.get(0),
)
.map_err(|e| format!("load events first cursor error: {e}"))?;
let next_cursor: u64 = conn
.query_row(
"SELECT COALESCE(MAX(seq) + 1, 0) FROM events WHERE run_id = ?1",
params![run_id],
|row| row.get(0),
)
.map_err(|e| format!("load events next cursor error: {e}"))?;
Ok(Some(
serde_json::json!({"run_id":run_id,"events":events,"next_cursor":next_cursor,"gap":cursor<first,"truncated":false,"dropped":first,"projection":"terminal_persisted_projection","replay_capability":"integrity_only","replayable_execution":false,"resume_supported":false}),
))
}
// ── Idempotency ─────────────────────────────────────────────────────
/// Look up a cached idempotent response together with the canonical request
/// digest it was bound to. A NULL digest is a pre-migration record and must
/// never be replayed for a new request.
pub fn check_idempotency(&self, key: &str) -> Result<Option<(Option<String>, Value)>, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let mut stmt = conn
.prepare("SELECT request_digest, result_json FROM idempotency_keys WHERE key = ?1 AND request_digest IS NOT NULL AND valid = 1")
.map_err(|e| format!("idempotency error: {e}"))?;
let result: Option<(Option<String>, String)> = stmt
.query_row(params![key], |row| Ok((row.get(0)?, row.get(1)?)))
.ok();
match result {
Some((request_digest, json_str)) => serde_json::from_str(&json_str)
.map(|result| Some((request_digest, result)))
.map_err(|e| format!("json parse error: {e}")),
None => Ok(None),
}
}
pub fn save_idempotency(
&self,
key: &str,
request_digest: &str,
result_json: &str,
) -> Result<bool, String> {
let conn = self.conn.lock().map_err(|e| format!("lock error: {e}"))?;
let inserted = conn.execute(
"INSERT OR IGNORE INTO idempotency_keys (key, request_digest, result_json) VALUES (?1, ?2, ?3)",
params![key, request_digest, result_json],
)
.map_err(|e| format!("idempotency error: {e}"))?;
Ok(inserted == 1)
}
pub fn data_dir(&self) -> Option<PathBuf> {
// Returns None since we don't store the path separately.
// The store is identified by its existence, not a path reference.
None
}
}
#[cfg(test)]
mod tests {
use super::*;
fn configure_test_integrity_key() {
let path = std::env::temp_dir().join("agent-graph-mcp-unit-integrity.key");
std::fs::write(&path, [0x5au8; 32]).expect("test integrity key");
std::env::set_var("AGENT_GRAPH_INTEGRITY_KEY_PATH", path);
}
#[test]
fn graph_delete_fault_rolls_back_graph_and_versions_together() {
let temp = tempfile::tempdir().expect("graph database");
let store = PersistentStore::open(temp.path()).expect("store");
store
.save_graph("atomic-delete", "{\"name\":\"atomic-delete\"}", "v1", false)
.expect("graph");
store.fail_graph_delete_after_graph_row();
assert!(store.delete_graph("atomic-delete").is_err());
assert!(store.load_graph("atomic-delete").unwrap().is_some());
assert_eq!(
store.list_graph_versions("atomic-delete").unwrap(),
vec!["v1"]
);
}
#[test]
fn checkpoint_persistence_fault_leaves_no_resumable_row() {
configure_test_integrity_key();
let temp = tempfile::tempdir().expect("checkpoint database");
let store = PersistentStore::open(temp.path()).expect("store");
store
.save_graph("checkpoint-fault", "{}", "version", false)
.expect("graph");
store
.save_execution_with_budgets(
"run-checkpoint-fault",
"checkpoint-fault",
"version",
"checkpointed",
"{}",
Some("null"),
)
.expect("execution");
store.fail_checkpoint_persistence();
let result = store.create_resume_checkpoint(
"run-checkpoint-fault",
"checkpoint-fault",
"version",
"entry",
&serde_json::json!({"__input__":null}),
&Value::Null,
&serde_json::json!({"nodes":0,"llm_calls":0,"wall_clock_ms":0}),
&serde_json::json!({"eligible":true}),
0,
0,
);
assert_eq!(result, Err(CheckpointError::Persistence));
assert_eq!(
store.load_resume_checkpoint(None, Some("run-checkpoint-fault")),
Ok(None)
);
}
}