use super::realm_profile::{RealmProfileStore, StoredRealmProfile};
use super::{
BeginPlacedSpawnResult, CommitPlacedSpawnResult, DeletePlacedSpawnResult,
ExternalBindingOverlayRecord, IdentityMemberEventCommitOutcome,
IdentityMemberTargetObservation, IdentityMemberTargetState, IdentityWiringEventCommitOutcome,
IdentityWiringTargetObservation, IdentityWiringTargetState, MobEventStore,
MobExternalDeliveryBeginOutcome, MobExternalDeliveryCompleteOutcome, MobExternalDeliveryIntent,
MobExternalDeliveryPhase, MobExternalDeliveryRecord, MobExternalDeliveryRepairOutcome,
MobExternalDeliveryRepairState, MobExternalDeliveryTerminal, MobHostAuthorityDeletionAuthority,
MobHostAuthorityPersistenceAuthority, MobHostAuthorityRecord, MobIdentityMemberStore,
MobIdentityStatusStore, MobIdentityStore, MobIdentityStoreClock, MobMemberEventCursorRecord,
MobMemberLiveCleanupRecord, MobMemberOperatorPruneAuthority, MobMemberOperatorRequestBegin,
MobMemberOperatorRequestKey, MobMemberOperatorRequestRecord, MobOperatorGrantDeletionAuthority,
MobOperatorGrantPersistenceAuthority, MobOperatorGrantRecord,
MobPlacedSpawnBindingPromotionAuthority, MobPlacedSpawnCarrierRecord,
MobPlacedSpawnCleanupAuthority, MobPlacedSpawnCommitPersistenceAuthority,
MobPlacedSpawnPendingPersistenceAuthority, MobRunStore, MobRuntimeMetadataStore, MobSpecStore,
MobStoreError, PlacedSpawnCarrierPhase, PromotePlacedSpawnBindingResult,
SupervisorAuthorityDeletionAuthority, SupervisorAuthorityPersistenceAuthority,
SupervisorAuthorityRecord, SystemMobIdentityStoreClock, external_delivery_repair_now_ms,
identity_member_target_state, identity_structural_projection_is_anchor,
identity_wiring_target_state, next_external_delivery_repair_state, private,
step_failed_event_identity, terminal_event_identity, validate_external_delivery_record,
validate_external_delivery_terminal, validate_identity_member_commit_authority,
validate_identity_wiring_commit_authority, validate_mob_event_write_authority,
};
#[cfg(feature = "runtime-adapter")]
use super::{
identity_runtime_fence_error, validate_identity_runtime_target_binding,
validate_identity_runtime_write_authority,
};
use crate::definition::MobDefinition;
use crate::error::MobError;
use crate::event::{
MobEvent, MobEventKind, NewMobEvent, decode_stored_mob_event, encode_stored_mob_event,
};
#[cfg(feature = "runtime-adapter")]
use crate::identity::DesiredSessionTarget;
use crate::identity::{
IDENTITY_INTENT_SCHEMA_VERSION, IDENTITY_LEASE_MAX_TTL_MS, IDENTITY_LEASE_SCHEMA_VERSION,
IDENTITY_OPERATION_RECEIPT_SCHEMA_VERSION, IdentityActuationPermit, IdentityActuatorTarget,
IdentityConvergenceStatus, IdentityIntent, IdentityIntentError, IdentityIntentRecord,
IdentityLeaseClaim, IdentityLeaseClaimOutcome, IdentityLeaseRecord, IdentityOperationKind,
IdentityOperationReceipt, IdentityOperationReceiptInsertOutcome,
IdentityOperationReceiptPayload, IdentityOperationSlot, IdentityOperationSubject,
IdentityRetirementPlan, IdentityStoredObservation, IdentityTargetObservationVersion,
};
use crate::ids::{
AgentIdentity, FlowId, FrameId, Generation, LoopId, LoopInstanceId, MobId, RunId, StepId,
};
use crate::machines::mob_machine as mob_dsl;
use crate::profile::Profile;
use crate::run::flow_run;
use crate::run::{
FailureLedgerEntry, FrameSnapshot, LoopIterationLedgerEntry, LoopSnapshot, MobRun,
MobRunProvenanceAuthority, MobRunRemoteTurnIntent, MobRunRemoteTurnReceipt, MobRunStatus,
StepLedgerEntry, mob_machine_run_status_is_terminal,
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use notify::{RecursiveMode, Watcher};
use rusqlite::{
Connection, ErrorCode, OpenFlags, OptionalExtension, Transaction, TransactionBehavior, params,
};
use serde::{Serialize, de::DeserializeOwned};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, Weak};
use std::thread;
use std::time::{Duration, Instant};
use tokio::sync::broadcast;
const EVENT_SUBSCRIPTION_CHANNEL_CAPACITY: usize = 4096;
const EVENT_WATCH_CATCH_UP_LIMIT: usize = 1024;
const EVENT_WATCH_SAFETY_SWEEP_MS: u64 = 5_000;
const EVENT_WATCH_RECOVERY_BACKOFF_MIN_MS: u64 = 100;
const EVENT_WATCH_RECOVERY_BACKOFF_MAX_MS: u64 = 30_000;
const IDENTITY_STORE_INSTANCE_KEY: &str = "store_instance_id";
const IDENTITY_RECEIPT_SLOT_KEY_VERSION: i64 = 1;
const CREATE_SCHEMA_SQL: &str = r"
CREATE TABLE IF NOT EXISTS mob_events (
cursor INTEGER PRIMARY KEY,
mob_id TEXT,
event_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_event_meta (
key TEXT PRIMARY KEY,
value INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_external_deliveries (
mob_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, idempotency_key)
);
CREATE TABLE IF NOT EXISTS mob_runs (
run_id TEXT PRIMARY KEY,
run_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_run_remote_turn_intents (
run_id TEXT NOT NULL,
dispatch_sequence BLOB NOT NULL,
intent_json BLOB NOT NULL,
PRIMARY KEY (run_id, dispatch_sequence)
);
CREATE TABLE IF NOT EXISTS mob_run_remote_turn_receipts (
run_id TEXT NOT NULL,
dispatch_sequence BLOB NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (run_id, dispatch_sequence)
);
CREATE TABLE IF NOT EXISTS mob_specs (
mob_id TEXT PRIMARY KEY,
spec_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_runtime_supervisors (
mob_id TEXT PRIMARY KEY,
record_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_runtime_host_authorities (
mob_id TEXT NOT NULL,
host_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, host_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_host_binding_generation_highwaters (
mob_id TEXT NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
PRIMARY KEY (mob_id, host_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_operator_requests (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation BLOB NOT NULL,
fence_token BLOB NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
member_session_id TEXT NOT NULL,
request_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_placed_spawns (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_runtime_operator_grants (
mob_id TEXT NOT NULL,
principal TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, principal)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_event_cursors (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_live_cleanups (
mob_id TEXT NOT NULL,
cleanup_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, cleanup_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_binding_overlays (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation INTEGER NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity, generation)
);
CREATE TABLE IF NOT EXISTS mob_identity_store_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_identity_intents (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
declaration_scope TEXT,
session_id TEXT,
lineage_id TEXT,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE UNIQUE INDEX IF NOT EXISTS mob_identity_intents_session
ON mob_identity_intents (session_id)
WHERE session_id IS NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS mob_identity_intents_lineage
ON mob_identity_intents (lineage_id)
WHERE lineage_id IS NOT NULL;
CREATE TABLE IF NOT EXISTS mob_identity_leases (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
epoch_highwater BLOB NOT NULL,
active_holder_id TEXT,
active_incarnation_id TEXT,
active_epoch BLOB,
active_renewed_at_ms BLOB,
active_expires_at_ms BLOB,
authority_digest TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_identity_operation_receipts (
mob_id TEXT NOT NULL,
subject_kind TEXT NOT NULL,
subject_id TEXT NOT NULL,
slot_kind TEXT NOT NULL,
slot_key_version INTEGER NOT NULL,
slot_digest TEXT NOT NULL,
subject_json BLOB NOT NULL,
slot_json BLOB NOT NULL,
subject_identity TEXT,
effect_kind TEXT NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, subject_kind, subject_id, slot_kind,
slot_key_version, slot_digest
)
);
CREATE INDEX IF NOT EXISTS mob_identity_receipts_identity
ON mob_identity_operation_receipts (mob_id, subject_identity);
CREATE TABLE IF NOT EXISTS mob_identity_statuses (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
status_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS realm_profiles (
name TEXT PRIMARY KEY,
profile_json BLOB NOT NULL,
revision INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)";
const RELEASED_0_8_10_CREATE_SCHEMA_SQL: &str = r"
CREATE TABLE IF NOT EXISTS mob_events (
cursor INTEGER PRIMARY KEY,
mob_id TEXT,
event_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_event_meta (
key TEXT PRIMARY KEY,
value INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_runs (
run_id TEXT PRIMARY KEY,
run_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_run_remote_turn_intents (
run_id TEXT NOT NULL,
dispatch_sequence BLOB NOT NULL,
intent_json BLOB NOT NULL,
PRIMARY KEY (run_id, dispatch_sequence)
);
CREATE TABLE IF NOT EXISTS mob_run_remote_turn_receipts (
run_id TEXT NOT NULL,
dispatch_sequence BLOB NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (run_id, dispatch_sequence)
);
CREATE TABLE IF NOT EXISTS mob_specs (
mob_id TEXT PRIMARY KEY,
spec_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_runtime_supervisors (
mob_id TEXT PRIMARY KEY,
record_json BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_runtime_host_authorities (
mob_id TEXT NOT NULL,
host_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, host_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_host_binding_generation_highwaters (
mob_id TEXT NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
PRIMARY KEY (mob_id, host_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_operator_requests (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation BLOB NOT NULL,
fence_token BLOB NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
member_session_id TEXT NOT NULL,
request_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_placed_spawns (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_runtime_operator_grants (
mob_id TEXT NOT NULL,
principal TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, principal)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_event_cursors (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_runtime_member_live_cleanups (
mob_id TEXT NOT NULL,
cleanup_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, cleanup_id)
);
CREATE TABLE IF NOT EXISTS mob_runtime_binding_overlays (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation INTEGER NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity, generation)
);
CREATE TABLE IF NOT EXISTS mob_identity_declaration_scopes (
mob_id TEXT NOT NULL,
scope_id TEXT NOT NULL,
head_json BLOB NOT NULL,
PRIMARY KEY (mob_id, scope_id)
);
CREATE TABLE IF NOT EXISTS mob_identity_store_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS mob_identity_intents (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
declaration_scope TEXT,
session_id TEXT,
lineage_id TEXT,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE UNIQUE INDEX IF NOT EXISTS mob_identity_intents_session
ON mob_identity_intents (session_id)
WHERE session_id IS NOT NULL;
CREATE UNIQUE INDEX IF NOT EXISTS mob_identity_intents_lineage
ON mob_identity_intents (lineage_id)
WHERE lineage_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS mob_identity_intents_scope
ON mob_identity_intents (mob_id, declaration_scope);
CREATE TABLE IF NOT EXISTS mob_identity_leases (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
epoch_highwater BLOB NOT NULL,
active_holder_id TEXT,
active_incarnation_id TEXT,
active_epoch BLOB,
active_renewed_at_ms BLOB,
active_expires_at_ms BLOB,
authority_digest TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS mob_identity_operation_receipts (
mob_id TEXT NOT NULL,
subject_kind TEXT NOT NULL,
subject_id TEXT NOT NULL,
slot_kind TEXT NOT NULL,
slot_key_version INTEGER NOT NULL,
slot_digest TEXT NOT NULL,
subject_json BLOB NOT NULL,
slot_json BLOB NOT NULL,
subject_scope_id TEXT,
subject_identity TEXT,
effect_kind TEXT NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, subject_kind, subject_id, slot_kind,
slot_key_version, slot_digest
)
);
CREATE INDEX IF NOT EXISTS mob_identity_receipts_scope
ON mob_identity_operation_receipts (mob_id, subject_scope_id);
CREATE INDEX IF NOT EXISTS mob_identity_receipts_identity
ON mob_identity_operation_receipts (mob_id, subject_identity);
CREATE TABLE IF NOT EXISTS mob_identity_statuses (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
status_json BLOB NOT NULL,
PRIMARY KEY (mob_id, agent_identity)
);
CREATE TABLE IF NOT EXISTS realm_profiles (
name TEXT PRIMARY KEY,
profile_json BLOB NOT NULL,
revision INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)";
fn se(error: impl std::fmt::Display) -> MobStoreError {
MobStoreError::Internal(error.to_string())
}
fn encode_json<T: Serialize>(value: &T) -> Result<Vec<u8>, MobStoreError> {
serde_json::to_vec(value).map_err(|e| MobStoreError::Serialization(e.to_string()))
}
fn decode_json<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, MobStoreError> {
serde_json::from_slice(bytes).map_err(|e| MobStoreError::Serialization(e.to_string()))
}
fn identity_raw_evidence_digest(bytes: &[u8]) -> String {
format!("sha256:{:x}", Sha256::digest(bytes))
}
fn identity_contract_error(error: IdentityIntentError) -> MobStoreError {
MobStoreError::Serialization(error.to_string())
}
fn identity_authority_blocked(
evidence_digest: Option<String>,
detail: impl Into<String>,
) -> MobStoreError {
MobStoreError::IdentityAuthorityBlocked {
evidence_digest,
detail: detail.into(),
}
}
fn classify_identity_blob<T>(
storage_type: &str,
bytes: &[u8],
expected_schema_version: Option<u32>,
validate: impl FnOnce(&T) -> Result<(), IdentityIntentError>,
physical_key_matches: impl FnOnce(&T) -> Result<(), String>,
) -> IdentityStoredObservation<T>
where
T: DeserializeOwned,
{
let evidence_digest = identity_raw_evidence_digest(bytes);
if storage_type != "blob" {
return IdentityStoredObservation::Malformed {
evidence_digest,
detail: format!(
"identity row uses SQLite storage class '{storage_type}', expected blob"
),
};
}
let json = match serde_json::from_slice::<serde_json::Value>(bytes) {
Ok(value) => value,
Err(error) => {
return IdentityStoredObservation::Malformed {
evidence_digest,
detail: format!("identity row JSON is malformed: {error}"),
};
}
};
if let Some(expected) = expected_schema_version
&& let Some(observed) = json
.get("schema_version")
.and_then(serde_json::Value::as_u64)
&& observed != u64::from(expected)
{
return IdentityStoredObservation::Unsupported {
evidence_digest,
detail: format!("unsupported identity row schema version {observed}"),
};
}
let value = match serde_json::from_value::<T>(json) {
Ok(value) => value,
Err(error) => {
return IdentityStoredObservation::Malformed {
evidence_digest,
detail: format!("identity row JSON does not match its current schema: {error}"),
};
}
};
if let Err(detail) = physical_key_matches(&value) {
return IdentityStoredObservation::Malformed {
evidence_digest,
detail,
};
}
match validate(&value) {
Ok(()) => IdentityStoredObservation::Valid(value),
Err(error @ IdentityIntentError::UnsupportedSchemaVersion { .. }) => {
IdentityStoredObservation::Unsupported {
evidence_digest,
detail: error.to_string(),
}
}
Err(error) => IdentityStoredObservation::Malformed {
evidence_digest,
detail: error.to_string(),
},
}
}
fn require_identity_authority<T>(
observation: IdentityStoredObservation<T>,
row_name: &str,
) -> Result<Option<T>, MobStoreError> {
match observation {
IdentityStoredObservation::Missing => Ok(None),
IdentityStoredObservation::Valid(value) => Ok(Some(value)),
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => Err(identity_authority_blocked(
Some(evidence_digest),
format!("{row_name}: {detail}"),
)),
}
}
fn validate_identity_store_text(field: &str, value: &str) -> Result<(), MobStoreError> {
if value.is_empty() || value.trim() != value {
return Err(MobStoreError::Serialization(format!(
"{field} must be nonempty canonical text"
)));
}
Ok(())
}
fn decode_optional_identity_text(
storage_type: &str,
bytes: Option<Vec<u8>>,
field: &str,
) -> Result<Option<String>, String> {
match (storage_type, bytes) {
("null", None) => Ok(None),
("text", Some(bytes)) => String::from_utf8(bytes)
.map(Some)
.map_err(|error| format!("{field} is not valid UTF-8 text: {error}")),
(observed, _) => Err(format!(
"{field} uses SQLite storage class '{observed}', expected text or null"
)),
}
}
fn decode_required_identity_text(
storage_type: &str,
bytes: Vec<u8>,
field: &str,
) -> Result<String, String> {
if storage_type != "text" {
return Err(format!(
"{field} uses SQLite storage class '{storage_type}', expected text"
));
}
String::from_utf8(bytes).map_err(|error| format!("{field} is not valid UTF-8 text: {error}"))
}
fn identity_intent_projection(
record: &IdentityIntentRecord,
) -> (Option<String>, Option<String>, Option<String>) {
let target = match &record.retirement_plan {
IdentityRetirementPlan::Targets { session, .. } => Some(session),
IdentityRetirementPlan::NoKnownRealization => match &record.intent {
IdentityIntent::Present { session, .. } => Some(session),
IdentityIntent::Absent { .. } => None,
},
};
(
record
.declaration_scope
.as_ref()
.map(|scope| scope.as_str().to_string()),
target.map(|target| target.session_id.to_string()),
target.map(|target| target.lineage_id.as_str().to_string()),
)
}
fn identity_intent_physical_key_matches(
record: &IdentityIntentRecord,
mob_id: &MobId,
identity: &AgentIdentity,
declaration_scope: Option<&str>,
session_id: Option<&str>,
lineage_id: Option<&str>,
) -> Result<(), String> {
if record.mob_id != *mob_id || record.intent.identity() != identity {
return Err(
"identity intent record does not match its physical mob/identity key".to_string(),
);
}
let expected = identity_intent_projection(record);
if expected.0.as_deref() != declaration_scope
|| expected.1.as_deref() != session_id
|| expected.2.as_deref() != lineage_id
{
return Err(
"identity intent record does not match its derived scope/session/lineage keys"
.to_string(),
);
}
Ok(())
}
fn identity_operation_kind_key(kind: IdentityOperationKind) -> &'static str {
match kind {
IdentityOperationKind::SessionCreationConsumed => "session_creation_consumed",
IdentityOperationKind::RetirementProven => "retirement_proven",
IdentityOperationKind::ExternalBinding => "external_binding",
IdentityOperationKind::InitialDelivery => "initial_delivery",
}
}
fn identity_receipt_keys(
subject: &IdentityOperationSubject,
slot: &IdentityOperationSlot,
) -> Result<(String, String, Vec<u8>, Vec<u8>), MobStoreError> {
let subject_text = serde_json::to_string(subject)
.map_err(|error| MobStoreError::Serialization(error.to_string()))?;
let slot_text = serde_json::to_string(slot)
.map_err(|error| MobStoreError::Serialization(error.to_string()))?;
Ok((
subject_text.clone(),
slot_text.clone(),
subject_text.into_bytes(),
slot_text.into_bytes(),
))
}
fn identity_receipt_subject_identity(subject: &IdentityOperationSubject) -> &str {
let IdentityOperationSubject::Identity { identity } = subject;
identity.as_str()
}
fn identity_receipt_sql_key(
subject: &IdentityOperationSubject,
slot: &IdentityOperationSlot,
) -> Result<(&'static str, String, &'static str, i64, String), MobStoreError> {
#[derive(Serialize)]
struct SlotDigestMaterial<'a> {
domain: &'static str,
version: i64,
slot: &'a IdentityOperationSlot,
}
let subject_id = identity_receipt_subject_identity(subject).to_string();
let slot_kind = identity_operation_kind_key(slot.kind());
let material = serde_json::to_vec(&SlotDigestMaterial {
domain: "meerkat.identity.operation_slot_key.v1",
version: IDENTITY_RECEIPT_SLOT_KEY_VERSION,
slot,
})
.map_err(|error| MobStoreError::Serialization(error.to_string()))?;
Ok((
"identity",
subject_id,
slot_kind,
IDENTITY_RECEIPT_SLOT_KEY_VERSION,
identity_raw_evidence_digest(&material),
))
}
fn identity_lease_authority_digest(
mob_id: &MobId,
identity: &AgentIdentity,
record: &IdentityLeaseRecord,
) -> Result<String, MobStoreError> {
#[derive(Serialize)]
struct LeaseAuthorityMaterial<'a> {
domain: &'static str,
mob_id: &'a MobId,
identity: &'a AgentIdentity,
schema_version: u32,
epoch_highwater: u64,
active: &'a Option<IdentityLeaseClaim>,
}
let bytes = serde_json::to_vec(&LeaseAuthorityMaterial {
domain: "meerkat.identity.lease.authority.v1",
mob_id,
identity,
schema_version: record.schema_version,
epoch_highwater: record.epoch_highwater,
active: &record.active,
})
.map_err(|error| MobStoreError::Serialization(error.to_string()))?;
Ok(identity_raw_evidence_digest(&bytes))
}
fn identity_receipt_target(receipt: &IdentityOperationReceipt) -> Option<IdentityActuatorTarget> {
match receipt.effect_kind {
IdentityOperationKind::SessionCreationConsumed => {
Some(IdentityActuatorTarget::SessionCreationReceipt)
}
IdentityOperationKind::RetirementProven => Some(IdentityActuatorTarget::RetirementReceipt),
IdentityOperationKind::ExternalBinding => {
Some(IdentityActuatorTarget::ExternalBindingReceipt)
}
IdentityOperationKind::InitialDelivery => {
Some(IdentityActuatorTarget::InitialDeliveryReceipt)
}
}
}
fn identity_actuator_receipt_matches_intent(
receipt: &IdentityOperationReceipt,
intent: &IdentityIntentRecord,
) -> bool {
use crate::identity::IdentityOperationReceiptPayload as Payload;
let normalized_tombstone = intent.tombstone_generation.unwrap_or(0);
match (&receipt.slot, &receipt.payload, &intent.intent) {
(
IdentityOperationSlot::SessionCreationConsumed {
tombstone_generation,
session_id,
lineage_id,
lineage_generation,
},
Payload::SessionCreationConsumed { authority },
IdentityIntent::Present { session, .. },
) => {
*tombstone_generation == normalized_tombstone
&& session_id == &session.session_id
&& lineage_id == &session.lineage_id
&& *lineage_generation == session.lineage_generation
&& authority.session_id() == &session.session_id
&& authority.validate().is_ok()
}
(
IdentityOperationSlot::RetirementProven {
tombstone_generation,
},
Payload::RetirementProven {
absent_authority_digest,
},
IdentityIntent::Absent { .. },
) => {
*tombstone_generation == normalized_tombstone
&& absent_authority_digest == &intent.authority_digest
}
(
IdentityOperationSlot::ExternalBinding {
tombstone_generation,
remote_signing_identity,
controller_signing_identity,
},
Payload::ExternalBinding {
expected_address,
expected_identity,
expected_controller_identity,
..
},
IdentityIntent::Present { member, .. },
) => {
let crate::identity::DesiredExecution::External { address, identity } =
member.execution()
else {
return false;
};
*tombstone_generation == normalized_tombstone
&& expected_address == address
&& expected_identity == identity
&& remote_signing_identity == identity
&& controller_signing_identity == expected_controller_identity
}
(
IdentityOperationSlot::InitialDelivery {
tombstone_generation,
session_id,
lineage_id,
lineage_generation,
delivery_generation,
},
Payload::InitialDelivery {
delivery_generation: payload_generation,
delivery_id,
message_digest,
},
IdentityIntent::Present {
session, member, ..
},
) => {
let Some(delivery) = &member.initial_delivery else {
return false;
};
*tombstone_generation == normalized_tombstone
&& session_id == &session.session_id
&& lineage_id == &session.lineage_id
&& *lineage_generation == session.lineage_generation
&& delivery_generation == payload_generation
&& *delivery_generation == delivery.delivery_generation
&& delivery_id == &delivery.delivery_id
&& message_digest == &delivery.message_digest
}
_ => false,
}
}
fn validate_member_operator_request_row(
record: &MobMemberOperatorRequestRecord,
key: &MobMemberOperatorRequestKey,
) -> Result<(), MobStoreError> {
record.validate()?;
if record.key().eq(key) {
return Ok(());
}
Err(MobStoreError::Internal(format!(
"member operator request row key does not match record identity/generation/fence/host/binding_generation/session/request_id: row=({},{},{},{},{},{},{}) record=({},{},{},{},{},{},{})",
key.agent_identity,
key.generation,
key.fence_token,
key.host_id,
key.host_binding_generation,
key.member_session_id,
key.request_id,
record.agent_identity,
record.generation,
record.fence_token,
record.host_id,
record.host_binding_generation,
record.member_session_id,
record.request_id
)))
}
fn cursor_to_i64(value: u64) -> Result<i64, MobStoreError> {
i64::try_from(value)
.map_err(|_| MobStoreError::Internal(format!("cursor value {value} exceeds i64::MAX")))
}
fn u64_to_sql_key(value: u64) -> Vec<u8> {
value.to_be_bytes().to_vec()
}
fn sql_key_to_u64(bytes: &[u8], field: &str) -> Result<u64, MobStoreError> {
let encoded: [u8; std::mem::size_of::<u64>()] = bytes.try_into().map_err(|_| {
MobStoreError::Internal(format!(
"{field} SQLite key has {} bytes; expected {}",
bytes.len(),
std::mem::size_of::<u64>()
))
})?;
Ok(u64::from_be_bytes(encoded))
}
fn i64_to_cursor(value: i64) -> u64 {
u64::try_from(value).unwrap_or(0)
}
fn stored_event_envelope_mob_id(bytes: &[u8]) -> Option<String> {
let value = serde_json::from_slice::<serde_json::Value>(bytes).ok()?;
let mob_id = value.get("event")?.get("mob_id")?.as_str()?.to_string();
(!mob_id.is_empty() && mob_id.trim() == mob_id).then_some(mob_id)
}
fn migration_0002_event_route(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
let has_mob_id = {
let mut statement = tx.prepare("PRAGMA table_info(mob_events)")?;
let columns = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<Result<Vec<_>, _>>()?;
columns.iter().any(|column| column == "mob_id")
};
if !has_mob_id {
tx.execute("ALTER TABLE mob_events ADD COLUMN mob_id TEXT", [])?;
}
tx.execute(
"CREATE INDEX IF NOT EXISTS mob_events_mob_cursor ON mob_events (mob_id, cursor)",
[],
)?;
Ok(())
}
fn heal_unrouted_events(conn: &Connection) -> Result<(), MobStoreError> {
let unrouted = {
let mut statement = conn
.prepare(
"SELECT cursor, CAST(event_json AS BLOB)
FROM mob_events
WHERE mob_id IS NULL
ORDER BY cursor",
)
.map_err(se)?;
statement
.query_map([], |row| {
Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
})
.map_err(se)?
.collect::<Result<Vec<_>, _>>()
.map_err(se)?
};
for (cursor, bytes) in unrouted {
if let Some(mob_id) = stored_event_envelope_mob_id(&bytes) {
conn.execute(
"UPDATE mob_events SET mob_id = ?2 WHERE cursor = ?1 AND mob_id IS NULL",
params![cursor, mob_id],
)
.map_err(se)?;
}
}
Ok(())
}
fn migration_0003_member_operator_execution_fence(
tx: &Transaction<'_>,
) -> Result<(), rusqlite::Error> {
let (has_host_id, has_binding_generation, has_member_session_id, primary_key) = {
let mut stmt = tx.prepare("PRAGMA table_info(mob_runtime_member_operator_requests)")?;
let columns = stmt.query_map([], |row| {
Ok((row.get::<_, String>(1)?, row.get::<_, i64>(5)?))
})?;
let mut found_host_id = false;
let mut found_binding_generation = false;
let mut found_member_session_id = false;
let mut primary_key = Vec::new();
for column in columns {
let (name, key_position) = column?;
match name.as_str() {
"host_id" => found_host_id = true,
"binding_generation" => found_binding_generation = true,
"member_session_id" => found_member_session_id = true,
_ => {}
}
if key_position > 0 {
primary_key.push((key_position, name));
}
}
primary_key.sort_by_key(|(position, _)| *position);
(
found_host_id,
found_binding_generation,
found_member_session_id,
primary_key
.into_iter()
.map(|(_, name)| name)
.collect::<Vec<_>>(),
)
};
let expected_primary_key = [
"mob_id",
"agent_identity",
"generation",
"fence_token",
"host_id",
"binding_generation",
"member_session_id",
"request_id",
];
if !has_host_id
|| !has_binding_generation
|| !has_member_session_id
|| !primary_key
.iter()
.map(String::as_str)
.eq(expected_primary_key)
{
tx.execute_batch(
"ALTER TABLE mob_runtime_member_operator_requests
RENAME TO mob_runtime_member_operator_requests_pre_execution_fence;
CREATE TABLE mob_runtime_member_operator_requests (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation BLOB NOT NULL,
fence_token BLOB NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
member_session_id TEXT NOT NULL,
request_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, agent_identity, generation, fence_token, host_id,
binding_generation, member_session_id, request_id
)
);
DROP TABLE mob_runtime_member_operator_requests_pre_execution_fence;",
)?;
}
Ok(())
}
fn migration_0004_release_0_8_11_schema(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch(
"CREATE TABLE mob_external_deliveries (
mob_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, idempotency_key)
);
DROP INDEX mob_identity_intents_scope;
DROP INDEX mob_identity_receipts_scope;
DROP TABLE mob_identity_declaration_scopes;
DROP INDEX mob_identity_receipts_identity;
ALTER TABLE mob_identity_operation_receipts
RENAME TO mob_identity_operation_receipts_v3;
CREATE TABLE mob_identity_operation_receipts (
mob_id TEXT NOT NULL,
subject_kind TEXT NOT NULL,
subject_id TEXT NOT NULL,
slot_kind TEXT NOT NULL,
slot_key_version INTEGER NOT NULL,
slot_digest TEXT NOT NULL,
subject_json BLOB NOT NULL,
slot_json BLOB NOT NULL,
subject_identity TEXT,
effect_kind TEXT NOT NULL,
receipt_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, subject_kind, subject_id, slot_kind,
slot_key_version, slot_digest
)
);
INSERT INTO mob_identity_operation_receipts (
mob_id, subject_kind, subject_id, slot_kind, slot_key_version,
slot_digest, subject_json, slot_json, subject_identity,
effect_kind, receipt_json
)
SELECT
mob_id, subject_kind, subject_id, slot_kind, slot_key_version,
slot_digest, subject_json, slot_json, subject_identity,
effect_kind, receipt_json
FROM mob_identity_operation_receipts_v3
WHERE subject_kind = 'identity'
AND subject_identity IS NOT NULL;
DROP TABLE mob_identity_operation_receipts_v3;
CREATE INDEX mob_identity_receipts_identity
ON mob_identity_operation_receipts (mob_id, subject_identity);",
)?;
Ok(())
}
fn migration_0001_mob_schema(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch(CREATE_SCHEMA_SQL)
}
fn initialize_current_mob_schema(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
migration_0001_mob_schema(tx)?;
migration_0002_event_route(tx)?;
migration_0003_member_operator_execution_fence(tx)
}
fn build_released_0_8_10_mob_schema(tx: &Transaction<'_>) -> Result<(), rusqlite::Error> {
tx.execute_batch(RELEASED_0_8_10_CREATE_SCHEMA_SQL)?;
migration_0002_event_route(tx)?;
migration_0003_member_operator_execution_fence(tx)
}
const RELEASED_0_8_10_MOB_OBJECTS: &[meerkat_sqlite::SchemaObject] = &[
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_events",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_event_meta",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runs",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_run_remote_turn_intents",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_run_remote_turn_receipts",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_specs",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_supervisors",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_host_authorities",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_host_binding_generation_highwaters",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_operator_requests",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_placed_spawns",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_operator_grants",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_event_cursors",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_live_cleanups",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_binding_overlays",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_declaration_scopes",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_store_meta",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_intents",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_leases",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_operation_receipts",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_statuses",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "realm_profiles",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_events_mob_cursor",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_session",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_lineage",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_scope",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_receipts_scope",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_receipts_identity",
},
];
fn verify_released_0_8_10_mob_schema(conn: &Connection) -> Result<(), String> {
meerkat_sqlite::verify_released_schema_fingerprint(
conn,
&MOB_DOMAIN,
RELEASED_0_8_10_MOB_OBJECTS,
build_released_0_8_10_mob_schema,
)
}
pub const MOB_DOMAIN: meerkat_sqlite::SchemaDomain = meerkat_sqlite::SchemaDomain {
name: "mob",
migrations: &[
meerkat_sqlite::Migration {
version: 1,
name: "base-schema",
apply: migration_0001_mob_schema,
},
meerkat_sqlite::Migration {
version: 2,
name: "event-route",
apply: migration_0002_event_route,
},
meerkat_sqlite::Migration {
version: 3,
name: "member-operator-execution-fence",
apply: migration_0003_member_operator_execution_fence,
},
meerkat_sqlite::Migration {
version: 4,
name: "release-0.8.11-schema",
apply: migration_0004_release_0_8_11_schema,
},
],
initialize_current: initialize_current_mob_schema,
allowed_existing_versions: &[3, 4],
released_predecessors: &[meerkat_sqlite::SchemaPredecessor {
version: 3,
verify: verify_released_0_8_10_mob_schema,
}],
owned_objects: &[
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_events",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_event_meta",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_external_deliveries",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runs",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_run_remote_turn_intents",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_run_remote_turn_receipts",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_specs",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_supervisors",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_host_authorities",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_host_binding_generation_highwaters",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_operator_requests",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_placed_spawns",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_operator_grants",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_event_cursors",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_member_live_cleanups",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_runtime_binding_overlays",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_store_meta",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_intents",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_leases",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_operation_receipts",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_statuses",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "realm_profiles",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_events_mob_cursor",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_session",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_lineage",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_receipts_identity",
},
],
retired_objects: &[
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Table,
name: "mob_identity_declaration_scopes",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_intents_scope",
},
meerkat_sqlite::SchemaObject {
kind: meerkat_sqlite::SchemaObjectKind::Index,
name: "mob_identity_receipts_scope",
},
],
};
#[cfg(test)]
mod schema_floor_tests {
use super::*;
fn released_v3() -> Connection {
let mut conn = Connection::open_in_memory().expect("open");
let tx = conn.transaction().expect("tx");
build_released_0_8_10_mob_schema(&tx).expect("released schema");
tx.commit().expect("commit");
conn.execute_batch(
"CREATE TABLE meerkat_schema (
domain TEXT PRIMARY KEY,
version INTEGER NOT NULL
);
INSERT INTO meerkat_schema VALUES ('mob', 3);",
)
.expect("ledger");
conn
}
#[test]
fn exact_released_v3_upgrades_to_current() {
let mut conn = released_v3();
let report =
meerkat_sqlite::apply_domain_migrations(&mut conn, &MOB_DOMAIN).expect("upgrade");
assert_eq!(report.from_version, 3);
assert_eq!(report.to_version, 4);
}
#[test]
fn released_v3_external_table_and_pruned_candidate_shapes_are_refused_unmutated() {
for candidate in [
"CREATE TABLE mob_external_deliveries (
mob_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (mob_id, idempotency_key)
)",
"DROP INDEX mob_identity_intents_scope;
DROP INDEX mob_identity_receipts_scope;
DROP TABLE mob_identity_declaration_scopes",
] {
let mut conn = released_v3();
conn.execute_batch(candidate).expect("candidate shape");
let err = meerkat_sqlite::apply_domain_migrations(&mut conn, &MOB_DOMAIN)
.expect_err("refuse candidate");
assert!(matches!(
err,
meerkat_sqlite::SqliteStoreError::SchemaFingerprintMismatch { version: 3, .. }
));
assert_eq!(
meerkat_sqlite::domain_version(&conn, MOB_DOMAIN.name).expect("ledger"),
Some(3)
);
}
}
}
struct MobConn {
conn: Connection,
_guard: meerkat_sqlite::OperationGuard,
}
impl std::ops::Deref for MobConn {
type Target = Connection;
fn deref(&self) -> &Connection {
&self.conn
}
}
impl std::ops::DerefMut for MobConn {
fn deref_mut(&mut self) -> &mut Connection {
&mut self.conn
}
}
fn open_connection(path: &Path) -> Result<MobConn, MobStoreError> {
let guard = meerkat_sqlite::OperationGuard::for_database(path).map_err(se)?;
let mut conn = meerkat_sqlite::open_with(
path,
meerkat_sqlite::ConnectionProfile::PRIMARY,
meerkat_sqlite::OpenOptions {
schema_preflight: &[&MOB_DOMAIN],
..meerkat_sqlite::OpenOptions::default()
},
)
.map_err(se)?;
meerkat_sqlite::apply_domain_migrations(&mut conn, &MOB_DOMAIN).map_err(se)?;
repair_next_event_cursor(&conn)?;
heal_unrouted_events(&conn)?;
Ok(MobConn {
conn,
_guard: guard,
})
}
fn open_existing_read_connection(path: &Path) -> Result<Connection, MobStoreError> {
meerkat_sqlite::open(path, meerkat_sqlite::ConnectionProfile::ReadOnly).map_err(se)
}
fn initialize_identity_store_instance(conn: &Connection) -> Result<String, MobStoreError> {
let candidate = uuid::Uuid::new_v4().to_string();
conn.execute(
"INSERT OR IGNORE INTO mob_identity_store_meta (key, value) VALUES (?1, ?2)",
params![IDENTITY_STORE_INSTANCE_KEY, candidate],
)
.map_err(se)?;
conn.query_row(
"SELECT value FROM mob_identity_store_meta WHERE key = ?1",
params![IDENTITY_STORE_INSTANCE_KEY],
|row| row.get(0),
)
.map_err(se)
}
fn verify_identity_store_instance(
conn: &Connection,
expected_store_instance_id: &str,
) -> Result<(), MobStoreError> {
let actual: String = conn
.query_row(
"SELECT value FROM mob_identity_store_meta WHERE key = ?1",
params![IDENTITY_STORE_INSTANCE_KEY],
|row| row.get(0),
)
.map_err(|error| {
MobStoreError::ReadFailed(format!(
"identity store instance marker is unavailable: {error}"
))
})?;
if actual != expected_store_instance_id {
return Err(identity_authority_blocked(
None,
"identity store database was replaced while this store handle was live",
));
}
Ok(())
}
fn open_existing_identity_read_connection(
path: &Path,
expected_store_instance_id: &str,
) -> Result<Connection, MobStoreError> {
let conn = open_existing_read_connection(path).map_err(|error| {
MobStoreError::ReadFailed(format!("identity store database is unavailable: {error}"))
})?;
verify_identity_store_instance(&conn, expected_store_instance_id)?;
Ok(conn)
}
fn open_existing_identity_write_connection(
path: &Path,
_expected_store_instance_id: &str,
) -> Result<MobConn, MobStoreError> {
let guard = meerkat_sqlite::OperationGuard::for_database(path).map_err(se)?;
let conn = meerkat_sqlite::open_with(
path,
meerkat_sqlite::ConnectionProfile::Primary { create: false },
meerkat_sqlite::OpenOptions {
schema_preflight: &[&MOB_DOMAIN],
..meerkat_sqlite::OpenOptions::default()
},
)
.map_err(|error| {
MobStoreError::WriteFailed(format!("identity store database is unavailable: {error}"))
})?;
Ok(MobConn {
conn,
_guard: guard,
})
}
fn begin_identity_immediate<'connection>(
conn: &'connection mut Connection,
expected_store_instance_id: &str,
) -> Result<Transaction<'connection>, MobStoreError> {
let tx = begin_immediate(conn)?;
verify_identity_store_instance(&tx, expected_store_instance_id)?;
Ok(tx)
}
fn begin_immediate(conn: &mut Connection) -> Result<Transaction<'_>, MobStoreError> {
conn.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(se)
}
fn lock_unpoisoned<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
mutex
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
fn latest_event_cursor_sync(path: &Path) -> Result<u64, MobStoreError> {
let conn = open_existing_read_connection(path)?;
let cursor: Option<i64> = conn
.query_row("SELECT MAX(cursor) FROM mob_events", [], |row| row.get(0))
.optional()
.map_err(se)?
.flatten();
Ok(cursor.map_or(0, i64_to_cursor))
}
fn poll_events_sync(
path: &Path,
after_cursor: u64,
limit: usize,
) -> Result<Vec<MobEvent>, MobStoreError> {
let conn = open_existing_read_connection(path)?;
poll_events_from_connection(&conn, after_cursor, limit)
}
fn poll_events_from_connection(
conn: &Connection,
after_cursor: u64,
limit: usize,
) -> Result<Vec<MobEvent>, MobStoreError> {
let mut stmt = conn
.prepare("SELECT event_json FROM mob_events WHERE cursor > ?1 ORDER BY cursor LIMIT ?2")
.map_err(se)?;
let rows = stmt
.query_map(
params![
cursor_to_i64(after_cursor)?,
i64::try_from(limit).map_err(|_| se("limit exceeds i64::MAX"))?
],
|row| row.get::<_, Vec<u8>>(0),
)
.map_err(se)?;
let mut result = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
result.push(
decode_stored_mob_event(&bytes)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?,
);
}
Ok(result)
}
fn load_run_bytes(tx: &Transaction<'_>, key: &str) -> Result<Option<Vec<u8>>, MobStoreError> {
tx.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)
}
fn write_run_json(tx: &Transaction<'_>, key: &str, run: &MobRun) -> Result<(), MobStoreError> {
let encoded = encode_json(run)?;
tx.execute(
"UPDATE mob_runs SET run_json = ?1 WHERE run_id = ?2",
params![encoded, key],
)
.map_err(se)?;
Ok(())
}
fn append_loop_iteration_ledger_if_absent(run: &mut MobRun, entry: LoopIterationLedgerEntry) {
if !run.loop_iteration_ledger.iter().any(|existing| {
existing.loop_instance_id == entry.loop_instance_id
&& existing.iteration == entry.iteration
&& existing.frame_id == entry.frame_id
}) {
run.loop_iteration_ledger.push(entry);
}
}
async fn run_sqlite_task<T>(
task: impl FnOnce() -> Result<T, MobStoreError> + Send + 'static,
) -> Result<T, MobStoreError>
where
T: Send + 'static,
{
tokio::task::spawn_blocking(task)
.await
.map_err(|error| MobStoreError::Internal(format!("sqlite task join failed: {error}")))?
}
static SQLITE_EVENT_BUSES: OnceLock<Mutex<HashMap<PathBuf, Weak<SqliteMobEventBus>>>> =
OnceLock::new();
fn sqlite_event_buses() -> &'static Mutex<HashMap<PathBuf, Weak<SqliteMobEventBus>>> {
SQLITE_EVENT_BUSES.get_or_init(|| Mutex::new(HashMap::new()))
}
fn sqlite_event_bus_for_path(path: PathBuf) -> Result<Arc<SqliteMobEventBus>, MobStoreError> {
let mut buses = lock_unpoisoned(sqlite_event_buses());
if let Some(bus) = buses.get(&path).and_then(Weak::upgrade) {
return Ok(bus);
}
let bus = SqliteMobEventBus::new(path.clone())?;
buses.insert(path, Arc::downgrade(&bus));
Ok(bus)
}
struct SqliteMobEventBus {
path: PathBuf,
event_tx: broadcast::Sender<MobEvent>,
latest_broadcast_cursor: Mutex<u64>,
catch_up_lock: Mutex<()>,
watcher: Mutex<Option<notify::RecommendedWatcher>>,
}
impl std::fmt::Debug for SqliteMobEventBus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteMobEventBus")
.field("path", &self.path)
.finish()
}
}
impl SqliteMobEventBus {
fn new(path: PathBuf) -> Result<Arc<Self>, MobStoreError> {
let latest_cursor = latest_event_cursor_sync(&path)?;
let (event_tx, _event_rx) = broadcast::channel(EVENT_SUBSCRIPTION_CHANNEL_CAPACITY);
let bus = Arc::new(Self {
path,
event_tx,
latest_broadcast_cursor: Mutex::new(latest_cursor),
catch_up_lock: Mutex::new(()),
watcher: Mutex::new(None),
});
bus.start_external_watch();
Ok(bus)
}
fn subscribe(&self) -> super::MobEventReceiver {
self.event_tx.subscribe()
}
fn publish_committed(&self, event: MobEvent) {
self.publish_committed_batch(std::slice::from_ref(&event));
}
fn publish_committed_batch(&self, events: &[MobEvent]) {
let current_cursor = *lock_unpoisoned(&self.latest_broadcast_cursor);
let mut expected_cursor = current_cursor.saturating_add(1);
let mut has_gap = false;
for event in events.iter().filter(|event| event.cursor > current_cursor) {
if event.cursor > expected_cursor {
has_gap = true;
break;
}
expected_cursor = event.cursor.saturating_add(1);
}
if has_gap {
if let Err(error) = self.publish_available_from_storage() {
tracing::warn!(
error = %error,
path = %self.path.display(),
"sqlite mob event gap catch-up failed before direct publish",
);
}
}
self.publish_committed_batch_unchecked(events);
}
fn publish_committed_batch_unchecked(&self, events: &[MobEvent]) {
let mut cursor = lock_unpoisoned(&self.latest_broadcast_cursor);
for event in events {
if event.cursor <= *cursor {
continue;
}
*cursor = event.cursor;
let _ = self.event_tx.send(event.clone());
}
}
fn publish_available_from_storage(&self) -> Result<(), MobStoreError> {
let _catch_up_guard = lock_unpoisoned(&self.catch_up_lock);
if !self.path.try_exists().map_err(se)? {
return Ok(());
}
loop {
let after_cursor = *lock_unpoisoned(&self.latest_broadcast_cursor);
let batch = match poll_events_sync(&self.path, after_cursor, EVENT_WATCH_CATCH_UP_LIMIT)
{
Ok(batch) => batch,
Err(error) => {
if !self.path.try_exists().map_err(se)? {
return Ok(());
}
return Err(error);
}
};
if batch.is_empty() {
return Ok(());
}
let is_complete = batch.len() < EVENT_WATCH_CATCH_UP_LIMIT;
self.publish_committed_batch_unchecked(&batch);
if is_complete {
return Ok(());
}
}
}
fn start_external_watch(self: &Arc<Self>) {
let Some(parent) = self.path.parent().map(Path::to_path_buf) else {
return;
};
let watched_paths = sqlite_watch_paths(&self.path);
let (wake_tx, wake_rx) = std::sync::mpsc::channel::<()>();
let thread_bus = Arc::downgrade(self);
let thread_builder = thread::Builder::new().name("sqlite-mob-event-watch".to_string());
if let Err(error) = thread_builder.spawn(move || {
let recovery_backoff_min = Duration::from_millis(EVENT_WATCH_RECOVERY_BACKOFF_MIN_MS);
let recovery_backoff_max = Duration::from_millis(EVENT_WATCH_RECOVERY_BACKOFF_MAX_MS);
let mut recovery_backoff = recovery_backoff_min;
let mut recovery_deadline = None;
loop {
let wait = recovery_deadline
.map(|deadline: Instant| deadline.saturating_duration_since(Instant::now()))
.unwrap_or_else(|| Duration::from_millis(EVENT_WATCH_SAFETY_SWEEP_MS));
let received_wake = match wake_rx.recv_timeout(wait) {
Ok(()) => true,
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => false,
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
};
if received_wake {
thread::sleep(Duration::from_millis(10));
}
while wake_rx.try_recv().is_ok() {}
let Some(bus) = thread_bus.upgrade() else {
break;
};
if recovery_deadline.is_some_and(|deadline| Instant::now() < deadline) {
continue;
}
if !received_wake && bus.event_tx.receiver_count() == 0 {
recovery_deadline = None;
recovery_backoff = recovery_backoff_min;
continue;
}
match bus.publish_available_from_storage() {
Ok(()) => {
recovery_deadline = None;
recovery_backoff = recovery_backoff_min;
}
Err(error) => {
tracing::warn!(
error = %error,
path = %bus.path.display(),
retry_after_ms = recovery_backoff.as_millis(),
"sqlite mob event watch catch-up failed",
);
recovery_deadline = Some(Instant::now() + recovery_backoff);
recovery_backoff =
recovery_backoff.saturating_mul(2).min(recovery_backoff_max);
}
}
}
}) {
tracing::warn!(
error = %error,
path = %self.path.display(),
"failed to start sqlite mob event watch thread",
);
return;
}
let callback_wake_tx = wake_tx.clone();
let callback_parent = parent.clone();
let mut watcher =
match notify::recommended_watcher(move |result: notify::Result<notify::Event>| {
match result {
Ok(event)
if sqlite_watch_event_relevant(
&event,
&callback_parent,
&watched_paths,
) =>
{
let _ = callback_wake_tx.send(());
}
Ok(_) => {}
Err(error) => {
tracing::warn!(
error = %error,
"sqlite mob event filesystem watch reported an error",
);
}
}
}) {
Ok(watcher) => watcher,
Err(error) => {
tracing::warn!(
error = %error,
path = %self.path.display(),
"failed to create sqlite mob event filesystem watcher",
);
return;
}
};
if let Err(error) = watcher.watch(&parent, RecursiveMode::NonRecursive) {
tracing::warn!(
error = %error,
path = %parent.display(),
"failed to watch sqlite mob event directory",
);
return;
}
*lock_unpoisoned(&self.watcher) = Some(watcher);
}
}
fn sqlite_sidecar_path(path: &Path, suffix: &str) -> PathBuf {
let mut value = path.as_os_str().to_os_string();
value.push(suffix);
PathBuf::from(value)
}
fn sqlite_watch_paths(path: &Path) -> Vec<PathBuf> {
vec![
path.to_path_buf(),
sqlite_sidecar_path(path, "-wal"),
sqlite_sidecar_path(path, "-shm"),
]
}
fn sqlite_watch_event_relevant(
event: ¬ify::Event,
parent: &Path,
watched_paths: &[PathBuf],
) -> bool {
if matches!(event.kind, notify::EventKind::Access(_)) {
return false;
}
if event.paths.is_empty() {
return true;
}
event.paths.iter().any(|path| {
path == parent
|| watched_paths.iter().any(|watched| {
path == watched
|| (path.parent() == watched.parent()
&& path.file_name() == watched.file_name())
})
})
}
#[derive(Debug, Clone)]
pub struct SqliteMobStores {
path: PathBuf,
event_bus: Arc<SqliteMobEventBus>,
identity_store_instance_id: String,
}
impl SqliteMobStores {
pub fn open(path: impl AsRef<Path>) -> Result<Self, MobError> {
let path = path.as_ref().to_path_buf();
let conn = open_connection(&path)?;
let identity_store_instance_id = initialize_identity_store_instance(&conn)?;
let path = std::fs::canonicalize(&path).map_err(se)?;
let event_bus = sqlite_event_bus_for_path(path.clone())?;
Ok(Self {
path,
event_bus,
identity_store_instance_id,
})
}
pub fn event_store(&self) -> SqliteMobEventStore {
SqliteMobEventStore {
path: self.path.clone(),
event_bus: Arc::clone(&self.event_bus),
}
}
pub fn run_store(&self) -> SqliteMobRunStore {
SqliteMobRunStore {
path: self.path.clone(),
}
}
pub fn spec_store(&self) -> SqliteMobSpecStore {
SqliteMobSpecStore {
path: self.path.clone(),
}
}
pub fn runtime_metadata_store(&self) -> SqliteMobRuntimeMetadataStore {
SqliteMobRuntimeMetadataStore {
path: self.path.clone(),
}
}
pub fn identity_store(&self) -> SqliteMobIdentityStore {
SqliteMobIdentityStore {
path: self.path.clone(),
store_instance_id: self.identity_store_instance_id.clone(),
clock: Arc::new(SystemMobIdentityStoreClock),
event_bus: Some(Arc::clone(&self.event_bus)),
}
}
#[cfg(any(test, feature = "test-support"))]
pub(crate) fn identity_store_with_clock(
&self,
clock: Arc<dyn MobIdentityStoreClock>,
) -> SqliteMobIdentityStore {
SqliteMobIdentityStore {
path: self.path.clone(),
store_instance_id: self.identity_store_instance_id.clone(),
clock,
event_bus: Some(Arc::clone(&self.event_bus)),
}
}
pub fn identity_status_store(&self) -> SqliteMobIdentityStatusStore {
SqliteMobIdentityStatusStore {
path: self.path.clone(),
store_instance_id: self.identity_store_instance_id.clone(),
}
}
pub fn realm_profile_store(&self) -> SqliteRealmProfileStore {
SqliteRealmProfileStore {
path: self.path.clone(),
}
}
}
#[derive(Clone)]
pub struct SqliteMobIdentityStore {
path: PathBuf,
store_instance_id: String,
clock: Arc<dyn MobIdentityStoreClock>,
event_bus: Option<Arc<SqliteMobEventBus>>,
}
#[cfg(feature = "runtime-adapter")]
struct SqliteIdentityRuntimeWriteFence {
path: PathBuf,
store_instance_id: String,
clock: Arc<dyn MobIdentityStoreClock>,
permit: IdentityActuationPermit,
expected_session: DesiredSessionTarget,
}
#[cfg(feature = "runtime-adapter")]
impl meerkat_runtime::RuntimeStoreWriteFence for SqliteIdentityRuntimeWriteFence {
fn execute_if_current(
&self,
operation: Box<dyn FnOnce() -> Result<(), meerkat_runtime::RuntimeStoreError> + '_>,
) -> Result<meerkat_runtime::RuntimeStoreWriteFenceOutcome, meerkat_runtime::RuntimeStoreError>
{
let _guard = match meerkat_sqlite::OperationGuard::for_database(&self.path) {
Ok(guard) => guard,
Err(error) => {
return Ok(meerkat_runtime::RuntimeStoreWriteFenceOutcome::Backoff {
reason: format!(
"identity runtime write fence deferred to the maintenance fence: {error}"
),
});
}
};
let mut conn = match meerkat_sqlite::open(
&self.path,
meerkat_sqlite::ConnectionProfile::Maintenance { write: true },
) {
Ok(conn) => conn,
Err(error) => {
return Ok(meerkat_runtime::RuntimeStoreWriteFenceOutcome::Backoff {
reason: format!(
"identity runtime write fence could not open the store: {error}"
),
});
}
};
let tx = match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
Ok(tx) => tx,
Err(error) if meerkat_sqlite::is_busy_or_locked(&error) => {
return Ok(meerkat_runtime::RuntimeStoreWriteFenceOutcome::Backoff {
reason: "identity runtime write fence authority guard is contended".to_string(),
});
}
Err(error) => {
return identity_runtime_fence_error(MobStoreError::WriteFailed(format!(
"identity runtime write fence could not begin its authority transaction: {error}"
)));
}
};
if let Err(error) = verify_identity_store_instance(&tx, &self.store_instance_id) {
return identity_runtime_fence_error(error);
}
let intent = match query_identity_intent_observation(
&tx,
&self.permit.mob_id,
&self.permit.identity,
)
.and_then(|observation| require_identity_authority(observation, "identity intent"))
{
Ok(intent) => intent,
Err(error) => return identity_runtime_fence_error(error),
};
let lease =
match query_identity_lease_observation(&tx, &self.permit.mob_id, &self.permit.identity)
.and_then(|observation| require_identity_authority(observation, "identity lease"))
{
Ok(lease) => lease,
Err(error) => return identity_runtime_fence_error(error),
};
let observed_at_ms = match self.clock.now_ms() {
Ok(observed_at_ms) => observed_at_ms,
Err(error) => return identity_runtime_fence_error(error),
};
if let Err(error) = validate_identity_runtime_write_authority(
&self.permit,
&self.expected_session,
intent.as_ref(),
lease.as_ref(),
observed_at_ms,
) {
return identity_runtime_fence_error(error);
}
operation()?;
Ok(meerkat_runtime::RuntimeStoreWriteFenceOutcome::Applied)
}
}
impl std::fmt::Debug for SqliteMobIdentityStore {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("SqliteMobIdentityStore")
.field("path", &self.path)
.field("store_instance_id", &self.store_instance_id)
.field("clock", &"<dyn MobIdentityStoreClock>")
.field("member_event_capability", &self.event_bus.is_some())
.finish()
}
}
impl SqliteMobIdentityStore {
#[cfg(any(test, feature = "test-support"))]
pub(crate) fn with_clock(
path: PathBuf,
store_instance_id: String,
clock: Arc<dyn MobIdentityStoreClock>,
) -> Self {
Self {
path,
store_instance_id,
clock,
event_bus: None,
}
}
}
#[derive(Debug, Clone)]
pub struct SqliteMobIdentityStatusStore {
path: PathBuf,
store_instance_id: String,
}
fn sqlite_identity_write_error(error: rusqlite::Error, context: &str) -> MobStoreError {
if matches!(
&error,
rusqlite::Error::SqliteFailure(sqlite_error, _)
if sqlite_error.code == ErrorCode::ConstraintViolation
&& matches!(
sqlite_error.extended_code,
rusqlite::ffi::SQLITE_CONSTRAINT_PRIMARYKEY
| rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE
)
) {
MobStoreError::CasConflict(format!("{context}: {error}"))
} else {
MobStoreError::WriteFailed(format!("{context}: {error}"))
}
}
fn query_identity_intent_observation(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityIntentRecord>, MobStoreError> {
let row: Option<(
String,
Option<Vec<u8>>,
String,
Option<Vec<u8>>,
String,
Option<Vec<u8>>,
String,
Vec<u8>,
)> = conn
.query_row(
"SELECT typeof(declaration_scope), CAST(declaration_scope AS BLOB),
typeof(session_id), CAST(session_id AS BLOB),
typeof(lineage_id), CAST(lineage_id AS BLOB),
typeof(record_json), CAST(record_json AS BLOB)
FROM mob_identity_intents
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), identity.as_str()],
|row| {
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)?,
))
},
)
.optional()
.map_err(se)?;
let Some((
declaration_scope_type,
declaration_scope_bytes,
session_id_type,
session_id_bytes,
lineage_id_type,
lineage_id_bytes,
storage_type,
bytes,
)) = row
else {
return Ok(IdentityStoredObservation::Missing);
};
let declaration_scope = decode_optional_identity_text(
&declaration_scope_type,
declaration_scope_bytes,
"identity intent declaration_scope",
);
let session_id = decode_optional_identity_text(
&session_id_type,
session_id_bytes,
"identity intent session_id",
);
let lineage_id = decode_optional_identity_text(
&lineage_id_type,
lineage_id_bytes,
"identity intent lineage_id",
);
let projection_error = declaration_scope
.as_ref()
.err()
.or_else(|| session_id.as_ref().err())
.or_else(|| lineage_id.as_ref().err())
.cloned();
let declaration_scope = declaration_scope.unwrap_or(None);
let session_id = session_id.unwrap_or(None);
let lineage_id = lineage_id.unwrap_or(None);
Ok(classify_identity_blob(
&storage_type,
&bytes,
Some(IDENTITY_INTENT_SCHEMA_VERSION),
IdentityIntentRecord::validate,
|record| {
if let Some(detail) = projection_error {
return Err(detail);
}
identity_intent_physical_key_matches(
record,
mob_id,
identity,
declaration_scope.as_deref(),
session_id.as_deref(),
lineage_id.as_deref(),
)
},
))
}
struct LoadedIdentityLeaseRow {
observation: IdentityStoredObservation<IdentityLeaseRecord>,
}
fn query_identity_lease_row(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<Option<LoadedIdentityLeaseRow>, MobStoreError> {
type LeaseSqlRow = (
Vec<u8>,
Option<String>,
Option<String>,
Option<Vec<u8>>,
Option<Vec<u8>>,
Option<Vec<u8>>,
String,
String,
Vec<u8>,
);
let row: Option<LeaseSqlRow> = conn
.query_row(
"SELECT epoch_highwater, active_holder_id, active_incarnation_id,
active_epoch, active_renewed_at_ms, active_expires_at_ms,
authority_digest, typeof(record_json), CAST(record_json AS BLOB)
FROM mob_identity_leases
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), identity.as_str()],
|row| {
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)?,
))
},
)
.optional()
.map_err(se)?;
let Some((
epoch_highwater,
active_holder_id,
active_incarnation_id,
active_epoch,
active_renewed_at_ms,
active_expires_at_ms,
authority_digest,
storage_type,
bytes,
)) = row
else {
return Ok(None);
};
let evidence_digest = identity_raw_evidence_digest(&bytes);
let authority = (|| -> Result<IdentityLeaseRecord, String> {
let epoch_highwater = sql_key_to_u64(&epoch_highwater, "identity lease epoch highwater")
.map_err(|error| error.to_string())?;
let active = match (
active_holder_id,
active_incarnation_id,
active_epoch,
active_renewed_at_ms,
active_expires_at_ms,
) {
(None, None, None, None, None) => None,
(Some(holder_id), Some(incarnation_id), Some(epoch), Some(renewed), Some(expires)) => {
Some(IdentityLeaseClaim {
holder_id,
incarnation_id,
epoch: sql_key_to_u64(&epoch, "identity lease active epoch")
.map_err(|error| error.to_string())?,
renewed_at_ms: sql_key_to_u64(&renewed, "identity lease renewed time")
.map_err(|error| error.to_string())?,
expires_at_ms: sql_key_to_u64(&expires, "identity lease expiry time")
.map_err(|error| error.to_string())?,
})
}
_ => {
return Err(
"identity lease normalized active-claim columns are partially populated"
.to_string(),
);
}
};
let record = IdentityLeaseRecord {
schema_version: IDENTITY_LEASE_SCHEMA_VERSION,
epoch_highwater,
active,
};
record.validate().map_err(|error| error.to_string())?;
let expected_digest = identity_lease_authority_digest(mob_id, identity, &record)
.map_err(|error| error.to_string())?;
if authority_digest != expected_digest {
return Err(
"identity lease normalized authority seal does not match its physical mob/identity key"
.to_string(),
);
}
Ok(record)
})();
let observation = match &authority {
Ok(authority) => classify_identity_blob(
&storage_type,
&bytes,
Some(IDENTITY_LEASE_SCHEMA_VERSION),
IdentityLeaseRecord::validate,
|record| {
if record == authority {
Ok(())
} else {
Err(
"identity lease projection differs from its sealed normalized authority"
.to_string(),
)
}
},
),
Err(detail) => IdentityStoredObservation::Malformed {
evidence_digest: evidence_digest.clone(),
detail: detail.clone(),
},
};
Ok(Some(LoadedIdentityLeaseRow { observation }))
}
fn query_identity_lease_observation(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityLeaseRecord>, MobStoreError> {
Ok(query_identity_lease_row(conn, mob_id, identity)?
.map_or(IdentityStoredObservation::Missing, |row| row.observation))
}
fn query_identity_lease_authority(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<Option<IdentityLeaseRecord>, MobStoreError> {
let Some(row) = query_identity_lease_row(conn, mob_id, identity)? else {
return Ok(None);
};
match row.observation {
IdentityStoredObservation::Valid(record) => Ok(Some(record)),
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => Err(identity_authority_blocked(
Some(evidence_digest),
format!("identity lease: {detail}"),
)),
IdentityStoredObservation::Missing => Err(identity_authority_blocked(
None,
"identity lease row disappeared while loading authority",
)),
}
}
enum IdentityStructuralEventLog {
Valid(Vec<MobEvent>),
Malformed {
evidence_digest: String,
detail: String,
},
}
fn query_identity_structural_event_log(
conn: &Connection,
mob_id: &MobId,
) -> Result<IdentityStructuralEventLog, MobStoreError> {
let mut statement = conn
.prepare(
"SELECT cursor, typeof(mob_id), CAST(mob_id AS BLOB),
typeof(event_json), event_json
FROM mob_events
WHERE mob_id = ?1
OR mob_id IS NULL
OR typeof(mob_id) <> 'text'
OR length(mob_id) = 0
OR trim(mob_id) <> mob_id
ORDER BY cursor",
)
.map_err(se)?;
let rows = statement
.query_map(params![mob_id.as_str()], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<Vec<u8>>>(2)?,
row.get::<_, String>(3)?,
row.get::<_, rusqlite::types::Value>(4)?,
))
})
.map_err(se)?;
let mut events = Vec::new();
let mut resettable_malformed: Option<(String, String)> = None;
let mut unordered_malformed: Option<(String, String)> = None;
for row in rows {
let (physical_cursor, route_type, route_bytes, storage_type, raw) = row.map_err(se)?;
let raw_bytes = match raw {
rusqlite::types::Value::Blob(bytes) => bytes,
rusqlite::types::Value::Text(text) => text.into_bytes(),
rusqlite::types::Value::Integer(value) => value.to_string().into_bytes(),
rusqlite::types::Value::Real(value) => value.to_string().into_bytes(),
rusqlite::types::Value::Null => Vec::new(),
};
let evidence_digest = identity_raw_evidence_digest(&raw_bytes);
let physical_route = match (route_type.as_str(), route_bytes) {
("text", Some(bytes)) => match String::from_utf8(bytes) {
Ok(route) if !route.is_empty() && route.trim() == route => Ok(Some(route)),
Ok(_) => Err("mob event row has a non-canonical physical mob route".to_string()),
Err(error) => Err(format!(
"mob event physical mob route is not UTF-8: {error}"
)),
},
("null", None) => Ok(None),
(observed, _) => Err(format!(
"mob event physical mob route uses SQLite storage class '{observed}', expected text or null"
)),
};
if physical_route
.as_ref()
.ok()
.and_then(|route| route.as_deref())
.is_some_and(|route| route != mob_id.as_str())
{
continue;
}
if !matches!(physical_route.as_ref(), Ok(Some(_)))
&& stored_event_envelope_mob_id(&raw_bytes)
.as_deref()
.is_some_and(|route| route != mob_id.as_str())
{
continue;
}
let physical_route = match physical_route {
Ok(route) => route,
Err(detail) => {
resettable_malformed.get_or_insert_with(|| (evidence_digest.clone(), detail));
None
}
};
let cursor = match u64::try_from(physical_cursor)
.ok()
.filter(|cursor| *cursor > 0)
{
Some(cursor) => cursor,
None => {
unordered_malformed.get_or_insert_with(|| {
(
evidence_digest,
format!("mob event row has invalid physical cursor {physical_cursor}"),
)
});
continue;
}
};
if storage_type != "blob" {
resettable_malformed.get_or_insert_with(|| {
(
evidence_digest,
format!(
"mob event row uses SQLite storage class '{storage_type}', expected blob"
),
)
});
continue;
}
let decoded = match decode_stored_mob_event(&raw_bytes) {
Ok(decoded) => decoded,
Err(error) => {
resettable_malformed.get_or_insert_with(|| {
(
evidence_digest,
format!("mob event row is malformed or unsupported: {error}"),
)
});
continue;
}
};
if decoded.mob_id != *mob_id {
if physical_route.is_some() {
resettable_malformed.get_or_insert_with(|| {
(
evidence_digest,
format!(
"mob event physical mob route '{}' differs from payload mob '{}'",
mob_id, decoded.mob_id
),
)
});
}
continue;
}
if decoded.cursor != cursor {
resettable_malformed.get_or_insert_with(|| {
(
evidence_digest,
format!(
"mob event row cursor mismatch: physical={cursor} payload={}",
decoded.cursor
),
)
});
continue;
}
if matches!(&decoded.kind, MobEventKind::MobReset) {
events.clear();
resettable_malformed = None;
}
events.push(decoded);
}
if let Some((evidence_digest, detail)) = unordered_malformed {
return Ok(IdentityStructuralEventLog::Malformed {
evidence_digest,
detail,
});
}
if let Some((evidence_digest, detail)) = resettable_malformed {
return Ok(IdentityStructuralEventLog::Malformed {
evidence_digest,
detail,
});
}
Ok(IdentityStructuralEventLog::Valid(events))
}
fn query_identity_member_target_state(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityMemberTargetState, MobStoreError> {
match query_identity_structural_event_log(conn, mob_id)? {
IdentityStructuralEventLog::Valid(events) => {
Ok(identity_member_target_state(&events, mob_id, identity))
}
IdentityStructuralEventLog::Malformed {
evidence_digest,
detail,
} => Ok(IdentityMemberTargetState {
observation: IdentityMemberTargetObservation::Malformed {
observed_version: Some(evidence_digest),
detail,
},
exact_current_spawn: None,
}),
}
}
fn query_identity_wiring_target_state(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityWiringTargetState, MobStoreError> {
match query_identity_structural_event_log(conn, mob_id)? {
IdentityStructuralEventLog::Valid(events) => {
Ok(identity_wiring_target_state(&events, mob_id, identity))
}
IdentityStructuralEventLog::Malformed {
evidence_digest,
detail,
} => Ok(IdentityWiringTargetState {
observation: IdentityWiringTargetObservation::Malformed {
observed_version: Some(evidence_digest),
detail,
},
}),
}
}
fn query_identity_receipt_observation(
conn: &Connection,
mob_id: &MobId,
subject: &IdentityOperationSubject,
slot: &IdentityOperationSlot,
) -> Result<IdentityStoredObservation<IdentityOperationReceipt>, MobStoreError> {
let (subject_kind, subject_id, slot_kind, slot_key_version, slot_digest) =
identity_receipt_sql_key(subject, slot)?;
let row: Option<(
String,
Vec<u8>,
String,
Vec<u8>,
Option<String>,
String,
String,
Vec<u8>,
)> = conn
.query_row(
"SELECT typeof(subject_json), CAST(subject_json AS BLOB),
typeof(slot_json), CAST(slot_json AS BLOB),
subject_identity, effect_kind,
typeof(receipt_json), CAST(receipt_json AS BLOB)
FROM mob_identity_operation_receipts
WHERE mob_id = ?1
AND subject_kind = ?2 AND subject_id = ?3
AND slot_kind = ?4 AND slot_key_version = ?5 AND slot_digest = ?6",
params![
mob_id.as_str(),
subject_kind,
subject_id,
slot_kind,
slot_key_version,
slot_digest,
],
|row| {
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)?,
))
},
)
.optional()
.map_err(se)?;
let Some((
subject_type,
stored_subject_key,
slot_type,
stored_slot_key,
subject_identity,
effect_kind,
receipt_type,
receipt_bytes,
)) = row
else {
return Ok(IdentityStoredObservation::Missing);
};
if subject_type != "blob"
|| slot_type != "blob"
|| stored_subject_key != serde_json::to_vec(subject).map_err(se)?
|| stored_slot_key != serde_json::to_vec(slot).map_err(se)?
{
return Ok(IdentityStoredObservation::Malformed {
evidence_digest: identity_raw_evidence_digest(&receipt_bytes),
detail: "identity operation receipt key is not canonical blob JSON".to_string(),
});
}
let expected_identity = identity_receipt_subject_identity(subject);
Ok(classify_identity_blob(
&receipt_type,
&receipt_bytes,
Some(IDENTITY_OPERATION_RECEIPT_SCHEMA_VERSION),
IdentityOperationReceipt::validate,
|receipt| {
if receipt.mob_id != *mob_id
|| receipt.subject != *subject
|| receipt.slot != *slot
|| receipt.effect_kind != slot.kind()
|| subject_identity.as_deref() != Some(expected_identity)
|| effect_kind != identity_operation_kind_key(receipt.effect_kind)
{
Err("identity operation receipt does not match its physical key".to_string())
} else {
Ok(())
}
},
))
}
fn query_identity_status_observation(
conn: &Connection,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityConvergenceStatus>, MobStoreError> {
let row: Option<(String, Vec<u8>)> = conn
.query_row(
"SELECT typeof(status_json), CAST(status_json AS BLOB)
FROM mob_identity_statuses
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), identity.as_str()],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()
.map_err(se)?;
let Some((storage_type, bytes)) = row else {
return Ok(IdentityStoredObservation::Missing);
};
Ok(classify_identity_blob(
&storage_type,
&bytes,
None,
|_| Ok(()),
|status: &IdentityConvergenceStatus| {
if status.identity == *identity {
Ok(())
} else {
Err("identity convergence status does not match its physical key".to_string())
}
},
))
}
fn write_identity_intent(
tx: &Transaction<'_>,
mob_id: &MobId,
identity: &AgentIdentity,
record: &IdentityIntentRecord,
) -> Result<(), MobStoreError> {
record.validate().map_err(identity_contract_error)?;
if record.mob_id != *mob_id || record.intent.identity() != identity {
return Err(identity_authority_blocked(
Some(identity_raw_evidence_digest(&encode_json(record)?)),
"identity intent write does not match its physical mob/identity key",
));
}
let (declaration_scope, session_id, lineage_id) = identity_intent_projection(record);
tx.execute(
"INSERT INTO mob_identity_intents
(mob_id, agent_identity, declaration_scope, session_id, lineage_id, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(mob_id, agent_identity) DO UPDATE SET
declaration_scope = excluded.declaration_scope,
session_id = excluded.session_id,
lineage_id = excluded.lineage_id,
record_json = excluded.record_json",
params![
mob_id.as_str(),
identity.as_str(),
declaration_scope,
session_id,
lineage_id,
encode_json(record)?,
],
)
.map_err(|error| sqlite_identity_write_error(error, "write identity intent"))?;
Ok(())
}
fn write_identity_lease(
tx: &Transaction<'_>,
mob_id: &MobId,
identity: &AgentIdentity,
record: &IdentityLeaseRecord,
) -> Result<(), MobStoreError> {
let record_json = encode_json(record)?;
let authority_digest = identity_lease_authority_digest(mob_id, identity, record)?;
let (active_holder_id, active_incarnation_id, active_epoch, active_renewed, active_expires) =
record
.active
.as_ref()
.map_or((None, None, None, None, None), |active| {
(
Some(active.holder_id.as_str()),
Some(active.incarnation_id.as_str()),
Some(u64_to_sql_key(active.epoch)),
Some(u64_to_sql_key(active.renewed_at_ms)),
Some(u64_to_sql_key(active.expires_at_ms)),
)
});
tx.execute(
"INSERT INTO mob_identity_leases
(mob_id, agent_identity, epoch_highwater, active_holder_id,
active_incarnation_id, active_epoch, active_renewed_at_ms,
active_expires_at_ms, authority_digest, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT(mob_id, agent_identity) DO UPDATE SET
epoch_highwater = excluded.epoch_highwater,
active_holder_id = excluded.active_holder_id,
active_incarnation_id = excluded.active_incarnation_id,
active_epoch = excluded.active_epoch,
active_renewed_at_ms = excluded.active_renewed_at_ms,
active_expires_at_ms = excluded.active_expires_at_ms,
authority_digest = excluded.authority_digest,
record_json = excluded.record_json",
params![
mob_id.as_str(),
identity.as_str(),
u64_to_sql_key(record.epoch_highwater),
active_holder_id,
active_incarnation_id,
active_epoch,
active_renewed,
active_expires,
authority_digest,
record_json,
],
)
.map_err(|error| sqlite_identity_write_error(error, "write identity lease"))?;
Ok(())
}
fn insert_identity_receipt(
tx: &Transaction<'_>,
receipt: &IdentityOperationReceipt,
) -> Result<(), MobStoreError> {
let (_, _, subject_key, slot_key) = identity_receipt_keys(&receipt.subject, &receipt.slot)?;
let (subject_kind, subject_id, slot_kind, slot_key_version, slot_digest) =
identity_receipt_sql_key(&receipt.subject, &receipt.slot)?;
let subject_identity = identity_receipt_subject_identity(&receipt.subject);
tx.execute(
"INSERT INTO mob_identity_operation_receipts
(mob_id, subject_kind, subject_id, slot_kind, slot_key_version, slot_digest,
subject_json, slot_json, subject_identity, effect_kind,
receipt_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
params![
receipt.mob_id.as_str(),
subject_kind,
subject_id,
slot_kind,
slot_key_version,
slot_digest,
subject_key,
slot_key,
subject_identity,
identity_operation_kind_key(receipt.effect_kind),
encode_json(receipt)?,
],
)
.map_err(|error| sqlite_identity_write_error(error, "insert identity operation receipt"))?;
Ok(())
}
#[async_trait]
impl MobIdentityStore for SqliteMobIdentityStore {
#[cfg(feature = "runtime-adapter")]
fn prepare_runtime_write_fence(
&self,
permit: IdentityActuationPermit,
expected_session: DesiredSessionTarget,
observed: &meerkat_runtime::RuntimeSessionLifecycleObservation,
) -> Result<Arc<dyn meerkat_runtime::RuntimeStoreWriteFence>, MobStoreError> {
validate_identity_runtime_target_binding(&permit, &expected_session, observed)?;
Ok(Arc::new(SqliteIdentityRuntimeWriteFence {
path: self.path.clone(),
store_instance_id: self.store_instance_id.clone(),
clock: Arc::clone(&self.clock),
permit,
expected_session,
}))
}
async fn observe_identity_intent(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityIntentRecord>, MobStoreError> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
query_identity_intent_observation(&conn, &mob_id, &identity)
})
.await
}
async fn list_identity_intents(
&self,
mob_id: &MobId,
) -> Result<
BTreeMap<AgentIdentity, IdentityStoredObservation<IdentityIntentRecord>>,
MobStoreError,
> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
let mut stmt = conn
.prepare(
"SELECT typeof(agent_identity), CAST(agent_identity AS BLOB),
typeof(declaration_scope), CAST(declaration_scope AS BLOB),
typeof(session_id), CAST(session_id AS BLOB),
typeof(lineage_id), CAST(lineage_id AS BLOB),
typeof(record_json), CAST(record_json AS BLOB)
FROM mob_identity_intents
WHERE mob_id = ?1
ORDER BY agent_identity",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<Vec<u8>>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, Option<Vec<u8>>>(5)?,
row.get::<_, String>(6)?,
row.get::<_, Option<Vec<u8>>>(7)?,
row.get::<_, String>(8)?,
row.get::<_, Vec<u8>>(9)?,
))
})
.map_err(se)?;
let mut observations = BTreeMap::new();
for row in rows {
let (
identity_type,
identity_bytes,
declaration_scope_type,
declaration_scope_bytes,
session_id_type,
session_id_bytes,
lineage_id_type,
lineage_id_bytes,
storage_type,
bytes,
) = row.map_err(se)?;
let identity_key = decode_required_identity_text(
&identity_type,
identity_bytes.clone(),
"agent_identity",
);
let identity_key_error = identity_key.as_ref().err().cloned();
let identity = AgentIdentity::from(identity_key.unwrap_or_else(|_| {
format!(
"malformed-physical-identity:{identity_type}:{}",
identity_raw_evidence_digest(&identity_bytes)
)
}));
let declaration_scope = decode_optional_identity_text(
&declaration_scope_type,
declaration_scope_bytes,
"identity intent declaration_scope",
);
let session_id = decode_optional_identity_text(
&session_id_type,
session_id_bytes,
"identity intent session_id",
);
let lineage_id = decode_optional_identity_text(
&lineage_id_type,
lineage_id_bytes,
"identity intent lineage_id",
);
let projection_error = identity_key_error
.or_else(|| declaration_scope.as_ref().err().cloned())
.or_else(|| session_id.as_ref().err().cloned())
.or_else(|| lineage_id.as_ref().err().cloned());
let declaration_scope = declaration_scope.unwrap_or(None);
let session_id = session_id.unwrap_or(None);
let lineage_id = lineage_id.unwrap_or(None);
let observation = classify_identity_blob(
&storage_type,
&bytes,
Some(IDENTITY_INTENT_SCHEMA_VERSION),
IdentityIntentRecord::validate,
|record| {
if let Some(detail) = projection_error {
return Err(detail);
}
identity_intent_physical_key_matches(
record,
&mob_id,
&identity,
declaration_scope.as_deref(),
session_id.as_deref(),
lineage_id.as_deref(),
)
},
);
observations.insert(identity, observation);
}
Ok(observations)
})
.await
}
async fn scan_identity_intents_page(
&self,
mob_id: &MobId,
after: Option<&super::IdentityIntentScanCursor>,
limit: usize,
) -> Result<super::IdentityIntentScanPage, MobStoreError> {
if limit == 0 || limit > super::IDENTITY_INTENT_SCAN_PAGE_MAX {
return Err(MobStoreError::Internal(format!(
"identity intent scan limit must be in 1..={}, got {limit}",
super::IDENTITY_INTENT_SCAN_PAGE_MAX
)));
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let after = after.cloned();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
let after_storage_class = after
.as_ref()
.map(super::IdentityIntentScanCursor::storage_class);
let after_identity = after
.as_ref()
.map(super::IdentityIntentScanCursor::physical_identity);
let fetch_limit = i64::try_from(limit.saturating_add(1)).map_err(|_| {
MobStoreError::Internal("identity intent scan limit overflow".to_string())
})?;
let mut stmt = conn
.prepare(
"SELECT typeof(agent_identity), CAST(agent_identity AS BLOB),
typeof(declaration_scope), CAST(declaration_scope AS BLOB),
typeof(session_id), CAST(session_id AS BLOB),
typeof(lineage_id), CAST(lineage_id AS BLOB),
typeof(record_json), CAST(record_json AS BLOB)
FROM mob_identity_intents
WHERE mob_id = ?1
AND (
?2 IS NULL
OR typeof(agent_identity) > ?2
OR (
typeof(agent_identity) = ?2
AND CAST(agent_identity AS BLOB) > ?3
)
)
ORDER BY typeof(agent_identity), CAST(agent_identity AS BLOB)
LIMIT ?4",
)
.map_err(se)?;
let rows = stmt
.query_map(
params![
mob_id.as_str(),
after_storage_class,
after_identity,
fetch_limit
],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, String>(2)?,
row.get::<_, Option<Vec<u8>>>(3)?,
row.get::<_, String>(4)?,
row.get::<_, Option<Vec<u8>>>(5)?,
row.get::<_, String>(6)?,
row.get::<_, Option<Vec<u8>>>(7)?,
row.get::<_, String>(8)?,
row.get::<_, Vec<u8>>(9)?,
))
},
)
.map_err(se)?;
let mut selected = rows.collect::<Result<Vec<_>, _>>().map_err(se)?;
let has_more = selected.len() > limit;
if has_more {
selected.pop();
}
let next = if has_more {
let Some((identity_type, identity_bytes, ..)) = selected.last() else {
return Err(MobStoreError::Internal(
"identity intent scan continuation has no returned row".to_string(),
));
};
Some(super::IdentityIntentScanCursor::new(
identity_type.clone(),
identity_bytes.clone(),
))
} else {
None
};
let mut observations = BTreeMap::new();
for (
identity_type,
identity_bytes,
declaration_scope_type,
declaration_scope_bytes,
session_id_type,
session_id_bytes,
lineage_id_type,
lineage_id_bytes,
storage_type,
bytes,
) in selected
{
let identity_key = decode_required_identity_text(
&identity_type,
identity_bytes.clone(),
"agent_identity",
);
let identity_key_error = identity_key.as_ref().err().cloned();
let identity = AgentIdentity::from(identity_key.unwrap_or_else(|_| {
format!(
"malformed-physical-identity:{identity_type}:{}",
identity_raw_evidence_digest(&identity_bytes)
)
}));
let declaration_scope = decode_optional_identity_text(
&declaration_scope_type,
declaration_scope_bytes,
"identity intent declaration_scope",
);
let session_id = decode_optional_identity_text(
&session_id_type,
session_id_bytes,
"identity intent session_id",
);
let lineage_id = decode_optional_identity_text(
&lineage_id_type,
lineage_id_bytes,
"identity intent lineage_id",
);
let projection_error = identity_key_error
.or_else(|| declaration_scope.as_ref().err().cloned())
.or_else(|| session_id.as_ref().err().cloned())
.or_else(|| lineage_id.as_ref().err().cloned());
let declaration_scope = declaration_scope.unwrap_or(None);
let session_id = session_id.unwrap_or(None);
let lineage_id = lineage_id.unwrap_or(None);
let observation = classify_identity_blob(
&storage_type,
&bytes,
Some(IDENTITY_INTENT_SCHEMA_VERSION),
IdentityIntentRecord::validate,
|record| {
if let Some(detail) = projection_error {
return Err(detail);
}
identity_intent_physical_key_matches(
record,
&mob_id,
&identity,
declaration_scope.as_deref(),
session_id.as_deref(),
lineage_id.as_deref(),
)
},
);
observations.insert(identity, observation);
}
Ok(super::IdentityIntentScanPage { observations, next })
})
.await
}
async fn observe_identity_lease(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityLeaseRecord>, MobStoreError> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
query_identity_lease_observation(&conn, &mob_id, &identity)
})
.await
}
async fn claim_or_renew_identity_lease(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
holder_id: &str,
incarnation_id: &str,
ttl_ms: u64,
) -> Result<IdentityLeaseClaimOutcome, MobStoreError> {
validate_identity_store_text("mob_id", mob_id.as_str())?;
validate_identity_store_text("identity", identity.as_str())?;
validate_identity_store_text("holder_id", holder_id)?;
validate_identity_store_text("incarnation_id", incarnation_id)?;
if ttl_ms == 0 || ttl_ms > IDENTITY_LEASE_MAX_TTL_MS {
return Err(identity_contract_error(
IdentityIntentError::InvalidLeaseLifetime,
));
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let clock = Arc::clone(&self.clock);
let mob_id = mob_id.clone();
let identity = identity.clone();
let holder_id = holder_id.to_string();
let incarnation_id = incarnation_id.to_string();
run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
let observed_at_ms = clock.now_ms()?;
let expires_at_ms = observed_at_ms.checked_add(ttl_ms).ok_or_else(|| {
MobStoreError::WriteFailed("identity lease expiry overflow".to_string())
})?;
let current = query_identity_lease_authority(&tx, &mob_id, &identity)?;
if let Some(record) = ¤t
&& let Some(active) = &record.active
&& observed_at_ms < active.expires_at_ms
{
if active.holder_id != holder_id || active.incarnation_id != incarnation_id {
return Ok(IdentityLeaseClaimOutcome::HeldByOther(active.clone()));
}
if observed_at_ms < active.renewed_at_ms {
return Err(MobStoreError::WriteFailed(
"identity lease clock regressed before the last renewal".to_string(),
));
}
let claim = IdentityLeaseClaim {
holder_id,
incarnation_id,
epoch: active.epoch,
renewed_at_ms: observed_at_ms,
expires_at_ms,
};
let record = IdentityLeaseRecord {
schema_version: IDENTITY_LEASE_SCHEMA_VERSION,
epoch_highwater: record.epoch_highwater,
active: Some(claim.clone()),
};
record.validate().map_err(identity_contract_error)?;
write_identity_lease(&tx, &mob_id, &identity, &record)?;
tx.commit().map_err(se)?;
return Ok(IdentityLeaseClaimOutcome::Renewed(claim));
}
let epoch = current
.as_ref()
.map_or(0, |record| record.epoch_highwater)
.checked_add(1)
.ok_or_else(|| MobStoreError::IdentityCounterExhausted {
counter: "identity lease epoch".to_string(),
})?;
let claim = IdentityLeaseClaim {
holder_id,
incarnation_id,
epoch,
renewed_at_ms: observed_at_ms,
expires_at_ms,
};
let record = IdentityLeaseRecord {
schema_version: IDENTITY_LEASE_SCHEMA_VERSION,
epoch_highwater: epoch,
active: Some(claim.clone()),
};
record.validate().map_err(identity_contract_error)?;
write_identity_lease(&tx, &mob_id, &identity, &record)?;
tx.commit().map_err(se)?;
Ok(IdentityLeaseClaimOutcome::Acquired(claim))
})
.await
}
async fn release_identity_lease(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
expected: &IdentityLeaseClaim,
) -> Result<bool, MobStoreError> {
let expected_record = IdentityLeaseRecord {
schema_version: IDENTITY_LEASE_SCHEMA_VERSION,
epoch_highwater: expected.epoch,
active: Some(expected.clone()),
};
expected_record
.validate()
.map_err(identity_contract_error)?;
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
let expected = expected.clone();
run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
let Some(current) = query_identity_lease_authority(&tx, &mob_id, &identity)? else {
return Ok(false);
};
if current.active.as_ref() != Some(&expected) {
return Ok(false);
}
let released = IdentityLeaseRecord {
schema_version: IDENTITY_LEASE_SCHEMA_VERSION,
epoch_highwater: current.epoch_highwater,
active: None,
};
released.validate().map_err(identity_contract_error)?;
write_identity_lease(&tx, &mob_id, &identity, &released)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
async fn validate_identity_actuation_permit(
&self,
permit: &IdentityActuationPermit,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let clock = Arc::clone(&self.clock);
let permit = permit.clone();
run_sqlite_task(move || {
let observed_at_ms = clock.now_ms()?;
permit.validate_for_write(observed_at_ms).map_err(|error| {
MobStoreError::CasConflict(format!(
"identity actuation permit is no longer current: {error}"
))
})?;
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
let intent = require_identity_authority(
query_identity_intent_observation(&conn, &permit.mob_id, &permit.identity)?,
"identity intent",
)?
.ok_or_else(|| {
MobStoreError::CasConflict(
"identity actuation observed no current intent".to_string(),
)
})?;
if intent.intent_revision != permit.intent_revision
|| intent.intent_digest != permit.intent_digest
|| intent.authority_digest != permit.intent_authority_digest
{
return Err(MobStoreError::CasConflict(
"identity actuation intent authority is stale".to_string(),
));
}
let lease = query_identity_lease_authority(&conn, &permit.mob_id, &permit.identity)?
.ok_or_else(|| {
MobStoreError::CasConflict(
"identity actuation observed no current lease".to_string(),
)
})?;
let active = lease.active.ok_or_else(|| {
MobStoreError::CasConflict(
"identity actuation observed no active lease".to_string(),
)
})?;
if active.epoch != permit.lease_epoch
|| active.holder_id != permit.lease_holder_id
|| active.incarnation_id != permit.lease_incarnation_id
|| active.expires_at_ms != permit.lease_expires_at_ms
|| observed_at_ms >= active.expires_at_ms
{
return Err(MobStoreError::CasConflict(
"identity actuation lease authority is stale".to_string(),
));
}
Ok(())
})
.await
}
async fn observe_identity_operation_receipt(
&self,
mob_id: &MobId,
subject: &IdentityOperationSubject,
slot: &IdentityOperationSlot,
) -> Result<IdentityStoredObservation<IdentityOperationReceipt>, MobStoreError> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let subject = subject.clone();
let slot = slot.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
query_identity_receipt_observation(&conn, &mob_id, &subject, &slot)
})
.await
}
async fn insert_identity_operation_receipt_if_absent(
&self,
receipt: &IdentityOperationReceipt,
permit: &IdentityActuationPermit,
) -> Result<IdentityOperationReceiptInsertOutcome, MobStoreError> {
receipt.validate().map_err(identity_contract_error)?;
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let clock = Arc::clone(&self.clock);
let receipt = receipt.clone();
let permit = permit.clone();
run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
if let Some(existing) = require_identity_authority(
query_identity_receipt_observation(
&tx,
&receipt.mob_id,
&receipt.subject,
&receipt.slot,
)?,
"identity operation receipt",
)? {
return Ok(if existing.request_digest == receipt.request_digest {
IdentityOperationReceiptInsertOutcome::ExistingExact(existing)
} else {
IdentityOperationReceiptInsertOutcome::Conflict(existing)
});
}
let observed_at_ms = clock.now_ms()?;
permit.validate_for_write(observed_at_ms).map_err(|error| {
MobStoreError::CasConflict(format!(
"identity receipt insertion permit is no longer current: {error}"
))
})?;
let IdentityOperationSubject::Identity { identity } = &receipt.subject;
if receipt.mob_id != permit.mob_id
|| identity != &permit.identity
|| identity_receipt_target(&receipt) != Some(permit.target)
|| receipt.intent_revision != Some(permit.intent_revision)
|| receipt.intent_digest.as_ref() != Some(&permit.intent_digest)
|| receipt.intent_authority_digest.as_ref() != Some(&permit.intent_authority_digest)
|| !matches!(
permit.target_observation,
IdentityTargetObservationVersion::InsertIfAbsent
)
{
return Err(MobStoreError::CasConflict(
"identity receipt insertion permit does not match the immutable receipt slot"
.to_string(),
));
}
let intent = require_identity_authority(
query_identity_intent_observation(&tx, &receipt.mob_id, identity)?,
"identity intent",
)?
.ok_or_else(|| {
MobStoreError::CasConflict(
"identity receipt insertion observed no current intent".to_string(),
)
})?;
if intent.intent_revision != permit.intent_revision
|| intent.intent_digest != permit.intent_digest
|| intent.authority_digest != permit.intent_authority_digest
|| !identity_actuator_receipt_matches_intent(&receipt, &intent)
|| receipt
.audit_lease_epoch
.is_some_and(|epoch| epoch != permit.lease_epoch)
{
return Err(MobStoreError::CasConflict(
"identity receipt insertion intent authority is stale".to_string(),
));
}
let lease = query_identity_lease_authority(&tx, &receipt.mob_id, identity)?
.ok_or_else(|| {
MobStoreError::CasConflict(
"identity receipt insertion observed no current lease".to_string(),
)
})?;
let active = lease.active.ok_or_else(|| {
MobStoreError::CasConflict(
"identity receipt insertion observed no active lease".to_string(),
)
})?;
if active.epoch != permit.lease_epoch
|| active.holder_id != permit.lease_holder_id
|| active.incarnation_id != permit.lease_incarnation_id
|| active.expires_at_ms != permit.lease_expires_at_ms
|| observed_at_ms >= active.expires_at_ms
{
return Err(MobStoreError::CasConflict(
"identity receipt insertion lease authority is stale".to_string(),
));
}
insert_identity_receipt(&tx, &receipt)?;
tx.commit().map_err(se)?;
Ok(IdentityOperationReceiptInsertOutcome::Inserted(receipt))
})
.await
}
}
#[async_trait]
impl MobIdentityMemberStore for SqliteMobIdentityStore {
async fn observe_identity_member_target(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityMemberTargetObservation, MobStoreError> {
if self.event_bus.is_none() {
return Err(MobStoreError::IdentityMemberAtomicPersistenceUnavailable);
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
Ok(query_identity_member_target_state(&conn, &mob_id, &identity)?.observation)
})
.await
}
async fn commit_identity_member_spawned(
&self,
permit: &IdentityActuationPermit,
event: &NewMobEvent,
) -> Result<IdentityMemberEventCommitOutcome, MobStoreError> {
let Some(event_bus) = self.event_bus.clone() else {
return Err(MobStoreError::IdentityMemberAtomicPersistenceUnavailable);
};
if let Err(error) = validate_mob_event_write_authority(&event.kind) {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: error.to_string(),
});
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let clock = Arc::clone(&self.clock);
let permit = permit.clone();
let event = event.clone();
let result = run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
let observed_at_ms = match clock.now_ms() {
Ok(observed_at_ms) => observed_at_ms,
Err(error) => {
return Ok(IdentityMemberEventCommitOutcome::Backoff {
detail: error.to_string(),
});
}
};
let proposed = MobEvent {
cursor: 0,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let intent =
match query_identity_intent_observation(&tx, &permit.mob_id, &permit.identity)? {
IdentityStoredObservation::Valid(intent) => intent,
IdentityStoredObservation::Missing => {
return Ok(IdentityMemberEventCommitOutcome::Conflict {
current: None,
detail: "identity member commit observed no current intent".to_string(),
});
}
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: Some(evidence_digest),
detail: format!(
"identity member commit observed unsafe intent authority: {detail}"
),
});
}
};
let lease =
match query_identity_lease_observation(&tx, &permit.mob_id, &permit.identity)? {
IdentityStoredObservation::Valid(lease) => lease,
IdentityStoredObservation::Missing => {
return Ok(IdentityMemberEventCommitOutcome::Conflict {
current: None,
detail: "identity member commit observed no current lease".to_string(),
});
}
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: Some(evidence_digest),
detail: format!(
"identity member commit observed unsafe lease authority: {detail}"
),
});
}
};
if let Err(outcome) = validate_identity_member_commit_authority(
&permit,
&intent,
&lease,
&proposed,
observed_at_ms,
) {
return Ok(outcome);
}
let current =
query_identity_member_target_state(&tx, &permit.mob_id, &permit.identity)?;
if let IdentityMemberTargetObservation::Malformed {
observed_version,
detail,
} = ¤t.observation
{
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: observed_version.clone(),
detail: detail.clone(),
});
}
let expected = current.observation.target_precondition();
if expected.as_ref() != Some(&permit.target_observation) {
tx.commit().map_err(se)?;
return Ok(IdentityMemberEventCommitOutcome::Conflict {
current: Some(current.observation),
detail: "identity member target observation is stale".to_string(),
});
}
if let Some(existing) = current.exact_current_spawn
&& existing.kind == proposed.kind
{
tx.commit().map_err(se)?;
return Ok(IdentityMemberEventCommitOutcome::AlreadyExact { event: existing });
}
if !matches!(
permit.target_observation,
IdentityTargetObservationVersion::Absent { .. }
) {
tx.commit().map_err(se)?;
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: current
.observation
.target_precondition()
.and_then(|target| match target {
IdentityTargetObservationVersion::Version { version } => Some(version),
_ => None,
}),
detail: "MemberSpawned append requires an exact absent-target witness"
.to_string(),
});
}
let cursor = next_event_cursor(&tx)?;
if cursor == 0 || cursor >= i64::MAX as u64 {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: "mob event cursor is exhausted".to_string(),
});
}
let stored = MobEvent {
cursor,
timestamp: proposed.timestamp,
mob_id: proposed.mob_id,
kind: proposed.kind,
};
let encoded = match encode_stored_mob_event(&stored) {
Ok(encoded) => encoded,
Err(error) => {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: format!("MemberSpawned event is not serializable: {error}"),
});
}
};
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
let Some(next_cursor) = cursor.checked_add(1) else {
return Ok(IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: "mob event cursor is exhausted".to_string(),
});
};
set_next_cursor(&tx, next_cursor)?;
tx.commit().map_err(se)?;
Ok(IdentityMemberEventCommitOutcome::Applied { event: stored })
})
.await;
let outcome = match result {
Ok(outcome) => outcome,
Err(MobStoreError::IdentityAuthorityBlocked {
evidence_digest,
detail,
}) => IdentityMemberEventCommitOutcome::RepairBlocked {
evidence_digest,
detail,
},
Err(error) => IdentityMemberEventCommitOutcome::Backoff {
detail: error.to_string(),
},
};
if let IdentityMemberEventCommitOutcome::Applied { event } = &outcome {
event_bus.publish_committed(event.clone());
}
Ok(outcome)
}
async fn observe_identity_wiring_target(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityWiringTargetObservation, MobStoreError> {
if self.event_bus.is_none() {
return Err(MobStoreError::IdentityMemberAtomicPersistenceUnavailable);
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
Ok(query_identity_wiring_target_state(&conn, &mob_id, &identity)?.observation)
})
.await
}
async fn commit_identity_wiring_event(
&self,
permit: &IdentityActuationPermit,
event: &NewMobEvent,
) -> Result<IdentityWiringEventCommitOutcome, MobStoreError> {
let Some(event_bus) = self.event_bus.clone() else {
return Err(MobStoreError::IdentityMemberAtomicPersistenceUnavailable);
};
if let Err(error) = validate_mob_event_write_authority(&event.kind) {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: error.to_string(),
});
}
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let clock = Arc::clone(&self.clock);
let permit = permit.clone();
let event = event.clone();
let result = run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
let observed_at_ms = match clock.now_ms() {
Ok(observed_at_ms) => observed_at_ms,
Err(error) => {
return Ok(IdentityWiringEventCommitOutcome::Backoff {
detail: error.to_string(),
});
}
};
let proposed = MobEvent {
cursor: 0,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let intent =
match query_identity_intent_observation(&tx, &permit.mob_id, &permit.identity)? {
IdentityStoredObservation::Valid(intent) => intent,
IdentityStoredObservation::Missing => {
return Ok(IdentityWiringEventCommitOutcome::Conflict {
current: None,
detail: "identity wiring commit observed no current intent".to_string(),
});
}
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: Some(evidence_digest),
detail: format!(
"identity wiring commit observed unsafe intent authority: {detail}"
),
});
}
};
let lease =
match query_identity_lease_observation(&tx, &permit.mob_id, &permit.identity)? {
IdentityStoredObservation::Valid(lease) => lease,
IdentityStoredObservation::Missing => {
return Ok(IdentityWiringEventCommitOutcome::Conflict {
current: None,
detail: "identity wiring commit observed no current lease".to_string(),
});
}
IdentityStoredObservation::Unsupported {
evidence_digest,
detail,
}
| IdentityStoredObservation::Malformed {
evidence_digest,
detail,
} => {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: Some(evidence_digest),
detail: format!(
"identity wiring commit observed unsafe lease authority: {detail}"
),
});
}
};
let (edge, adding) = match validate_identity_wiring_commit_authority(
&permit,
&intent,
&lease,
&proposed,
observed_at_ms,
) {
Ok(validated) => validated,
Err(outcome) => return Ok(outcome),
};
let current =
query_identity_wiring_target_state(&tx, &permit.mob_id, &permit.identity)?;
if let IdentityWiringTargetObservation::Malformed {
observed_version,
detail,
} = ¤t.observation
{
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: observed_version.clone(),
detail: detail.clone(),
});
}
if current.observation.target_precondition().as_ref()
!= Some(&permit.target_observation)
{
tx.commit().map_err(se)?;
return Ok(IdentityWiringEventCommitOutcome::Conflict {
current: Some(current.observation),
detail: "identity wiring target observation is stale".to_string(),
});
}
if current.observation.contains(&edge) == adding {
tx.commit().map_err(se)?;
return Ok(IdentityWiringEventCommitOutcome::AlreadyExact {
current: current.observation,
});
}
let cursor = next_event_cursor(&tx)?;
if cursor == 0 || cursor >= i64::MAX as u64 {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: "mob event cursor is exhausted".to_string(),
});
}
let stored = MobEvent {
cursor,
timestamp: proposed.timestamp,
mob_id: proposed.mob_id,
kind: proposed.kind,
};
let encoded = match encode_stored_mob_event(&stored) {
Ok(encoded) => encoded,
Err(error) => {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: format!("identity wiring event is not serializable: {error}"),
});
}
};
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
let Some(next_cursor) = cursor.checked_add(1) else {
return Ok(IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest: None,
detail: "mob event cursor is exhausted".to_string(),
});
};
set_next_cursor(&tx, next_cursor)?;
tx.commit().map_err(se)?;
Ok(IdentityWiringEventCommitOutcome::Applied { event: stored })
})
.await;
let outcome = match result {
Ok(outcome) => outcome,
Err(MobStoreError::IdentityAuthorityBlocked {
evidence_digest,
detail,
}) => IdentityWiringEventCommitOutcome::RepairBlocked {
evidence_digest,
detail,
},
Err(error) => IdentityWiringEventCommitOutcome::Backoff {
detail: error.to_string(),
},
};
if let IdentityWiringEventCommitOutcome::Applied { event } = &outcome {
event_bus.publish_committed(event.clone());
}
Ok(outcome)
}
}
#[async_trait]
impl MobIdentityStatusStore for SqliteMobIdentityStatusStore {
async fn load_identity_convergence_status(
&self,
mob_id: &MobId,
identity: &AgentIdentity,
) -> Result<IdentityStoredObservation<IdentityConvergenceStatus>, MobStoreError> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let identity = identity.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
query_identity_status_observation(&conn, &mob_id, &identity)
})
.await
}
async fn list_identity_convergence_statuses(
&self,
mob_id: &MobId,
) -> Result<
BTreeMap<AgentIdentity, IdentityStoredObservation<IdentityConvergenceStatus>>,
MobStoreError,
> {
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_existing_identity_read_connection(&path, &store_instance_id)?;
let mut stmt = conn
.prepare(
"SELECT agent_identity, typeof(status_json), CAST(status_json AS BLOB)
FROM mob_identity_statuses
WHERE mob_id = ?1
ORDER BY agent_identity",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Vec<u8>>(2)?,
))
})
.map_err(se)?;
let mut observations = BTreeMap::new();
for row in rows {
let (identity, storage_type, bytes) = row.map_err(se)?;
let identity = AgentIdentity::from(identity);
let observation = classify_identity_blob(
&storage_type,
&bytes,
None,
|_| Ok(()),
|status: &IdentityConvergenceStatus| {
if status.identity == identity {
Ok(())
} else {
Err(
"identity convergence status does not match its physical key"
.to_string(),
)
}
},
);
observations.insert(identity, observation);
}
Ok(observations)
})
.await
}
async fn replace_identity_convergence_status(
&self,
mob_id: &MobId,
status: &IdentityConvergenceStatus,
) -> Result<(), MobStoreError> {
validate_identity_store_text("mob_id", mob_id.as_str())?;
validate_identity_store_text("identity", status.identity.as_str())?;
let path = self.path.clone();
let store_instance_id = self.store_instance_id.clone();
let mob_id = mob_id.clone();
let status = status.clone();
run_sqlite_task(move || {
let mut conn = open_existing_identity_write_connection(&path, &store_instance_id)?;
let tx = begin_identity_immediate(&mut conn, &store_instance_id)?;
tx.execute(
"INSERT INTO mob_identity_statuses (mob_id, agent_identity, status_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(mob_id, agent_identity) DO UPDATE SET
status_json = excluded.status_json",
params![
mob_id.as_str(),
status.identity.as_str(),
encode_json(&status)?,
],
)
.map_err(|error| {
sqlite_identity_write_error(error, "replace identity convergence status")
})?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
}
#[derive(Clone)]
pub struct SqliteMobRuntimeMetadataStore {
path: PathBuf,
}
impl std::fmt::Debug for SqliteMobRuntimeMetadataStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteMobRuntimeMetadataStore")
.field("path", &self.path)
.finish()
}
}
#[async_trait]
impl MobRuntimeMetadataStore for SqliteMobRuntimeMetadataStore {
async fn load_supervisor_authority(
&self,
mob_id: &MobId,
) -> Result<Option<SupervisorAuthorityRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let row: Option<Vec<u8>> = conn
.query_row(
"SELECT record_json FROM mob_runtime_supervisors WHERE mob_id = ?1",
params![mob_id.as_str()],
|row| row.get(0),
)
.optional()
.map_err(se)?;
row.map(|bytes| decode_json(&bytes)).transpose()
})
.await
}
async fn put_supervisor_authority(
&self,
mob_id: &MobId,
record: &SupervisorAuthorityRecord,
authority: &SupervisorAuthorityPersistenceAuthority,
) -> Result<(), MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_supervisors (mob_id, record_json) VALUES (?1, ?2)
ON CONFLICT(mob_id) DO UPDATE SET record_json = excluded.record_json",
params![mob_id.as_str(), encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn compare_and_put_supervisor_authority(
&self,
mob_id: &MobId,
expected: &SupervisorAuthorityRecord,
record: &SupervisorAuthorityRecord,
authority: &SupervisorAuthorityPersistenceAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"UPDATE mob_runtime_supervisors
SET record_json = ?2
WHERE mob_id = ?1 AND record_json = ?3",
params![
mob_id.as_str(),
encode_json(&record)?,
encode_json(&expected)?
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn put_supervisor_authority_if_absent(
&self,
mob_id: &MobId,
record: &SupervisorAuthorityRecord,
authority: &SupervisorAuthorityPersistenceAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"INSERT OR IGNORE INTO mob_runtime_supervisors (mob_id, record_json) VALUES (?1, ?2)",
params![mob_id.as_str(), encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn delete_supervisor_authority(
&self,
mob_id: &MobId,
expected: &SupervisorAuthorityRecord,
authority: &SupervisorAuthorityDeletionAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(expected)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_supervisors WHERE mob_id = ?1 AND record_json = ?2",
params![mob_id.as_str(), encode_json(&expected)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn load_mob_host_authority(
&self,
mob_id: &MobId,
host_id: &str,
) -> Result<Option<MobHostAuthorityRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let host_id = host_id.to_string();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let row: Option<Vec<u8>> = conn
.query_row(
"SELECT record_json FROM mob_runtime_host_authorities
WHERE mob_id = ?1 AND host_id = ?2",
params![mob_id.as_str(), host_id],
|row| row.get(0),
)
.optional()
.map_err(se)?;
row.map(|bytes| decode_json(&bytes)).transpose()
})
.await
}
async fn list_mob_host_authorities(
&self,
mob_id: &MobId,
) -> Result<Vec<MobHostAuthorityRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT record_json FROM mob_runtime_host_authorities
WHERE mob_id = ?1
ORDER BY host_id",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
records.push(decode_json(&bytes)?);
}
Ok(records)
})
.await
}
async fn put_mob_host_authority(
&self,
mob_id: &MobId,
record: &MobHostAuthorityRecord,
authority: &MobHostAuthorityPersistenceAuthority,
) -> Result<(), MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_host_authorities (mob_id, host_id, record_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(mob_id, host_id) DO UPDATE SET record_json = excluded.record_json",
params![mob_id.as_str(), record.host_id, encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn put_mob_host_authority_if_absent(
&self,
mob_id: &MobId,
record: &MobHostAuthorityRecord,
authority: &MobHostAuthorityPersistenceAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"INSERT OR IGNORE INTO mob_runtime_host_authorities (mob_id, host_id, record_json)
VALUES (?1, ?2, ?3)",
params![mob_id.as_str(), record.host_id, encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn compare_and_put_mob_host_authority(
&self,
mob_id: &MobId,
expected: &MobHostAuthorityRecord,
record: &MobHostAuthorityRecord,
authority: &MobHostAuthorityPersistenceAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"UPDATE mob_runtime_host_authorities
SET record_json = ?3
WHERE mob_id = ?1 AND host_id = ?2 AND record_json = ?4",
params![
mob_id.as_str(),
record.host_id,
encode_json(&record)?,
encode_json(&expected)?
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn delete_mob_host_authority(
&self,
mob_id: &MobId,
expected: &MobHostAuthorityRecord,
authority: &MobHostAuthorityDeletionAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_record(expected)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_host_authorities
WHERE mob_id = ?1 AND host_id = ?2 AND record_json = ?3",
params![mob_id.as_str(), expected.host_id, encode_json(&expected)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn put_mob_host_binding_generation_highwater(
&self,
mob_id: &MobId,
expected: &MobHostAuthorityRecord,
authority: &MobHostAuthorityDeletionAuthority,
) -> Result<(), MobStoreError> {
authority.verify_record(expected)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let host_id = expected.host_id.clone();
let binding_generation = u64_to_sql_key(expected.binding_generation);
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_host_binding_generation_highwaters
(mob_id, host_id, binding_generation)
VALUES (?1, ?2, ?3)
ON CONFLICT(mob_id, host_id) DO UPDATE SET binding_generation =
CASE
WHEN excluded.binding_generation > mob_runtime_host_binding_generation_highwaters.binding_generation
THEN excluded.binding_generation
ELSE mob_runtime_host_binding_generation_highwaters.binding_generation
END",
params![mob_id.as_str(), host_id, binding_generation],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn list_mob_host_binding_generation_highwaters(
&self,
mob_id: &MobId,
) -> Result<Vec<(String, u64)>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT host_id, binding_generation
FROM mob_runtime_host_binding_generation_highwaters
WHERE mob_id = ?1
ORDER BY host_id",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
})
.map_err(se)?;
let mut highwaters = Vec::new();
for row in rows {
let (host_id, encoded) = row.map_err(se)?;
highwaters.push((host_id, sql_key_to_u64(&encoded, "binding_generation")?));
}
Ok(highwaters)
})
.await
}
async fn begin_member_operator_request(
&self,
mob_id: &MobId,
record: &MobMemberOperatorRequestRecord,
) -> Result<MobMemberOperatorRequestBegin, MobStoreError> {
record.validate_pending()?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let generation = u64_to_sql_key(record.generation);
let fence_token = u64_to_sql_key(record.fence_token);
let binding_generation = u64_to_sql_key(record.host_binding_generation);
let existing_bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1 AND agent_identity = ?2
AND generation = ?3 AND fence_token = ?4
AND host_id = ?5 AND binding_generation = ?6
AND member_session_id = ?7 AND request_id = ?8",
params![
mob_id.as_str(),
&record.agent_identity,
generation,
fence_token,
&record.host_id,
binding_generation,
&record.member_session_id,
&record.request_id,
],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let result = if let Some(bytes) = existing_bytes {
let existing: MobMemberOperatorRequestRecord = decode_json(&bytes)?;
validate_member_operator_request_row(&existing, &record.key())?;
MobMemberOperatorRequestBegin::Existing(existing)
} else {
let incarnation_rows: i64 = tx
.query_row(
"SELECT COUNT(*) FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1 AND agent_identity = ?2
AND generation = ?3 AND fence_token = ?4
AND host_id = ?5 AND binding_generation = ?6
AND member_session_id = ?7",
params![
mob_id.as_str(),
&record.agent_identity,
generation,
fence_token,
&record.host_id,
binding_generation,
&record.member_session_id,
],
|row| row.get(0),
)
.map_err(se)?;
if incarnation_rows
>= i64::try_from(super::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION)
.unwrap_or(i64::MAX)
{
return Err(MobStoreError::WriteFailed(format!(
"member operator request quota exhausted for '{}' generation {} fence {} host '{}' binding generation {} session '{}' (max {})",
record.agent_identity,
record.generation,
record.fence_token,
record.host_id,
record.host_binding_generation,
record.member_session_id,
super::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION,
)));
}
let mob_rows: i64 = tx
.query_row(
"SELECT COUNT(*) FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1",
params![mob_id.as_str()],
|row| row.get(0),
)
.map_err(se)?;
if mob_rows
>= i64::try_from(super::MEMBER_OPERATOR_REQUEST_MAX_PER_MOB)
.unwrap_or(i64::MAX)
{
return Err(MobStoreError::WriteFailed(format!(
"member operator request quota exhausted for mob '{}' (max {})",
mob_id,
super::MEMBER_OPERATOR_REQUEST_MAX_PER_MOB,
)));
}
tx.execute(
"INSERT INTO mob_runtime_member_operator_requests
(mob_id, agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
mob_id.as_str(),
&record.agent_identity,
generation,
fence_token,
&record.host_id,
binding_generation,
&record.member_session_id,
&record.request_id,
encode_json(&record)?,
],
)
.map_err(se)?;
MobMemberOperatorRequestBegin::Started
};
tx.commit().map_err(se)?;
Ok(result)
})
.await
}
async fn load_member_operator_request(
&self,
mob_id: &MobId,
key: &MobMemberOperatorRequestKey,
) -> Result<Option<MobMemberOperatorRequestRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let key = key.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let row: Option<Vec<u8>> = conn
.query_row(
"SELECT record_json FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1 AND agent_identity = ?2
AND generation = ?3 AND fence_token = ?4
AND host_id = ?5 AND binding_generation = ?6
AND member_session_id = ?7 AND request_id = ?8",
params![
mob_id.as_str(),
&key.agent_identity,
u64_to_sql_key(key.generation),
u64_to_sql_key(key.fence_token),
&key.host_id,
u64_to_sql_key(key.host_binding_generation),
&key.member_session_id,
&key.request_id,
],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let record: Option<MobMemberOperatorRequestRecord> =
row.map(|bytes| decode_json(&bytes)).transpose()?;
if let Some(record) = &record {
validate_member_operator_request_row(record, &key)?;
}
Ok(record)
})
.await
}
async fn compare_and_put_member_operator_request(
&self,
mob_id: &MobId,
expected: &MobMemberOperatorRequestRecord,
record: &MobMemberOperatorRequestRecord,
) -> Result<bool, MobStoreError> {
expected.validate_terminal_transition(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"UPDATE mob_runtime_member_operator_requests
SET record_json = ?1
WHERE mob_id = ?2 AND agent_identity = ?3
AND generation = ?4 AND fence_token = ?5
AND host_id = ?6 AND binding_generation = ?7
AND member_session_id = ?8 AND request_id = ?9
AND record_json = ?10",
params![
encode_json(&record)?,
mob_id.as_str(),
expected.agent_identity,
u64_to_sql_key(expected.generation),
u64_to_sql_key(expected.fence_token),
expected.host_id,
u64_to_sql_key(expected.host_binding_generation),
expected.member_session_id,
expected.request_id,
encode_json(&expected)?,
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed == 1)
})
.await
}
async fn list_member_operator_requests(
&self,
mob_id: &MobId,
) -> Result<Vec<MobMemberOperatorRequestRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id, record_json
FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1
ORDER BY agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, Vec<u8>>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Vec<u8>>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, Vec<u8>>(7)?,
))
})
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let (
agent_identity,
generation,
fence_token,
host_id,
binding_generation,
member_session_id,
request_id,
bytes,
) =
row.map_err(se)?;
let generation = sql_key_to_u64(
&generation,
&format!("member operator request '{request_id}' generation"),
)?;
let fence_token = sql_key_to_u64(
&fence_token,
&format!("member operator request '{request_id}' fence token"),
)?;
let binding_generation = sql_key_to_u64(
&binding_generation,
&format!(
"member operator request '{request_id}' binding generation"
),
)?;
let record: MobMemberOperatorRequestRecord = decode_json(&bytes)?;
let key = MobMemberOperatorRequestKey::new(
agent_identity,
generation,
fence_token,
host_id,
binding_generation,
member_session_id,
request_id,
);
validate_member_operator_request_row(&record, &key)?;
records.push(record);
}
Ok(records)
})
.await
}
async fn prune_stale_member_operator_requests(
&self,
mob_id: &MobId,
authority: &MobMemberOperatorPruneAuthority,
) -> Result<u64, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let authority = authority.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let stale = {
let mut stmt = tx
.prepare(
"SELECT agent_identity, generation, fence_token, host_id,
binding_generation, member_session_id, request_id, record_json
FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, Vec<u8>>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Vec<u8>>(4)?,
row.get::<_, String>(5)?,
row.get::<_, String>(6)?,
row.get::<_, Vec<u8>>(7)?,
))
})
.map_err(se)?;
let mut stale = Vec::new();
for row in rows {
let (
agent_identity,
generation,
fence_token,
host_id,
binding_generation,
member_session_id,
request_id,
bytes,
) = row.map_err(se)?;
let generation = sql_key_to_u64(
&generation,
&format!("member operator request '{request_id}' generation"),
)?;
let fence_token = sql_key_to_u64(
&fence_token,
&format!("member operator request '{request_id}' fence token"),
)?;
let binding_generation = sql_key_to_u64(
&binding_generation,
&format!("member operator request '{request_id}' binding generation"),
)?;
let record: MobMemberOperatorRequestRecord = decode_json(&bytes)?;
let key = MobMemberOperatorRequestKey::new(
agent_identity,
generation,
fence_token,
host_id,
binding_generation,
member_session_id,
request_id,
);
validate_member_operator_request_row(&record, &key)?;
if !authority.preserves(&record) {
stale.push(record);
}
}
stale
};
let mut deleted = 0usize;
for record in stale {
deleted += tx
.execute(
"DELETE FROM mob_runtime_member_operator_requests
WHERE mob_id = ?1 AND agent_identity = ?2
AND generation = ?3 AND fence_token = ?4
AND host_id = ?5 AND binding_generation = ?6
AND member_session_id = ?7 AND request_id = ?8",
params![
mob_id.as_str(),
record.agent_identity,
u64_to_sql_key(record.generation),
u64_to_sql_key(record.fence_token),
record.host_id,
u64_to_sql_key(record.host_binding_generation),
record.member_session_id,
record.request_id,
],
)
.map_err(se)?;
}
tx.commit().map_err(se)?;
u64::try_from(deleted).map_err(|_| {
MobStoreError::Internal(
"member operator request SQLite prune count overflow".to_string(),
)
})
})
.await
}
async fn delete_member_operator_requests(&self, mob_id: &MobId) -> Result<u64, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let deleted = tx
.execute(
"DELETE FROM mob_runtime_member_operator_requests WHERE mob_id = ?1",
params![mob_id.as_str()],
)
.map_err(se)?;
tx.commit().map_err(se)?;
u64::try_from(deleted).map_err(|_| {
MobStoreError::Internal("member operator request delete count overflow".to_string())
})
})
.await
}
async fn load_placed_spawn(
&self,
mob_id: &MobId,
agent_identity: &str,
) -> Result<Option<MobPlacedSpawnCarrierRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let agent_identity = agent_identity.to_string();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let row: Option<Vec<u8>> = conn
.query_row(
"SELECT record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), agent_identity],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let record: Option<MobPlacedSpawnCarrierRecord> =
row.map(|bytes| decode_json(&bytes)).transpose()?;
if let Some(record) = &record {
record.validate_for_store_key(&mob_id, &agent_identity)?;
}
Ok(record)
})
.await
}
async fn list_placed_spawns(
&self,
mob_id: &MobId,
) -> Result<Vec<MobPlacedSpawnCarrierRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT agent_identity, record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 ORDER BY agent_identity",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
})
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let (agent_identity, bytes) = row.map_err(se)?;
let record: MobPlacedSpawnCarrierRecord = decode_json(&bytes)?;
record.validate_for_store_key(&mob_id, &agent_identity)?;
records.push(record);
}
Ok(records)
})
.await
}
async fn begin_placed_spawn_if_absent(
&self,
mob_id: &MobId,
record: &MobPlacedSpawnCarrierRecord,
authority: &MobPlacedSpawnPendingPersistenceAuthority,
) -> Result<BeginPlacedSpawnResult, MobStoreError> {
record.validate_for_mob(mob_id)?;
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), record.agent_identity],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let result = if let Some(bytes) = row {
let existing: MobPlacedSpawnCarrierRecord = decode_json(&bytes)?;
existing.validate_for_store_key(&mob_id, &record.agent_identity)?;
if !existing.same_attempt_as(&record) {
BeginPlacedSpawnResult::Conflict
} else {
match existing.phase {
PlacedSpawnCarrierPhase::Pending => {
BeginPlacedSpawnResult::ExistingExactPending
}
PlacedSpawnCarrierPhase::Committed(_) => {
BeginPlacedSpawnResult::ExistingExactCommitted
}
}
}
} else {
tx.execute(
"INSERT INTO mob_runtime_placed_spawns
(mob_id, agent_identity, record_json) VALUES (?1, ?2, ?3)",
params![
mob_id.as_str(),
record.agent_identity,
encode_json(&record)?
],
)
.map_err(se)?;
BeginPlacedSpawnResult::Inserted
};
tx.commit().map_err(se)?;
Ok(result)
})
.await
}
async fn compare_and_commit_placed_spawn(
&self,
mob_id: &MobId,
expected_pending: &MobPlacedSpawnCarrierRecord,
committed: &MobPlacedSpawnCarrierRecord,
authority: &MobPlacedSpawnCommitPersistenceAuthority,
) -> Result<CommitPlacedSpawnResult, MobStoreError> {
expected_pending.validate_for_mob(mob_id)?;
committed.validate_for_mob(mob_id)?;
authority.verify_record(committed)?;
if !matches!(expected_pending.phase, PlacedSpawnCarrierPhase::Pending)
|| !expected_pending.same_attempt_as(committed)
{
return Err(MobStoreError::Internal(
"placed-spawn commit CAS changed its attempt tuple or expected non-pending state"
.to_string(),
));
}
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected_pending = expected_pending.clone();
let committed = committed.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), expected_pending.agent_identity],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let result = match row {
None => CommitPlacedSpawnResult::Conflict,
Some(bytes) => {
let existing: MobPlacedSpawnCarrierRecord = decode_json(&bytes)?;
existing.validate_for_store_key(&mob_id, &expected_pending.agent_identity)?;
if existing == committed {
CommitPlacedSpawnResult::AlreadyCommittedExact
} else if existing == expected_pending {
tx.execute(
"UPDATE mob_runtime_placed_spawns SET record_json = ?3
WHERE mob_id = ?1 AND agent_identity = ?2",
params![
mob_id.as_str(),
expected_pending.agent_identity,
encode_json(&committed)?
],
)
.map_err(se)?;
CommitPlacedSpawnResult::Committed
} else if existing.same_attempt_as(&expected_pending)
&& matches!(existing.phase, PlacedSpawnCarrierPhase::Pending)
{
CommitPlacedSpawnResult::StillPending
} else {
CommitPlacedSpawnResult::Conflict
}
}
};
tx.commit().map_err(se)?;
Ok(result)
})
.await
}
async fn compare_and_promote_placed_spawn_binding(
&self,
mob_id: &MobId,
expected: &MobPlacedSpawnCarrierRecord,
promoted: &MobPlacedSpawnCarrierRecord,
authority: &MobPlacedSpawnBindingPromotionAuthority,
) -> Result<PromotePlacedSpawnBindingResult, MobStoreError> {
expected.validate_for_mob(mob_id)?;
promoted.validate_for_mob(mob_id)?;
authority.verify_records(expected, promoted)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
let promoted = promoted.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), expected.agent_identity],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let result = match row {
None => PromotePlacedSpawnBindingResult::Conflict,
Some(bytes) => {
let existing: MobPlacedSpawnCarrierRecord = decode_json(&bytes)?;
existing.validate_for_store_key(&mob_id, &expected.agent_identity)?;
if existing == promoted {
PromotePlacedSpawnBindingResult::AlreadyPromotedExact
} else if existing == expected {
tx.execute(
"UPDATE mob_runtime_placed_spawns SET record_json = ?3
WHERE mob_id = ?1 AND agent_identity = ?2",
params![
mob_id.as_str(),
expected.agent_identity,
encode_json(&promoted)?
],
)
.map_err(se)?;
PromotePlacedSpawnBindingResult::Promoted
} else {
PromotePlacedSpawnBindingResult::Conflict
}
}
};
tx.commit().map_err(se)?;
Ok(result)
})
.await
}
async fn compare_and_delete_placed_spawn(
&self,
mob_id: &MobId,
expected: &MobPlacedSpawnCarrierRecord,
authority: &MobPlacedSpawnCleanupAuthority,
) -> Result<DeletePlacedSpawnResult, MobStoreError> {
expected.validate_for_mob(mob_id)?;
authority.verify_record(expected)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), expected.agent_identity],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let result = match row {
None => DeletePlacedSpawnResult::AlreadyAbsent,
Some(bytes) => {
let existing: MobPlacedSpawnCarrierRecord = decode_json(&bytes)?;
existing.validate_for_store_key(&mob_id, &expected.agent_identity)?;
if existing != expected {
DeletePlacedSpawnResult::Conflict
} else {
tx.execute(
"DELETE FROM mob_runtime_placed_spawns
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), expected.agent_identity],
)
.map_err(se)?;
DeletePlacedSpawnResult::Deleted
}
}
};
tx.commit().map_err(se)?;
Ok(result)
})
.await
}
async fn list_mob_operator_grants(
&self,
mob_id: &MobId,
) -> Result<Vec<MobOperatorGrantRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT record_json FROM mob_runtime_operator_grants
WHERE mob_id = ?1
ORDER BY principal",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
records.push(decode_json(&bytes)?);
}
Ok(records)
})
.await
}
async fn put_mob_operator_grant(
&self,
mob_id: &MobId,
record: &MobOperatorGrantRecord,
authority: &MobOperatorGrantPersistenceAuthority,
) -> Result<(), MobStoreError> {
authority.verify_record(record)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_operator_grants (mob_id, principal, record_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(mob_id, principal) DO UPDATE SET record_json = excluded.record_json",
params![mob_id.as_str(), record.principal, encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn delete_mob_operator_grant(
&self,
mob_id: &MobId,
principal: &str,
authority: &MobOperatorGrantDeletionAuthority,
) -> Result<bool, MobStoreError> {
authority.verify_principal(principal)?;
let path = self.path.clone();
let mob_id = mob_id.clone();
let principal = principal.to_string();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_operator_grants
WHERE mob_id = ?1 AND principal = ?2",
params![mob_id.as_str(), principal],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn delete_mob_operator_grants(&self, mob_id: &MobId) -> Result<u64, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_operator_grants WHERE mob_id = ?1",
params![mob_id.as_str()],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed as u64)
})
.await
}
async fn list_member_event_cursors(
&self,
mob_id: &MobId,
) -> Result<Vec<MobMemberEventCursorRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT record_json FROM mob_runtime_member_event_cursors
WHERE mob_id = ?1
ORDER BY agent_identity",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
records.push(decode_json(&bytes)?);
}
Ok(records)
})
.await
}
async fn put_member_event_cursor(
&self,
mob_id: &MobId,
record: &MobMemberEventCursorRecord,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_member_event_cursors (mob_id, agent_identity, record_json)
VALUES (?1, ?2, ?3)
ON CONFLICT(mob_id, agent_identity) DO UPDATE SET record_json = excluded.record_json",
params![mob_id.as_str(), record.agent_identity, encode_json(&record)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn delete_member_event_cursor(
&self,
mob_id: &MobId,
agent_identity: &str,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let agent_identity = agent_identity.to_string();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_member_event_cursors
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), agent_identity],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn delete_member_event_cursors(&self, mob_id: &MobId) -> Result<u64, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_member_event_cursors WHERE mob_id = ?1",
params![mob_id.as_str()],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed as u64)
})
.await
}
async fn list_member_live_cleanup_records(
&self,
mob_id: &MobId,
) -> Result<Vec<MobMemberLiveCleanupRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT record_json FROM mob_runtime_member_live_cleanups
WHERE mob_id = ?1 ORDER BY cleanup_id",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
records.push(decode_json(&row.map_err(se)?)?);
}
Ok(records)
})
.await
}
async fn put_member_live_cleanup_record_if_absent(
&self,
mob_id: &MobId,
record: &MobMemberLiveCleanupRecord,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let encoded = encode_json(&record)?;
let changed = tx
.execute(
"INSERT OR IGNORE INTO mob_runtime_member_live_cleanups
(mob_id, cleanup_id, record_json) VALUES (?1, ?2, ?3)",
params![mob_id.as_str(), record.cleanup_id.as_str(), encoded],
)
.map_err(se)?;
if changed == 0 {
let existing: Vec<u8> = tx
.query_row(
"SELECT record_json FROM mob_runtime_member_live_cleanups
WHERE mob_id = ?1 AND cleanup_id = ?2",
params![mob_id.as_str(), record.cleanup_id.as_str()],
|row| row.get(0),
)
.map_err(se)?;
let existing: MobMemberLiveCleanupRecord = decode_json(&existing)?;
if existing != record {
return Err(MobStoreError::Internal(format!(
"member-live cleanup id '{}' was reused for a conflicting record",
record.cleanup_id
)));
}
}
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn delete_member_live_cleanup_record(
&self,
mob_id: &MobId,
expected: &MobMemberLiveCleanupRecord,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let expected = expected.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let expected_encoded = encode_json(&expected)?;
let changed = tx
.execute(
"DELETE FROM mob_runtime_member_live_cleanups
WHERE mob_id = ?1 AND cleanup_id = ?2 AND record_json = ?3",
params![
mob_id.as_str(),
expected.cleanup_id.as_str(),
expected_encoded.as_slice()
],
)
.map_err(se)?;
if changed == 0 {
let existing: Option<Vec<u8>> = tx
.query_row(
"SELECT record_json FROM mob_runtime_member_live_cleanups
WHERE mob_id = ?1 AND cleanup_id = ?2",
params![mob_id.as_str(), expected.cleanup_id.as_str()],
|row| row.get(0),
)
.optional()
.map_err(se)?;
if existing.is_some_and(|bytes| bytes != expected_encoded) {
return Err(MobStoreError::Internal(format!(
"member-live cleanup id '{}' no longer matches its expected immutable record",
expected.cleanup_id
)));
}
}
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn list_external_binding_overlays(
&self,
mob_id: &MobId,
) -> Result<Vec<ExternalBindingOverlayRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT record_json
FROM mob_runtime_binding_overlays
WHERE mob_id = ?1
ORDER BY agent_identity, generation",
)
.map_err(se)?;
let rows = stmt
.query_map(params![mob_id.as_str()], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut records = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
records.push(decode_json(&bytes)?);
}
Ok(records)
})
.await
}
async fn put_external_binding_overlay_if_absent(
&self,
mob_id: &MobId,
record: &ExternalBindingOverlayRecord,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let changed = tx
.execute(
"INSERT OR IGNORE INTO mob_runtime_binding_overlays
(mob_id, agent_identity, generation, record_json)
VALUES (?1, ?2, ?3, ?4)",
params![
mob_id.as_str(),
record.agent_identity.as_str(),
i64::try_from(record.generation.get()).map_err(|_| {
MobStoreError::Internal(format!(
"generation {} exceeds i64::MAX",
record.generation.get()
))
})?,
encode_json(&record)?,
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(changed > 0)
})
.await
}
async fn upsert_external_binding_overlay(
&self,
mob_id: &MobId,
record: &ExternalBindingOverlayRecord,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let record = record.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_runtime_binding_overlays
(mob_id, agent_identity, generation, record_json)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(mob_id, agent_identity, generation)
DO UPDATE SET record_json = excluded.record_json",
params![
mob_id.as_str(),
record.agent_identity.as_str(),
i64::try_from(record.generation.get()).map_err(|_| {
MobStoreError::Internal(format!(
"generation {} exceeds i64::MAX",
record.generation.get()
))
})?,
encode_json(&record)?,
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn delete_external_binding_overlay(
&self,
mob_id: &MobId,
agent_identity: &AgentIdentity,
generation: Generation,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let agent_identity = agent_identity.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"DELETE FROM mob_runtime_binding_overlays
WHERE mob_id = ?1 AND agent_identity = ?2 AND generation = ?3",
params![
mob_id.as_str(),
agent_identity.as_str(),
i64::try_from(generation.get()).map_err(|_| {
MobStoreError::Internal(format!(
"generation {} exceeds i64::MAX",
generation.get()
))
})?,
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn delete_external_binding_overlays(&self, mob_id: &MobId) -> Result<(), MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"DELETE FROM mob_runtime_binding_overlays WHERE mob_id = ?1",
params![mob_id.as_str()],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
}
#[derive(Clone)]
pub struct SqliteMobEventStore {
path: PathBuf,
event_bus: Arc<SqliteMobEventBus>,
}
impl std::fmt::Debug for SqliteMobEventStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteMobEventStore")
.field("path", &self.path)
.finish()
}
}
const EVENT_CURSOR_KEY: &str = "next_cursor";
impl private::MobEventStoreSealed for SqliteMobEventStore {}
#[async_trait]
impl MobEventStore for SqliteMobEventStore {
async fn begin_external_delivery(
&self,
intent: &MobExternalDeliveryIntent,
) -> Result<MobExternalDeliveryBeginOutcome, MobStoreError> {
intent.validate()?;
let path = self.path.clone();
let intent = intent.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let existing = tx
.query_row(
"SELECT CAST(record_json AS BLOB)
FROM mob_external_deliveries
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str()
],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()
.map_err(se)?
.map(|bytes| decode_json::<MobExternalDeliveryRecord>(&bytes))
.transpose()?;
if let Some(existing) = existing {
validate_external_delivery_record(
&existing,
&intent.mob_id,
&intent.identity.idempotency_key,
)?;
if existing.intent != intent {
return Err(MobStoreError::ExternalDeliveryConflict {
mob_id: intent.mob_id,
idempotency_key: intent.identity.idempotency_key,
detail: "stored correlation, target kind, or action digest differs"
.to_string(),
});
}
tx.commit().map_err(se)?;
return Ok(match existing.phase {
MobExternalDeliveryPhase::Begun { repair } => {
MobExternalDeliveryBeginOutcome::ExistingBegun { repair }
}
MobExternalDeliveryPhase::Terminal { terminal } => {
MobExternalDeliveryBeginOutcome::ExistingTerminal(terminal)
}
});
}
let record = MobExternalDeliveryRecord {
intent: intent.clone(),
phase: MobExternalDeliveryPhase::Begun {
repair: MobExternalDeliveryRepairState::default(),
},
};
tx.execute(
"INSERT INTO mob_external_deliveries
(mob_id, idempotency_key, record_json)
VALUES (?1, ?2, ?3)",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str(),
encode_json(&record)?
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(MobExternalDeliveryBeginOutcome::Begun)
})
.await
}
async fn complete_external_delivery(
&self,
intent: &MobExternalDeliveryIntent,
terminal: &MobExternalDeliveryTerminal,
) -> Result<MobExternalDeliveryCompleteOutcome, MobStoreError> {
intent.validate()?;
validate_external_delivery_terminal(terminal)?;
let path = self.path.clone();
let intent = intent.clone();
let terminal = terminal.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes = tx
.query_row(
"SELECT CAST(record_json AS BLOB)
FROM mob_external_deliveries
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str()
],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()
.map_err(se)?
.ok_or_else(|| {
MobStoreError::NotFound(format!(
"mob external delivery '{}'",
intent.identity.idempotency_key
))
})?;
let mut record = decode_json::<MobExternalDeliveryRecord>(&bytes)?;
validate_external_delivery_record(
&record,
&intent.mob_id,
&intent.identity.idempotency_key,
)?;
if record.intent != intent {
return Err(MobStoreError::ExternalDeliveryConflict {
mob_id: intent.mob_id,
idempotency_key: intent.identity.idempotency_key,
detail: "terminal authority does not match stored admission".to_string(),
});
}
let outcome = match &record.phase {
MobExternalDeliveryPhase::Begun { .. } => {
record.phase = MobExternalDeliveryPhase::Terminal {
terminal: terminal.clone(),
};
MobExternalDeliveryCompleteOutcome::Completed
}
MobExternalDeliveryPhase::Terminal { terminal: existing }
if existing == &terminal =>
{
MobExternalDeliveryCompleteOutcome::AlreadyCompleted
}
MobExternalDeliveryPhase::Terminal { .. } => {
return Err(MobStoreError::ExternalDeliveryConflict {
mob_id: intent.mob_id,
idempotency_key: intent.identity.idempotency_key,
detail: "a different terminal is already committed".to_string(),
});
}
};
if outcome == MobExternalDeliveryCompleteOutcome::Completed {
tx.execute(
"UPDATE mob_external_deliveries
SET record_json = ?3
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str(),
encode_json(&record)?
],
)
.map_err(se)?;
}
tx.commit().map_err(se)?;
Ok(outcome)
})
.await
}
async fn schedule_external_delivery_repair(
&self,
intent: &MobExternalDeliveryIntent,
) -> Result<MobExternalDeliveryRepairOutcome, MobStoreError> {
intent.validate()?;
let path = self.path.clone();
let intent = intent.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes = tx
.query_row(
"SELECT CAST(record_json AS BLOB)
FROM mob_external_deliveries
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str()
],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()
.map_err(se)?
.ok_or_else(|| {
MobStoreError::NotFound(format!(
"mob external delivery '{}'",
intent.identity.idempotency_key
))
})?;
let mut record = decode_json::<MobExternalDeliveryRecord>(&bytes)?;
validate_external_delivery_record(
&record,
&intent.mob_id,
&intent.identity.idempotency_key,
)?;
if record.intent != intent {
return Err(MobStoreError::ExternalDeliveryConflict {
mob_id: intent.mob_id,
idempotency_key: intent.identity.idempotency_key,
detail: "repair authority does not match stored admission".to_string(),
});
}
let outcome = match &mut record.phase {
MobExternalDeliveryPhase::Begun { repair } => {
*repair = next_external_delivery_repair_state(
*repair,
external_delivery_repair_now_ms()?,
);
MobExternalDeliveryRepairOutcome::Scheduled(*repair)
}
MobExternalDeliveryPhase::Terminal { terminal } => {
tx.commit().map_err(se)?;
return Ok(MobExternalDeliveryRepairOutcome::ExistingTerminal(
terminal.clone(),
));
}
};
tx.execute(
"UPDATE mob_external_deliveries
SET record_json = ?3
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![
intent.mob_id.as_str(),
intent.identity.idempotency_key.as_str(),
encode_json(&record)?
],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(outcome)
})
.await
}
async fn load_external_delivery(
&self,
mob_id: &MobId,
idempotency_key: &str,
) -> Result<Option<MobExternalDeliveryRecord>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let idempotency_key = idempotency_key.to_string();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let record = conn
.query_row(
"SELECT CAST(record_json AS BLOB)
FROM mob_external_deliveries
WHERE mob_id = ?1 AND idempotency_key = ?2",
params![mob_id.as_str(), idempotency_key.as_str()],
|row| row.get::<_, Vec<u8>>(0),
)
.optional()
.map_err(se)?
.map(|bytes| decode_json::<MobExternalDeliveryRecord>(&bytes))
.transpose()?;
if let Some(record) = &record {
validate_external_delivery_record(record, &mob_id, &idempotency_key)?;
}
Ok(record)
})
.await
}
async fn append(&self, event: NewMobEvent) -> Result<MobEvent, MobStoreError> {
validate_mob_event_write_authority(&event.kind)?;
let path = self.path.clone();
let stored = run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let cursor = next_event_cursor(&tx)?;
let stored = MobEvent {
cursor,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let encoded = encode_stored_mob_event(&stored)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
set_next_cursor(&tx, checked_event_cursor_successor(cursor)?)?;
tx.commit().map_err(se)?;
Ok(stored)
})
.await?;
self.event_bus.publish_committed(stored.clone());
Ok(stored)
}
async fn append_terminal_event_if_absent(
&self,
event: NewMobEvent,
) -> Result<Option<MobEvent>, MobStoreError> {
let Some((run_id, flow_id)) = terminal_event_identity(&event.kind) else {
return Err(MobStoreError::Internal(
"append_terminal_event_if_absent requires a terminal flow event".to_string(),
));
};
let run_id = run_id.clone();
let flow_id = flow_id.clone();
let mob_id = event.mob_id.clone();
let path = self.path.clone();
let stored = run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let mut stmt = tx
.prepare("SELECT event_json FROM mob_events ORDER BY cursor")
.map_err(se)?;
let rows = stmt
.query_map([], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
for row in rows {
let bytes = row.map_err(se)?;
let existing = decode_stored_mob_event(&bytes)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
if existing.mob_id == mob_id
&& terminal_event_identity(&existing.kind).is_some_and(
|(existing_run_id, existing_flow_id)| {
existing_run_id == &run_id && existing_flow_id == &flow_id
},
)
{
return Ok(None);
}
}
drop(stmt);
let cursor = next_event_cursor(&tx)?;
let stored = MobEvent {
cursor,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let encoded = encode_stored_mob_event(&stored)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
set_next_cursor(&tx, checked_event_cursor_successor(cursor)?)?;
tx.commit().map_err(se)?;
Ok(Some(stored))
})
.await?;
if let Some(stored) = stored.as_ref() {
self.event_bus.publish_committed(stored.clone());
}
Ok(stored)
}
async fn append_step_failed_event_if_absent(
&self,
event: NewMobEvent,
) -> Result<Option<MobEvent>, MobStoreError> {
validate_mob_event_write_authority(&event.kind)?;
let Some((run_id, step_id, _)) = step_failed_event_identity(&event.kind) else {
return Err(MobStoreError::Internal(
"append_step_failed_event_if_absent requires a StepFailed event".to_string(),
));
};
let run_id = run_id.clone();
let step_id = step_id.clone();
let mob_id = event.mob_id.clone();
let path = self.path.clone();
let stored = run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let mut stmt = tx
.prepare("SELECT event_json FROM mob_events ORDER BY cursor")
.map_err(se)?;
let rows = stmt
.query_map([], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut exact_replay = false;
for row in rows {
let bytes = row.map_err(se)?;
let existing = decode_stored_mob_event(&bytes)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
if existing.mob_id != mob_id {
continue;
}
let Some((existing_run_id, existing_step_id, _)) =
step_failed_event_identity(&existing.kind)
else {
continue;
};
if existing_run_id == &run_id && existing_step_id == &step_id {
if existing.kind != event.kind {
return Err(MobStoreError::Internal(format!(
"StepFailed event conflict for run '{run_id}' step '{step_id}'"
)));
}
exact_replay = true;
}
}
drop(stmt);
if exact_replay {
return Ok(None);
}
let cursor = next_event_cursor(&tx)?;
let stored = MobEvent {
cursor,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let encoded = encode_stored_mob_event(&stored)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
set_next_cursor(&tx, checked_event_cursor_successor(cursor)?)?;
tx.commit().map_err(se)?;
Ok(Some(stored))
})
.await?;
if let Some(stored) = stored.as_ref() {
self.event_bus.publish_committed(stored.clone());
}
Ok(stored)
}
async fn append_batch(&self, batch: Vec<NewMobEvent>) -> Result<Vec<MobEvent>, MobStoreError> {
for event in &batch {
validate_mob_event_write_authority(&event.kind)?;
}
let path = self.path.clone();
let stored = run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let mut cursor = next_event_cursor(&tx)?;
let mut results = Vec::with_capacity(batch.len());
for event in batch {
let stored = MobEvent {
cursor,
timestamp: event.timestamp.unwrap_or_else(Utc::now),
mob_id: event.mob_id,
kind: event.kind,
};
let encoded = encode_stored_mob_event(&stored)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
results.push(stored);
cursor = checked_event_cursor_successor(cursor)?;
}
set_next_cursor(&tx, cursor)?;
tx.commit().map_err(se)?;
Ok(results)
})
.await?;
self.event_bus.publish_committed_batch(&stored);
Ok(stored)
}
async fn poll(&self, after_cursor: u64, limit: usize) -> Result<Vec<MobEvent>, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || poll_events_sync(&path, after_cursor, limit)).await
}
async fn replay_all(&self) -> Result<Vec<MobEvent>, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let mut stmt = conn
.prepare("SELECT event_json FROM mob_events ORDER BY cursor")
.map_err(se)?;
let rows = stmt
.query_map([], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut result = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
result.push(
decode_stored_mob_event(&bytes)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?,
);
}
Ok(result)
})
.await
}
async fn latest_cursor(&self) -> Result<u64, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || latest_event_cursor_sync(&path)).await
}
fn subscribe(&self) -> Result<super::MobEventReceiver, MobStoreError> {
Ok(self.event_bus.subscribe())
}
async fn clear(&self) -> Result<(), MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute("DELETE FROM mob_events", []).map_err(se)?;
tx.execute(
"INSERT OR REPLACE INTO mob_event_meta (key, value) VALUES (?1, ?2)",
params![EVENT_CURSOR_KEY, 1i64],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn prune(&self, older_than: DateTime<Utc>) -> Result<u64, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let mut stmt = tx
.prepare("SELECT cursor, event_json FROM mob_events ORDER BY cursor")
.map_err(se)?;
let rows: Vec<(i64, Vec<u8>)> = stmt
.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
.map_err(se)?
.collect::<Result<_, _>>()
.map_err(se)?;
drop(stmt);
let mut removed = 0u64;
for (cursor_val, bytes) in rows {
let event: MobEvent = decode_stored_mob_event(&bytes)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
if event.timestamp < older_than
&& !identity_structural_projection_is_anchor(&event.kind)
{
tx.execute(
"DELETE FROM mob_events WHERE cursor = ?1",
params![cursor_val],
)
.map_err(se)?;
removed = removed.saturating_add(1);
}
}
tx.commit().map_err(se)?;
Ok(removed)
})
.await
}
}
fn get_next_cursor(conn: &Connection) -> Result<u64, MobStoreError> {
repair_next_event_cursor(conn)
}
fn repair_next_event_cursor(conn: &Connection) -> Result<u64, MobStoreError> {
let physical_max: Option<i64> = conn
.query_row("SELECT MAX(cursor) FROM mob_events", [], |row| row.get(0))
.map_err(se)?;
let physical_floor = match physical_max {
Some(max) if max > 0 => {
let cursor = u64::try_from(max).map_err(se)?;
if cursor >= i64::MAX as u64 {
i64::MAX as u64
} else {
cursor + 1
}
}
_ => 1,
};
let stored: Option<rusqlite::types::Value> = conn
.query_row(
"SELECT value FROM mob_event_meta WHERE key = ?1",
params![EVENT_CURSOR_KEY],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let stored_next = match &stored {
Some(rusqlite::types::Value::Integer(value)) if *value > 0 => u64::try_from(*value).ok(),
_ => None,
};
let next = stored_next.map_or(physical_floor, |stored| stored.max(physical_floor));
if stored_next != Some(next) {
set_next_cursor(conn, next)?;
}
Ok(next)
}
fn next_event_cursor(tx: &Transaction<'_>) -> Result<u64, MobStoreError> {
let cursor = get_next_cursor(tx)?;
if cursor >= i64::MAX as u64 {
return Err(MobStoreError::Internal(
"mob event cursor is exhausted".to_string(),
));
}
Ok(cursor)
}
fn checked_event_cursor_successor(cursor: u64) -> Result<u64, MobStoreError> {
cursor
.checked_add(1)
.filter(|next| i64::try_from(*next).is_ok())
.ok_or_else(|| MobStoreError::Internal("mob event cursor is exhausted".to_string()))
}
fn set_next_cursor(conn: &Connection, value: u64) -> Result<(), MobStoreError> {
conn.execute(
"INSERT OR REPLACE INTO mob_event_meta (key, value) VALUES (?1, ?2)",
params![EVENT_CURSOR_KEY, cursor_to_i64(value)?],
)
.map_err(se)?;
Ok(())
}
#[derive(Clone)]
pub struct SqliteMobRunStore {
path: PathBuf,
}
#[derive(Debug, Clone, Copy)]
enum MissingRunCasBehavior {
ReturnFalse,
NotFound,
}
impl std::fmt::Debug for SqliteMobRunStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteMobRunStore")
.field("path", &self.path)
.finish()
}
}
impl SqliteMobRunStore {
async fn update_run_with_authority_if<F>(
&self,
run_id: &RunId,
missing_behavior: MissingRunCasBehavior,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
update: F,
) -> Result<bool, MobStoreError>
where
F: FnOnce(&mut MobRun) -> Result<bool, MobStoreError> + Send + 'static,
{
let path = self.path.clone();
let key = run_id.to_string();
let run_id = run_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes = load_run_bytes(&tx, &key)?;
let Some(bytes) = bytes else {
return match missing_behavior {
MissingRunCasBehavior::ReturnFalse => Ok(false),
MissingRunCasBehavior::NotFound => {
Err(MobStoreError::NotFound(format!("run not found: {run_id}")))
}
};
};
let mut run: MobRun = decode_json(&bytes)?;
if !update(&mut run)? {
return Ok(false);
}
run.append_flow_authority_inputs(authority_inputs)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
write_run_json(&tx, &key, &run)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
}
#[async_trait]
impl MobRunStore for SqliteMobRunStore {
async fn create_run(&self, run: MobRun) -> Result<(), MobStoreError> {
let path = self.path.clone();
let key = run.run_id.to_string();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let exists: bool = tx
.query_row(
"SELECT 1 FROM mob_runs WHERE run_id = ?1",
params![key],
|_| Ok(true),
)
.optional()
.map_err(se)?
.unwrap_or(false);
if exists {
return Err(MobStoreError::Internal(format!(
"run already exists: {}",
run.run_id
)));
}
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let encoded = encode_json(&run)?;
tx.execute(
"INSERT INTO mob_runs (run_id, run_json) VALUES (?1, ?2)",
params![key, encoded],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn get_run(&self, run_id: &RunId) -> Result<Option<MobRun>, MobStoreError> {
let path = self.path.clone();
let key = run_id.to_string();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let bytes: Option<Vec<u8>> = conn
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
match bytes {
Some(b) => Ok(Some(decode_json(&b)?)),
None => Ok(None),
}
})
.await
}
async fn list_runs(
&self,
mob_id: &MobId,
flow_id: Option<&FlowId>,
) -> Result<Vec<MobRun>, MobStoreError> {
let path = self.path.clone();
let mob_id = mob_id.clone();
let flow_id = flow_id.cloned();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let mut stmt = conn.prepare("SELECT run_json FROM mob_runs").map_err(se)?;
let rows = stmt
.query_map([], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut runs = Vec::new();
for row in rows {
let bytes = row.map_err(se)?;
let run: MobRun = decode_json(&bytes)?;
if run.mob_id == mob_id && flow_id.as_ref().is_none_or(|fid| run.flow_id == *fid) {
runs.push(run);
}
}
Ok(runs)
})
.await
}
async fn put_remote_turn_intent(
&self,
run_id: &RunId,
intent: &MobRunRemoteTurnIntent,
) -> Result<bool, MobStoreError> {
if &intent.obligation.run_id != run_id || intent.obligation.dispatch_sequence == 0 {
return Err(MobStoreError::Internal(format!(
"remote-turn intent does not match run '{run_id}' or has sequence zero"
)));
}
let path = self.path.clone();
let run_key = run_id.to_string();
let sequence = u64_to_sql_key(intent.obligation.dispatch_sequence);
let intent = intent.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let run_json: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![run_key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(run_json) = run_json else {
return Err(MobStoreError::NotFound(format!(
"run not found: {}",
intent.obligation.run_id
)));
};
let run: MobRun = decode_json(&run_json)?;
intent
.validate_for(&run.run_id, &run.mob_id)
.map_err(MobStoreError::Internal)?;
let existing: Option<Vec<u8>> = tx
.query_row(
"SELECT intent_json FROM mob_run_remote_turn_intents \
WHERE run_id = ?1 AND dispatch_sequence = ?2",
params![run_key, sequence],
|row| row.get(0),
)
.optional()
.map_err(se)?;
if let Some(bytes) = existing {
let existing: MobRunRemoteTurnIntent = decode_json(&bytes)?;
if existing != intent {
return Err(MobStoreError::Internal(format!(
"remote-turn intent sequence {} conflicts for run '{}'",
intent.obligation.dispatch_sequence, intent.obligation.run_id
)));
}
return Ok(false);
}
tx.execute(
"INSERT INTO mob_run_remote_turn_intents \
(run_id, dispatch_sequence, intent_json) VALUES (?1, ?2, ?3)",
params![run_key, sequence, encode_json(&intent)?],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
async fn delete_remote_turn_intent(
&self,
run_id: &RunId,
dispatch_sequence: u64,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let run_key = run_id.to_string();
let sequence = u64_to_sql_key(dispatch_sequence);
run_sqlite_task(move || {
let conn = open_connection(&path)?;
Ok(conn
.execute(
"DELETE FROM mob_run_remote_turn_intents \
WHERE run_id = ?1 AND dispatch_sequence = ?2",
params![run_key, sequence],
)
.map_err(se)?
> 0)
})
.await
}
async fn list_remote_turn_intents(
&self,
run_id: &RunId,
) -> Result<Vec<MobRunRemoteTurnIntent>, MobStoreError> {
let path = self.path.clone();
let run_key = run_id.to_string();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT intent_json FROM mob_run_remote_turn_intents \
WHERE run_id = ?1 ORDER BY dispatch_sequence ASC",
)
.map_err(se)?;
let rows = stmt
.query_map(params![run_key], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut intents = Vec::new();
for row in rows {
intents.push(decode_json(&row.map_err(se)?)?);
}
Ok(intents)
})
.await
}
async fn put_remote_turn_receipt(
&self,
run_id: &RunId,
receipt: &MobRunRemoteTurnReceipt,
) -> Result<bool, MobStoreError> {
if &receipt.obligation.run_id != run_id || receipt.obligation.dispatch_sequence == 0 {
return Err(MobStoreError::Internal(format!(
"remote-turn receipt does not match run '{run_id}' or has sequence zero"
)));
}
let path = self.path.clone();
let run_key = run_id.to_string();
let sequence = u64_to_sql_key(receipt.obligation.dispatch_sequence);
let receipt = receipt.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let run_json: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![run_key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(run_json) = run_json else {
return Err(MobStoreError::NotFound(format!(
"run not found: {}",
receipt.obligation.run_id
)));
};
let run: MobRun = decode_json(&run_json)?;
let intent_json: Option<Vec<u8>> = tx
.query_row(
"SELECT intent_json FROM mob_run_remote_turn_intents \
WHERE run_id = ?1 AND dispatch_sequence = ?2",
params![run_key, sequence],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let intent: MobRunRemoteTurnIntent = intent_json
.map(|bytes| decode_json(&bytes))
.transpose()?
.ok_or_else(|| {
MobStoreError::Internal(
"remote-turn receipt has no exact durable intent".to_string(),
)
})?;
receipt
.validate_for(&run.run_id, &run.mob_id, &intent)
.map_err(MobStoreError::Internal)?;
let existing: Option<Vec<u8>> = tx
.query_row(
"SELECT receipt_json FROM mob_run_remote_turn_receipts \
WHERE run_id = ?1 AND dispatch_sequence = ?2",
params![run_key, sequence],
|row| row.get(0),
)
.optional()
.map_err(se)?;
if let Some(bytes) = existing {
let existing: MobRunRemoteTurnReceipt = decode_json(&bytes)?;
if existing != receipt {
return Err(MobStoreError::Internal(format!(
"remote-turn receipt sequence {} conflicts for run '{}'",
receipt.obligation.dispatch_sequence, receipt.obligation.run_id
)));
}
return Ok(false);
}
let bytes = encode_json(&receipt)?;
tx.execute(
"INSERT INTO mob_run_remote_turn_receipts \
(run_id, dispatch_sequence, receipt_json) VALUES (?1, ?2, ?3)",
params![run_key, sequence, bytes],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
async fn list_remote_turn_receipts(
&self,
run_id: &RunId,
) -> Result<Vec<MobRunRemoteTurnReceipt>, MobStoreError> {
let path = self.path.clone();
let run_key = run_id.to_string();
run_sqlite_task(move || {
let conn = open_existing_read_connection(&path)?;
let mut stmt = conn
.prepare(
"SELECT receipt_json FROM mob_run_remote_turn_receipts \
WHERE run_id = ?1 ORDER BY dispatch_sequence ASC",
)
.map_err(se)?;
let rows = stmt
.query_map(params![run_key], |row| row.get::<_, Vec<u8>>(0))
.map_err(se)?;
let mut receipts = Vec::new();
for row in rows {
receipts.push(decode_json(&row.map_err(se)?)?);
}
Ok(receipts)
})
.await
}
async fn delete_remote_turn_receipt(
&self,
run_id: &RunId,
dispatch_sequence: u64,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let run_key = run_id.to_string();
let sequence = u64_to_sql_key(dispatch_sequence);
run_sqlite_task(move || {
let conn = open_connection(&path)?;
Ok(conn
.execute(
"DELETE FROM mob_run_remote_turn_receipts \
WHERE run_id = ?1 AND dispatch_sequence = ?2",
params![run_key, sequence],
)
.map_err(se)?
> 0)
})
.await
}
async fn cas_flow_state_with_authority(
&self,
run_id: &RunId,
expected: &flow_run::State,
next: &flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let expected = expected.clone();
let next = next.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::ReturnFalse,
authority_inputs,
move |run| {
if run.flow_state != expected {
return Ok(false);
}
run.flow_state = next;
Ok(true)
},
)
.await
}
async fn cas_run_snapshot_with_authority(
&self,
run_id: &RunId,
expected_status: MobRunStatus,
expected_flow_state: &flow_run::State,
next_status: MobRunStatus,
next_flow_state: &flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let expected_flow_state = expected_flow_state.clone();
let next_flow_state = next_flow_state.clone();
let terminality_run_id = run_id.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::ReturnFalse,
authority_inputs,
move |run| {
let current_terminal =
mob_machine_run_status_is_terminal(&terminality_run_id, &run.status)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
if run.status != expected_status
|| current_terminal
|| run.flow_state != expected_flow_state
{
return Ok(false);
}
let terminal =
mob_machine_run_status_is_terminal(&terminality_run_id, &next_status)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.status = next_status;
run.flow_state = next_flow_state;
if terminal && run.completed_at.is_none() {
run.completed_at = Some(Utc::now());
}
Ok(true)
},
)
.await
}
async fn cas_run_snapshot_and_append_terminal_event_with_authority(
&self,
run_id: &RunId,
expected_status: MobRunStatus,
expected_flow_state: &flow_run::State,
next_status: MobRunStatus,
next_flow_state: &flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
_events: &dyn MobEventStore,
terminal_event: NewMobEvent,
) -> Result<Option<MobEvent>, MobStoreError> {
validate_mob_event_write_authority(&terminal_event.kind)?;
if terminal_event_identity(&terminal_event.kind).is_none() {
return Err(MobStoreError::Internal(
"cas_run_snapshot_and_append_terminal_event_with_authority requires a terminal flow event".to_string(),
));
}
let path = self.path.clone();
let key = run_id.to_string();
let terminality_run_id = run_id.clone();
let expected_flow_state = expected_flow_state.clone();
let next_flow_state = next_flow_state.clone();
let event_bus = sqlite_event_bus_for_path(self.path.clone())?;
let stored = run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes = load_run_bytes(&tx, &key)?;
let Some(bytes) = bytes else {
return Ok(None);
};
let mut run: MobRun = decode_json(&bytes)?;
let current_terminal =
mob_machine_run_status_is_terminal(&terminality_run_id, &run.status)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
if run.status != expected_status
|| current_terminal
|| run.flow_state != expected_flow_state
{
return Ok(None);
}
let terminal = mob_machine_run_status_is_terminal(&terminality_run_id, &next_status)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.status = next_status;
run.flow_state = next_flow_state;
if terminal && run.completed_at.is_none() {
run.completed_at = Some(Utc::now());
}
run.append_flow_authority_inputs(authority_inputs)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
write_run_json(&tx, &key, &run)?;
let cursor = next_event_cursor(&tx)?;
let stored = MobEvent {
cursor,
timestamp: terminal_event.timestamp.unwrap_or_else(Utc::now),
mob_id: terminal_event.mob_id,
kind: terminal_event.kind,
};
let encoded = encode_stored_mob_event(&stored)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?;
tx.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, ?2, ?3)",
params![cursor_to_i64(cursor)?, stored.mob_id.as_str(), encoded],
)
.map_err(se)?;
set_next_cursor(&tx, checked_event_cursor_successor(cursor)?)?;
tx.commit().map_err(se)?;
Ok(Some(stored))
})
.await?;
if let Some(stored) = stored.as_ref() {
event_bus.publish_committed(stored.clone());
}
Ok(stored)
}
async fn append_step_entry_with_authority(
&self,
run_id: &RunId,
entry: StepLedgerEntry,
authority: MobRunProvenanceAuthority,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let key = run_id.to_string();
let run_id = run_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(bytes) = bytes else {
return Err(MobStoreError::NotFound(format!("run not found: {run_id}")));
};
let mut run: MobRun = decode_json(&bytes)?;
authority
.validate_step_entry(&run, &entry)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.step_ledger.push(entry);
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let encoded = encode_json(&run)?;
tx.execute(
"UPDATE mob_runs SET run_json = ?1 WHERE run_id = ?2",
params![encoded, key],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn append_step_entry_if_absent_with_authority(
&self,
run_id: &RunId,
entry: StepLedgerEntry,
authority: MobRunProvenanceAuthority,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let key = run_id.to_string();
let run_id = run_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(bytes) = bytes else {
return Err(MobStoreError::NotFound(format!("run not found: {run_id}")));
};
let mut run: MobRun = decode_json(&bytes)?;
authority
.validate_step_entry(&run, &entry)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let is_duplicate = run.step_ledger.iter().any(|existing| {
existing.step_id == entry.step_id
&& existing.agent_identity == entry.agent_identity
&& existing.status == entry.status
});
if is_duplicate {
return Ok(false);
}
run.step_ledger.push(entry);
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let encoded = encode_json(&run)?;
tx.execute(
"UPDATE mob_runs SET run_json = ?1 WHERE run_id = ?2",
params![encoded, key],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
async fn append_failure_entry_with_authority(
&self,
run_id: &RunId,
entry: FailureLedgerEntry,
authority: MobRunProvenanceAuthority,
) -> Result<(), MobStoreError> {
let path = self.path.clone();
let key = run_id.to_string();
let run_id = run_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(bytes) = bytes else {
return Err(MobStoreError::NotFound(format!("run not found: {run_id}")));
};
let mut run: MobRun = decode_json(&bytes)?;
authority
.validate_failure_entry(&run, &entry)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
run.failure_ledger.push(entry);
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let encoded = encode_json(&run)?;
tx.execute(
"UPDATE mob_runs SET run_json = ?1 WHERE run_id = ?2",
params![encoded, key],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(())
})
.await
}
async fn append_failure_entry_if_absent_with_authority(
&self,
run_id: &RunId,
entry: FailureLedgerEntry,
authority: MobRunProvenanceAuthority,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let key = run_id.to_string();
let run_id = run_id.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT run_json FROM mob_runs WHERE run_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(bytes) = bytes else {
return Err(MobStoreError::NotFound(format!("run not found: {run_id}")));
};
let mut run: MobRun = decode_json(&bytes)?;
authority
.validate_failure_entry(&run, &entry)
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
if let Some(existing) = run
.failure_ledger
.iter()
.find(|existing| existing.step_id == entry.step_id)
{
if existing.reason == entry.reason
&& existing.error_report == entry.error_report
&& existing.error == entry.error
{
return Ok(false);
}
return Err(MobStoreError::Internal(format!(
"failure ledger conflict for run '{run_id}' step '{}'",
entry.step_id
)));
}
run.failure_ledger.push(entry);
run.validate_flow_authority_projection()
.map_err(|error| MobStoreError::Internal(error.to_string()))?;
let encoded = encode_json(&run)?;
tx.execute(
"UPDATE mob_runs SET run_json = ?1 WHERE run_id = ?2",
params![encoded, key],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
async fn cas_frame_state_with_authority(
&self,
run_id: &RunId,
frame_id: &FrameId,
expected: Option<&FrameSnapshot>,
next: FrameSnapshot,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let frame_id = frame_id.clone();
let expected = expected.cloned();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
let current = run.frames.get(&frame_id);
let matches = match (expected.as_ref(), current) {
(None, None) => true,
(Some(exp), Some(cur)) => exp == cur,
_ => false,
};
if !matches {
return Ok(false);
}
run.frames.insert(frame_id, next);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_grant_node_slot_with_authority(
&self,
run_id: &RunId,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
frame_id: &FrameId,
expected_frame: &FrameSnapshot,
next_frame: FrameSnapshot,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let expected_run_state = expected_run_state.clone();
let frame_id = frame_id.clone();
let expected_frame = expected_frame.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.frames.get(&frame_id) != Some(&expected_frame) {
return Ok(false);
}
run.flow_state = next_run_state;
run.frames.insert(frame_id, next_frame);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_complete_step_and_record_output_with_authority(
&self,
run_id: &RunId,
frame_id: &FrameId,
expected_frame: &FrameSnapshot,
next_frame: FrameSnapshot,
step_output_key: String,
step_output: serde_json::Value,
loop_context: Option<(&LoopId, u64)>,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let frame_id = frame_id.clone();
let expected_frame = expected_frame.clone();
let loop_context = loop_context.map(|(loop_id, iteration)| (loop_id.clone(), iteration));
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.frames.get(&frame_id) != Some(&expected_frame) {
return Ok(false);
}
run.frames.insert(frame_id, next_frame);
match loop_context {
None => {
run.root_step_outputs
.insert(StepId::from(step_output_key.as_str()), step_output);
}
Some((loop_id, iteration)) => {
let iteration_index = usize::try_from(iteration).map_err(|_| {
MobStoreError::Internal(format!(
"loop iteration index {iteration} exceeds usize::MAX on this target"
))
})?;
let outputs = run.loop_iteration_outputs.entry(loop_id).or_default();
while outputs.len() <= iteration_index {
outputs.push(indexmap::IndexMap::new());
}
outputs[iteration_index]
.insert(StepId::from(step_output_key.as_str()), step_output);
}
}
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_start_loop_with_authority(
&self,
run_id: &RunId,
loop_instance_id: &LoopInstanceId,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
frame_id: &FrameId,
expected_frame: &FrameSnapshot,
next_frame: FrameSnapshot,
initial_loop: LoopSnapshot,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let loop_instance_id = loop_instance_id.clone();
let expected_run_state = expected_run_state.clone();
let frame_id = frame_id.clone();
let expected_frame = expected_frame.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.frames.get(&frame_id) != Some(&expected_frame) {
return Ok(false);
}
if run.loops.contains_key(&loop_instance_id) {
return Ok(false);
}
run.flow_state = next_run_state;
run.frames.insert(frame_id, next_frame);
run.loops.insert(loop_instance_id, initial_loop);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_loop_request_body_frame_with_authority(
&self,
run_id: &RunId,
loop_instance_id: &LoopInstanceId,
expected_loop: &LoopSnapshot,
next_loop: LoopSnapshot,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let loop_instance_id = loop_instance_id.clone();
let expected_loop = expected_loop.clone();
let expected_run_state = expected_run_state.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.loops.get(&loop_instance_id) != Some(&expected_loop) {
return Ok(false);
}
run.flow_state = next_run_state;
run.loops.insert(loop_instance_id, next_loop);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_grant_body_frame_start_with_authority(
&self,
run_id: &RunId,
loop_instance_id: &LoopInstanceId,
expected_loop: &LoopSnapshot,
next_loop: LoopSnapshot,
frame_id: &FrameId,
initial_frame: FrameSnapshot,
ledger_entry: LoopIterationLedgerEntry,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let loop_instance_id = loop_instance_id.clone();
let expected_loop = expected_loop.clone();
let frame_id = frame_id.clone();
let expected_run_state = expected_run_state.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.loops.get(&loop_instance_id) != Some(&expected_loop) {
return Ok(false);
}
if run.frames.contains_key(&frame_id) {
return Ok(false);
}
run.flow_state = next_run_state;
run.loops.insert(loop_instance_id, next_loop);
run.frames.insert(frame_id, initial_frame);
append_loop_iteration_ledger_if_absent(run, ledger_entry);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_complete_body_frame_with_authority(
&self,
run_id: &RunId,
loop_instance_id: &LoopInstanceId,
expected_loop: &LoopSnapshot,
next_loop: LoopSnapshot,
frame_id: &FrameId,
expected_frame: &FrameSnapshot,
next_frame: FrameSnapshot,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let loop_instance_id = loop_instance_id.clone();
let expected_loop = expected_loop.clone();
let frame_id = frame_id.clone();
let expected_frame = expected_frame.clone();
let expected_run_state = expected_run_state.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.loops.get(&loop_instance_id) != Some(&expected_loop) {
return Ok(false);
}
if run.frames.get(&frame_id) != Some(&expected_frame) {
return Ok(false);
}
run.flow_state = next_run_state;
run.loops.insert(loop_instance_id, next_loop);
run.frames.insert(frame_id, next_frame);
Ok(true)
},
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn cas_complete_loop_with_authority(
&self,
run_id: &RunId,
loop_instance_id: &LoopInstanceId,
expected_loop: &LoopSnapshot,
next_loop: LoopSnapshot,
frame_id: &FrameId,
expected_frame: &FrameSnapshot,
next_frame: FrameSnapshot,
expected_run_state: &flow_run::State,
next_run_state: flow_run::State,
authority_inputs: Vec<mob_dsl::MobMachineInput>,
) -> Result<bool, MobStoreError> {
let loop_instance_id = loop_instance_id.clone();
let expected_loop = expected_loop.clone();
let frame_id = frame_id.clone();
let expected_frame = expected_frame.clone();
let expected_run_state = expected_run_state.clone();
self.update_run_with_authority_if(
run_id,
MissingRunCasBehavior::NotFound,
authority_inputs,
move |run| {
if run.flow_state != expected_run_state {
return Ok(false);
}
if run.loops.get(&loop_instance_id) != Some(&expected_loop) {
return Ok(false);
}
if run.frames.get(&frame_id) != Some(&expected_frame) {
return Ok(false);
}
run.flow_state = next_run_state;
run.loops.insert(loop_instance_id, next_loop);
run.frames.insert(frame_id, next_frame);
Ok(true)
},
)
.await
}
}
#[derive(Clone)]
pub struct SqliteMobSpecStore {
path: PathBuf,
}
impl std::fmt::Debug for SqliteMobSpecStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteMobSpecStore")
.field("path", &self.path)
.finish()
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct StoredSpec {
definition: MobDefinition,
revision: u64,
}
#[async_trait]
impl MobSpecStore for SqliteMobSpecStore {
async fn put_spec(
&self,
mob_id: &MobId,
definition: &MobDefinition,
revision: Option<u64>,
) -> Result<u64, MobStoreError> {
let path = self.path.clone();
let key = mob_id.to_string();
let mob_id = mob_id.clone();
let definition = definition.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let current: Option<Vec<u8>> = tx
.query_row(
"SELECT spec_json FROM mob_specs WHERE mob_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let current_revision = match current {
Some(bytes) => decode_json::<StoredSpec>(&bytes)?.revision,
None => 0,
};
if let Some(expected) = revision
&& expected != current_revision
{
return Err(MobStoreError::SpecRevisionConflict {
mob_id,
expected: revision,
actual: current_revision,
});
}
let next_revision = current_revision + 1;
let payload = StoredSpec {
definition,
revision: next_revision,
};
let encoded = encode_json(&payload)?;
tx.execute(
"INSERT OR REPLACE INTO mob_specs (mob_id, spec_json) VALUES (?1, ?2)",
params![key, encoded],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(next_revision)
})
.await
}
async fn get_spec(
&self,
mob_id: &MobId,
) -> Result<Option<(MobDefinition, u64)>, MobStoreError> {
let path = self.path.clone();
let key = mob_id.to_string();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let bytes: Option<Vec<u8>> = conn
.query_row(
"SELECT spec_json FROM mob_specs WHERE mob_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
match bytes {
Some(b) => {
let stored: StoredSpec = decode_json(&b)?;
Ok(Some((stored.definition, stored.revision)))
}
None => Ok(None),
}
})
.await
}
async fn list_specs(&self) -> Result<Vec<MobId>, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn.prepare("SELECT mob_id FROM mob_specs").map_err(se)?;
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.map_err(se)?;
let mut result = Vec::new();
for row in rows {
let id = row.map_err(se)?;
result.push(MobId::from(id));
}
Ok(result)
})
.await
}
async fn delete_spec(
&self,
mob_id: &MobId,
revision: Option<u64>,
) -> Result<bool, MobStoreError> {
let path = self.path.clone();
let key = mob_id.to_string();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let bytes: Option<Vec<u8>> = tx
.query_row(
"SELECT spec_json FROM mob_specs WHERE mob_id = ?1",
params![key],
|row| row.get(0),
)
.optional()
.map_err(se)?;
let Some(bytes) = bytes else {
return Ok(false);
};
let stored: StoredSpec = decode_json(&bytes)?;
if let Some(expected) = revision
&& expected != stored.revision
{
return Ok(false);
}
tx.execute("DELETE FROM mob_specs WHERE mob_id = ?1", params![key])
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(true)
})
.await
}
}
#[derive(Clone)]
pub struct SqliteRealmProfileStore {
path: PathBuf,
}
impl SqliteRealmProfileStore {
pub fn open(db_path: &std::path::Path) -> Result<Self, MobStoreError> {
let conn = open_connection(db_path)?;
drop(conn);
Ok(Self {
path: db_path.to_path_buf(),
})
}
}
impl std::fmt::Debug for SqliteRealmProfileStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SqliteRealmProfileStore")
.field("path", &self.path)
.finish()
}
}
#[async_trait]
impl RealmProfileStore for SqliteRealmProfileStore {
async fn create(
&self,
name: &str,
profile: &Profile,
) -> Result<StoredRealmProfile, MobStoreError> {
let path = self.path.clone();
let name = name.to_string();
let profile = profile.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let exists: bool = tx
.query_row(
"SELECT 1 FROM realm_profiles WHERE name = ?1",
params![name],
|_| Ok(true),
)
.optional()
.map_err(se)?
.unwrap_or(false);
if exists {
return Err(MobStoreError::CasConflict(format!(
"realm profile already exists: {name}"
)));
}
let now = Utc::now();
let now_str = now.to_rfc3339();
let profile_json = encode_json(&profile)?;
tx.execute(
"INSERT INTO realm_profiles (name, profile_json, revision, created_at, updated_at) VALUES (?1, ?2, 1, ?3, ?4)",
params![name, profile_json, now_str, now_str],
)
.map_err(se)?;
tx.commit().map_err(se)?;
Ok(StoredRealmProfile {
name,
profile,
revision: 1,
created_at: now,
updated_at: now,
})
})
.await
}
async fn get(&self, name: &str) -> Result<Option<StoredRealmProfile>, MobStoreError> {
let path = self.path.clone();
let name = name.to_string();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let row: Option<(Vec<u8>, i64, String, String)> = conn
.query_row(
"SELECT profile_json, revision, created_at, updated_at FROM realm_profiles WHERE name = ?1",
params![name],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.optional()
.map_err(se)?;
match row {
Some((bytes, revision, created_at_str, updated_at_str)) => {
let profile: Profile = decode_json(&bytes)?;
let created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
let updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
Ok(Some(StoredRealmProfile {
name,
profile,
revision: revision as u64,
created_at,
updated_at,
}))
}
None => Ok(None),
}
})
.await
}
async fn list(&self) -> Result<Vec<StoredRealmProfile>, MobStoreError> {
let path = self.path.clone();
run_sqlite_task(move || {
let conn = open_connection(&path)?;
let mut stmt = conn
.prepare("SELECT name, profile_json, revision, created_at, updated_at FROM realm_profiles ORDER BY name")
.map_err(se)?;
let rows = stmt
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, i64>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
))
})
.map_err(se)?;
let mut result = Vec::new();
for row in rows {
let (name, bytes, revision, created_at_str, updated_at_str) = row.map_err(se)?;
let profile: Profile = decode_json(&bytes)?;
let created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
let updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
result.push(StoredRealmProfile {
name,
profile,
revision: revision as u64,
created_at,
updated_at,
});
}
Ok(result)
})
.await
}
async fn update(
&self,
name: &str,
profile: &Profile,
expected_revision: u64,
) -> Result<StoredRealmProfile, MobStoreError> {
let path = self.path.clone();
let name = name.to_string();
let profile = profile.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<(i64, String)> = tx
.query_row(
"SELECT revision, created_at FROM realm_profiles WHERE name = ?1",
params![name],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()
.map_err(se)?;
let (current_revision, created_at_str) = row.ok_or_else(|| {
MobStoreError::NotFound(format!("realm profile not found: {name}"))
})?;
if current_revision as u64 != expected_revision {
return Err(MobStoreError::CasConflict(format!(
"realm profile '{name}' revision conflict: expected {expected_revision}, actual {current_revision}"
)));
}
let next_revision = expected_revision + 1;
let now = Utc::now();
let now_str = now.to_rfc3339();
let profile_json = encode_json(&profile)?;
tx.execute(
"UPDATE realm_profiles SET profile_json = ?1, revision = ?2, updated_at = ?3 WHERE name = ?4",
params![profile_json, next_revision as i64, now_str, name],
)
.map_err(se)?;
tx.commit().map_err(se)?;
let created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
Ok(StoredRealmProfile {
name,
profile,
revision: next_revision,
created_at,
updated_at: now,
})
})
.await
}
async fn delete(
&self,
name: &str,
expected_revision: u64,
) -> Result<StoredRealmProfile, MobStoreError> {
let path = self.path.clone();
let name = name.to_string();
run_sqlite_task(move || {
let mut conn = open_connection(&path)?;
let tx = begin_immediate(&mut conn)?;
let row: Option<(Vec<u8>, i64, String, String)> = tx
.query_row(
"SELECT profile_json, revision, created_at, updated_at FROM realm_profiles WHERE name = ?1",
params![name],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.optional()
.map_err(se)?;
let (bytes, current_revision, created_at_str, updated_at_str) = row.ok_or_else(|| {
MobStoreError::NotFound(format!("realm profile not found: {name}"))
})?;
if current_revision as u64 != expected_revision {
return Err(MobStoreError::CasConflict(format!(
"realm profile '{name}' revision conflict: expected {expected_revision}, actual {current_revision}"
)));
}
let profile: Profile = decode_json(&bytes)?;
tx.execute(
"DELETE FROM realm_profiles WHERE name = ?1",
params![name],
)
.map_err(se)?;
tx.commit().map_err(se)?;
let created_at = chrono::DateTime::parse_from_rfc3339(&created_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
let updated_at = chrono::DateTime::parse_from_rfc3339(&updated_at_str)
.map_err(|e| MobStoreError::Serialization(e.to_string()))?
.with_timezone(&Utc);
Ok(StoredRealmProfile {
name,
profile,
revision: expected_revision,
created_at,
updated_at,
})
})
.await
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use crate::definition::{BackendConfig, FlowSpec, WiringRules};
use crate::event::{MemberRef, MobEventKind};
use crate::ids::{AgentIdentity, Generation, ProfileName};
use crate::profile::{Profile, ProfileBinding, ToolConfig};
use crate::run::StepRunStatus;
use crate::store::{
ExternalBindingOverlayStatus, MobExternalDeliveryIdentity, MobExternalDeliveryTargetKind,
};
use futures::future::join_all;
use indexmap::IndexMap;
use meerkat_contracts::wire::{
PortableDefinitionExtract, PortableProfile, PortableSystemPrompt,
};
use meerkat_core::ops::OperationId;
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Debug)]
struct TestIdentityClock {
now_ms: AtomicU64,
}
impl TestIdentityClock {
fn new(now_ms: u64) -> Self {
Self {
now_ms: AtomicU64::new(now_ms),
}
}
fn set(&self, now_ms: u64) {
self.now_ms.store(now_ms, Ordering::SeqCst);
}
}
impl MobIdentityStoreClock for TestIdentityClock {
fn now_ms(&self) -> Result<u64, MobStoreError> {
Ok(self.now_ms.load(Ordering::SeqCst))
}
}
fn sqlite_external_delivery_intent(key: &str, action: &str) -> MobExternalDeliveryIntent {
MobExternalDeliveryIntent::new(
MobId::from("mob"),
MobExternalDeliveryIdentity::new(
key,
uuid::Uuid::from_u128(0xD311_0000_0000_0000_0000_0000_0000_0002).to_string(),
)
.expect("valid external delivery identity"),
MobExternalDeliveryTargetKind::Flow,
action.as_bytes(),
)
.expect("valid external delivery intent")
}
#[tokio::test]
async fn sqlite_external_delivery_ledger_survives_reopen_and_rejects_action_conflict() {
let (_dir, path) = temp_db_path();
let intent = sqlite_external_delivery_intent("schedule:sqlite", "action-a");
let terminal = MobExternalDeliveryTerminal::Completed;
{
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
assert_eq!(
events.begin_external_delivery(&intent).await.unwrap(),
MobExternalDeliveryBeginOutcome::Begun
);
assert!(matches!(
events
.schedule_external_delivery_repair(&intent)
.await
.unwrap(),
MobExternalDeliveryRepairOutcome::Scheduled(MobExternalDeliveryRepairState {
attempt: 1,
..
})
));
}
{
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
assert!(matches!(
events.begin_external_delivery(&intent).await.unwrap(),
MobExternalDeliveryBeginOutcome::ExistingBegun {
repair: MobExternalDeliveryRepairState { attempt: 1, .. }
}
));
assert_eq!(
events
.complete_external_delivery(&intent, &terminal)
.await
.unwrap(),
MobExternalDeliveryCompleteOutcome::Completed
);
}
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
assert_eq!(
events.begin_external_delivery(&intent).await.unwrap(),
MobExternalDeliveryBeginOutcome::ExistingTerminal(terminal)
);
let conflict = sqlite_external_delivery_intent("schedule:sqlite", "action-b");
assert!(matches!(
events.begin_external_delivery(&conflict).await,
Err(MobStoreError::ExternalDeliveryConflict { .. })
));
}
fn default_bridge_protocol_version()
-> meerkat_contracts::wire::supervisor_bridge::BridgeProtocolVersion {
meerkat_contracts::wire::supervisor_bridge::supervisor_bridge_default_protocol_version()
}
fn temp_db_path() -> (tempfile::TempDir, std::path::PathBuf) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("mob.db");
(dir, path)
}
#[test]
fn released_v3_migrates_directly_to_canonical_v4_schema() {
let (_dir, path) = temp_db_path();
{
let mut conn = Connection::open(&path).unwrap();
let tx = conn.transaction().unwrap();
build_released_0_8_10_mob_schema(&tx).unwrap();
tx.execute_batch(
"CREATE TABLE meerkat_schema (
domain TEXT PRIMARY KEY,
version INTEGER NOT NULL
);
INSERT INTO meerkat_schema VALUES ('mob', 3);
INSERT INTO mob_identity_operation_receipts (
mob_id, subject_kind, subject_id, slot_kind,
slot_key_version, slot_digest, subject_json, slot_json,
subject_identity, effect_kind, receipt_json,
subject_scope_id
) VALUES (
'mob', 'identity', 'member-a', 'ensure', 1, 'digest',
X'7b7d', X'7b7d', 'member-a', 'completed', X'7b7d',
'obsolete-scope'
);
INSERT INTO mob_identity_operation_receipts (
mob_id, subject_kind, subject_id, slot_kind,
slot_key_version, slot_digest, subject_json, slot_json,
subject_identity, effect_kind, receipt_json,
subject_scope_id
) VALUES (
'mob', 'declaration_scope', 'obsolete-scope',
'apply_declaration_manifest', 1, 'obsolete-digest',
X'7b7d', X'7b7d', NULL, 'apply_declaration_manifest',
X'7b7d', 'obsolete-scope'
);",
)
.unwrap();
tx.commit().unwrap();
}
drop(SqliteMobStores::open(&path).unwrap());
let conn = Connection::open(&path).unwrap();
assert_eq!(
conn.query_row(
"SELECT version FROM meerkat_schema WHERE domain = 'mob'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
4
);
for retired in [
"mob_identity_declaration_scopes",
"mob_identity_receipts_scope",
"mob_identity_intents_scope",
] {
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE name = ?1",
[retired],
|row| row.get::<_, i64>(0),
)
.unwrap(),
0,
"{retired} must not survive the released-v3 migration"
);
}
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM sqlite_master
WHERE type = 'table' AND name = 'mob_external_deliveries'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
1
);
assert!(
!conn
.prepare("PRAGMA table_info(mob_identity_operation_receipts)")
.unwrap()
.query_map([], |row| row.get::<_, String>(1))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap()
.iter()
.any(|column| column == "subject_scope_id")
);
assert_eq!(
conn.query_row(
"SELECT subject_identity, receipt_json
FROM mob_identity_operation_receipts
WHERE mob_id = 'mob'",
[],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?)),
)
.unwrap(),
("member-a".to_string(), b"{}".to_vec())
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*)
FROM mob_identity_operation_receipts
WHERE subject_kind = 'declaration_scope'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
0,
"retired declaration-facade receipts must not become undecodable poison rows"
);
}
fn operator_request(
identity: &str,
generation: u64,
sequence: usize,
) -> MobMemberOperatorRequestRecord {
MobMemberOperatorRequestRecord::pending(
MobMemberOperatorRequestKey::new(
identity,
generation,
1,
"host-a",
1,
"member-session-a",
format!("request-{sequence}"),
),
"0".repeat(64),
)
}
fn seed_operator_requests(
path: &Path,
mob_id: &MobId,
records: &[MobMemberOperatorRequestRecord],
) {
let mut conn = open_connection(path).expect("open quota seed connection");
let tx = begin_immediate(&mut conn).expect("begin quota seed transaction");
{
let mut insert = tx
.prepare(
"INSERT INTO mob_runtime_member_operator_requests
(mob_id, agent_identity, generation, fence_token, host_id, binding_generation, member_session_id, request_id, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
)
.expect("prepare quota seed insert");
for record in records {
record.validate_pending().expect("valid quota seed row");
insert
.execute(params![
mob_id.as_str(),
&record.agent_identity,
u64_to_sql_key(record.generation),
u64_to_sql_key(record.fence_token),
&record.host_id,
u64_to_sql_key(record.host_binding_generation),
&record.member_session_id,
&record.request_id,
encode_json(record).expect("encode quota seed row"),
])
.expect("insert quota seed row");
}
}
tx.commit().expect("commit quota seed transaction");
}
fn operator_request_count(path: &Path, mob_id: &MobId) -> usize {
let conn = open_connection(path).expect("open quota count connection");
let count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM mob_runtime_member_operator_requests WHERE mob_id = ?1",
params![mob_id.as_str()],
|row| row.get(0),
)
.expect("count operator request rows");
usize::try_from(count).expect("non-negative operator request count")
}
fn sample_definition() -> MobDefinition {
let mut profiles = std::collections::BTreeMap::new();
profiles.insert(
ProfileName::from("worker"),
ProfileBinding::Inline(Box::new(Profile {
model: "model".to_string(),
provider: None,
self_hosted_server_id: None,
image_generation_provider: None,
auto_compact_threshold: None,
resume_overrides: Vec::new(),
skills: Vec::new(),
tools: ToolConfig::default(),
peer_description: "worker".to_string(),
external_addressable: false,
backend: None,
runtime_mode: crate::MobRuntimeMode::AutonomousHost,
max_inline_peer_notifications: None,
output_schema: None,
provider_params: None,
})),
);
let mut definition = MobDefinition::explicit("mob");
definition.profiles = profiles;
definition.flows = {
let mut flows = std::collections::BTreeMap::new();
flows.insert(
FlowId::from("flow-a"),
FlowSpec::new(None, IndexMap::new(), None),
);
flows
};
definition
}
fn sample_run(status: MobRunStatus) -> MobRun {
MobRun::authority_backed_for_steps(
RunId::new(),
MobId::from("mob"),
FlowId::from("flow-a"),
[crate::ids::StepId::from("step-1")],
status,
serde_json::json!({"a":1}),
)
.expect("authority-backed sample run")
}
#[tokio::test]
async fn sqlite_identity_live_handle_never_recreates_a_deleted_database() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let store = stores.identity_store();
drop(stores);
std::fs::remove_file(&path).unwrap();
assert!(matches!(
store
.observe_identity_intent(
&MobId::from("deleted-db"),
&AgentIdentity::from("member-a"),
)
.await,
Err(MobStoreError::ReadFailed(_))
));
assert!(
!path.exists(),
"identity read must not recreate the database"
);
}
#[tokio::test]
async fn sqlite_identity_future_schema_is_unsupported_before_typed_decode() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let mob_id = MobId::from("sqlite-identity-future-schema");
let bytes = br#"{"schema_version":2,"future_shape":true}"#.to_vec();
{
let conn = open_connection(&stores.path).unwrap();
conn.execute(
"INSERT INTO mob_identity_intents
(mob_id, agent_identity, declaration_scope, session_id, lineage_id, record_json)
VALUES (?1, ?2, NULL, NULL, NULL, ?3)",
params![mob_id.as_str(), "future-member", bytes],
)
.unwrap();
}
match stores
.identity_store()
.observe_identity_intent(&mob_id, &AgentIdentity::from("future-member"))
.await
.unwrap()
{
IdentityStoredObservation::Unsupported {
evidence_digest, ..
} => assert_eq!(
evidence_digest,
identity_raw_evidence_digest(br#"{"schema_version":2,"future_shape":true}"#)
),
other => panic!("expected unsupported future schema, got {other:?}"),
}
}
#[tokio::test]
async fn sqlite_identity_intent_scan_is_keyset_paged_and_strictly_bounded() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let mob_id = MobId::from("sqlite-identity-page");
{
let conn = open_connection(&stores.path).unwrap();
for identity in ["member-c", "member-a", "member-e", "member-b", "member-d"] {
conn.execute(
"INSERT INTO mob_identity_intents
(mob_id, agent_identity, declaration_scope, session_id, lineage_id, record_json)
VALUES (?1, ?2, NULL, NULL, NULL, ?3)",
params![
mob_id.as_str(),
identity,
br#"{"schema_version":2,"future_shape":true}"#.as_slice()
],
)
.unwrap();
}
}
let store = stores.identity_store();
let first = store
.scan_identity_intents_page(&mob_id, None, 2)
.await
.unwrap();
assert_eq!(
first
.observations
.keys()
.map(AgentIdentity::as_str)
.collect::<Vec<_>>(),
["member-a", "member-b"]
);
let second = store
.scan_identity_intents_page(&mob_id, first.next.as_ref(), 2)
.await
.unwrap();
assert_eq!(
second
.observations
.keys()
.map(AgentIdentity::as_str)
.collect::<Vec<_>>(),
["member-c", "member-d"]
);
let third = store
.scan_identity_intents_page(&mob_id, second.next.as_ref(), 2)
.await
.unwrap();
assert_eq!(
third
.observations
.keys()
.map(AgentIdentity::as_str)
.collect::<Vec<_>>(),
["member-e"]
);
assert!(third.next.is_none());
}
#[tokio::test]
async fn sqlite_identity_intent_scan_advances_past_malformed_physical_keys() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let mob_id = MobId::from("sqlite-identity-malformed-key-page");
{
let conn = open_connection(&stores.path).unwrap();
for identity in [
rusqlite::types::Value::Blob(b"member-blob".to_vec()),
rusqlite::types::Value::Text("member-text".to_string()),
] {
conn.execute(
"INSERT INTO mob_identity_intents
(mob_id, agent_identity, declaration_scope, session_id, lineage_id, record_json)
VALUES (?1, ?2, NULL, NULL, NULL, ?3)",
params![
mob_id.as_str(),
identity,
br#"{"schema_version":2,"future_shape":true}"#.as_slice()
],
)
.unwrap();
}
}
let store = stores.identity_store();
let malformed = store
.scan_identity_intents_page(&mob_id, None, 1)
.await
.unwrap();
assert_eq!(malformed.observations.len(), 1);
assert!(
malformed
.observations
.keys()
.next()
.is_some_and(|identity| identity
.as_str()
.starts_with("malformed-physical-identity:blob:"))
);
let valid = store
.scan_identity_intents_page(&mob_id, malformed.next.as_ref(), 1)
.await
.unwrap();
assert_eq!(
valid.observations.keys().next().map(AgentIdentity::as_str),
Some("member-text")
);
assert!(valid.next.is_none());
}
#[tokio::test]
async fn sqlite_identity_lease_takeover_strictly_advances_epoch() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let clock = Arc::new(TestIdentityClock::new(100));
let store = SqliteMobIdentityStore::with_clock(
stores.path.clone(),
stores.identity_store_instance_id.clone(),
clock.clone(),
);
let mob_id = MobId::from("sqlite-identity-lease");
let identity = AgentIdentity::from("member-a");
let first = match store
.claim_or_renew_identity_lease(&mob_id, &identity, "controller", "inc-a", 10)
.await
.unwrap()
{
IdentityLeaseClaimOutcome::Acquired(claim) => claim,
other => panic!("expected acquire, got {other:?}"),
};
assert!(matches!(
store
.claim_or_renew_identity_lease(&mob_id, &identity, "controller", "inc-b", 10)
.await
.unwrap(),
IdentityLeaseClaimOutcome::HeldByOther(_)
));
clock.set(first.expires_at_ms);
let takeover = match store
.claim_or_renew_identity_lease(&mob_id, &identity, "controller", "inc-b", 10)
.await
.unwrap()
{
IdentityLeaseClaimOutcome::Acquired(claim) => claim,
other => panic!("expected takeover, got {other:?}"),
};
assert_eq!(takeover.epoch, first.epoch + 1);
assert!(
!store
.release_identity_lease(&mob_id, &identity, &first)
.await
.unwrap()
);
}
#[tokio::test]
async fn sqlite_identity_lease_refuses_to_overwrite_malformed_authority() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let clock = Arc::new(TestIdentityClock::new(100));
let store = SqliteMobIdentityStore::with_clock(
stores.path.clone(),
stores.identity_store_instance_id.clone(),
clock.clone(),
);
let mob_id = MobId::from("sqlite-identity-lease-projection-repair");
let identity = AgentIdentity::from("member-a");
let first = match store
.claim_or_renew_identity_lease(&mob_id, &identity, "controller", "inc-a", 10)
.await
.unwrap()
{
IdentityLeaseClaimOutcome::Acquired(claim) => claim,
other => panic!("expected acquire, got {other:?}"),
};
{
let conn = open_connection(&stores.path).unwrap();
conn.execute(
"UPDATE mob_identity_leases SET record_json = ?3
WHERE mob_id = ?1 AND agent_identity = ?2",
params![mob_id.as_str(), identity.as_str(), b"{".as_slice()],
)
.unwrap();
}
assert!(matches!(
store
.observe_identity_lease(&mob_id, &identity)
.await
.unwrap(),
IdentityStoredObservation::Malformed { .. }
));
clock.set(first.expires_at_ms);
assert!(matches!(
store
.claim_or_renew_identity_lease(&mob_id, &identity, "controller", "inc-b", 10)
.await,
Err(MobStoreError::IdentityAuthorityBlocked { .. })
));
assert!(matches!(
store
.observe_identity_lease(&mob_id, &identity)
.await
.unwrap(),
IdentityStoredObservation::Malformed { .. }
));
}
#[tokio::test]
async fn sqlite_identity_wiring_is_mob_local_and_reset_heals_target_garbage() {
let (_dir, path) = temp_db_path();
let target_mob = MobId::from("sqlite-wiring-target");
let malformed_mob = MobId::from("sqlite-wiring-malformed");
let identity = AgentIdentity::from("member-a");
let stores = SqliteMobStores::open(&path).unwrap();
let encoded = encode_stored_mob_event(&MobEvent {
cursor: 1,
timestamp: Utc::now(),
mob_id: malformed_mob.clone(),
kind: MobEventKind::MobCompleted,
})
.unwrap();
let mut malformed_value = serde_json::from_slice::<serde_json::Value>(&encoded).unwrap();
malformed_value["event"]["kind"]["type"] =
serde_json::Value::String("future_structural_event".to_string());
let malformed_bytes = serde_json::to_vec(&malformed_value).unwrap();
let malformed_digest = identity_raw_evidence_digest(&malformed_bytes);
{
let conn = Connection::open(&path).unwrap();
conn.execute(
"INSERT INTO mob_events (cursor, mob_id, event_json) VALUES (?1, NULL, ?2)",
params![1i64, &malformed_bytes],
)
.unwrap();
}
drop(stores);
let stores = SqliteMobStores::open(&path).unwrap();
let physical_route: String = Connection::open(&path)
.unwrap()
.query_row(
"SELECT mob_id FROM mob_events WHERE cursor = 1",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(physical_route, malformed_mob.as_str());
assert!(matches!(
stores
.identity_store()
.observe_identity_wiring_target(&target_mob, &identity)
.await
.unwrap(),
IdentityWiringTargetObservation::Absent { .. }
));
match stores
.identity_store()
.observe_identity_wiring_target(&malformed_mob, &identity)
.await
.unwrap()
{
IdentityWiringTargetObservation::Malformed {
observed_version, ..
} => assert_eq!(observed_version.as_deref(), Some(malformed_digest.as_str())),
other => panic!("expected target-local malformed event evidence, got {other:?}"),
}
let reset = stores
.event_store()
.append(NewMobEvent {
mob_id: malformed_mob.clone(),
timestamp: None,
kind: MobEventKind::MobReset,
})
.await
.unwrap();
assert_eq!(reset.cursor, 2);
assert!(matches!(
stores
.identity_store()
.observe_identity_wiring_target(&malformed_mob, &identity)
.await
.unwrap(),
IdentityWiringTargetObservation::Absent { .. }
));
}
#[tokio::test]
async fn sqlite_event_cursor_repairs_regressed_meta_in_transaction_and_missing_meta_on_open() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
let first = events
.append(NewMobEvent {
mob_id: MobId::from("cursor-repair"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
assert_eq!(first.cursor, 1);
{
let mut conn = Connection::open(&path).unwrap();
conn.execute(
"UPDATE mob_event_meta SET value = 1 WHERE key = ?1",
params![EVENT_CURSOR_KEY],
)
.unwrap();
let tx = begin_immediate(&mut conn).unwrap();
assert_eq!(next_event_cursor(&tx).unwrap(), 2);
tx.commit().unwrap();
assert_eq!(get_next_cursor(&conn).unwrap(), 2);
}
let second = events
.append(NewMobEvent {
mob_id: MobId::from("cursor-repair"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
assert_eq!(second.cursor, 2);
{
let conn = Connection::open(&path).unwrap();
conn.execute(
"DELETE FROM mob_event_meta WHERE key = ?1",
params![EVENT_CURSOR_KEY],
)
.unwrap();
}
drop(events);
drop(stores);
let reopened = SqliteMobStores::open(&path).unwrap();
let repaired_meta: i64 = Connection::open(&path)
.unwrap()
.query_row(
"SELECT value FROM mob_event_meta WHERE key = ?1",
params![EVENT_CURSOR_KEY],
|row| row.get(0),
)
.unwrap();
assert_eq!(repaired_meta, 3);
let third = reopened
.event_store()
.append(NewMobEvent {
mob_id: MobId::from("cursor-repair"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
assert_eq!(third.cursor, 3);
assert_eq!(
reopened
.event_store()
.replay_all()
.await
.unwrap()
.into_iter()
.map(|event| event.cursor)
.collect::<Vec<_>>(),
vec![1, 2, 3]
);
}
#[tokio::test]
async fn test_sqlite_event_store_append_poll_replay_prune() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().event_store();
let now = Utc::now();
store
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: Some(now - chrono::Duration::minutes(10)),
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
store
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: Some(now),
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
let polled = store.poll(0, 10).await.unwrap();
assert_eq!(polled.len(), 2);
let removed = store
.prune(now - chrono::Duration::minutes(1))
.await
.unwrap();
assert_eq!(removed, 1);
let replayed = store.replay_all().await.unwrap();
assert_eq!(replayed.len(), 1);
}
#[test]
fn test_sqlite_event_bus_catch_up_does_not_recreate_deleted_database() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
assert!(
path.exists(),
"opening the store should create the database"
);
std::fs::remove_file(&path).unwrap();
assert!(
!path.exists(),
"test setup should remove the database before watcher catch-up"
);
stores.event_bus.publish_available_from_storage().unwrap();
assert!(
!path.exists(),
"watcher catch-up must not recreate intentionally deleted mob storage"
);
}
#[test]
fn test_sqlite_event_bus_catch_up_open_does_not_recreate_database_deleted_after_precheck() {
let (_dir, path) = temp_db_path();
let _stores = SqliteMobStores::open(&path).unwrap();
assert!(
path.try_exists().unwrap(),
"watcher preflight should observe the database before the simulated race"
);
std::fs::remove_file(&path).unwrap();
let error = poll_events_sync(&path, 0, EVENT_WATCH_CATCH_UP_LIMIT)
.expect_err("open-existing catch-up must reject a concurrently deleted database");
assert!(
!path.exists(),
"passive catch-up must not recreate a database deleted after its preflight check: {error}"
);
}
#[tokio::test]
async fn test_sqlite_terminal_reads_do_not_recreate_deleted_database() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let event_store = stores.event_store();
let run_store = stores.run_store();
let runtime_store = stores.runtime_metadata_store();
std::fs::remove_file(&path).unwrap();
assert!(event_store.poll(0, 1).await.is_err());
assert!(event_store.replay_all().await.is_err());
assert!(event_store.latest_cursor().await.is_err());
let run_id = RunId::new();
assert!(run_store.get_run(&run_id).await.is_err());
assert!(
run_store
.list_runs(&MobId::from("deleted-mob"), None)
.await
.is_err()
);
assert!(run_store.list_remote_turn_intents(&run_id).await.is_err());
assert!(run_store.list_remote_turn_receipts(&run_id).await.is_err());
assert!(
runtime_store
.load_placed_spawn(&MobId::from("deleted-mob"), "deleted-agent")
.await
.is_err()
);
assert!(
runtime_store
.list_placed_spawns(&MobId::from("deleted-mob"))
.await
.is_err()
);
assert!(
!path.exists(),
"terminal storage reads must not recreate removed mob storage"
);
}
#[tokio::test]
async fn test_sqlite_event_store_subscribe_receives_appended_events() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().event_store();
let mut rx = store.subscribe().expect("subscribe");
let stored = store
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
let observed = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
.await
.expect("subscription should receive appended event")
.expect("subscription should stay open");
assert_eq!(observed.cursor, stored.cursor);
assert!(matches!(observed.kind, MobEventKind::MobCompleted));
}
#[tokio::test]
async fn test_sqlite_event_store_subscribe_receives_appends_from_separate_open() {
let (_dir, path) = temp_db_path();
let subscriber_store = SqliteMobStores::open(&path).unwrap().event_store();
let writer_store = SqliteMobStores::open(&path).unwrap().event_store();
let mut rx = subscriber_store.subscribe().expect("subscribe");
let stored = writer_store
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
let observed = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
.await
.expect("subscription should receive event from separately opened writer")
.expect("subscription should stay open");
assert_eq!(observed.cursor, stored.cursor);
assert!(matches!(observed.kind, MobEventKind::MobCompleted));
}
#[tokio::test]
async fn test_sqlite_event_store_subscribe_receives_raw_sqlite_append() {
let (_dir, path) = temp_db_path();
let subscriber_store = SqliteMobStores::open(&path).unwrap().event_store();
let mut rx = subscriber_store.subscribe().expect("subscribe");
let stored = MobEvent {
cursor: 1,
timestamp: Utc::now(),
mob_id: MobId::from("mob"),
kind: MobEventKind::MobCompleted,
};
let encoded = encode_stored_mob_event(&stored).unwrap();
let writer_path = path.clone();
run_sqlite_task(move || {
let mut conn = open_connection(&writer_path)?;
let tx = begin_immediate(&mut conn)?;
tx.execute(
"INSERT INTO mob_events (cursor, event_json) VALUES (?1, ?2)",
params![cursor_to_i64(stored.cursor)?, encoded],
)
.map_err(se)?;
set_next_cursor(&tx, stored.cursor.saturating_add(1))?;
tx.commit().map_err(se)?;
Ok(())
})
.await
.unwrap();
let observed = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
.await
.expect("subscription should receive raw sqlite append")
.expect("subscription should stay open");
assert_eq!(observed.cursor, 1);
assert!(matches!(observed.kind, MobEventKind::MobCompleted));
}
#[tokio::test]
async fn test_sqlite_event_store_rejects_pre_0_6_unversioned_history() {
let (_dir, path) = temp_db_path();
let raw_event = serde_json::to_vec(&MobEvent {
cursor: 1,
timestamp: Utc::now(),
mob_id: MobId::from("mob"),
kind: MobEventKind::MobCompleted,
})
.unwrap();
{
let conn = open_connection(&path).unwrap();
conn.execute(
"INSERT INTO mob_events (cursor, event_json) VALUES (?1, ?2)",
params![1i64, raw_event],
)
.unwrap();
}
let store = SqliteMobStores::open(&path).unwrap().event_store();
let error = store
.replay_all()
.await
.expect_err("pre-0.6 unversioned history must be rejected");
match error {
MobStoreError::Serialization(message) => {
assert!(message.contains("pre-0.6 mob event history is unsupported"));
}
other => panic!("expected serialization error, got {other:?}"),
}
}
#[tokio::test]
async fn test_sqlite_run_store_cas_and_dedup() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().run_store();
let run = sample_run(MobRunStatus::Running);
let run_id = run.run_id.clone();
let expected_flow_state = run.flow_state.clone();
let (completed_flow_state, completed_authority_input) = run
.flow_run_command_projection_for_test(
crate::run::MobMachineFlowRunCommand::TerminalizeCompleted(
crate::run::flow_run::inputs::TerminalizeCompleted {},
),
)
.expect("project completed run state");
let (failed_flow_state, failed_authority_input) = run
.flow_run_command_projection_for_test(
crate::run::MobMachineFlowRunCommand::TerminalizeFailed(
crate::run::flow_run::inputs::TerminalizeFailed {},
),
)
.expect("project failed run state");
store.create_run(run).await.unwrap();
let s1 = store.clone();
let rid1 = run_id.clone();
let state1 = expected_flow_state.clone();
let completed_state = completed_flow_state.clone();
let completed_input = completed_authority_input.clone();
let s2 = store.clone();
let rid2 = run_id.clone();
let state2 = expected_flow_state.clone();
let failed_state = failed_flow_state.clone();
let failed_input = failed_authority_input.clone();
let outcomes = join_all(vec![
tokio::spawn(async move {
s1.cas_run_snapshot_with_authority(
&rid1,
MobRunStatus::Running,
&state1,
MobRunStatus::Completed,
&completed_state,
vec![completed_input],
)
.await
.unwrap()
}),
tokio::spawn(async move {
s2.cas_run_snapshot_with_authority(
&rid2,
MobRunStatus::Running,
&state2,
MobRunStatus::Failed,
&failed_state,
vec![failed_input],
)
.await
.unwrap()
}),
])
.await;
let winners = outcomes
.into_iter()
.map(|join| join.unwrap())
.filter(|value| *value)
.count();
assert_eq!(winners, 1);
let ledger_run = sample_run(MobRunStatus::Running);
let ledger_run_id = ledger_run.run_id.clone();
let step_id = StepId::from("step-1");
let ledger_expected_flow_state = ledger_run.flow_state.clone();
let (dispatched_flow_state, dispatched_authority_input) = ledger_run
.flow_run_command_projection_for_test(
crate::run::MobMachineFlowRunCommand::DispatchStep(
crate::run::flow_run::inputs::DispatchStep {
step_id: step_id.clone(),
},
),
)
.expect("project dispatched step state");
store.create_run(ledger_run).await.unwrap();
assert!(
store
.cas_flow_state_with_authority(
&ledger_run_id,
&ledger_expected_flow_state,
&dispatched_flow_state,
vec![dispatched_authority_input.clone()],
)
.await
.unwrap()
);
let dispatched_authority =
MobRunProvenanceAuthority::from_flow_authority_input(dispatched_authority_input)
.expect("dispatch input is provenance authority");
let entry = StepLedgerEntry {
step_id,
agent_identity: AgentIdentity::flow_system_provenance(),
status: StepRunStatus::Dispatched,
output: None,
timestamp: Utc::now(),
};
assert!(
store
.append_step_entry_if_absent_with_authority(
&ledger_run_id,
entry.clone(),
dispatched_authority.clone(),
)
.await
.unwrap()
);
assert!(
!store
.append_step_entry_if_absent_with_authority(
&ledger_run_id,
entry,
dispatched_authority.clone(),
)
.await
.unwrap()
);
store
.append_step_entry_with_authority(
&ledger_run_id,
StepLedgerEntry {
step_id: StepId::from("step-1"),
agent_identity: AgentIdentity::flow_system_provenance(),
status: StepRunStatus::Dispatched,
output: None,
timestamp: Utc::now(),
},
dispatched_authority,
)
.await
.expect_err("one dispatch authority must not authorize duplicate step ledger rows");
}
#[tokio::test]
async fn test_sqlite_run_store_create_rejects_preset_provenance_ledgers_without_authority() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().run_store();
let mut run = sample_run(MobRunStatus::Running);
run.step_ledger.push(StepLedgerEntry {
step_id: StepId::from("step-1"),
agent_identity: AgentIdentity::from("worker-1"),
status: StepRunStatus::Dispatched,
output: None,
timestamp: Utc::now(),
});
let error = store
.create_run(run)
.await
.expect_err("raw step provenance must be rejected");
assert!(error.to_string().contains("step ledger entry"));
let mut run = sample_run(MobRunStatus::Running);
run.failure_ledger.push(FailureLedgerEntry {
step_id: StepId::from("step-1"),
reason: "caller-injected failure".to_string(),
error_report: None,
error: None,
timestamp: Utc::now(),
});
let error = store
.create_run(run)
.await
.expect_err("raw failure provenance must be rejected");
assert!(error.to_string().contains("failure ledger entry"));
let mut run = sample_run(MobRunStatus::Running);
run.schema_version = crate::run::mob_run_schema_version() - 1;
let error = store
.create_run(run)
.await
.expect_err("caller-controlled schema version must be rejected");
assert!(error.to_string().contains("schema_version"));
}
#[tokio::test]
async fn test_sqlite_run_store_snapshot_rejects_missing_authority_without_mutation() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().run_store();
let run = sample_run(MobRunStatus::Running);
let run_id = run.run_id.clone();
let expected_flow_state = run.flow_state.clone();
let authority_input_count = run.flow_authority_inputs.len();
store.create_run(run).await.unwrap();
let error = store
.cas_run_snapshot_with_authority(
&run_id,
MobRunStatus::Running,
&expected_flow_state,
MobRunStatus::Completed,
&expected_flow_state,
Vec::new(),
)
.await
.expect_err("missing machine authority must reject snapshot CAS");
assert!(
error
.to_string()
.contains("store mutation missing MobMachine authority input")
);
let stored = store.get_run(&run_id).await.unwrap().unwrap();
assert_eq!(stored.status, MobRunStatus::Running);
assert!(stored.completed_at.is_none());
assert_eq!(stored.flow_authority_inputs.len(), authority_input_count);
}
#[tokio::test]
async fn test_sqlite_terminal_snapshot_and_event_commit_atomically() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
let store = stores.run_store();
let events = stores.event_store();
let run = sample_run(MobRunStatus::Running);
let run_id = run.run_id.clone();
let expected_flow_state = run.flow_state.clone();
let (completed_flow_state, completed_authority_input) = run
.flow_run_command_projection_for_test(
crate::run::MobMachineFlowRunCommand::TerminalizeCompleted(
crate::run::flow_run::inputs::TerminalizeCompleted {},
),
)
.expect("project completed run state");
store.create_run(run).await.unwrap();
let stored = store
.cas_run_snapshot_and_append_terminal_event_with_authority(
&run_id,
MobRunStatus::Running,
&expected_flow_state,
MobRunStatus::Completed,
&completed_flow_state,
vec![completed_authority_input.clone()],
&events,
crate::event::NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: crate::event::MobEventKind::FlowCompleted {
run_id: run_id.clone(),
flow_id: FlowId::from("flow-a"),
structured_output: None,
},
},
)
.await
.expect("atomic terminal commit");
assert!(stored.is_some(), "winning CAS must commit the event");
let run = store.get_run(&run_id).await.unwrap().unwrap();
assert_eq!(run.status, MobRunStatus::Completed);
assert!(run.completed_at.is_some());
let replayed = events.replay_all().await.unwrap();
assert!(replayed.iter().any(|event| matches!(
&event.kind,
crate::event::MobEventKind::FlowCompleted { run_id: event_run, .. } if event_run == &run_id
)));
let lost = store
.cas_run_snapshot_and_append_terminal_event_with_authority(
&run_id,
MobRunStatus::Running,
&expected_flow_state,
MobRunStatus::Completed,
&completed_flow_state,
vec![completed_authority_input],
&events,
crate::event::NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: crate::event::MobEventKind::FlowCompleted {
run_id: run_id.clone(),
flow_id: FlowId::from("flow-a"),
structured_output: None,
},
},
)
.await
.expect("lost CAS is not an error");
assert!(lost.is_none());
assert_eq!(
events.replay_all().await.unwrap().len(),
1,
"a lost CAS must not append a terminal event"
);
}
#[tokio::test]
async fn test_sqlite_spec_store_revision_conflict() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().spec_store();
let definition = sample_definition();
let revision = store
.put_spec(&MobId::from("mob"), &definition, None)
.await
.unwrap();
assert_eq!(revision, 1);
let conflict = store
.put_spec(&MobId::from("mob"), &definition, Some(0))
.await
.expect_err("revision conflict expected");
assert!(matches!(
conflict,
MobStoreError::SpecRevisionConflict { .. }
));
let loaded = store.get_spec(&MobId::from("mob")).await.unwrap();
assert!(loaded.is_some());
}
#[tokio::test]
async fn test_sqlite_member_operator_request_ledger_survives_reopen_and_replays_terminal() {
use meerkat_contracts::wire::supervisor_bridge::{
MemberOperatorOp, MemberOperatorOutcome, MemberOperatorReply, WireOpaqueJson,
};
let (_dir, path) = temp_db_path();
let mob_id = MobId::from("mob-upcall-ledger");
let op = MemberOperatorOp::MobRunFlow {
flow_id: "release".to_string(),
params: None,
};
let pending = MobMemberOperatorRequestRecord::pending(
MobMemberOperatorRequestKey::new(
"worker-1",
u64::MAX - 1,
u64::MAX,
"host-a",
u64::MAX - 2,
"member-session-a",
"req-reopen-1",
),
crate::store::member_operator_op_digest(&op).expect("operation digest"),
);
{
let store = SqliteMobStores::open(&path)
.expect("open initial store")
.runtime_metadata_store();
assert_eq!(
store
.begin_member_operator_request(&mob_id, &pending)
.await
.expect("persist Pending"),
MobMemberOperatorRequestBegin::Started
);
}
let terminal_reply = MemberOperatorReply {
request_id: pending.request_id.clone(),
outcome: MemberOperatorOutcome::Completed {
result: WireOpaqueJson::from_value(&serde_json::json!({"ok": true})),
},
};
let terminal = pending
.terminal(terminal_reply.clone())
.expect("legal terminal transition");
{
let reopened = SqliteMobStores::open(&path)
.expect("reopen after Pending")
.runtime_metadata_store();
assert_eq!(
reopened
.load_member_operator_request(&mob_id, &pending.key())
.await
.expect("load reopened Pending"),
Some(pending.clone())
);
assert!(
reopened
.compare_and_put_member_operator_request(&mob_id, &pending, &terminal)
.await
.expect("terminal CAS")
);
}
let reopened = SqliteMobStores::open(&path)
.expect("reopen after Terminal")
.runtime_metadata_store();
let loaded = reopened
.load_member_operator_request(&mob_id, &pending.key())
.await
.expect("load reopened Terminal")
.expect("terminal row survives");
assert_eq!(loaded, terminal);
assert_eq!(loaded.terminal_reply(), Some(&terminal_reply));
assert_eq!(
reopened
.list_member_operator_requests(&mob_id)
.await
.expect("list full-domain terminal row"),
vec![terminal.clone()]
);
assert_eq!(
reopened
.begin_member_operator_request(&mob_id, &pending)
.await
.expect("duplicate begin reads terminal"),
MobMemberOperatorRequestBegin::Existing(loaded)
);
}
#[tokio::test]
async fn sqlite_member_operator_request_key_separates_host_generation_and_member_session() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path)
.expect("open execution-fence key store")
.runtime_metadata_store();
let mob_id = MobId::from("sqlite-operator-execution-fence-key");
let base = MobMemberOperatorRequestRecord::pending(
MobMemberOperatorRequestKey::new(
"member-a",
7,
11,
"host-a",
1,
"member-session-a",
"same-request-id",
),
"0".repeat(64),
);
let mut host_generation_two = base.clone();
host_generation_two.host_binding_generation = 2;
let mut replacement_session = host_generation_two.clone();
replacement_session.member_session_id = "member-session-b".to_string();
for record in [&base, &host_generation_two, &replacement_session] {
assert_eq!(
store
.begin_member_operator_request(&mob_id, record)
.await
.expect("insert exact SQLite execution-fence key"),
MobMemberOperatorRequestBegin::Started,
);
}
assert!(matches!(
store
.begin_member_operator_request(&mob_id, &base)
.await
.expect("exact SQLite duplicate replays"),
MobMemberOperatorRequestBegin::Existing(existing) if existing == base
));
let reopened = SqliteMobStores::open(&path)
.expect("reopen execution-fence key store")
.runtime_metadata_store();
for record in [&base, &host_generation_two, &replacement_session] {
assert_eq!(
reopened
.load_member_operator_request(&mob_id, &record.key())
.await
.expect("load exact SQLite execution-fence key"),
Some(record.clone()),
);
}
assert_eq!(
reopened
.list_member_operator_requests(&mob_id)
.await
.expect("list distinct SQLite execution-fence rows")
.len(),
3,
);
}
#[tokio::test]
async fn sqlite_refuses_unledgered_pre_execution_fence_shape_without_mutation() {
let (_dir, path) = temp_db_path();
{
let conn = Connection::open(&path).expect("open legacy member-operator database");
conn.execute_batch(
"CREATE TABLE mob_runtime_member_operator_requests (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation BLOB NOT NULL,
fence_token BLOB NOT NULL,
request_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, agent_identity, generation, fence_token, request_id
)
);",
)
.expect("create pre-execution-fence ledger");
conn.execute(
"INSERT INTO mob_runtime_member_operator_requests
(mob_id, agent_identity, generation, fence_token, request_id, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
"legacy-mob",
"member-a",
u64_to_sql_key(7),
u64_to_sql_key(11),
"ambiguous-request",
b"{}".as_slice(),
],
)
.expect("seed ambiguous pre-fence row");
}
let err = match SqliteMobStores::open(&path) {
Ok(_) => panic!("unledgered owned shape must be refused"),
Err(error) => error,
};
assert!(
err.to_string().contains("no ledger row"),
"unexpected refusal: {err}"
);
let conn = Connection::open(&path).expect("inspect refused legacy schema");
let row_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM mob_runtime_member_operator_requests",
[],
|row| row.get(0),
)
.expect("legacy row remains");
assert_eq!(row_count, 1, "refusal mutated legacy rows");
let mut statement = conn
.prepare("PRAGMA table_info(mob_runtime_member_operator_requests)")
.expect("prepare table-info inspection");
let columns = statement
.query_map([], |row| {
Ok((row.get::<_, String>(1)?, row.get::<_, i64>(5)?))
})
.expect("read legacy table info")
.map(|row| row.expect("read legacy column"))
.collect::<std::collections::BTreeMap<_, _>>();
assert!(!columns.contains_key("host_id"));
assert!(!columns.contains_key("binding_generation"));
assert!(!columns.contains_key("member_session_id"));
assert_eq!(columns.get("request_id"), Some(&5));
assert_eq!(
meerkat_sqlite::domain_version(&conn, MOB_DOMAIN.name).expect("ledger"),
None
);
}
#[tokio::test]
async fn sqlite_refuses_unledgered_malformed_execution_fence_key_without_mutation() {
let (_dir, path) = temp_db_path();
{
let conn = Connection::open(&path).expect("open malformed member-operator database");
conn.execute_batch(
"CREATE TABLE mob_runtime_member_operator_requests (
mob_id TEXT NOT NULL,
agent_identity TEXT NOT NULL,
generation BLOB NOT NULL,
fence_token BLOB NOT NULL,
host_id TEXT NOT NULL,
binding_generation BLOB NOT NULL,
member_session_id TEXT NOT NULL,
request_id TEXT NOT NULL,
record_json BLOB NOT NULL,
PRIMARY KEY (
mob_id, agent_identity, generation, fence_token, request_id
)
);",
)
.expect("create ledger whose fence columns are not key columns");
conn.execute(
"INSERT INTO mob_runtime_member_operator_requests
(mob_id, agent_identity, generation, fence_token, host_id,
binding_generation, member_session_id, request_id, record_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
"malformed-key-mob",
"member-a",
u64_to_sql_key(7),
u64_to_sql_key(11),
"host-a",
u64_to_sql_key(3),
"member-session-a",
"ambiguous-request",
b"{}".as_slice(),
],
)
.expect("seed row under malformed execution-fence key");
}
let err = match SqliteMobStores::open(&path) {
Ok(_) => panic!("unledgered malformed key shape must be refused"),
Err(error) => error,
};
assert!(
err.to_string().contains("no ledger row"),
"unexpected refusal: {err}"
);
let conn = Connection::open(&path).expect("inspect refused malformed schema");
let row_count: i64 = conn
.query_row(
"SELECT COUNT(*) FROM mob_runtime_member_operator_requests",
[],
|row| row.get(0),
)
.expect("malformed-key row remains");
assert_eq!(row_count, 1, "refusal mutated malformed-key rows");
let mut statement = conn
.prepare("PRAGMA table_info(mob_runtime_member_operator_requests)")
.expect("prepare malformed table-info inspection");
let primary_key = statement
.query_map([], |row| {
Ok((row.get::<_, i64>(5)?, row.get::<_, String>(1)?))
})
.expect("read malformed table info")
.map(|row| row.expect("read malformed column"))
.filter(|(position, _)| *position > 0)
.collect::<std::collections::BTreeMap<_, _>>();
assert_eq!(
primary_key.into_values().collect::<Vec<_>>(),
vec![
"mob_id",
"agent_identity",
"generation",
"fence_token",
"request_id",
],
);
assert_eq!(
meerkat_sqlite::domain_version(&conn, MOB_DOMAIN.name).expect("ledger"),
None
);
}
#[test]
fn sqlite_u64_blob_keys_round_trip_and_sort_the_full_domain() {
let (_dir, path) = temp_db_path();
let conn = open_connection(&path).expect("open full-domain key store");
let expected = [1, i64::MAX as u64, i64::MAX as u64 + 1, u64::MAX];
for value in expected {
conn.execute(
"INSERT INTO mob_run_remote_turn_intents \
(run_id, dispatch_sequence, intent_json) VALUES (?1, ?2, ?3)",
params!["full-domain-run", u64_to_sql_key(value), Vec::<u8>::new()],
)
.expect("insert full-domain dispatch key");
}
let mut statement = conn
.prepare(
"SELECT dispatch_sequence FROM mob_run_remote_turn_intents \
WHERE run_id = ?1 ORDER BY dispatch_sequence ASC",
)
.expect("prepare ordered full-domain key query");
let rows = statement
.query_map(params!["full-domain-run"], |row| row.get::<_, Vec<u8>>(0))
.expect("query full-domain keys");
let actual = rows
.map(|row| {
let key = row.expect("read full-domain key");
sql_key_to_u64(&key, "test dispatch sequence").expect("decode full-domain key")
})
.collect::<Vec<_>>();
assert_eq!(actual, expected);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_sqlite_member_operator_incarnation_quota_is_atomic_and_replays_at_ceiling() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).expect("open SQLite quota store");
let mob_id = MobId::from("operator-incarnation-quota");
let oversized = MobMemberOperatorRequestRecord::pending(
MobMemberOperatorRequestKey::new(
"member-a",
1,
1,
"host-a",
1,
"member-session-a",
"x".repeat(crate::store::MEMBER_OPERATOR_REQUEST_ID_MAX_BYTES + 1),
),
"0".repeat(64),
);
assert!(
stores
.runtime_metadata_store()
.begin_member_operator_request(&mob_id, &oversized)
.await
.is_err(),
"SQLite must reject oversized untrusted idempotency keys before persistence"
);
assert_eq!(operator_request_count(&path, &mob_id), 0);
let seed = (0..crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION - 1)
.map(|sequence| operator_request("member-a", 1, sequence))
.collect::<Vec<_>>();
seed_operator_requests(&path, &mob_id, &seed);
let candidate_a = operator_request(
"member-a",
1,
crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION - 1,
);
let candidate_b = operator_request(
"member-a",
1,
crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION,
);
let store_a = stores.runtime_metadata_store();
let store_b = stores.runtime_metadata_store();
let (result_a, result_b) = tokio::join!(
store_a.begin_member_operator_request(&mob_id, &candidate_a),
store_b.begin_member_operator_request(&mob_id, &candidate_b),
);
assert_eq!(
[&result_a, &result_b]
.into_iter()
.filter(|result| matches!(result, Ok(MobMemberOperatorRequestBegin::Started)))
.count(),
1,
"BEGIN IMMEDIATE must serialize the count+insert so only one contender owns execution"
);
assert_eq!(
[&result_a, &result_b]
.into_iter()
.filter(|result| matches!(result, Err(MobStoreError::WriteFailed(_))))
.count(),
1,
"the serialized loser must observe the committed ceiling and fail closed"
);
assert_eq!(
operator_request_count(&path, &mob_id),
crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION,
"concurrent begins must not oversubscribe the incarnation ceiling"
);
let winner = if matches!(result_a, Ok(MobMemberOperatorRequestBegin::Started)) {
&candidate_a
} else {
&candidate_b
};
assert_eq!(
stores
.runtime_metadata_store()
.begin_member_operator_request(&mob_id, winner)
.await
.expect("the winning key must replay at the ceiling"),
MobMemberOperatorRequestBegin::Existing(winner.clone()),
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_sqlite_member_operator_global_quota_is_atomic_and_replays_at_ceiling() {
let (_dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).expect("open SQLite quota store");
let mob_id = MobId::from("operator-global-quota");
let seed = (0..crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_MOB - 1)
.map(|sequence| {
let incarnation =
sequence / crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION;
let request = sequence % crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION;
operator_request(&format!("member-{incarnation}"), 1, request)
})
.collect::<Vec<_>>();
seed_operator_requests(&path, &mob_id, &seed);
let candidate_a = operator_request("member-overflow-a", 1, 0);
let candidate_b = operator_request("member-overflow-b", 1, 0);
let store_a = stores.runtime_metadata_store();
let store_b = stores.runtime_metadata_store();
let (result_a, result_b) = tokio::join!(
store_a.begin_member_operator_request(&mob_id, &candidate_a),
store_b.begin_member_operator_request(&mob_id, &candidate_b),
);
assert_eq!(
[&result_a, &result_b]
.into_iter()
.filter(|result| matches!(result, Ok(MobMemberOperatorRequestBegin::Started)))
.count(),
1,
"only one concurrent contender may claim the final per-mob slot"
);
assert_eq!(
[&result_a, &result_b]
.into_iter()
.filter(|result| matches!(result, Err(MobStoreError::WriteFailed(_))))
.count(),
1,
"the other contender must observe the committed global ceiling"
);
assert_eq!(
operator_request_count(&path, &mob_id),
crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_MOB,
"concurrent begins must not oversubscribe the per-mob ceiling"
);
let winner = if matches!(result_a, Ok(MobMemberOperatorRequestBegin::Started)) {
&candidate_a
} else {
&candidate_b
};
assert_eq!(
stores
.runtime_metadata_store()
.begin_member_operator_request(&mob_id, winner)
.await
.expect("the globally winning key must replay at the ceiling"),
MobMemberOperatorRequestBegin::Existing(winner.clone()),
);
}
#[tokio::test]
async fn sqlite_recovery_prune_frees_stale_global_capacity_and_preserves_current_rows() {
use meerkat_contracts::wire::supervisor_bridge::{
MemberOperatorOutcome, MemberOperatorReply, WireOpaqueJson,
};
let (_dir, path) = temp_db_path();
let mob_id = MobId::from("sqlite-operator-recovery-prune");
let current = (0..crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION)
.map(|sequence| operator_request("member-current", 1, sequence));
let stale = (0..3).flat_map(|incarnation| {
(0..crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION).map(move |sequence| {
operator_request(&format!("member-stale-{incarnation}"), 1, sequence)
})
});
let rows = current.chain(stale).collect::<Vec<_>>();
seed_operator_requests(&path, &mob_id, &rows);
let current_pending = operator_request("member-current", 1, 0);
let current_terminal = current_pending
.terminal(MemberOperatorReply {
request_id: current_pending.request_id.clone(),
outcome: MemberOperatorOutcome::Completed {
result: WireOpaqueJson::from_value(&serde_json::json!({"ok": true})),
},
})
.expect("terminalize current SQLite replay row");
{
let store = SqliteMobStores::open(&path)
.expect("open before simulated crash")
.runtime_metadata_store();
assert!(
store
.compare_and_put_member_operator_request(
&mob_id,
¤t_pending,
¤t_terminal,
)
.await
.expect("persist current SQLite terminal")
);
assert!(
store
.begin_member_operator_request(&mob_id, &operator_request("member-new", 1, 0),)
.await
.is_err(),
"the recovered store starts at the global fail-closed ceiling"
);
}
let recovered = SqliteMobStores::open(&path)
.expect("reopen after simulated crash")
.runtime_metadata_store();
let authority = MobMemberOperatorPruneAuthority::from_actor_current_residencies(
std::collections::BTreeSet::from([
crate::store::MobMemberOperatorResidency {
agent_identity: "member-current".to_string(),
generation: 1,
fence_token: 1,
host_id: "host-a".to_string(),
host_binding_generation: 1,
member_session_id: "member-session-a".to_string(),
},
crate::store::MobMemberOperatorResidency {
agent_identity: "member-new".to_string(),
generation: 1,
fence_token: 1,
host_id: "host-a".to_string(),
host_binding_generation: 1,
member_session_id: "member-session-a".to_string(),
},
]),
);
assert_eq!(
recovered
.prune_stale_member_operator_requests(&mob_id, &authority)
.await
.expect("recovered actor-authorized SQLite prune"),
3 * crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION as u64,
);
assert_eq!(
recovered
.begin_member_operator_request(&mob_id, ¤t_pending)
.await
.expect("current SQLite terminal still replays"),
MobMemberOperatorRequestBegin::Existing(current_terminal),
);
assert!(
recovered
.begin_member_operator_request(
&mob_id,
&operator_request(
"member-current",
1,
crate::store::MEMBER_OPERATOR_REQUEST_MAX_PER_INCARNATION,
),
)
.await
.is_err(),
"recovery pruning must preserve the current incarnation ceiling"
);
assert_eq!(
recovered
.begin_member_operator_request(&mob_id, &operator_request("member-new", 1, 0))
.await
.expect("recovery pruning frees global SQLite capacity"),
MobMemberOperatorRequestBegin::Started,
);
}
#[tokio::test]
async fn test_sqlite_runtime_metadata_store_roundtrips_supervisor_and_overlay_records() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
let mob_id = MobId::from("mob");
let supervisor = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let supervisor_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&supervisor)
.unwrap();
let overlay = ExternalBindingOverlayRecord {
agent_identity: AgentIdentity::from("worker-1"),
generation: Generation::new(2),
normalized_member_ref: Some(MemberRef::BackendPeer {
peer_id: "peer-worker-1".to_string(),
address: "tcp://worker-1".to_string(),
bootstrap_token: None,
session_id: None,
pubkey: [7u8; 32],
}),
bootstrap_token: None,
status: ExternalBindingOverlayStatus::Normalized,
updated_at: Utc::now(),
};
store
.put_supervisor_authority(&mob_id, &supervisor, &supervisor_authority)
.await
.unwrap();
let loaded_supervisor = store
.load_supervisor_authority(&mob_id)
.await
.unwrap()
.expect("supervisor should persist");
assert_eq!(loaded_supervisor, supervisor);
assert!(
store
.put_external_binding_overlay_if_absent(&mob_id, &overlay)
.await
.unwrap(),
"first overlay insert should win"
);
assert!(
!store
.put_external_binding_overlay_if_absent(&mob_id, &overlay)
.await
.unwrap(),
"duplicate overlay insert should be ignored"
);
let overlays = store.list_external_binding_overlays(&mob_id).await.unwrap();
assert_eq!(overlays, vec![overlay.clone()]);
let failed_overlay = ExternalBindingOverlayRecord {
status: ExternalBindingOverlayStatus::Failed {
reason: "normalization failed".to_string(),
},
normalized_member_ref: None,
updated_at: Utc::now(),
..overlay
};
store
.upsert_external_binding_overlay(&mob_id, &failed_overlay)
.await
.unwrap();
let overlays = store.list_external_binding_overlays(&mob_id).await.unwrap();
assert_eq!(overlays, vec![failed_overlay]);
store
.delete_external_binding_overlays(&mob_id)
.await
.unwrap();
assert!(
store
.list_external_binding_overlays(&mob_id)
.await
.unwrap()
.is_empty(),
"overlay delete should clear all records for the mob"
);
let delete_authority =
crate::store::supervisor_authority_deletion_authority_for_record(&supervisor).unwrap();
assert!(
store
.delete_supervisor_authority(&mob_id, &supervisor, &delete_authority)
.await
.unwrap()
);
assert!(
store
.load_supervisor_authority(&mob_id)
.await
.unwrap()
.is_none(),
"supervisor delete should remove the stored record"
);
}
#[tokio::test]
async fn test_sqlite_runtime_metadata_store_put_supervisor_if_absent_preserves_existing_record()
{
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
let mob_id = MobId::from("mob");
let first = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let second = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let first_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&first).unwrap();
let second_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&second).unwrap();
assert!(
store
.put_supervisor_authority_if_absent(&mob_id, &first, &first_authority)
.await
.unwrap()
);
assert!(
!store
.put_supervisor_authority_if_absent(&mob_id, &second, &second_authority)
.await
.unwrap()
);
assert_eq!(
store.load_supervisor_authority(&mob_id).await.unwrap(),
Some(first)
);
}
#[tokio::test]
async fn test_sqlite_runtime_metadata_store_compare_and_put_supervisor_authority() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
let mob_id = MobId::from("mob-cas");
let first = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let second = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let third = SupervisorAuthorityRecord::generate(default_bridge_protocol_version());
let first_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&first).unwrap();
let second_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&second).unwrap();
let third_authority =
crate::store::supervisor_authority_persistence_authority_for_record(&third).unwrap();
store
.put_supervisor_authority(&mob_id, &first, &first_authority)
.await
.unwrap();
assert!(
!store
.compare_and_put_supervisor_authority(&mob_id, &second, &third, &third_authority)
.await
.unwrap(),
"mismatched expected authority must not update"
);
assert_eq!(
store.load_supervisor_authority(&mob_id).await.unwrap(),
Some(first.clone())
);
assert!(
store
.compare_and_put_supervisor_authority(&mob_id, &first, &second, &second_authority)
.await
.unwrap(),
"matching expected authority should update"
);
assert_eq!(
store.load_supervisor_authority(&mob_id).await.unwrap(),
Some(second)
);
}
#[tokio::test]
async fn test_sqlite_runtime_metadata_store_roundtrips_mob_host_authority_records() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
let mob_id = MobId::from("mob-hosts");
let other_mob = MobId::from("mob-other");
let host_b = crate::store::sample_mob_host_authority_record("host-peer-b", 1);
let host_c = crate::store::sample_mob_host_authority_record("host-peer-c", 2);
let host_b_authority =
crate::store::mob_host_authority_persistence_authority_for_record(&host_b).unwrap();
let host_c_authority =
crate::store::mob_host_authority_persistence_authority_for_record(&host_c).unwrap();
assert!(
store
.put_mob_host_authority_if_absent(&mob_id, &host_b, &host_b_authority)
.await
.unwrap()
);
assert!(
!store
.put_mob_host_authority_if_absent(&mob_id, &host_b, &host_b_authority)
.await
.unwrap(),
"duplicate (mob, host) insert must be ignored"
);
store
.put_mob_host_authority(&mob_id, &host_c, &host_c_authority)
.await
.unwrap();
store
.put_mob_host_authority(&other_mob, &host_b, &host_b_authority)
.await
.unwrap();
assert_eq!(
store
.load_mob_host_authority(&mob_id, "host-peer-b")
.await
.unwrap(),
Some(host_b.clone())
);
assert_eq!(
store.list_mob_host_authorities(&mob_id).await.unwrap(),
vec![host_b.clone(), host_c.clone()],
"listing is mob-scoped and host-id ordered"
);
let host_b_rebound = MobHostAuthorityRecord {
authority_epoch: 2,
live_endpoint: None,
..host_b.clone()
};
let rebound_authority =
crate::store::mob_host_authority_persistence_authority_for_record(&host_b_rebound)
.unwrap();
assert!(
!store
.compare_and_put_mob_host_authority(
&mob_id,
&host_b_rebound,
&host_b_rebound,
&rebound_authority
)
.await
.unwrap(),
"mismatched expected record must not update"
);
assert!(
store
.compare_and_put_mob_host_authority(
&mob_id,
&host_b,
&host_b_rebound,
&rebound_authority
)
.await
.unwrap()
);
drop(store);
let reopened = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
assert_eq!(
reopened.list_mob_host_authorities(&mob_id).await.unwrap(),
vec![host_b_rebound.clone(), host_c],
"host bindings must survive a store reopen"
);
let deletion =
crate::store::mob_host_authority_deletion_authority_for_record(&host_b_rebound)
.unwrap();
reopened
.put_mob_host_binding_generation_highwater(&mob_id, &host_b_rebound, &deletion)
.await
.unwrap();
assert!(
!reopened
.delete_mob_host_authority(&mob_id, &host_b, &deletion)
.await
.unwrap(),
"stale expected record must not delete"
);
assert!(
reopened
.delete_mob_host_authority(&mob_id, &host_b_rebound, &deletion)
.await
.unwrap()
);
assert!(
reopened
.load_mob_host_authority(&mob_id, "host-peer-b")
.await
.unwrap()
.is_none()
);
drop(reopened);
let reopened = SqliteMobStores::open(&path)
.unwrap()
.runtime_metadata_store();
assert_eq!(
reopened
.list_mob_host_binding_generation_highwaters(&mob_id)
.await
.unwrap(),
vec![("host-peer-b".to_string(), host_b_rebound.binding_generation)],
"the generation tombstone survives both active-row deletion and reopen",
);
assert_eq!(
reopened
.list_mob_host_authorities(&other_mob)
.await
.unwrap(),
vec![host_b],
"another mob's binding for the same host must survive"
);
}
#[tokio::test]
async fn test_sqlite_stores_share_single_database_path() {
let (_dir, path) = temp_db_path();
let shared = SqliteMobStores::open(&path).unwrap();
let events = shared.event_store();
let runs = shared.run_store();
let specs = shared.spec_store();
events
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
let run = sample_run(MobRunStatus::Pending);
let run_id = run.run_id.clone();
runs.create_run(run).await.unwrap();
let fetched_run = runs.get_run(&run_id).await.unwrap();
assert!(fetched_run.is_some(), "run store should share same db path");
let definition = sample_definition();
let revision = specs
.put_spec(&MobId::from("mob"), &definition, None)
.await
.unwrap();
assert_eq!(revision, 1);
assert_eq!(events.replay_all().await.unwrap().len(), 1);
}
#[tokio::test]
#[ignore] async fn integration_real_sqlite_event_store_clear_and_prune_large_keyset() {
let (_dir, path) = temp_db_path();
let store = SqliteMobStores::open(&path).unwrap().event_store();
let now = Utc::now();
for i in 0..2_000 {
store
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: Some(now - chrono::Duration::minutes((i % 10) as i64)),
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
}
let removed = store
.prune(now - chrono::Duration::minutes(5))
.await
.unwrap();
assert!(removed > 0, "expected stale events to be pruned");
store.clear().await.unwrap();
let replayed = store.replay_all().await.unwrap();
assert!(
replayed.is_empty(),
"clear should remove all persisted events"
);
}
#[tokio::test]
async fn test_sqlite_reopen_after_drop_same_path() {
let (_dir, path) = temp_db_path();
{
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
events
.append(NewMobEvent {
mob_id: MobId::from("mob"),
timestamp: None,
kind: MobEventKind::MobCompleted,
})
.await
.unwrap();
}
{
let stores = SqliteMobStores::open(&path).unwrap();
let events = stores.event_store();
let all = events.replay_all().await.unwrap();
assert_eq!(all.len(), 1, "data from first open must survive reopen");
}
}
use crate::store::realm_profile::contract_tests;
fn sqlite_realm_profile_store() -> (tempfile::TempDir, SqliteRealmProfileStore) {
let (dir, path) = temp_db_path();
let stores = SqliteMobStores::open(&path).unwrap();
(dir, stores.realm_profile_store())
}
#[tokio::test]
async fn realm_profile_create_and_get() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_create_and_get(&store).await;
}
#[tokio::test]
async fn realm_profile_get_nonexistent() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_get_nonexistent(&store).await;
}
#[tokio::test]
async fn realm_profile_create_duplicate_fails() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_create_duplicate_fails(&store).await;
}
#[tokio::test]
async fn realm_profile_update_correct_revision() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_update_with_correct_revision(&store).await;
}
#[tokio::test]
async fn realm_profile_update_wrong_revision() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_update_with_wrong_revision(&store).await;
}
#[tokio::test]
async fn realm_profile_update_nonexistent() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_update_nonexistent(&store).await;
}
#[tokio::test]
async fn realm_profile_delete_correct_revision() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_delete_with_correct_revision(&store).await;
}
#[tokio::test]
async fn realm_profile_delete_wrong_revision() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_delete_with_wrong_revision(&store).await;
}
#[tokio::test]
async fn realm_profile_delete_nonexistent() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_delete_nonexistent(&store).await;
}
#[tokio::test]
async fn realm_profile_list() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_list(&store).await;
}
#[tokio::test]
async fn realm_profile_list_empty() {
let (_dir, store) = sqlite_realm_profile_store();
contract_tests::test_list_empty(&store).await;
}
#[tokio::test]
async fn sqlite_realm_profile_store_open_creates_directory() {
let dir = tempfile::tempdir().unwrap();
let nested = dir.path().join("a").join("b").join("realm_profiles.db");
assert!(!nested.parent().unwrap().exists());
let _store = SqliteRealmProfileStore::open(&nested).unwrap();
assert!(nested.parent().unwrap().exists());
}
#[tokio::test]
async fn sqlite_realm_profile_store_open_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("realm_profiles.db");
{
let store = SqliteRealmProfileStore::open(&db_path).unwrap();
let profile = Profile {
model: "claude-sonnet-4-5".into(),
provider: None,
self_hosted_server_id: None,
image_generation_provider: None,
auto_compact_threshold: None,
resume_overrides: Vec::new(),
skills: Vec::new(),
tools: ToolConfig::default(),
peer_description: String::new(),
external_addressable: false,
backend: None,
runtime_mode: crate::MobRuntimeMode::AutonomousHost,
max_inline_peer_notifications: None,
output_schema: None,
provider_params: None,
};
store.create("test-profile", &profile).await.unwrap();
}
{
let store = SqliteRealmProfileStore::open(&db_path).unwrap();
let got = store.get("test-profile").await.unwrap();
assert!(got.is_some(), "profile must survive reopen");
assert_eq!(got.unwrap().profile.model, "claude-sonnet-4-5");
}
}
#[tokio::test]
async fn sqlite_realm_profile_store_open_contract_create_and_get() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("realm_profiles.db");
let store = SqliteRealmProfileStore::open(&db_path).unwrap();
contract_tests::test_create_and_get(&store).await;
}
#[tokio::test]
async fn sqlite_realm_profile_store_open_contract_get_nonexistent() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("realm_profiles.db");
let store = SqliteRealmProfileStore::open(&db_path).unwrap();
contract_tests::test_get_nonexistent(&store).await;
}
#[tokio::test]
async fn sqlite_realm_profile_store_open_contract_list() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("realm_profiles.db");
let store = SqliteRealmProfileStore::open(&db_path).unwrap();
contract_tests::test_list(&store).await;
}
}