use super::*;
use std::num::NonZeroUsize;
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn limits() -> Result<WaitingLimits, Box<dyn std::error::Error>> {
Ok(WaitingLimits {
drafts: NonZeroUsize::new(8).ok_or("test bound is zero")?,
words_bytes: NonZeroUsize::new(1024).ok_or("test bound is zero")?,
identifier_bytes: NonZeroUsize::new(256).ok_or("test bound is zero")?,
})
}
#[test]
fn oldest_draft_is_visible_with_count_and_exact_words() -> TestResult {
let mut record = WaitingRecord::default();
record.hold("newer", "template β", 20, limits()?)?;
record.hold("older", "site α", 10, limits()?)?;
record.hold("older", "site α", 10, limits()?)?;
assert_eq!(
record.snapshot()?,
serde_json::json!({
"state":"waiting", "draft":"older", "words":"site α", "since":10, "count":2,
"last_outcome":null
})
);
assert!(!format!("{record:?}").contains("site α"));
Ok(())
}
#[test]
fn resolving_once_preserves_other_drafts_and_sends_exchange_provenance() -> TestResult {
let mut record = WaitingRecord::default();
record.hold("one", "first words", 1, limits()?)?;
record.hold("two", "second words", 2, limits()?)?;
let outcome = DraftOutcome::refused("one", "TargetRetired", "Lys's exact reason");
let message = record
.resolve(&outcome, "exchange-a", false, limits()?)?
.ok_or("missing inbox")?;
assert_eq!(message["meta"]["from"], "exchange-a");
assert_eq!(message["reason"], "Lys's exact reason");
assert_eq!(record.snapshot()?["draft"], "two");
assert_eq!(
record.resolve(&outcome, "exchange-a", false, limits()?)?,
None
);
assert!(!record.hold("one", "first words", 3, limits()?)?);
assert!(record.hold("one", "changed words", 3, limits()?).is_err());
record.resolve(
&DraftOutcome::executed("two"),
"exchange-a",
true,
limits()?,
)?;
assert_eq!(record.snapshot()?["state"], "gone");
assert_eq!(record.snapshot()?["outcome"], "executed");
Ok(())
}
#[test]
fn unknown_or_conflicting_resolution_is_refused_without_losing_waiting() -> TestResult {
let mut record = WaitingRecord::default();
record.hold("one", "exact words", 1, limits()?)?;
assert!(
record
.resolve(
&DraftOutcome::executed("unknown"),
"exchange",
false,
limits()?
)
.is_err()
);
assert!(record.hold("one", "changed", 1, limits()?).is_err());
assert_eq!(record.snapshot()?["words"], "exact words");
Ok(())
}
use crate::auth_lys::live::drafts::{
AgentDraftStore, DraftExecutor, DraftLimits, DraftPuller, PullSignal,
};
pub fn pull_limits() -> Result<DraftLimits, Box<dyn std::error::Error>> {
Ok(DraftLimits {
pages: NonZeroUsize::new(4).ok_or("test bound is zero")?,
drafts: NonZeroUsize::new(16).ok_or("test bound is zero")?,
approvers: NonZeroUsize::new(2).ok_or("test bound is zero")?,
words_bytes: NonZeroUsize::new(4096).ok_or("test bound is zero")?,
metadata_bytes: NonZeroUsize::new(1024).ok_or("test bound is zero")?,
})
}
use crate::agents::records::{AgentAct, AgentEvent, AgentHarness, AgentSpec};
use crate::auth_lys::live::drafts::admin::tests::verified_holder;
use crate::auth_lys::live::drafts::tests::{Issuer, closed, page};
use crate::auth_lys::live::{PreparedDraftKeeper, PreparedOwner};
use crate::services::lifecycle::{Lifecycle, Placer};
use crate::services::lys_http::{HttpLys, LysHttp};
use liminal::channel::ChannelSupervisor;
use liminal::durability::{CountingStore, DurableStore, HaematiteStore, bridge::block_on};
use lys_pass::drafts::{ApprovedDraft, DraftRequest, execute_once, request_digest};
pub struct DiskFixture {
pub directory: tempfile::TempDir,
pub store: Arc<CountingStore<HaematiteStore>>,
}
impl DiskFixture {
pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
let directory = crate::test_fixture::fixture_copy("agt004-approval-store", |path| {
let database = haematite::Database::create(
haematite::DatabaseConfig {
data_dir: path.into(),
shard_count: 1,
distributed: None,
executor_threads: None,
node_cache_budget: Some(haematite::NodeCacheBudget::Unlimited),
retention: haematite::Retention::reclaim(0),
},
haematite::RuntimeOptions {
long_wait_interval: std::time::Duration::from_secs(5),
reclaim_start: haematite::ReclaimStart::AtOpen,
},
)?;
let store = HaematiteStore::new(Arc::new(haematite::EventStore::new(database)));
block_on(store.flush())??;
drop(store);
Ok(())
})?;
let database = haematite::Database::open(
directory.path(),
haematite::RuntimeOptions {
long_wait_interval: std::time::Duration::from_secs(5),
reclaim_start: haematite::ReclaimStart::AtOpen,
},
)?;
Ok(Self {
directory,
store: Arc::new(CountingStore::new(HaematiteStore::new(Arc::new(
haematite::EventStore::new(database),
)))),
})
}
pub fn owners(
&self,
placer: Arc<dyn Placer>,
) -> Result<Arc<durable::ApprovalAgents>, DraftError> {
let supervisor = ChannelSupervisor::new().map_err(|_| DraftError::Store)?;
durable::ApprovalAgents::open(
self.store.clone(),
journal_limits()?,
durable::AgentBindings {
bounds: crate::agents::config::AgentsConfig::default(),
network: "fixture-network".into(),
supervisor,
device: "fixture-exchange".into(),
placer: Some(placer),
seats: None,
},
)
}
pub fn lifecycle(
&self,
placer: Arc<dyn Placer>,
) -> Result<Arc<Lifecycle>, crate::services::lifecycle::LifecycleError> {
Lifecycle::open(
&[],
self.directory.path().join("sites"),
"fixture-network".into(),
Some(placer),
)
}
}
fn journal_limits() -> Result<durable::JournalLimits, DraftError> {
let positive = |value| NonZeroUsize::new(value).ok_or(DraftError::Bound);
Ok(durable::JournalLimits {
rows: positive(64)?,
operations: positive(64)?,
snapshot_bytes: positive(1024 * 1024)?,
replay_page_rows: positive(2)?,
})
}
#[derive(Debug)]
struct FixturePlace(HttpLys);
impl Placer for FixturePlace {
fn place(
&self,
operation: &str,
child: (&str, &str),
parent: (&str, &str),
) -> Result<(), String> {
let answer = self.0.post("connector-fixture", "apps/liminal/placements", &json!({"operation":operation,"child":{"kind":child.0,"id":child.1},"parent":{"kind":parent.0,"id":parent.1}})).map_err(|_| "fixture_transport_failed")?;
if answer.succeeded() {
Ok(())
} else {
Err(answer.refusal())
}
}
}
pub fn placer(issuer: &Issuer) -> Result<Arc<dyn Placer>, Box<dyn std::error::Error>> {
Ok(Arc::new(FixturePlace(HttpLys::new(
&issuer.api,
tokio::runtime::Handle::current(),
)?)))
}
pub fn agent_draft(
id: &str,
name: &str,
change: AgentEvent,
action: &str,
) -> Result<ApprovedDraft, Box<dyn std::error::Error>> {
let target = lys_pass::Target::new("liminal.agent", name, action)?;
let identity = verified_holder("fixture-owner", "person", &target)?;
let words = serde_json::to_string(&AgentAct::prepare(name.into(), change, &identity, 1000)?)?;
Ok(ApprovedDraft {
id: id.into(),
app: "liminal".into(),
grant: "grant-prepared".into(),
target,
request_digest: request_digest(&words),
approved_by: vec!["fixture-owner".into()],
mode: lys_pass::Mode::ByDraft,
words,
})
}
pub fn create_agent(id: &str) -> Result<ApprovedDraft, Box<dyn std::error::Error>> {
agent_draft(
id,
"builder",
AgentEvent::Create(AgentSpec {
harness: AgentHarness::Cdx,
purpose: "Build the approved service.".into(),
service: None,
}),
"create_agent",
)
}
pub fn keeper(
owners: Arc<durable::ApprovalAgents>,
lifecycle: Arc<Lifecycle>,
holder: &str,
kind: &str,
draft: &ApprovedDraft,
) -> Result<PreparedOwner, Box<dyn std::error::Error>> {
Ok(PreparedOwner::new(
owners,
lifecycle,
verified_holder(holder, kind, &draft.target)?,
pull_limits()?,
limits()?,
))
}
pub fn held(keeper: &PreparedOwner, draft: &ApprovedDraft) -> TestResult {
let request = DraftRequest {
operation: format!("op:{}", draft.id),
grant: draft.grant.clone(),
target: draft.target.clone(),
request_digest: draft.request_digest.clone(),
words: draft.words.clone(),
};
keeper.prepare("fixture-owner", &request, 1000)?;
keeper.held("fixture-owner", &draft.id, &request, 1000)?;
Ok(())
}
pub fn store(owners: &durable::ApprovalAgents) -> Result<AgentDraftStore, DraftError> {
Ok(AgentDraftStore::new(
owners,
AgentDelivery::new(
owners,
BTreeMap::new(),
"fixture-exchange".into(),
limits().map_err(|_| DraftError::Bound)?,
)?,
))
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_agent_approval_is_one_durable_transition_and_replay_is_read_only() -> TestResult {
let issuer = Issuer::new(vec![(200, json!({})); 4])?;
let disk = DiskFixture::new()?;
let placer = placer(&issuer)?;
let owners = disk.owners(placer.clone())?;
let lifecycle = disk.lifecycle(placer)?;
let draft = create_agent("agent-create")?;
held(
&keeper(owners.clone(), lifecycle, "fixture-owner", "person", &draft)?,
&draft,
)?;
let mut executor = DraftExecutor {
store: store(&owners)?,
limits: pull_limits()?,
at_ms: 2000,
};
let before = disk.store.snapshot();
let receipt = execute_once(&draft, &mut executor)?;
let counts = disk.store.snapshot().since(&before);
assert_eq!(
(
counts.apply_routed_batch_calls,
counts.flushes,
counts.append_all_calls
),
(1, 1, 0)
);
let approval = owners
.journal()
.read(|snapshot| snapshot.outcomes[&draft.id].approval.clone())?;
assert_eq!(approval.draft, draft.id);
assert_eq!(approval.digest, draft.request_digest);
assert_eq!(approval.approved_by, ["fixture-owner"]);
assert_eq!(approval.executed_at_ms, 2000);
assert_eq!(
owners
.agents()
.registry()
.record("builder")?
.transitions
.len(),
1
);
assert_eq!(execute_once(&draft, &mut executor)?, receipt);
assert_eq!(
disk.store
.snapshot()
.since(&before)
.apply_routed_batch_calls,
1
);
let reopened = durable::DraftJournal::open(disk.store, journal_limits()?)?;
assert_eq!(
reopened.read(|snapshot| snapshot.outcomes[&draft.id].receipt.clone())?,
receipt
);
issuer.finish()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connector_pulls_native_owner_once_and_records_execution_once() -> TestResult {
let draft = create_agent("agent-pull")?;
let issuer = Issuer::new(vec![
(200, page(&draft, None)),
(200, json!({})),
(200, json!({})),
(200, json!({})),
(200, json!({})),
(200, closed(&draft, "executed")),
(200, page(&draft, None)),
])?;
let disk = DiskFixture::new()?;
let placer = placer(&issuer)?;
let owners = disk.owners(placer.clone())?;
held(
&keeper(
owners.clone(),
disk.lifecycle(placer)?,
"fixture-owner",
"person",
&draft,
)?,
&draft,
)?;
let mut puller = DraftPuller::new(&issuer.api, store(&owners)?, pull_limits()?)?;
let first = puller
.pull("connector-fixture", PullSignal::Startup, 2000)
.await?;
assert_eq!((first.executed, first.acknowledged), (1, 1));
let second = puller
.pull("connector-fixture", PullSignal::Console, 3000)
.await?;
assert_eq!((second.replayed, second.acknowledged), (1, 0));
assert_eq!(
owners
.agents()
.registry()
.record("builder")?
.transitions
.len(),
1
);
assert_eq!(issuer.calls()?, 7);
issuer.finish()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn old_agent_install_migrates_receipts_and_never_reads_legacy_again() -> TestResult {
let issuer = Issuer::new(vec![(200, json!({})); 4])?;
let disk = DiskFixture::new()?;
let draft = create_agent("old-create")?;
let native = crate::agents::registry::AgentRegistry::open(
disk.store.clone(),
crate::agents::config::AgentsConfig::default(),
"fixture-network".into(),
Some(placer(&issuer)?),
None,
)?;
let receipt = native.execute_approved_at(&draft, 1000)?;
drop(native);
let original = disk
.store
.read_from(crate::agents::registry::REGISTRY_STREAM, 0, 1)
.await?;
let journal = durable::DraftJournal::open(disk.store.clone(), journal_limits()?)?;
assert_eq!(
journal.read(|snapshot| snapshot.outcomes[&draft.id].receipt.clone())?,
receipt
);
assert_eq!(journal.read(|snapshot| snapshot.rows.len())?, 1);
assert_eq!(
journal
.read_from(crate::agents::registry::REGISTRY_STREAM, 0, 1)
.await?[0]
.timestamp,
original[0].timestamp
);
drop(journal);
let before = disk.store.snapshot();
let restored = durable::DraftJournal::open(disk.store.clone(), journal_limits()?)?;
assert_eq!(restored.read(|snapshot| snapshot.rows.len())?, 1);
let counts = disk.store.snapshot().since(&before);
assert_eq!(
(
counts.scan_page_calls,
counts.read_from_calls,
counts.apply_routed_batch_calls,
counts.flushes
),
(0, 0, 0, 0)
);
issuer.finish()
}
#[tokio::test]
async fn missing_native_target_keeps_a_named_refusal_and_never_attempts_it_again() -> TestResult {
let issuer = Issuer::new(vec![])?;
let disk = DiskFixture::new()?;
let placement = placer(&issuer)?;
let owners = disk.owners(placement.clone())?;
let draft = agent_draft("target-gone", "missing", AgentEvent::Stop, "stop")?;
held(
&keeper(
owners.clone(),
disk.lifecycle(placement)?,
"fixture-owner",
"person",
&draft,
)?,
&draft,
)?;
let mut executor = DraftExecutor {
store: store(&owners)?,
limits: pull_limits()?,
at_ms: 2000,
};
let before = disk.store.snapshot();
let receipt = execute_once(&draft, &mut executor)?;
assert!(
matches!(&receipt, lys_pass::drafts::Execution::RefusedOnExecution { refusal, reason, .. } if refusal == "unknown_agent" && reason == "unknown_agent: missing")
);
assert_eq!(owners.journal().read(|snapshot| snapshot.rows.len())?, 0);
let counts = disk.store.snapshot().since(&before);
assert_eq!((counts.apply_routed_batch_calls, counts.flushes), (1, 1));
executor.at_ms = 3000;
assert_eq!(execute_once(&draft, &mut executor)?, receipt);
assert_eq!(
disk.store
.snapshot()
.since(&before)
.apply_routed_batch_calls,
1
);
assert_eq!(issuer.calls()?, 0);
issuer.finish()
}
#[path = "waiting_sites_fixture.rs"]
mod sites;
pub use sites::configure_sites;