use anyhow::{Result, anyhow, ensure};
use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior, params};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use uuid::Uuid;
use super::{
BindingCandidate, BindingValue, STRATEGY_SCHEMA_VERSION, StrategyAuthorization,
StrategyDefinition, StrategyDefinitionSummary, StrategyDiagnostic, StrategyProjection,
};
use crate::ASSISTANT_TEMPORARY_DEFINITION_PREFIX;
use licoup_foundation::core::sqlite_contract::ColumnVariant;
use licoup_workflow::{
BindingKind, CommandStatus, FailureClass, GraphState, GraphStateKind, ReducerEvent, RunCommand,
RunSnapshot, StrategyRunStatus, Transition, TransitionEvent, TransitionMode,
WorkflowDefinition, compile_workflow, reduce,
};
const DATABASE_FILE: &str = "strategies.sqlite3";
#[derive(Clone, Debug)]
pub struct StrategyStore {
db_path: PathBuf,
package_revisions_root: Option<PathBuf>,
}
impl StrategyStore {
pub fn open(portable_root: &Path) -> Result<Self> {
let root = portable_root.join("client-state").join("adaptive-flywheel");
let db_path = root.join(DATABASE_FILE);
preflight_existing_store(&db_path)?;
licoup_foundation::platform::file_security::ensure_private_dir(
&portable_root.join("client-state"),
)?;
licoup_foundation::platform::file_security::ensure_private_dir(&root)?;
let existed = db_path.exists();
let store = Self {
db_path,
package_revisions_root: Some(root.join("strategy-packages").join("revisions")),
};
let retired = store.with_connection(|connection| {
if existed {
validate_current_schema(connection).map(|()| Vec::new())
} else {
initialize_schema(connection)
}
})?;
licoup_foundation::platform::file_security::harden_private_path(&store.db_path)?;
store.remove_retired_revision_trees(&retired);
Ok(store)
}
pub fn open_for_migration(portable_root: &Path) -> Result<Self> {
let root = portable_root.join("client-state").join("adaptive-flywheel");
licoup_foundation::platform::file_security::ensure_private_dir(
&portable_root.join("client-state"),
)?;
licoup_foundation::platform::file_security::ensure_private_dir(&root)?;
let store = Self {
db_path: root.join(DATABASE_FILE),
package_revisions_root: Some(root.join("strategy-packages").join("revisions")),
};
let retired = store.with_connection(initialize_schema)?;
licoup_foundation::platform::file_security::harden_private_path(&store.db_path)?;
store.remove_retired_revision_trees(&retired);
Ok(store)
}
pub fn open_in_memory() -> Result<Self> {
let path =
std::env::temp_dir().join(format!("lico-adaptive-flywheel-{}.sqlite3", Uuid::new_v4()));
let store = Self {
db_path: path,
package_revisions_root: None,
};
store.with_connection(initialize_schema)?;
Ok(store)
}
pub fn db_path(&self) -> &Path {
&self.db_path
}
pub fn durable_queue(&self) -> super::DurableQueue {
super::DurableQueue::from_store(self.clone())
}
pub fn durable_subscriptions(&self) -> super::DurableSubscriptionStore {
super::DurableSubscriptionStore::from_store(self.clone())
}
pub fn durable_control(&self) -> super::DurableControlledStore {
super::DurableControlledStore::from_store(self.clone())
}
pub fn with_connection<T>(
&self,
operation: impl FnOnce(&mut Connection) -> Result<T>,
) -> Result<T> {
self.with_connection_typed(operation)
}
pub(crate) fn with_connection_typed<T, E>(
&self,
operation: impl FnOnce(&mut Connection) -> std::result::Result<T, E>,
) -> std::result::Result<T, E>
where
E: From<anyhow::Error>,
{
let mut connection = Connection::open(&self.db_path)
.map_err(|_| E::from(anyhow!("strategy_database_open_failed")))?;
configure_connection(&connection).map_err(E::from)?;
operation(&mut connection)
}
pub fn register_definition(
&self,
revision_digest: &str,
semantics_digest: &str,
workflow: &WorkflowDefinition,
asset_count: usize,
imported_at_unix_ms: i64,
) -> Result<StrategyDefinition> {
let compiled = compile_workflow(workflow.clone())?;
let workflow_json = serde_json::to_string(compiled.definition())?;
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute(
"INSERT INTO strategy_definitions(
definition_id, revision_digest, semantics_digest, name, version,
workflow_json, asset_count, imported_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
ON CONFLICT(revision_digest) DO NOTHING",
params![
compiled.definition().metadata.id,
revision_digest,
semantics_digest,
compiled.definition().metadata.name,
compiled.definition().metadata.version,
workflow_json,
asset_count as i64,
imported_at_unix_ms,
],
)?;
let existing: Option<(String, String)> = transaction
.query_row(
"SELECT definition_id, semantics_digest FROM strategy_definitions
WHERE revision_digest=?1",
params![revision_digest],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
ensure!(
existing
== Some((
compiled.definition().metadata.id.clone(),
semantics_digest.to_owned()
)),
"strategy_revision_conflict"
);
transaction.commit()?;
self.definition_by_revision(revision_digest)
})
}
pub fn list_definitions(&self) -> Result<Vec<StrategyDefinitionSummary>> {
self.with_connection(|connection| {
let mut statement = connection.prepare(
"SELECT definition_id, revision_digest, semantics_digest, name, version, imported_at,
EXISTS(
SELECT 1 FROM strategy_authorizations
WHERE revision_digest=strategy_definitions.revision_digest AND active=1
)
FROM strategy_definitions
WHERE definition_id NOT LIKE 'assistant-temporary%'
ORDER BY imported_at DESC, revision_digest ASC",
)?;
let rows = statement.query_map([], summary_from_row)?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(Into::into)
})
}
pub fn definition_by_revision(&self, revision_digest: &str) -> Result<StrategyDefinition> {
self.with_connection(|connection| definition_by_revision(connection, revision_digest))
}
pub fn latest_definition(&self, definition_id: &str) -> Result<StrategyDefinition> {
self.with_connection(|connection| {
let revision: Option<String> = connection
.query_row(
"SELECT revision_digest FROM strategy_definitions
WHERE definition_id=?1 ORDER BY imported_at DESC, revision_digest ASC LIMIT 1",
params![definition_id],
|row| row.get(0),
)
.optional()?;
definition_by_revision(
connection,
revision
.as_deref()
.ok_or_else(|| anyhow!("strategy_definition_not_found"))?,
)
})
}
pub fn update_binding(
&self,
revision_digest: &str,
slot_id: &str,
value_id: &str,
model: &str,
reasoning_effort: &str,
expected_revision: Option<u64>,
) -> Result<BindingValue> {
self.replace_slot_bindings(
revision_digest,
slot_id,
&[BindingCandidate {
value_id: value_id.to_owned(),
model: model.to_owned(),
reasoning_effort: reasoning_effort.to_owned(),
}],
expected_revision,
)?
.into_iter()
.next()
.ok_or_else(|| anyhow!("strategy_binding_incomplete"))
}
pub fn replace_slot_bindings(
&self,
revision_digest: &str,
slot_id: &str,
candidates: &[BindingCandidate],
expected_revision: Option<u64>,
) -> Result<Vec<BindingValue>> {
validate_opaque_id(slot_id, "strategy_binding_slot_invalid")?;
ensure!(candidates.len() <= 16, "strategy_binding_limit");
for candidate in candidates {
validate_opaque_id(&candidate.value_id, "strategy_binding_value_invalid")?;
validate_optional_text(&candidate.model, "strategy_binding_model_invalid")?;
validate_optional_text(
&candidate.reasoning_effort,
"strategy_binding_reasoning_effort_invalid",
)?;
}
let definition = self.definition_by_revision(revision_digest)?;
ensure!(
definition
.workflow
.actor_slots
.iter()
.any(|slot| slot.id == slot_id),
"strategy_binding_slot_unknown"
);
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let current: Option<u64> = transaction
.query_row(
"SELECT MAX(revision) FROM strategy_bindings
WHERE revision_digest=?1 AND slot_id=?2",
params![revision_digest, slot_id],
|row| row.get::<_, Option<i64>>(0),
)
.optional()?
.flatten()
.map(|value| value as u64);
if let Some(expected) = expected_revision {
ensure!(current.unwrap_or(0) == expected, "strategy_revision_conflict");
}
let revision = current.unwrap_or(0) + 1;
transaction.execute(
"DELETE FROM strategy_bindings WHERE revision_digest=?1 AND slot_id=?2",
params![revision_digest, slot_id],
)?;
let mut values = Vec::with_capacity(candidates.len());
for (ordinal, candidate) in candidates.iter().enumerate() {
let ordinal = u8::try_from(ordinal).map_err(|_| anyhow!("strategy_binding_limit"))?;
transaction.execute(
"INSERT INTO strategy_bindings(
revision_digest, slot_id, ordinal, value_id, model, reasoning_effort, revision
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
revision_digest,
slot_id,
ordinal as i64,
candidate.value_id,
candidate.model,
candidate.reasoning_effort,
revision as i64
],
)?;
values.push(BindingValue {
slot_id: slot_id.to_owned(),
ordinal,
value_id: candidate.value_id.clone(),
model: candidate.model.clone(),
reasoning_effort: candidate.reasoning_effort.clone(),
revision,
});
}
transaction.execute(
"UPDATE strategy_authorizations SET active=0 WHERE revision_digest=?1 AND active=1",
params![revision_digest],
)?;
transaction.commit()?;
Ok(values)
})
}
pub fn remove_binding(
&self,
revision_digest: &str,
slot_id: &str,
expected_revision: Option<u64>,
) -> Result<()> {
validate_opaque_id(slot_id, "strategy_binding_slot_invalid")?;
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let current: Option<u64> = transaction
.query_row(
"SELECT MAX(revision) FROM strategy_bindings
WHERE revision_digest=?1 AND slot_id=?2",
params![revision_digest, slot_id],
|row| row.get::<_, Option<i64>>(0),
)
.optional()?
.flatten()
.map(|value| value as u64);
if let Some(expected) = expected_revision {
ensure!(
current.unwrap_or(0) == expected,
"strategy_revision_conflict"
);
}
transaction.execute(
"DELETE FROM strategy_bindings WHERE revision_digest=?1 AND slot_id=?2",
params![revision_digest, slot_id],
)?;
transaction.execute(
"UPDATE strategy_authorizations SET active=0
WHERE revision_digest=?1 AND active=1",
params![revision_digest],
)?;
transaction.commit()?;
Ok(())
})
}
pub fn authorization_preview(&self, revision_digest: &str) -> Result<StrategyAuthorization> {
let definition = self.definition_by_revision(revision_digest)?;
ensure!(
bindings_complete(&definition),
"strategy_binding_incomplete"
);
let binding_digest = binding_digest(&definition.bindings)?;
let next_revision = definition
.authorization
.as_ref()
.map_or(1, |authorization| authorization.revision + 1);
let authorization_digest = authorization_digest(
revision_digest,
&definition.summary.semantics_digest,
&binding_digest,
next_revision,
);
Ok(StrategyAuthorization {
definition_digest: revision_digest.to_owned(),
semantics_digest: definition.summary.semantics_digest,
binding_digest,
authorization_digest,
revision: next_revision,
active: false,
})
}
pub fn grant_authorization(
&self,
revision_digest: &str,
expected_authorization_digest: &str,
) -> Result<StrategyAuthorization> {
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let definition = definition_by_revision(&transaction, revision_digest)?;
ensure!(
bindings_complete(&definition),
"strategy_binding_incomplete"
);
let binding_digest = binding_digest(&definition.bindings)?;
let next_revision = definition
.authorization
.as_ref()
.map_or(1, |authorization| authorization.revision + 1);
let authorization_digest = authorization_digest(
revision_digest,
&definition.summary.semantics_digest,
&binding_digest,
next_revision,
);
ensure!(
authorization_digest == expected_authorization_digest,
"strategy_authorization_stale"
);
transaction.execute(
"UPDATE strategy_authorizations SET active=0 WHERE revision_digest=?1 AND active=1",
params![revision_digest],
)?;
transaction.execute(
"INSERT INTO strategy_authorizations(
revision_digest, revision, semantics_digest, binding_digest,
authorization_digest, active, created_at
) VALUES (?1, ?2, ?3, ?4, ?5, 1, ?6)",
params![
revision_digest,
next_revision as i64,
definition.summary.semantics_digest,
binding_digest,
authorization_digest,
now_ms(),
],
)?;
transaction.commit()?;
Ok(StrategyAuthorization {
definition_digest: revision_digest.to_owned(),
semantics_digest: definition.summary.semantics_digest,
binding_digest,
authorization_digest,
revision: next_revision,
active: true,
})
})
}
pub fn revoke_authorization(&self, revision_digest: &str) -> Result<()> {
self.with_connection(|connection| {
connection.execute(
"UPDATE strategy_authorizations SET active=0 WHERE revision_digest=?1 AND active=1",
params![revision_digest],
)?;
Ok(())
})
}
pub fn start_run(
&self,
revision_digest: &str,
input: Value,
idempotency_key: &str,
conversation_id: Option<&str>,
cwd: Option<String>,
) -> Result<RunSnapshot> {
self.start_run_with_assistant_context(
revision_digest,
input,
idempotency_key,
conversation_id,
cwd,
None,
None,
)
}
pub(crate) fn start_run_with_assistant_context(
&self,
revision_digest: &str,
input: Value,
idempotency_key: &str,
conversation_id: Option<&str>,
cwd: Option<String>,
assistant_membership_id: Option<&str>,
route_receipt: Option<Value>,
) -> Result<RunSnapshot> {
validate_opaque_id(idempotency_key, "strategy_idempotency_key_invalid")?;
if let Some(conversation_id) = conversation_id {
validate_opaque_id(conversation_id, "strategy_conversation_id_invalid")?;
}
if let Some(membership_id) = assistant_membership_id {
validate_opaque_id(membership_id, "strategy_membership_id_invalid")?;
}
let input_bytes = serde_json::to_vec(&input)?;
let mut digest_source = Vec::from(revision_digest.as_bytes());
digest_source.push(0);
digest_source.extend_from_slice(&input_bytes);
digest_source.push(0);
digest_source.extend_from_slice(conversation_id.unwrap_or("").as_bytes());
digest_source.push(0);
digest_source.extend_from_slice(cwd.as_deref().unwrap_or("").as_bytes());
digest_source.push(0);
digest_source.extend_from_slice(assistant_membership_id.unwrap_or("").as_bytes());
digest_source.push(0);
digest_source.extend_from_slice(&serde_json::to_vec(&route_receipt)?);
let request_digest = sha256_hex(&digest_source);
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let existing: Option<(String, String)> = transaction
.query_row(
"SELECT run_id, request_digest FROM strategy_runs WHERE idempotency_key=?1",
params![idempotency_key],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((run_id, existing_digest)) = existing {
ensure!(
existing_digest == request_digest,
"strategy_idempotency_conflict"
);
let snapshot = load_run(&transaction, &run_id)?;
transaction.commit()?;
return Ok(snapshot);
}
let definition = definition_by_revision(&transaction, revision_digest)?;
ensure!(
bindings_complete(&definition),
"strategy_binding_incomplete"
);
let authorization = current_authorization(&definition)?;
ensure!(
authorization.binding_digest == binding_digest(&definition.bindings)?,
"strategy_authorization_stale"
);
let slot_candidate_counts = slot_candidate_counts(&definition.bindings);
let semantics_digest = definition.summary.semantics_digest.clone();
let compiled = compile_workflow(definition.workflow)?;
let run_id = format!("run-{}", Uuid::new_v4());
let empty = RunSnapshot::empty(&run_id, revision_digest, &semantics_digest);
let event = ReducerEvent::Start { input };
let output = reduce(&compiled, &empty, event.clone())?;
let mut snapshot = output.snapshot;
snapshot.conversation_id = conversation_id.map(str::to_owned);
snapshot.assistant_membership_id = assistant_membership_id.map(str::to_owned);
snapshot.route_receipt = route_receipt;
snapshot.cwd = cwd;
snapshot.slot_candidate_counts = slot_candidate_counts;
let now = now_ms();
transaction.execute(
"INSERT INTO strategy_runs(
run_id, revision_digest, semantics_digest, idempotency_key,
request_digest, snapshot_json, conversation_id, terminal,
created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)",
params![
run_id,
revision_digest,
semantics_digest,
idempotency_key,
request_digest,
serde_json::to_string(&snapshot)?,
snapshot.conversation_id,
i64::from(run_is_terminal(snapshot.status)),
now,
],
)?;
persist_event_and_commands(
&transaction,
&empty,
&snapshot,
&event,
&output.emitted_commands,
now,
)?;
transaction.commit()?;
Ok(snapshot)
})
}
#[allow(clippy::too_many_arguments)]
pub fn admit_assistant_run(
&self,
revision_digest: &str,
semantics_digest: &str,
workflow: &WorkflowDefinition,
bindings: &[BindingValue],
input: Value,
idempotency_key: &str,
conversation_id: &str,
assistant_membership_id: &str,
route_receipt: Value,
) -> Result<(RunSnapshot, bool)> {
validate_opaque_id(idempotency_key, "strategy_idempotency_key_invalid")?;
validate_opaque_id(conversation_id, "strategy_conversation_id_invalid")?;
validate_opaque_id(assistant_membership_id, "strategy_membership_id_invalid")?;
ensure!(
workflow
.metadata
.id
.starts_with(ASSISTANT_TEMPORARY_DEFINITION_PREFIX),
"graph_identity_rejected"
);
let compiled = compile_workflow(workflow.clone())?;
let workflow_json = serde_json::to_string(compiled.definition())?;
let mut expected_bindings = bindings.to_vec();
expected_bindings.sort_by(|left, right| {
left.slot_id
.cmp(&right.slot_id)
.then_with(|| left.ordinal.cmp(&right.ordinal))
});
for binding in &expected_bindings {
validate_opaque_id(&binding.slot_id, "strategy_binding_slot_invalid")?;
validate_opaque_id(&binding.value_id, "strategy_binding_value_invalid")?;
ensure!(binding.ordinal < 16, "strategy_binding_limit");
validate_optional_text(&binding.model, "strategy_binding_model_invalid")?;
validate_optional_text(
&binding.reasoning_effort,
"strategy_binding_reasoning_effort_invalid",
)?;
}
let mut request_source = Vec::from(revision_digest.as_bytes());
request_source.push(0);
request_source.extend_from_slice(&serde_json::to_vec(&input)?);
request_source.push(0);
request_source.extend_from_slice(conversation_id.as_bytes());
request_source.push(0);
request_source.extend_from_slice(assistant_membership_id.as_bytes());
request_source.push(0);
request_source.extend_from_slice(&serde_json::to_vec(&route_receipt)?);
let request_digest = sha256_hex(&request_source);
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let existing: Option<(String, String)> = transaction
.query_row(
"SELECT run_id, request_digest FROM strategy_runs WHERE idempotency_key=?1",
params![idempotency_key],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((run_id, existing_digest)) = existing {
ensure!(
existing_digest == request_digest,
"strategy_idempotency_conflict"
);
let snapshot = load_run(&transaction, &run_id)?;
transaction.commit()?;
return Ok((snapshot, false));
}
transaction.execute(
"INSERT INTO strategy_definitions(
definition_id, revision_digest, semantics_digest, name, version,
workflow_json, asset_count, imported_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7)
ON CONFLICT(revision_digest) DO NOTHING",
params![
compiled.definition().metadata.id,
revision_digest,
semantics_digest,
compiled.definition().metadata.name,
compiled.definition().metadata.version,
workflow_json,
now_ms(),
],
)?;
let stored_identity: (String, String, String) = transaction.query_row(
"SELECT definition_id, semantics_digest, workflow_json
FROM strategy_definitions WHERE revision_digest=?1",
params![revision_digest],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
ensure!(
stored_identity
== (
compiled.definition().metadata.id.clone(),
semantics_digest.to_owned(),
serde_json::to_string(compiled.definition())?,
),
"strategy_revision_conflict"
);
for binding in &expected_bindings {
transaction.execute(
"INSERT OR IGNORE INTO strategy_bindings(
revision_digest, slot_id, ordinal, value_id, model,
reasoning_effort, revision
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1)",
params![
revision_digest,
binding.slot_id,
binding.ordinal as i64,
binding.value_id,
binding.model,
binding.reasoning_effort,
],
)?;
}
let definition = definition_by_revision(&transaction, revision_digest)?;
let mut stored_bindings = definition.bindings.clone();
stored_bindings.sort_by(|left, right| {
left.slot_id
.cmp(&right.slot_id)
.then_with(|| left.ordinal.cmp(&right.ordinal))
});
ensure!(
stored_bindings.len() == expected_bindings.len()
&& stored_bindings
.iter()
.zip(&expected_bindings)
.all(|(stored, expected)| {
stored.slot_id == expected.slot_id
&& stored.ordinal == expected.ordinal
&& stored.value_id == expected.value_id
&& stored.model == expected.model
&& stored.reasoning_effort == expected.reasoning_effort
})
&& bindings_complete(&definition),
"strategy_revision_conflict"
);
let frozen_binding_digest = binding_digest(&stored_bindings)?;
let frozen_authorization_digest =
authorization_digest(revision_digest, semantics_digest, &frozen_binding_digest, 1);
transaction.execute(
"INSERT OR IGNORE INTO strategy_authorizations(
revision_digest, revision, semantics_digest, binding_digest,
authorization_digest, active, created_at
) VALUES (?1, 1, ?2, ?3, ?4, 1, ?5)",
params![
revision_digest,
semantics_digest,
frozen_binding_digest,
frozen_authorization_digest,
now_ms(),
],
)?;
let authorization = load_authorization(&transaction, revision_digest)?
.ok_or_else(|| anyhow!("strategy_authorization_stale"))?;
ensure!(
authorization.active
&& authorization.revision == 1
&& authorization.semantics_digest == semantics_digest
&& authorization.binding_digest == frozen_binding_digest
&& authorization.authorization_digest == frozen_authorization_digest,
"strategy_authorization_stale"
);
let slot_candidate_counts = slot_candidate_counts(&stored_bindings);
let run_id = format!("run-{}", Uuid::new_v4());
let empty = RunSnapshot::empty(&run_id, revision_digest, semantics_digest);
let event = ReducerEvent::Start { input };
let output = reduce(&compiled, &empty, event.clone())?;
let mut snapshot = output.snapshot;
snapshot.conversation_id = Some(conversation_id.to_owned());
snapshot.assistant_membership_id = Some(assistant_membership_id.to_owned());
snapshot.route_receipt = Some(route_receipt);
snapshot.slot_candidate_counts = slot_candidate_counts;
let now = now_ms();
transaction.execute(
"INSERT INTO strategy_runs(
run_id, revision_digest, semantics_digest, idempotency_key,
request_digest, snapshot_json, conversation_id, terminal,
created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)",
params![
run_id,
revision_digest,
semantics_digest,
idempotency_key,
request_digest,
serde_json::to_string(&snapshot)?,
conversation_id,
i64::from(run_is_terminal(snapshot.status)),
now,
],
)?;
persist_event_and_commands(
&transaction,
&empty,
&snapshot,
&event,
&output.emitted_commands,
now,
)?;
transaction.commit()?;
Ok((snapshot, true))
})
}
pub fn run_id_by_idempotency_key(&self, idempotency_key: &str) -> Result<Option<String>> {
validate_opaque_id(idempotency_key, "strategy_idempotency_key_invalid")?;
self.with_connection(|connection| {
connection
.query_row(
"SELECT run_id FROM strategy_runs WHERE idempotency_key=?1",
params![idempotency_key],
|row| row.get(0),
)
.optional()
.map_err(Into::into)
})
}
pub fn run(&self, run_id: &str) -> Result<RunSnapshot> {
self.with_connection(|connection| load_run(connection, run_id))
}
pub fn apply_event(&self, run_id: &str, event: ReducerEvent) -> Result<RunSnapshot> {
let committed = self.apply_event_with_commit(run_id, event)?;
match committed {
Some(committed) => Ok(committed.after),
None => self.run(run_id),
}
}
pub(crate) fn apply_event_with_commit(
&self,
run_id: &str,
event: ReducerEvent,
) -> Result<Option<super::CommittedTransition>> {
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let previous = load_run(&transaction, run_id)?;
let workflow = workflow_for_revision(&transaction, &previous.definition_digest)?;
let compiled = compile_workflow(workflow)?;
let output = reduce(&compiled, &previous, event.clone())?;
if output.applied {
let now = now_ms();
persist_event_and_commands(
&transaction,
&previous,
&output.snapshot,
&event,
&output.emitted_commands,
now,
)?;
transaction.execute(
"UPDATE strategy_runs SET snapshot_json=?2, conversation_id=?3,
terminal=?4, updated_at=?5 WHERE run_id=?1",
params![
run_id,
serde_json::to_string(&output.snapshot)?,
output.snapshot.conversation_id,
i64::from(run_is_terminal(output.snapshot.status)),
now
],
)?;
}
transaction.commit()?;
if output.applied {
Ok(Some(super::CommittedTransition {
run_id: run_id.to_owned(),
sequence: output.snapshot.sequence,
event,
before: previous,
after: output.snapshot,
created_at_unix_ms: now_ms(),
}))
} else {
Ok(None)
}
})
}
pub fn claim_next_command(
&self,
run_id: &str,
claimant: &str,
lease_until_unix_ms: i64,
) -> Result<Option<RunCommand>> {
validate_opaque_id(claimant, "strategy_claimant_invalid")?;
let now = now_ms();
ensure!(lease_until_unix_ms > now, "strategy_lease_invalid");
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let active: i64 = transaction.query_row(
"SELECT COUNT(*) FROM strategy_commands
WHERE status IN ('claimed', 'running') AND lease_until>?1",
params![now],
|row| row.get(0),
)?;
if active >= licoup_workflow::MAX_ACTIVE_EFFECTS as i64 {
transaction.commit()?;
return Ok(None);
}
let previous = load_run(&transaction, run_id)?;
if !matches!(
previous.status,
StrategyRunStatus::Running
| StrategyRunStatus::Waiting
| StrategyRunStatus::Retryable
) {
transaction.commit()?;
return Ok(None);
}
let workflow = workflow_for_revision(&transaction, &previous.definition_digest)?;
let compiled = compile_workflow(workflow)?;
let run_active: i64 = transaction.query_row(
"SELECT COUNT(*) FROM strategy_commands
WHERE run_id=?1 AND status IN ('claimed', 'running') AND lease_until>?2",
params![run_id, now],
|row| row.get(0),
)?;
if run_active >= compiled.definition().limits.max_parallelism as i64 {
transaction.commit()?;
return Ok(None);
}
let value: Option<String> = transaction
.query_row(
"SELECT command_json FROM strategy_commands
WHERE run_id=?1 AND status='pending' AND kind!='authorization'
ORDER BY command_id ASC LIMIT 1",
params![run_id],
|row| row.get(0),
)
.optional()?;
let Some(value) = value else {
transaction.commit()?;
return Ok(None);
};
let command: RunCommand = serde_json::from_str(&value)?;
let event = ReducerEvent::CommandClaimed {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
};
let output = reduce(&compiled, &previous, event.clone())?;
ensure!(output.applied, "strategy_command_not_claimable");
persist_event_and_commands(
&transaction,
&previous,
&output.snapshot,
&event,
&output.emitted_commands,
now,
)?;
transaction.execute(
"UPDATE strategy_runs SET snapshot_json=?2, conversation_id=?3,
terminal=?4, updated_at=?5 WHERE run_id=?1",
params![
run_id,
serde_json::to_string(&output.snapshot)?,
output.snapshot.conversation_id,
i64::from(run_is_terminal(output.snapshot.status)),
now
],
)?;
transaction.execute(
"UPDATE strategy_commands SET status='claimed', lease_owner=?2,
lease_until=?3, command_json=?4, updated_at=?5 WHERE command_id=?1",
params![
command.id,
claimant,
lease_until_unix_ms,
serde_json::to_string(&output.snapshot.commands[&command.id])?,
now,
],
)?;
let claimed = output.snapshot.commands.get(&command.id).cloned();
transaction.commit()?;
Ok(claimed)
})
}
pub fn renew_command_lease(
&self,
command_id: &str,
claimant: &str,
lease_until_unix_ms: i64,
) -> Result<()> {
validate_opaque_id(claimant, "strategy_claimant_invalid")?;
ensure!(lease_until_unix_ms > now_ms(), "strategy_lease_invalid");
self.with_connection(|connection| {
let changed = connection.execute(
"UPDATE strategy_commands SET lease_until=?3, updated_at=?4
WHERE command_id=?1 AND lease_owner=?2
AND status IN ('claimed', 'running')",
params![command_id, claimant, lease_until_unix_ms, now_ms()],
)?;
ensure!(changed == 1, "strategy_lease_lost");
Ok(())
})
}
pub fn authorize_effect(
&self,
run_id: &str,
command_id: &str,
attempt_token: &str,
expected_authorization_digest: &str,
claimant: &str,
lease_until_unix_ms: i64,
) -> Result<()> {
validate_opaque_id(claimant, "strategy_claimant_invalid")?;
let now = now_ms();
ensure!(lease_until_unix_ms > now, "strategy_lease_invalid");
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let lease: Option<(String, String, Option<String>, Option<i64>)> = transaction
.query_row(
"SELECT status, attempt_token, lease_owner, lease_until
FROM strategy_commands WHERE command_id=?1 AND run_id=?2",
params![command_id, run_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
)
.optional()?;
ensure!(
lease.is_some_and(|(status, token, owner, lease_until)| {
status == "running"
&& token == attempt_token
&& owner.as_deref() == Some(claimant)
&& lease_until.is_some_and(|value| value > now)
}),
"strategy_lease_lost"
);
let snapshot = load_run(&transaction, run_id)?;
let command = snapshot
.commands
.get(command_id)
.filter(|command| {
command.status == CommandStatus::Running
&& command.attempt_token == attempt_token
})
.ok_or_else(|| anyhow!("strategy_callback_stale"))?;
ensure!(command.id == command_id, "strategy_callback_stale");
let definition = definition_by_revision(&transaction, &snapshot.definition_digest)?;
ensure!(
definition.summary.semantics_digest == snapshot.semantics_digest,
"strategy_authorization_stale"
);
ensure!(
bindings_complete(&definition),
"strategy_binding_incomplete"
);
let authorization = current_authorization(&definition)?;
ensure!(
authorization.binding_digest == binding_digest(&definition.bindings)?
&& authorization.authorization_digest == expected_authorization_digest,
"strategy_authorization_stale"
);
let renewed = transaction.execute(
"UPDATE strategy_commands SET lease_until=?4, updated_at=?5
WHERE command_id=?1 AND run_id=?2 AND lease_owner=?3
AND status='running' AND lease_until>?5",
params![command_id, run_id, claimant, lease_until_unix_ms, now],
)?;
ensure!(renewed == 1, "strategy_lease_lost");
transaction.commit()?;
Ok(())
})
}
pub fn recover_next_expired_command(&self, run_id: &str) -> Result<bool> {
let Some(command_id) = self.next_expired_leased_command_id(run_id)? else {
return Ok(false);
};
self.recover_leased_command(run_id, &command_id)?;
Ok(true)
}
fn next_expired_leased_command_id(&self, run_id: &str) -> Result<Option<String>> {
self.with_connection(|connection| {
connection
.query_row(
"SELECT command_id FROM strategy_commands
WHERE run_id=?1 AND status IN ('claimed', 'running')
AND lease_until IS NOT NULL AND lease_until<=?2
ORDER BY command_id ASC LIMIT 1",
params![run_id, now_ms()],
|row| row.get(0),
)
.optional()
.map_err(Into::into)
})
}
fn recover_leased_command(&self, run_id: &str, command_id: &str) -> Result<()> {
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let value: Option<(String, String)> = transaction
.query_row(
"SELECT command_json, status FROM strategy_commands
WHERE run_id=?1 AND command_id=?2 AND status IN ('claimed', 'running')",
params![run_id, command_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
let Some((command_json, persisted_status)) = value else {
transaction.commit()?;
return Ok(());
};
let command: RunCommand = serde_json::from_str(&command_json)?;
let expected_status = enum_wire(command.status)?;
ensure!(
expected_status == persisted_status
&& matches!(
command.status,
CommandStatus::Claimed | CommandStatus::Running
),
"strategy_recovery_state_conflict"
);
let previous = load_run(&transaction, run_id)?;
ensure!(
previous.commands.get(&command.id).is_some_and(|current| {
current.status == command.status
&& current.attempt_token == command.attempt_token
}),
"strategy_recovery_state_conflict"
);
let workflow = workflow_for_revision(&transaction, &previous.definition_digest)?;
let compiled = compile_workflow(workflow)?;
let (class, code) = match command.status {
CommandStatus::Claimed => (FailureClass::Transient, "lease_expired_before_start"),
_ => (FailureClass::InDoubt, "effect_outcome_unknown"),
};
let failure_event = ReducerEvent::CommandFailed {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
class,
code: code.to_owned(),
};
let failure = reduce(&compiled, &previous, failure_event.clone())?;
let now = now_ms();
persist_event_and_commands(
&transaction,
&previous,
&failure.snapshot,
&failure_event,
&failure.emitted_commands,
now,
)?;
let final_snapshot = if failure
.snapshot
.commands
.get(&command.id)
.is_some_and(|current| current.status == CommandStatus::Retryable)
{
let retry_event = ReducerEvent::RetryRequested {
command_id: command.id.clone(),
};
let retry = reduce(&compiled, &failure.snapshot, retry_event.clone())?;
persist_event_and_commands(
&transaction,
&failure.snapshot,
&retry.snapshot,
&retry_event,
&retry.emitted_commands,
now,
)?;
retry.snapshot
} else {
failure.snapshot
};
transaction.execute(
"UPDATE strategy_runs SET snapshot_json=?2, conversation_id=?3,
terminal=?4, updated_at=?5 WHERE run_id=?1",
params![
run_id,
serde_json::to_string(&final_snapshot)?,
final_snapshot.conversation_id,
i64::from(run_is_terminal(final_snapshot.status)),
now
],
)?;
transaction.execute(
"UPDATE strategy_commands SET lease_owner=NULL, lease_until=NULL
WHERE command_id=?1",
params![&command.id],
)?;
transaction.commit()?;
Ok(())
})
}
pub fn projection_for_definition(&self, revision_digest: &str) -> Result<StrategyProjection> {
let definition = self.definition_by_revision(revision_digest)?;
let complete = bindings_complete(&definition);
let mut allowed = BTreeSet::from([
"strategy.definition.inspect".into(),
"strategy.binding.update".into(),
"strategy.binding.replace".into(),
"strategy.authorization.preview".into(),
]);
let status = if complete {
if definition
.authorization
.as_ref()
.is_some_and(|value| value.active)
{
allowed.extend([
"strategy.authorization.revoke".into(),
"strategy.run.start".into(),
]);
StrategyRunStatus::Pending
} else {
allowed.insert("strategy.authorization.grant".into());
StrategyRunStatus::AuthorizationRequired
}
} else {
StrategyRunStatus::Pending
};
Ok(StrategyProjection {
schema: STRATEGY_SCHEMA_VERSION.into(),
definition: definition.summary,
run_id: None,
status,
current_states: BTreeSet::new(),
neighbor_states: BTreeSet::new(),
allowed_operations: allowed,
bindings: definition.bindings,
diagnostic: if complete {
None
} else {
Some(StrategyDiagnostic {
code: "binding_incomplete".into(),
component: "strategy_binding".into(),
retryable: true,
recovery: "Bind every required actor and runtime slot.".into(),
arguments: BTreeMap::new(),
})
},
history_count: 0,
fallbacks: Vec::new(),
pending_callbacks: Vec::new(),
needs_human_input: false,
entry_session_id: None,
})
}
pub fn projection_for_run(&self, run_id: &str) -> Result<StrategyProjection> {
let snapshot = self.run(run_id)?;
let definition = self.definition_by_revision(&snapshot.definition_digest)?;
let compiled = compile_workflow(definition.workflow.clone())?;
let neighbors = snapshot
.active_states
.iter()
.flat_map(|state| {
compiled
.outgoing(state)
.map(|transition| transition.to.clone())
})
.collect();
let mut allowed = BTreeSet::from(["strategy.run.inspect".into()]);
match snapshot.status {
StrategyRunStatus::AuthorizationRequired
| StrategyRunStatus::RuntimeMissing
| StrategyRunStatus::Waiting => {
allowed.insert("strategy.run.resume".into());
allowed.insert("strategy.run.cancel".into());
}
StrategyRunStatus::Running => {
allowed.insert("strategy.run.resume".into());
allowed.insert("strategy.run.cancel".into());
}
StrategyRunStatus::Retryable => {
allowed.insert("strategy.run.retry".into());
allowed.insert("strategy.run.cancel".into());
}
StrategyRunStatus::CancelRequested => {
allowed.insert("strategy.run.inspect".into());
}
_ => {}
}
let history_count = self.with_connection(|connection| -> Result<u64> {
connection
.query_row(
"SELECT COUNT(*) FROM strategy_run_events WHERE run_id=?1",
params![run_id],
|row| row.get::<_, i64>(0).map(|value| value as u64),
)
.map_err(Into::into)
})?;
let pending_callbacks = snapshot.pending_callbacks.clone();
Ok(StrategyProjection {
schema: STRATEGY_SCHEMA_VERSION.into(),
definition: definition.summary,
run_id: Some(run_id.to_owned()),
status: snapshot.status,
current_states: snapshot.active_states,
neighbor_states: neighbors,
allowed_operations: allowed,
bindings: definition.bindings,
diagnostic: snapshot.diagnostic_code.map(|code| StrategyDiagnostic {
component: "strategy_runtime".into(),
retryable: matches!(
snapshot.status,
StrategyRunStatus::Retryable
| StrategyRunStatus::RuntimeMissing
| StrategyRunStatus::AuthorizationRequired
),
recovery: recovery_for_status(snapshot.status).into(),
code,
arguments: BTreeMap::new(),
}),
history_count,
fallbacks: snapshot.fallbacks,
pending_callbacks: pending_callbacks.clone(),
needs_human_input: !pending_callbacks.is_empty()
|| matches!(
snapshot.status,
StrategyRunStatus::AuthorizationRequired
| StrategyRunStatus::RuntimeMissing
| StrategyRunStatus::Waiting
| StrategyRunStatus::Retryable
),
entry_session_id: definition
.workflow
.actor_slots
.iter()
.find(|slot| slot.kind == licoup_workflow::BindingKind::Actor && slot.entry)
.and_then(|slot| {
let prefix = format!("{}\0", slot.id);
snapshot
.actor_sessions
.iter()
.filter(|(key, _)| key.starts_with(&prefix))
.filter_map(|(key, session)| {
snapshot
.merge_sources
.get(&format!("session\0{key}"))
.map(|source| (source, key, session))
})
.max_by(|left, right| left.0.cmp(right.0).then(left.1.cmp(right.1)))
.map(|(_, _, session)| session.clone())
}),
})
}
pub fn active_run_for_conversation(
&self,
revision_digest: &str,
conversation_id: &str,
) -> Result<Option<RunSnapshot>> {
validate_opaque_id(conversation_id, "strategy_conversation_id_invalid")?;
self.with_connection(|connection| {
connection
.query_row(
"SELECT snapshot_json FROM strategy_runs
WHERE revision_digest=?1 AND conversation_id=?2 AND terminal=0
ORDER BY updated_at DESC LIMIT 1",
params![revision_digest, conversation_id],
|row| row.get::<_, String>(0),
)
.optional()?
.map(|snapshot_json| serde_json::from_str(&snapshot_json).map_err(Into::into))
.transpose()
})
}
pub fn bind_conversation_if_absent(&self, run_id: &str, conversation_id: &str) -> Result<()> {
validate_opaque_id(conversation_id, "strategy_conversation_id_invalid")?;
self.with_connection(|connection| {
let transaction =
connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let mut snapshot = load_run(&transaction, run_id)?;
if snapshot
.conversation_id
.as_deref()
.is_some_and(|value| !value.is_empty())
{
transaction.commit()?;
return Ok(());
}
snapshot.conversation_id = Some(conversation_id.to_owned());
transaction.execute(
"UPDATE strategy_runs SET snapshot_json=?2, conversation_id=?3, updated_at=?4
WHERE run_id=?1",
params![
run_id,
serde_json::to_string(&snapshot)?,
snapshot.conversation_id,
now_ms()
],
)?;
transaction.commit()?;
Ok(())
})
}
fn remove_retired_revision_trees(&self, digests: &[String]) {
for digest in digests {
let Some(root) = &self.package_revisions_root else {
return;
};
if digest.len() != 64
|| !digest
.chars()
.all(|character| character.is_ascii_hexdigit())
{
continue;
}
let path = root.join(digest);
let Ok(metadata) = fs::symlink_metadata(&path) else {
continue;
};
if metadata.file_type().is_symlink() {
continue;
}
remove_directory_tree(&path);
}
}
}
fn initialize_schema(connection: &mut Connection) -> Result<Vec<String>> {
let retired = initialize_schema_through(connection)?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
migrate_legacy_workflow_definitions(&transaction)?;
transaction.execute("UPDATE strategy_meta SET value='3' WHERE key='version'", [])?;
transaction.commit()?;
super::queue::initialize_schema(connection)?;
super::subscriptions::initialize_schema(connection)?;
super::commit::initialize_schema(connection)?;
super::controlled::initialize_schema(connection)?;
Ok(retired)
}
fn initialize_schema_through(connection: &mut Connection) -> Result<Vec<String>> {
connection.execute_batch(
"CREATE TABLE IF NOT EXISTS strategy_meta(
key TEXT PRIMARY KEY, value TEXT NOT NULL
);
INSERT INTO strategy_meta(key, value) VALUES ('version', '3')
ON CONFLICT(key) DO NOTHING;
CREATE TABLE IF NOT EXISTS strategy_definitions(
definition_id TEXT NOT NULL,
revision_digest TEXT PRIMARY KEY,
semantics_digest TEXT NOT NULL,
name TEXT NOT NULL,
version TEXT NOT NULL,
workflow_json TEXT NOT NULL,
asset_count INTEGER NOT NULL,
imported_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS strategy_definitions_id_idx
ON strategy_definitions(definition_id, imported_at DESC);
CREATE TABLE IF NOT EXISTS strategy_bindings(
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest) ON DELETE CASCADE,
slot_id TEXT NOT NULL,
ordinal INTEGER NOT NULL DEFAULT 0,
value_id TEXT NOT NULL,
model TEXT NOT NULL DEFAULT '',
reasoning_effort TEXT NOT NULL DEFAULT '',
revision INTEGER NOT NULL,
PRIMARY KEY(revision_digest, slot_id, ordinal)
);
CREATE TABLE IF NOT EXISTS strategy_authorizations(
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest) ON DELETE CASCADE,
revision INTEGER NOT NULL,
semantics_digest TEXT NOT NULL,
binding_digest TEXT NOT NULL,
authorization_digest TEXT NOT NULL,
active INTEGER NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY(revision_digest, revision)
);
CREATE UNIQUE INDEX IF NOT EXISTS strategy_authorization_active_idx
ON strategy_authorizations(revision_digest) WHERE active=1;
CREATE TABLE IF NOT EXISTS strategy_runs(
run_id TEXT PRIMARY KEY,
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest),
semantics_digest TEXT NOT NULL,
idempotency_key TEXT NOT NULL UNIQUE,
request_digest TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
conversation_id TEXT,
terminal INTEGER NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS strategy_runs_revision_idx
ON strategy_runs(revision_digest, updated_at DESC);
CREATE TABLE IF NOT EXISTS strategy_run_events(
run_id TEXT NOT NULL REFERENCES strategy_runs(run_id) ON DELETE CASCADE,
sequence INTEGER NOT NULL,
event_type TEXT NOT NULL,
event_json TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY(run_id, sequence)
);
CREATE TABLE IF NOT EXISTS strategy_commands(
command_id TEXT PRIMARY KEY,
run_id TEXT NOT NULL REFERENCES strategy_runs(run_id) ON DELETE CASCADE,
state_id TEXT NOT NULL,
kind TEXT NOT NULL,
status TEXT NOT NULL,
attempt INTEGER NOT NULL,
attempt_token TEXT NOT NULL,
command_json TEXT NOT NULL,
lease_owner TEXT,
lease_until INTEGER,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS strategy_commands_ready_idx
ON strategy_commands(status, command_id);
CREATE INDEX IF NOT EXISTS strategy_commands_lease_idx
ON strategy_commands(lease_until) WHERE status IN ('claimed', 'running');",
)?;
ensure_column(
connection,
"strategy_bindings",
"model",
"TEXT NOT NULL DEFAULT ''",
)?;
ensure_column(
connection,
"strategy_bindings",
"reasoning_effort",
"TEXT NOT NULL DEFAULT ''",
)?;
ensure_column(connection, "strategy_runs", "conversation_id", "TEXT")?;
ensure_column(connection, "strategy_runs", "terminal", "INTEGER")?;
let retired = migrate_retired_builtin_definition(connection)?;
backfill_run_query_columns(connection)?;
connection.execute_batch(
"CREATE INDEX IF NOT EXISTS strategy_runs_active_conversation_idx
ON strategy_runs(revision_digest, conversation_id, terminal, updated_at DESC);",
)?;
migrate_bindings_ordinal_primary_key(connection)?;
Ok(retired)
}
fn migrate_retired_builtin_definition(connection: &mut Connection) -> Result<Vec<String>> {
const RETIRED_DEFINITION_ID: &str = "licoup-basic";
const RETIRED_DEFINITION_NAME: &str = "LicoUp Basic Strategy";
let mut statement = connection.prepare(
"SELECT revision_digest FROM strategy_definitions
WHERE definition_id=?1 OR name=?2",
)?;
let digests = statement
.query_map(
params![RETIRED_DEFINITION_ID, RETIRED_DEFINITION_NAME],
|row| row.get::<_, String>(0),
)?
.collect::<rusqlite::Result<Vec<_>>>()?;
drop(statement);
for digest in &digests {
connection.execute(
"DELETE FROM strategy_runs WHERE revision_digest=?1",
params![digest],
)?;
connection.execute(
"DELETE FROM strategy_definitions WHERE revision_digest=?1",
params![digest],
)?;
}
Ok(digests)
}
fn validate_current_schema(connection: &mut Connection) -> Result<()> {
configure_connection(connection)?;
let version: String = connection.query_row(
"SELECT value FROM strategy_meta WHERE key='version'",
[],
|row| row.get(0),
)?;
ensure!(version == "3", "strategy_schema_migration_required");
super::queue::initialize_schema(connection)?;
super::subscriptions::initialize_schema(connection)?;
super::commit::initialize_schema(connection)?;
super::controlled::initialize_schema(connection)?;
Ok(())
}
pub(crate) fn preflight_existing_store(path: &Path) -> Result<()> {
let metadata = match fs::symlink_metadata(path) {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(error) => return Err(error.into()),
};
ensure!(
metadata.is_file() && !metadata.file_type().is_symlink(),
"unsupported_state_shape"
);
let connection = Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
let version: String = connection.query_row(
"SELECT value FROM strategy_meta WHERE key='version'",
[],
|row| row.get(0),
)?;
ensure!(
matches!(version.as_str(), "2" | "3"),
"unsupported_state_shape"
);
ensure!(version == "3", "strategy_schema_migration_required");
validate_published_core_layout(&connection, &version)
}
pub fn validate_published_core_layout(
connection: &Connection,
expected_meta_version: &str,
) -> Result<()> {
let version: Option<String> = connection
.query_row(
"SELECT value FROM strategy_meta WHERE key='version'",
[],
|row| row.get(0),
)
.optional()?;
ensure!(
version.as_deref() == Some(expected_meta_version),
"unsupported_state_shape"
);
let mut reference = Connection::open_in_memory()?;
initialize_schema_through(&mut reference)?;
let required = licoup_foundation::core::sqlite_contract::tables(&reference)?;
super::queue::initialize_schema(&reference)?;
super::subscriptions::initialize_schema(&reference)?;
super::commit::initialize_schema(&reference)?;
super::controlled::initialize_schema(&reference)?;
let existing = licoup_foundation::core::sqlite_contract::tables(connection)?;
for table in licoup_foundation::core::sqlite_contract::tables(&reference)? {
if !existing.contains(&table) && !required.contains(&table) {
continue;
}
licoup_foundation::core::sqlite_contract::validate_table(
connection,
&reference,
&table,
match table.as_str() {
"strategy_runs" => Some(ColumnVariant::Nullable("terminal")),
"strategy_bindings" => Some(ColumnVariant::MissingZeroDefault("ordinal")),
_ => None,
},
)?;
}
Ok(())
}
fn backfill_run_query_columns(connection: &mut Connection) -> Result<()> {
let mut statement = connection
.prepare("SELECT run_id, snapshot_json FROM strategy_runs WHERE terminal IS NULL")?;
let runs = statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
drop(statement);
for (run_id, snapshot_json) in runs {
let snapshot: RunSnapshot = serde_json::from_str(&snapshot_json)?;
connection.execute(
"UPDATE strategy_runs SET conversation_id=?2, terminal=?3 WHERE run_id=?1",
params![
run_id,
snapshot.conversation_id,
i64::from(run_is_terminal(snapshot.status))
],
)?;
}
Ok(())
}
fn run_is_terminal(status: StrategyRunStatus) -> bool {
matches!(
status,
StrategyRunStatus::Completed
| StrategyRunStatus::Failed
| StrategyRunStatus::Cancelled
| StrategyRunStatus::Blocked
| StrategyRunStatus::CancelInDoubt
)
}
fn migrate_bindings_ordinal_primary_key(connection: &mut Connection) -> Result<()> {
let mut statement = connection.prepare("PRAGMA table_info(strategy_bindings)")?;
let columns = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<rusqlite::Result<Vec<_>>>()?;
if columns.iter().any(|column| column == "ordinal") {
return Ok(());
}
connection.execute_batch(
"CREATE TABLE strategy_bindings_v2 (
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest) ON DELETE CASCADE,
slot_id TEXT NOT NULL,
ordinal INTEGER NOT NULL,
value_id TEXT NOT NULL,
model TEXT NOT NULL DEFAULT '',
reasoning_effort TEXT NOT NULL DEFAULT '',
revision INTEGER NOT NULL,
PRIMARY KEY(revision_digest, slot_id, ordinal)
);
INSERT INTO strategy_bindings_v2(
revision_digest, slot_id, ordinal, value_id, model, reasoning_effort, revision
)
SELECT revision_digest, slot_id, 0, value_id, model, reasoning_effort, revision
FROM strategy_bindings;
DROP TABLE strategy_bindings;
ALTER TABLE strategy_bindings_v2 RENAME TO strategy_bindings;
UPDATE strategy_authorizations SET active=0 WHERE active=1;
UPDATE strategy_meta SET value='2' WHERE key='version';",
)?;
Ok(())
}
fn migrate_legacy_workflow_definitions(connection: &Connection) -> Result<()> {
let mut statement = connection.prepare(
"SELECT revision_digest, workflow_json FROM strategy_definitions ORDER BY revision_digest",
)?;
let definitions = statement
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
drop(statement);
for (revision, workflow_json) in definitions {
let Ok(workflow) = serde_json::from_str::<WorkflowDefinition>(&workflow_json) else {
continue;
};
let Some(workflow) = normalize_legacy_workflow(workflow) else {
continue;
};
let compiled = compile_workflow(workflow)?;
let canonical = serde_json::to_string(compiled.definition())?;
connection.execute(
"UPDATE strategy_definitions
SET workflow_json=?2
WHERE revision_digest=?1",
params![revision, canonical],
)?;
}
Ok(())
}
pub fn normalize_legacy_workflow(mut workflow: WorkflowDefinition) -> Option<WorkflowDefinition> {
let mut changed = false;
if !workflow
.actor_slots
.iter()
.any(|slot| slot.kind == BindingKind::Actor && slot.entry)
&& let Some(entry) = workflow
.actor_slots
.iter_mut()
.find(|slot| slot.kind == BindingKind::Actor)
{
entry.entry = true;
changed = true;
}
let effect_ids = workflow
.states
.iter()
.filter(|state| {
matches!(
state.kind,
GraphStateKind::Authorization
| GraphStateKind::Actor
| GraphStateKind::Script
| GraphStateKind::Workset
)
})
.map(|state| state.id.clone())
.collect::<BTreeSet<_>>();
let mut success = BTreeSet::new();
let mut failure = BTreeSet::new();
for transition in &workflow.transitions {
if effect_ids.contains(&transition.from) {
match transition.event {
TransitionEvent::Success => _ = success.insert(transition.from.clone()),
TransitionEvent::Failure => _ = failure.insert(transition.from.clone()),
TransitionEvent::Complete => {}
}
}
}
for transition in &mut workflow.transitions {
if !effect_ids.contains(&transition.from) || transition.event != TransitionEvent::Complete {
continue;
}
changed = true;
if success.insert(transition.from.clone()) {
transition.event = TransitionEvent::Success;
} else if failure.insert(transition.from.clone()) {
transition.event = TransitionEvent::Failure;
}
}
let before = workflow.transitions.len();
workflow.transitions.retain(|transition| {
!(effect_ids.contains(&transition.from) && transition.event == TransitionEvent::Complete)
});
changed |= workflow.transitions.len() != before;
let missing_failure = effect_ids
.iter()
.filter(|state| success.contains(*state) && !failure.contains(*state))
.cloned()
.collect::<Vec<_>>();
if !missing_failure.is_empty() {
changed = true;
let terminal = workflow
.states
.iter()
.find(|state| matches!(state.kind, GraphStateKind::Blocked | GraphStateKind::Fail))
.map(|state| state.id.clone())
.unwrap_or_else(|| {
let mut id = "blocked".to_owned();
let mut suffix = 0u32;
while workflow.states.iter().any(|state| state.id == id) {
suffix += 1;
id = format!("blocked-{suffix}");
}
workflow.states.push(GraphState {
id: id.clone(),
kind: GraphStateKind::Blocked,
label: "Blocked".into(),
instruction: String::new(),
binding: None,
runtime: None,
entry: None,
workset: None,
retry: Default::default(),
});
id
});
let mut transition_ids = workflow
.transitions
.iter()
.map(|transition| transition.id.clone())
.collect::<BTreeSet<_>>();
for state in missing_failure {
let mut id = format!("{state}-legacy-failure");
let mut suffix = 0u32;
while !transition_ids.insert(id.clone()) {
suffix += 1;
id = format!("{state}-legacy-failure-{suffix}");
}
workflow.transitions.push(Transition {
id,
from: state,
to: terminal.clone(),
event: TransitionEvent::Failure,
mode: TransitionMode::Flow,
guard: None,
});
}
}
changed.then_some(workflow)
}
fn configure_connection(connection: &Connection) -> Result<()> {
connection.execute_batch(
"PRAGMA foreign_keys=ON;
PRAGMA journal_mode=WAL;
PRAGMA synchronous=FULL;
PRAGMA busy_timeout=5000;",
)?;
Ok(())
}
fn definition_by_revision(connection: &Connection, revision: &str) -> Result<StrategyDefinition> {
let base: Option<(String, String, String, String, String, String, i64, i64)> = connection
.query_row(
"SELECT definition_id, revision_digest, semantics_digest, name, version,
workflow_json, asset_count, imported_at FROM strategy_definitions
WHERE revision_digest=?1",
params![revision],
|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()?;
let Some((
definition_id,
revision_digest,
semantics_digest,
name,
version,
workflow_json,
asset_count,
imported_at,
)) = base
else {
return Err(anyhow!("strategy_definition_not_found"));
};
let workflow: WorkflowDefinition = serde_json::from_str(&workflow_json)?;
let bindings = load_bindings(connection, &revision_digest)?;
let authorization = load_authorization(connection, &revision_digest)?;
Ok(StrategyDefinition {
summary: StrategyDefinitionSummary {
definition_id,
revision_digest,
semantics_digest,
name,
version,
imported_at_unix_ms: imported_at,
authorized: authorization.as_ref().is_some_and(|value| value.active),
},
workflow,
asset_count: asset_count as usize,
bindings,
authorization,
})
}
fn summary_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<StrategyDefinitionSummary> {
Ok(StrategyDefinitionSummary {
definition_id: row.get(0)?,
revision_digest: row.get(1)?,
semantics_digest: row.get(2)?,
name: row.get(3)?,
version: row.get(4)?,
imported_at_unix_ms: row.get(5)?,
authorized: row
.get::<_, i64>(6)
.map(|value| value != 0)
.unwrap_or(false),
})
}
fn load_bindings(connection: &Connection, revision: &str) -> Result<Vec<BindingValue>> {
let mut statement = connection.prepare(
"SELECT slot_id, ordinal, value_id, model, reasoning_effort, revision
FROM strategy_bindings
WHERE revision_digest=?1 ORDER BY slot_id ASC, ordinal ASC",
)?;
let rows = statement.query_map(params![revision], |row| {
Ok(BindingValue {
slot_id: row.get(0)?,
ordinal: row.get::<_, i64>(1)? as u8,
value_id: row.get(2)?,
model: row.get(3)?,
reasoning_effort: row.get(4)?,
revision: row.get::<_, i64>(5)? as u64,
})
})?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(Into::into)
}
fn load_authorization(
connection: &Connection,
revision: &str,
) -> Result<Option<StrategyAuthorization>> {
connection
.query_row(
"SELECT revision, semantics_digest, binding_digest, authorization_digest, active
FROM strategy_authorizations WHERE revision_digest=?1
ORDER BY revision DESC LIMIT 1",
params![revision],
|row| {
Ok(StrategyAuthorization {
definition_digest: revision.to_owned(),
revision: row.get::<_, i64>(0)? as u64,
semantics_digest: row.get(1)?,
binding_digest: row.get(2)?,
authorization_digest: row.get(3)?,
active: row.get::<_, i64>(4)? != 0,
})
},
)
.optional()
.map_err(Into::into)
}
fn current_authorization(definition: &StrategyDefinition) -> Result<&StrategyAuthorization> {
definition
.authorization
.as_ref()
.filter(|authorization| {
authorization.active
&& authorization.definition_digest == definition.summary.revision_digest
&& authorization.semantics_digest == definition.summary.semantics_digest
})
.ok_or_else(|| anyhow!("strategy_authorization_required"))
}
fn bindings_complete(definition: &StrategyDefinition) -> bool {
let bound = definition
.bindings
.iter()
.filter(|binding| binding.ordinal == 0)
.map(|binding| binding.slot_id.as_str())
.collect::<BTreeSet<_>>();
definition
.workflow
.actor_slots
.iter()
.filter(|slot| slot.required)
.all(|slot| bound.contains(slot.id.as_str()))
}
fn binding_digest(bindings: &[BindingValue]) -> Result<String> {
let mut sorted = bindings.to_vec();
sorted.sort_by(|left, right| {
left.slot_id
.cmp(&right.slot_id)
.then(left.ordinal.cmp(&right.ordinal))
});
Ok(sha256_hex(&serde_json::to_vec(&sorted)?))
}
fn slot_candidate_counts(bindings: &[BindingValue]) -> BTreeMap<String, u8> {
let mut counts = BTreeMap::new();
for binding in bindings {
let entry = counts.entry(binding.slot_id.clone()).or_insert(0);
*entry = (*entry).max(binding.ordinal.saturating_add(1));
}
counts
}
fn authorization_digest(
revision: &str,
semantics: &str,
bindings: &str,
authorization_revision: u64,
) -> String {
sha256_hex(
format!(
"licoup-strategy-authorization-v1\0{revision}\0{semantics}\0{bindings}\0{authorization_revision}"
)
.as_bytes(),
)
}
fn workflow_for_revision(connection: &Connection, revision: &str) -> Result<WorkflowDefinition> {
let value: String = connection
.query_row(
"SELECT workflow_json FROM strategy_definitions WHERE revision_digest=?1",
params![revision],
|row| row.get(0),
)
.optional()?
.ok_or_else(|| anyhow!("strategy_definition_not_found"))?;
serde_json::from_str(&value).map_err(Into::into)
}
fn load_run(connection: &Connection, run_id: &str) -> Result<RunSnapshot> {
let value: Option<String> = connection
.query_row(
"SELECT snapshot_json FROM strategy_runs WHERE run_id=?1",
params![run_id],
|row| row.get(0),
)
.optional()?;
serde_json::from_str(
value
.as_deref()
.ok_or_else(|| anyhow!("strategy_run_not_found"))?,
)
.map_err(Into::into)
}
fn persist_event_and_commands(
transaction: &Transaction<'_>,
before: &RunSnapshot,
snapshot: &RunSnapshot,
event: &ReducerEvent,
emitted: &[RunCommand],
now: i64,
) -> Result<()> {
let event_json = serde_json::to_string(event)?;
let event_type = event_json.split('"').nth(3).unwrap_or("event").to_owned();
transaction.execute(
"INSERT INTO strategy_run_events(run_id, sequence, event_type, event_json, created_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
params![
snapshot.run_id,
snapshot.sequence as i64,
event_type,
event_json,
now
],
)?;
super::commit::insert_intent(
transaction,
&snapshot.run_id,
snapshot.sequence,
event,
before,
snapshot,
now,
)?;
for command in emitted {
transaction.execute(
"INSERT INTO strategy_commands(
command_id, run_id, state_id, kind, status, attempt,
attempt_token, command_json, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
params![
command.id,
snapshot.run_id,
command.state_id,
enum_wire(command.kind)?,
enum_wire(command.status)?,
command.attempt as i64,
command.attempt_token,
serde_json::to_string(command)?,
now,
],
)?;
}
for command in snapshot.commands.values() {
transaction.execute(
"UPDATE strategy_commands SET status=?2, command_json=?3, updated_at=?4
WHERE command_id=?1",
params![
command.id,
enum_wire(command.status)?,
serde_json::to_string(command)?,
now,
],
)?;
}
Ok(())
}
fn enum_wire(value: impl serde::Serialize) -> Result<String> {
serde_json::to_value(value)?
.as_str()
.map(str::to_owned)
.ok_or_else(|| anyhow!("strategy_enum_invalid"))
}
fn validate_opaque_id(value: &str, code: &'static str) -> Result<()> {
ensure!(
value == value.trim()
&& !value.is_empty()
&& value.len() <= 160
&& !value.chars().any(char::is_control),
code
);
Ok(())
}
fn validate_optional_text(value: &str, code: &'static str) -> Result<()> {
ensure!(
value == value.trim() && value.len() <= 160 && !value.chars().any(char::is_control),
code
);
Ok(())
}
fn ensure_column(
connection: &Connection,
table: &str,
column: &str,
declaration: &str,
) -> Result<()> {
let mut statement = connection.prepare(&format!("PRAGMA table_info({table})"))?;
let existing = statement
.query_map([], |row| row.get::<_, String>(1))?
.collect::<rusqlite::Result<Vec<_>>>()?;
if !existing.iter().any(|value| value == column) {
connection.execute_batch(&format!(
"ALTER TABLE {table} ADD COLUMN {column} {declaration}"
))?;
}
Ok(())
}
fn recovery_for_status(status: StrategyRunStatus) -> &'static str {
match status {
StrategyRunStatus::AuthorizationRequired => "Review and authorize the current semantics.",
StrategyRunStatus::RuntimeMissing => "Bind an available verified local runtime.",
StrategyRunStatus::Retryable => "Retry the failed command.",
StrategyRunStatus::CancelInDoubt => "Inspect the external effect before continuing.",
_ => "Inspect the run history.",
}
}
fn sha256_hex(bytes: &[u8]) -> String {
Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or(0)
}
fn remove_directory_tree(path: &Path) {
let mut stack = vec![path.to_path_buf()];
while let Some(current) = stack.pop() {
if let Ok(metadata) = fs::symlink_metadata(¤t) {
make_path_writable(¤t, metadata.permissions());
if metadata.is_dir()
&& let Ok(entries) = fs::read_dir(¤t)
{
stack.extend(entries.flatten().map(|entry| entry.path()));
}
}
}
let _ = fs::remove_dir_all(path);
}
fn make_path_writable(path: &Path, mut permissions: fs::Permissions) {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
permissions.set_mode(permissions.mode() | 0o200);
}
#[cfg(not(unix))]
permissions.set_readonly(false);
let _ = fs::set_permissions(path, permissions);
}
#[cfg(test)]
mod tests {
use super::*;
use licoup_workflow::{
ActorSlot, GraphState, GraphStateKind, RetryPolicy, Transition, TransitionEvent,
TransitionMode, WorkflowLimits, WorkflowMetadata,
};
use serde_json::json;
fn workflow() -> WorkflowDefinition {
WorkflowDefinition {
schema: licoup_workflow::WORKFLOW_SCHEMA_VERSION.into(),
metadata: WorkflowMetadata {
id: "test".into(),
name: "Test".into(),
version: "1".into(),
description: String::new(),
},
limits: WorkflowLimits::default(),
actor_slots: vec![ActorSlot::required_actor("worker", "Worker")],
runtimes: vec![],
worksets: vec![],
initial: "work".into(),
states: vec![
GraphState {
id: "work".into(),
kind: GraphStateKind::Actor,
label: "Work".into(),
instruction: String::new(),
binding: Some("worker".into()),
runtime: None,
entry: None,
workset: None,
retry: RetryPolicy {
max_attempts: 2,
transient_only: true,
},
},
GraphState {
id: "done".into(),
kind: GraphStateKind::Succeed,
label: "Done".into(),
instruction: String::new(),
binding: None,
runtime: None,
entry: None,
workset: None,
retry: RetryPolicy::default(),
},
GraphState {
id: "fail".into(),
kind: GraphStateKind::Fail,
label: "Fail".into(),
instruction: String::new(),
binding: None,
runtime: None,
entry: None,
workset: None,
retry: RetryPolicy::default(),
},
],
transitions: vec![
Transition {
id: "done".into(),
from: "work".into(),
to: "done".into(),
event: TransitionEvent::Success,
mode: TransitionMode::Flow,
guard: None,
},
Transition {
id: "failed".into(),
from: "work".into(),
to: "fail".into(),
event: TransitionEvent::Failure,
mode: TransitionMode::Flow,
guard: None,
},
],
}
}
#[test]
fn binding_ordinal_migration_invalidates_existing_authorizations() {
let mut connection = Connection::open_in_memory().unwrap();
connection
.execute_batch(
"CREATE TABLE strategy_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO strategy_meta(key, value) VALUES ('version', '1');
CREATE TABLE strategy_definitions(
definition_id TEXT NOT NULL,
revision_digest TEXT PRIMARY KEY,
semantics_digest TEXT NOT NULL,
name TEXT NOT NULL,
version TEXT NOT NULL,
workflow_json TEXT NOT NULL,
asset_count INTEGER NOT NULL,
imported_at INTEGER NOT NULL
);
CREATE TABLE strategy_bindings(
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest) ON DELETE CASCADE,
slot_id TEXT NOT NULL,
value_id TEXT NOT NULL,
model TEXT NOT NULL DEFAULT '',
reasoning_effort TEXT NOT NULL DEFAULT '',
revision INTEGER NOT NULL,
PRIMARY KEY(revision_digest, slot_id)
);
CREATE TABLE strategy_authorizations(
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest) ON DELETE CASCADE,
revision INTEGER NOT NULL,
semantics_digest TEXT NOT NULL,
binding_digest TEXT NOT NULL,
authorization_digest TEXT NOT NULL,
active INTEGER NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY(revision_digest, revision)
);
INSERT INTO strategy_definitions VALUES (
'strategy', 'revision', 'semantics', 'Strategy', '1', '{}', 0, 1
);
INSERT INTO strategy_bindings VALUES (
'revision', 'worker', 'agent:test', '', '', 1
);
INSERT INTO strategy_authorizations VALUES (
'revision', 1, 'semantics', 'old-bindings', 'authorization', 1, 1
);",
)
.unwrap();
let retired = initialize_schema(&mut connection).unwrap();
assert!(retired.is_empty());
let (ordinal, active, version): (i64, i64, String) = connection
.query_row(
"SELECT b.ordinal, a.active, m.value
FROM strategy_bindings b
JOIN strategy_authorizations a USING (revision_digest)
JOIN strategy_meta m ON m.key='version'",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.unwrap();
assert_eq!(ordinal, 0);
assert_eq!(active, 0);
assert_eq!(version, "3");
validate_published_core_layout(&connection, "3").unwrap();
}
#[test]
fn ordinary_open_rejects_a_legacy_strategy_schema() {
let root = std::env::temp_dir().join(format!("lico-strategy-legacy-{}", Uuid::new_v4()));
let database = root.join("client-state/adaptive-flywheel/strategies.sqlite3");
fs::create_dir_all(database.parent().unwrap()).unwrap();
let connection = Connection::open(&database).unwrap();
connection
.execute_batch(
"CREATE TABLE strategy_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO strategy_meta(key, value) VALUES ('version', '1');",
)
.unwrap();
drop(connection);
assert!(StrategyStore::open(&root).is_err());
StrategyStore::open_for_migration(&root).unwrap();
StrategyStore::open(&root).unwrap();
let _ = fs::remove_dir_all(root);
}
#[test]
fn typed_migration_materializes_legacy_workflow_routing_once() {
let root =
std::env::temp_dir().join(format!("lico-strategy-workflow-routing-{}", Uuid::new_v4()));
let database = root.join("client-state/adaptive-flywheel/strategies.sqlite3");
fs::create_dir_all(database.parent().unwrap()).unwrap();
let connection = Connection::open(&database).unwrap();
connection
.execute_batch(
"CREATE TABLE strategy_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO strategy_meta(key, value) VALUES ('version', '2');
CREATE TABLE strategy_definitions(
definition_id TEXT NOT NULL,
revision_digest TEXT PRIMARY KEY,
semantics_digest TEXT NOT NULL,
name TEXT NOT NULL,
version TEXT NOT NULL,
workflow_json TEXT NOT NULL,
asset_count INTEGER NOT NULL,
imported_at INTEGER NOT NULL
);",
)
.unwrap();
let mut success_only = workflow();
success_only.actor_slots[0].entry = false;
success_only
.transitions
.retain(|transition| transition.event != TransitionEvent::Failure);
let mut complete_as_failure = success_only.clone();
complete_as_failure.transitions.push(Transition {
id: "legacy-complete".into(),
from: "work".into(),
to: "fail".into(),
event: TransitionEvent::Complete,
mode: TransitionMode::Flow,
guard: None,
});
let mut redundant_complete = workflow();
redundant_complete.actor_slots[0].entry = false;
redundant_complete.transitions.push(Transition {
id: "legacy-complete".into(),
from: "work".into(),
to: "fail".into(),
event: TransitionEvent::Complete,
mode: TransitionMode::Flow,
guard: None,
});
for (revision, definition) in [
("legacy-success", success_only),
("legacy-convert", complete_as_failure),
("legacy-redundant", redundant_complete),
] {
connection
.execute(
"INSERT INTO strategy_definitions VALUES (?1, ?1, 'legacy', ?1, '1', ?2, 0, 1)",
params![revision, serde_json::to_string(&definition).unwrap()],
)
.unwrap();
}
drop(connection);
assert!(StrategyStore::open(&root).is_err());
StrategyStore::open_for_migration(&root).unwrap();
let store = StrategyStore::open(&root).unwrap();
for revision in ["legacy-success", "legacy-convert", "legacy-redundant"] {
let definition = store.definition_by_revision(revision).unwrap();
assert!(definition.workflow.actor_slots[0].entry);
assert_eq!(
definition
.workflow
.transitions
.iter()
.filter(|transition| {
transition.from == "work" && transition.event == TransitionEvent::Failure
})
.count(),
1
);
assert!(definition.workflow.transitions.iter().all(|transition| {
transition.from != "work" || transition.event != TransitionEvent::Complete
}));
assert!(compile_workflow(definition.workflow).is_ok());
}
let _ = fs::remove_dir_all(root);
}
#[test]
fn workflow_migration_preserves_live_run_and_authorization_identity() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
let semantics = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee";
let current = workflow();
store
.register_definition(revision, semantics, ¤t, 1, 1)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
let authorization = store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "migration-live-run", None, None)
.unwrap();
let claimed = store
.claim_next_command(&run.run_id, "migration-host", now_ms() + 60_000)
.unwrap()
.unwrap();
store
.apply_event(
&run.run_id,
ReducerEvent::CommandStarted {
command_id: claimed.id.clone(),
attempt_token: claimed.attempt_token.clone(),
},
)
.unwrap();
let mut legacy = current;
legacy.actor_slots[0].entry = false;
legacy
.transitions
.retain(|transition| transition.event != TransitionEvent::Failure);
let before = store
.with_connection(|connection| {
connection.execute(
"UPDATE strategy_definitions SET workflow_json=?2 WHERE revision_digest=?1",
params![revision, serde_json::to_string(&legacy)?],
)?;
connection.execute("UPDATE strategy_meta SET value='2' WHERE key='version'", [])?;
let run_row = connection.query_row(
"SELECT semantics_digest, snapshot_json FROM strategy_runs WHERE run_id=?1",
params![run.run_id],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)?;
let command_row = connection.query_row(
"SELECT status, attempt, attempt_token, command_json, lease_owner, lease_until
FROM strategy_commands WHERE command_id=?1",
params![claimed.id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<i64>>(5)?,
))
},
)?;
let authorization_row = connection.query_row(
"SELECT semantics_digest, binding_digest, authorization_digest, active
FROM strategy_authorizations WHERE revision_digest=?1",
params![revision],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
))
},
)?;
Ok((run_row, command_row, authorization_row))
})
.unwrap();
store.with_connection(initialize_schema).unwrap();
let after = store
.with_connection(|connection| {
let run_row = connection.query_row(
"SELECT semantics_digest, snapshot_json FROM strategy_runs WHERE run_id=?1",
params![run.run_id],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)?;
let command_row = connection.query_row(
"SELECT status, attempt, attempt_token, command_json, lease_owner, lease_until
FROM strategy_commands WHERE command_id=?1",
params![claimed.id],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<i64>>(5)?,
))
},
)?;
let authorization_row = connection.query_row(
"SELECT semantics_digest, binding_digest, authorization_digest, active
FROM strategy_authorizations WHERE revision_digest=?1",
params![revision],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
))
},
)?;
Ok((run_row, command_row, authorization_row))
})
.unwrap();
assert_eq!(after, before);
assert_eq!(after.0.0, semantics);
assert_eq!(after.1.0, "running");
assert_eq!(after.1.2, claimed.attempt_token);
assert_eq!(after.2.2, authorization.authorization_digest);
assert_eq!(after.2.3, 1);
let migrated = store.definition_by_revision(revision).unwrap();
assert_eq!(migrated.summary.revision_digest, revision);
assert_eq!(migrated.summary.semantics_digest, semantics);
assert!(migrated.workflow.actor_slots[0].entry);
assert!(compile_workflow(migrated.workflow).is_ok());
}
#[test]
fn workflow_routing_migration_rolls_back_all_rows_when_a_later_row_is_invalid() {
let store = StrategyStore::open_in_memory().unwrap();
let mut legacy = workflow();
legacy.actor_slots[0].entry = false;
legacy
.transitions
.retain(|transition| transition.event != TransitionEvent::Failure);
let legacy_json = serde_json::to_string(&legacy).unwrap();
let mut invalid = legacy.clone();
invalid.initial = "missing-state".into();
let invalid_json = serde_json::to_string(&invalid).unwrap();
store
.with_connection(|connection| {
connection.execute("UPDATE strategy_meta SET value='2' WHERE key='version'", [])?;
for (definition, revision, workflow_json) in [
("first", "a-valid-legacy", &legacy_json),
("later", "z-invalid-legacy", &invalid_json),
] {
connection.execute(
"INSERT INTO strategy_definitions(
definition_id, revision_digest, semantics_digest, name, version,
workflow_json, asset_count, imported_at
) VALUES (?1, ?2, 'legacy-semantics', ?1, '1', ?3, 0, 1)",
params![definition, revision, workflow_json],
)?;
}
Ok(())
})
.unwrap();
assert!(store.with_connection(initialize_schema).is_err());
store
.with_connection(|connection| {
let version: String = connection.query_row(
"SELECT value FROM strategy_meta WHERE key='version'",
[],
|row| row.get(0),
)?;
let first: String = connection.query_row(
"SELECT workflow_json FROM strategy_definitions WHERE revision_digest='a-valid-legacy'",
[],
|row| row.get(0),
)?;
assert_eq!(version, "2");
assert_eq!(first, legacy_json);
Ok(())
})
.unwrap();
}
#[test]
fn event_snapshot_and_outbox_commit_together() {
let store = StrategyStore::open_in_memory().unwrap();
store
.register_definition(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
&workflow(),
1,
1,
)
.unwrap();
let binding = store
.update_binding(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"worker",
"agent:test",
"gpt-5",
"high",
None,
)
.unwrap();
assert_eq!(binding.model, "gpt-5");
assert_eq!(binding.reasoning_effort, "high");
let persisted = store
.definition_by_revision(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
)
.unwrap();
assert_eq!(persisted.bindings[0].ordinal, 0);
assert_eq!(persisted.bindings[0].model, "gpt-5");
assert_eq!(persisted.bindings[0].reasoning_effort, "high");
let chain = store
.replace_slot_bindings(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"worker",
&[
BindingCandidate {
value_id: "agent:primary".into(),
model: "model-a".into(),
reasoning_effort: "high".into(),
},
BindingCandidate {
value_id: "agent:fallback".into(),
model: "model-b".into(),
reasoning_effort: String::new(),
},
],
Some(binding.revision),
)
.unwrap();
assert_eq!(chain.len(), 2);
assert_eq!(chain[0].ordinal, 0);
assert_eq!(chain[1].ordinal, 1);
assert_eq!(chain[1].value_id, "agent:fallback");
let preview = store
.authorization_preview(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
)
.unwrap();
store
.grant_authorization(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
&preview.authorization_digest,
)
.unwrap();
let run = store
.start_run(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
json!({}),
"idempotency-test",
Some("conversation:test"),
None,
)
.unwrap();
assert_eq!(run.commands.len(), 1);
assert_eq!(
store
.active_run_for_conversation(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"conversation:test",
)
.unwrap()
.map(|snapshot| snapshot.run_id),
Some(run.run_id.clone())
);
let indexed: (String, i64) = store
.with_connection(|connection| {
connection
.query_row(
"SELECT conversation_id, terminal FROM strategy_runs WHERE run_id=?1",
params![run.run_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.map_err(Into::into)
})
.unwrap();
assert_eq!(indexed, ("conversation:test".to_owned(), 0));
let replay = store
.start_run(
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
json!({}),
"idempotency-test",
Some("conversation:test"),
None,
)
.unwrap();
assert_eq!(run, replay);
}
#[test]
fn effect_authorization_revalidates_digest_owner_and_live_lease() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee";
store
.register_definition(revision, revision, &workflow(), 1, 1)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
let authorization = store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "authorization-fence", None, None)
.unwrap();
let command = store
.claim_next_command(&run.run_id, "claimant", now_ms() + 60_000)
.unwrap()
.unwrap();
store
.apply_event(
&run.run_id,
ReducerEvent::CommandStarted {
command_id: command.id.clone(),
attempt_token: command.attempt_token.clone(),
},
)
.unwrap();
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&authorization.authorization_digest,
"claimant",
now_ms() + 60_000,
)
.unwrap();
assert!(
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&authorization.authorization_digest,
"other-claimant",
now_ms() + 60_000,
)
.is_err()
);
store.revoke_authorization(revision).unwrap();
assert!(
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&authorization.authorization_digest,
"claimant",
now_ms() + 60_000,
)
.is_err()
);
let next = store.authorization_preview(revision).unwrap();
let next = store
.grant_authorization(revision, &next.authorization_digest)
.unwrap();
assert!(
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&authorization.authorization_digest,
"claimant",
now_ms() + 60_000,
)
.is_err()
);
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&next.authorization_digest,
"claimant",
now_ms() + 60_000,
)
.unwrap();
let connection = Connection::open(store.db_path()).unwrap();
connection
.execute(
"UPDATE strategy_commands SET lease_until=0 WHERE command_id=?1",
params![&command.id],
)
.unwrap();
assert!(
store
.authorize_effect(
&run.run_id,
&command.id,
&command.attempt_token,
&next.authorization_digest,
"claimant",
now_ms() + 60_000,
)
.is_err(),
"an expired running lease must never issue an effect permit"
);
}
#[test]
fn expired_claim_recovery_is_atomic_and_retries_before_start() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
store
.register_definition(
revision,
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
&workflow(),
1,
1,
)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "recovery-test", None, None)
.unwrap();
let claimed = store
.claim_next_command(&run.run_id, "claimant", now_ms() + 60_000)
.unwrap()
.unwrap();
let connection = Connection::open(store.db_path()).unwrap();
connection
.execute(
"UPDATE strategy_commands SET lease_until=0 WHERE command_id=?1",
params![claimed.id],
)
.unwrap();
assert!(store.recover_next_expired_command(&run.run_id).unwrap());
assert!(!store.recover_next_expired_command(&run.run_id).unwrap());
assert!(
store
.renew_command_lease(&claimed.id, "claimant", now_ms() + 60_000)
.is_err()
);
let recovered = store.run(&run.run_id).unwrap();
assert_eq!(
recovered.commands[&claimed.id].status,
CommandStatus::Cancelled
);
assert!(
recovered.commands.values().any(|command| {
command.attempt == 2 && command.status == CommandStatus::Pending
})
);
}
#[test]
fn expired_running_effect_is_fenced_in_doubt_without_retry() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
store
.register_definition(
revision,
"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
&workflow(),
1,
1,
)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "running-recovery-test", None, None)
.unwrap();
let claimed = store
.claim_next_command(&run.run_id, "claimant", now_ms() + 60_000)
.unwrap()
.unwrap();
store
.apply_event(
&run.run_id,
ReducerEvent::CommandStarted {
command_id: claimed.id.clone(),
attempt_token: claimed.attempt_token.clone(),
},
)
.unwrap();
let connection = Connection::open(store.db_path()).unwrap();
connection
.execute(
"UPDATE strategy_commands SET lease_until=0 WHERE command_id=?1",
params![claimed.id],
)
.unwrap();
assert!(store.recover_next_expired_command(&run.run_id).unwrap());
let recovered = store.run(&run.run_id).unwrap();
assert_eq!(recovered.status, StrategyRunStatus::CancelInDoubt);
assert_eq!(
recovered.commands[&claimed.id].status,
CommandStatus::InDoubt
);
assert!(
!recovered
.commands
.values()
.any(|command| command.attempt == 2)
);
}
#[test]
fn an_unexpired_claim_stays_held_when_this_host_becomes_the_driver() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee";
store
.register_definition(
revision,
"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
&workflow(),
1,
1,
)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "abandoned-host-recovery", None, None)
.unwrap();
let claimed = store
.claim_next_command(&run.run_id, "claimant", now_ms() + 60_000)
.unwrap()
.unwrap();
store
.apply_event(
&run.run_id,
ReducerEvent::CommandStarted {
command_id: claimed.id.clone(),
attempt_token: claimed.attempt_token.clone(),
},
)
.unwrap();
assert!(!store.recover_next_expired_command(&run.run_id).unwrap());
let held = store.run(&run.run_id).unwrap();
assert_eq!(held.commands[&claimed.id].status, CommandStatus::Running);
assert!(held.commands[&claimed.id].failure_code.is_none());
assert!(
!held.commands.values().any(|command| command.attempt == 2),
"an unexpired in-flight effect is never repeated"
);
assert_eq!(held.status, StrategyRunStatus::Running);
}
#[test]
fn a_released_unexpired_claim_never_becomes_retryable() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "5656565656565656565656565656565656565656565656565656565656565656";
store
.register_definition(
revision,
"7878787878787878787878787878787878787878787878787878787878787878",
&workflow(),
1,
1,
)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "held-claim", None, None)
.unwrap();
let claimed = store
.claim_next_command(&run.run_id, "claimant", now_ms() + 60_000)
.unwrap()
.unwrap();
store
.apply_event(
&run.run_id,
ReducerEvent::CommandStarted {
command_id: claimed.id.clone(),
attempt_token: claimed.attempt_token.clone(),
},
)
.unwrap();
assert!(
store
.apply_event(
&run.run_id,
ReducerEvent::CommandSucceeded {
command_id: claimed.id.clone(),
attempt_token: String::from("attempt:someone-else"),
output: json!({ "ok": true }),
},
)
.is_err(),
"a stale owner result cannot settle an attempt it does not own"
);
let unchanged = store.run(&run.run_id).unwrap();
assert_eq!(
unchanged.commands[&claimed.id].status,
CommandStatus::Running
);
assert!(unchanged.commands[&claimed.id].output_digest.is_none());
}
#[test]
fn bind_conversation_if_absent_fills_an_unbound_run() {
let store = StrategyStore::open_in_memory().unwrap();
let revision = "1111111111111111111111111111111111111111111111111111111111111111";
store
.register_definition(
revision,
"2222222222222222222222222222222222222222222222222222222222222222",
&workflow(),
1,
1,
)
.unwrap();
store
.update_binding(revision, "worker", "agent:test", "", "", None)
.unwrap();
let preview = store.authorization_preview(revision).unwrap();
store
.grant_authorization(revision, &preview.authorization_digest)
.unwrap();
let run = store
.start_run(revision, json!({}), "bind-conversation", None, None)
.unwrap();
assert!(run.conversation_id.is_none());
store
.bind_conversation_if_absent(&run.run_id, "conversation:group")
.unwrap();
store
.bind_conversation_if_absent(&run.run_id, "conversation:other")
.unwrap();
let bound = store.run(&run.run_id).unwrap();
assert_eq!(bound.conversation_id.as_deref(), Some("conversation:group"));
}
fn digest(fill: char) -> String {
fill.to_string().repeat(64)
}
#[test]
fn retired_builtin_definition_is_deleted_by_the_typed_migration() {
let root =
std::env::temp_dir().join(format!("lico-adaptive-flywheel-retire-{}", Uuid::new_v4()));
let database = root.join("client-state/adaptive-flywheel/strategies.sqlite3");
fs::create_dir_all(database.parent().unwrap()).unwrap();
let retired_id_digest = digest('a');
let retired_name_digest = digest('b');
let kept_digest = digest('c');
let connection = Connection::open(&database).unwrap();
connection
.execute_batch(
"CREATE TABLE strategy_meta(key TEXT PRIMARY KEY, value TEXT NOT NULL);
INSERT INTO strategy_meta(key, value) VALUES ('version', '1');
CREATE TABLE strategy_definitions(
definition_id TEXT NOT NULL,
revision_digest TEXT PRIMARY KEY,
semantics_digest TEXT NOT NULL,
name TEXT NOT NULL,
version TEXT NOT NULL,
workflow_json TEXT NOT NULL,
asset_count INTEGER NOT NULL,
imported_at INTEGER NOT NULL
);
CREATE TABLE strategy_runs(
run_id TEXT PRIMARY KEY,
revision_digest TEXT NOT NULL REFERENCES strategy_definitions(revision_digest),
semantics_digest TEXT NOT NULL,
idempotency_key TEXT NOT NULL UNIQUE,
request_digest TEXT NOT NULL,
snapshot_json TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);",
)
.unwrap();
let workflow_json = serde_json::to_string(&workflow()).unwrap();
for (definition_id, revision, name) in [
(
"licoup-basic",
retired_id_digest.clone(),
"LicoUp Basic Strategy",
),
(
"other-id",
retired_name_digest.clone(),
"LicoUp Basic Strategy",
),
("imported-graph", kept_digest.clone(), "Imported Graph"),
] {
connection
.execute(
"INSERT INTO strategy_definitions(
definition_id, revision_digest, semantics_digest, name, version,
workflow_json, asset_count, imported_at
) VALUES (?1, ?2, ?3, ?4, '1', ?5, 1, 1)",
params![definition_id, revision, digest('d'), name, workflow_json],
)
.unwrap();
}
let snapshot_json = serde_json::to_string(&RunSnapshot::empty(
"run-retired",
&retired_id_digest,
"semantics",
))
.unwrap();
connection
.execute(
"INSERT INTO strategy_runs(
run_id, revision_digest, semantics_digest, idempotency_key,
request_digest, snapshot_json, created_at, updated_at
) VALUES ('run-retired', ?1, 'semantics', 'key-retired', 'request', ?2, 1, 1)",
params![retired_id_digest, snapshot_json],
)
.unwrap();
drop(connection);
let revisions = root
.join("client-state")
.join("adaptive-flywheel")
.join("strategy-packages")
.join("revisions");
let retired_tree = revisions.join(&retired_id_digest);
fs::create_dir_all(retired_tree.join("content")).unwrap();
fs::write(retired_tree.join("content").join("marker.txt"), "retired").unwrap();
let kept_tree = revisions.join(&kept_digest);
fs::create_dir_all(kept_tree.join("content")).unwrap();
fs::write(kept_tree.join("content").join("marker.txt"), "kept").unwrap();
assert!(
StrategyStore::open(&root).is_err(),
"a legacy schema must pass through the migration path first"
);
StrategyStore::open_for_migration(&root).unwrap();
let store = StrategyStore::open(&root).unwrap();
let listed = store.list_definitions().unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].definition_id, "imported-graph");
assert_eq!(listed[0].name, "Imported Graph");
assert_eq!(listed[0].revision_digest, kept_digest);
assert!(store.definition_by_revision(&retired_id_digest).is_err());
assert!(store.definition_by_revision(&retired_name_digest).is_err());
assert!(!retired_tree.exists());
assert!(kept_tree.join("content").join("marker.txt").exists());
remove_directory_tree(&root);
}
}