use super::*;
mod pending_approval;
#[derive(Clone)]
struct FixtureIdentityBroker {
requirement: IdentityBrokerRequirement,
}
#[async_trait::async_trait]
impl ConsumerBoundIdentityBroker for FixtureIdentityBroker {
fn facts(&self) -> IdentityBrokerFacts {
identity_facts(&self.requirement)
}
async fn acquire(
&self,
requirement: &IdentityBrokerRequirement,
) -> Result<OpaqueIdentityHandle, IdentityBrokerError> {
if requirement != &self.requirement {
return Err(IdentityBrokerError::new(
"identity_binding_mismatch",
"fixture broker binding mismatch",
));
}
Ok(OpaqueIdentityHandle::new(
requirement,
crate::secrets::SecretBytes::from(SECRET_CANARY),
Some(DEADLINE_MS),
))
}
}
struct IdentityConsumingExecutor {
requirement: IdentityBrokerRequirement,
model_calls: Arc<AtomicUsize>,
}
#[derive(Clone)]
struct MaterialIdentityBroker {
requirement: IdentityBrokerRequirement,
material: &'static str,
}
struct SequencedIdentityBroker {
facts: IdentityBrokerFacts,
acquisitions: Arc<AtomicUsize>,
}
#[async_trait::async_trait]
impl ConsumerBoundIdentityBroker for SequencedIdentityBroker {
fn facts(&self) -> IdentityBrokerFacts {
self.facts.clone()
}
async fn acquire(
&self,
requirement: &IdentityBrokerRequirement,
) -> Result<OpaqueIdentityHandle, IdentityBrokerError> {
let attempt = self.acquisitions.fetch_add(1, Ordering::SeqCst);
Ok(OpaqueIdentityHandle::new(
requirement,
crate::secrets::SecretBytes::from(SECRET_CANARY),
Some(if attempt == 0 {
NOW_MS - 1
} else {
DEADLINE_MS
}),
))
}
}
#[async_trait::async_trait]
impl ConsumerBoundIdentityBroker for MaterialIdentityBroker {
fn facts(&self) -> IdentityBrokerFacts {
identity_facts(&self.requirement)
}
async fn acquire(
&self,
requirement: &IdentityBrokerRequirement,
) -> Result<OpaqueIdentityHandle, IdentityBrokerError> {
Ok(OpaqueIdentityHandle::new(
requirement,
crate::secrets::SecretBytes::from(self.material),
Some(DEADLINE_MS),
))
}
}
enum ProviderIdentityExecutor {
Bedrock,
Vertex,
}
#[async_trait::async_trait]
impl PreparedRunExecutor for ProviderIdentityExecutor {
type Output = String;
type Error = String;
async fn execute(&self, _authority: &AuthorityUse) -> Result<Self::Output, Self::Error> {
match self {
Self::Bedrock => crate::llm::providers::bedrock::resolve_bedrock_credentials(
"us-east-1",
"https://bedrock.local",
)
.await
.map(|credentials| credentials.access_key_id)
.map_err(|error| error.to_string()),
Self::Vertex => crate::llm::providers::vertex::resolve_vertex_token("", "project-a")
.await
.map_err(|error| error.to_string()),
}
}
}
#[async_trait::async_trait]
impl PreparedRunExecutor for IdentityConsumingExecutor {
type Output = usize;
type Error = String;
async fn execute(&self, _authority: &AuthorityUse) -> Result<Self::Output, Self::Error> {
let material_len = consume_provider_identity(
&self.requirement.binding.provider,
&self.requirement.binding.audience,
self.requirement.binding.tenant.as_deref(),
|material| Ok(material.as_ref().len()),
)
.await
.map_err(|error| error.code)?
.ok_or_else(|| "prepared_identity_context_missing".to_string())?;
self.model_calls.fetch_add(1, Ordering::SeqCst);
Ok(material_len)
}
}
#[tokio::test]
async fn identity_broker_is_readiness_bound_and_handle_is_opaque_and_consumer_exact() {
let identity = identity_requirement();
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), identity_facts(&identity));
let model_calls = Arc::new(AtomicUsize::new(0));
let broker = Arc::new(FixtureIdentityBroker {
requirement: identity.clone(),
});
let mut brokers = IdentityBrokerRegistry::default();
brokers.insert(identity.broker_id.clone(), broker.clone());
let run = PreparedRun::with_clock(
IdentityConsumingExecutor {
requirement: identity.clone(),
model_calls: model_calls.clone(),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
)
.with_identity_brokers(brokers, identity.binding.consumer.clone());
let lease = approved_lease(&run, &run_intent, &host);
assert_eq!(broker.facts(), identity_facts(&identity));
match run.execute(lease).await {
ExecutionOutcome::Completed { output, receipt } => {
assert_eq!(output, SECRET_CANARY.len());
assert!(receipt.used.iter().any(|authority| {
authority.requirement == AuthorityRequirement::IdentityBroker(identity.clone())
}));
let serialized = serde_json::to_string(&receipt).expect("receipt JSON");
assert!(!serialized.contains(SECRET_CANARY));
assert!(serialized.contains("burin-workload"));
}
ExecutionOutcome::ExecutorFailed { error, .. }
| ExecutionOutcome::AuthorityFailed { error, .. } => {
panic!("identity run failed: {error}")
}
}
assert_eq!(model_calls.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn bedrock_and_vertex_resolvers_consume_the_prepared_provider_identity() {
let cases = [
(
"bedrock",
"bedrock.local",
None,
PlatformIdentitySourceKind::SdkProfile,
r#"{"access_key_id":"brokered-access","secret_access_key":"brokered-secret"}"#,
"brokered-access",
ProviderIdentityExecutor::Bedrock,
),
(
"vertex",
"https://www.googleapis.com/auth/cloud-platform",
Some("project-a"),
PlatformIdentitySourceKind::WorkloadIdentity,
"brokered-vertex-token",
"brokered-vertex-token",
ProviderIdentityExecutor::Vertex,
),
];
for (provider, audience, tenant, source, material, expected, executor) in cases {
let identity = IdentityBrokerRequirement {
reference: PlatformIdentityReference::parse(&format!(
"harn-identity://burin.provider_auth/{provider}"
))
.expect("identity reference"),
broker_id: format!("{provider}-broker"),
source,
renewal: IdentityRenewalMode::BrokerManaged,
binding: IdentityBrokerBinding {
provider: provider.to_string(),
audience: audience.to_string(),
tenant: tenant.map(str::to_string),
consumer: SecretConsumerBinding {
kind: SecretConsumerKind::Provider,
id: provider.to_string(),
environment_name: None,
},
},
};
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), identity_facts(&identity));
let broker = Arc::new(MaterialIdentityBroker {
requirement: identity.clone(),
material,
});
let mut brokers = IdentityBrokerRegistry::default();
brokers.insert(identity.broker_id.clone(), broker);
let run = PreparedRun::with_clock(
executor,
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
)
.with_identity_brokers(brokers, identity.binding.consumer.clone());
let lease = approved_lease(&run, &run_intent, &host);
match run.execute(lease).await {
ExecutionOutcome::Completed { output, receipt } => {
assert_eq!(output, expected);
assert_eq!(receipt.used.len(), 1);
assert!(!serde_json::to_string(&receipt)
.expect("receipt JSON")
.contains(material));
}
ExecutionOutcome::ExecutorFailed { error, .. }
| ExecutionOutcome::AuthorityFailed { error, .. } => {
panic!("{provider} prepared identity failed: {error}")
}
}
}
}
#[tokio::test]
async fn broker_managed_identity_renews_once_and_runtime_fact_drift_fails_closed() {
let mut identity = identity_requirement();
identity.source = PlatformIdentitySourceKind::InstanceMetadata;
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), identity_facts(&identity));
let acquisitions = Arc::new(AtomicUsize::new(0));
let mut brokers = IdentityBrokerRegistry::default();
brokers.insert(
identity.broker_id.clone(),
Arc::new(SequencedIdentityBroker {
facts: identity_facts(&identity),
acquisitions: acquisitions.clone(),
}),
);
let run = PreparedRun::with_clock(
IdentityConsumingExecutor {
requirement: identity.clone(),
model_calls: Arc::new(AtomicUsize::new(0)),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
)
.with_identity_brokers(brokers, identity.binding.consumer.clone());
match run.execute(approved_lease(&run, &run_intent, &host)).await {
ExecutionOutcome::Completed { receipt, .. } => {
assert!(receipt.used.iter().any(|authority| authority.requirement
== AuthorityRequirement::IdentityBroker(identity.clone())));
}
ExecutionOutcome::ExecutorFailed { error, .. }
| ExecutionOutcome::AuthorityFailed { error, .. } => {
panic!("broker-managed renewal failed: {error}")
}
}
assert_eq!(acquisitions.load(Ordering::SeqCst), 2);
let mut drifted_facts = identity_facts(&identity);
drifted_facts
.sources
.remove(&PlatformIdentitySourceKind::InstanceMetadata);
let model_calls = Arc::new(AtomicUsize::new(0));
let mut drifted_brokers = IdentityBrokerRegistry::default();
drifted_brokers.insert(
identity.broker_id.clone(),
Arc::new(SequencedIdentityBroker {
facts: drifted_facts,
acquisitions: Arc::new(AtomicUsize::new(0)),
}),
);
let drifted_run = PreparedRun::with_clock(
IdentityConsumingExecutor {
requirement: identity.clone(),
model_calls: model_calls.clone(),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
)
.with_identity_brokers(drifted_brokers, identity.binding.consumer.clone());
match drifted_run
.execute(approved_lease(&drifted_run, &run_intent, &host))
.await
{
ExecutionOutcome::ExecutorFailed { receipt, .. } => {
assert!(receipt.used.is_empty());
assert_eq!(receipt.denied.len(), 1);
}
ExecutionOutcome::Completed { .. } => panic!("metadata drift reached model spend"),
ExecutionOutcome::AuthorityFailed { error, .. } => {
panic!("unexpected authority failure: {error}")
}
}
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn prepared_identity_missing_or_wrong_consumer_fails_before_model_spend() {
for wrong_consumer in [false, true] {
let identity = identity_requirement();
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), identity_facts(&identity));
let model_calls = Arc::new(AtomicUsize::new(0));
let mut brokers = IdentityBrokerRegistry::default();
if wrong_consumer {
brokers.insert(
identity.broker_id.clone(),
Arc::new(FixtureIdentityBroker {
requirement: identity.clone(),
}),
);
}
let consumer = if wrong_consumer {
SecretConsumerBinding {
id: "different-consumer".to_string(),
..identity.binding.consumer.clone()
}
} else {
identity.binding.consumer.clone()
};
let run = PreparedRun::with_clock(
IdentityConsumingExecutor {
requirement: identity.clone(),
model_calls: model_calls.clone(),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
)
.with_identity_brokers(brokers, consumer);
let lease = approved_lease(&run, &run_intent, &host);
match run.execute(lease).await {
ExecutionOutcome::ExecutorFailed { receipt, .. } => {
assert!(receipt.used.is_empty());
assert!(!receipt.denied.is_empty());
}
ExecutionOutcome::Completed { .. } => {
panic!("prepared identity failure must not reach model spend")
}
ExecutionOutcome::AuthorityFailed { error, .. } => {
panic!("provider identity refusal must remain an executor failure: {error}")
}
}
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
}
}
#[tokio::test]
async fn opaque_identity_handle_rejects_a_different_consumer_binding() {
let identity = identity_requirement();
let broker = FixtureIdentityBroker {
requirement: identity.clone(),
};
let handle = broker.acquire(&identity).await.expect("opaque handle");
let mut drifted = identity;
drifted.binding.consumer.id = "different-consumer".to_string();
let error = handle
.consume(&drifted, NOW_MS, |_| ())
.expect_err("a handle cannot cross consumer bindings");
assert_eq!(error.code, "identity_binding_mismatch");
}
#[test]
fn identity_broker_host_properties_are_readiness_facts() {
let identity = identity_requirement();
let cases = [
("noninteractive", "identity_broker_interactivity"),
("gui", "identity_broker_interactivity"),
("sandbox", "identity_broker_sandbox_boundary"),
("opaque", "identity_broker_sandbox_boundary"),
("source", "identity_source_mismatch"),
("renewal", "identity_renewal_mismatch"),
];
for (case, expected_code) in cases {
let mut facts = identity_facts(&identity);
match case {
"noninteractive" => facts.supports_non_interactive = false,
"gui" => facts.may_prompt_gui = true,
"sandbox" => facts.material_outside_sandbox = false,
"opaque" => facts.opaque_process_local_handles = false,
"source" => facts.sources.clear(),
"renewal" => facts.renewal_modes.clear(),
_ => unreachable!(),
}
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), facts);
let run = PreparedRun::with_clock(
FixtureExecutor {
requirements: Vec::new(),
model_calls: Arc::new(AtomicUsize::new(0)),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
);
match run.prepare(run_intent, host) {
PreparationOutcome::Blocked { diagnostics, .. } => assert!(
diagnostics.iter().any(|item| item.code == expected_code),
"{case} must emit {expected_code}: {diagnostics:#?}"
),
other => panic!("{case} must block readiness, got {other:?}"),
}
}
}
#[test]
fn provider_audience_tenant_and_consumer_drift_each_block_before_spend() {
let original = identity_requirement();
for drift in ["provider", "audience", "tenant", "consumer"] {
let mut changed = original.clone();
match drift {
"provider" => changed.binding.provider = "other-provider".to_string(),
"audience" => changed.binding.audience = "https://other.invalid".to_string(),
"tenant" => changed.binding.tenant = Some("tenant-b".to_string()),
"consumer" => changed.binding.consumer.id = "other-consumer".to_string(),
_ => unreachable!(),
}
let mut run_intent = intent();
run_intent.identity_brokers = vec![changed];
let mut host = host_facts();
host.identity_brokers
.insert(original.broker_id.clone(), identity_facts(&original));
let model_calls = Arc::new(AtomicUsize::new(0));
let run = PreparedRun::with_clock(
FixtureExecutor {
requirements: Vec::new(),
model_calls: model_calls.clone(),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
);
match run.prepare(run_intent, host) {
PreparationOutcome::Blocked { diagnostics, .. } => assert!(diagnostics
.iter()
.any(|item| item.code == "identity_consumer_binding")),
other => panic!("{drift} drift must block readiness, got {other:?}"),
}
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
}
}
#[test]
fn platform_identity_reference_rejects_values_and_ambiguous_paths() {
assert!(PlatformIdentityReference::parse(SECRET_CANARY).is_err());
assert!(PlatformIdentityReference::parse("harn-identity://tenant/name/extra").is_err());
assert!(serde_json::from_value::<PlatformIdentityReference>(json!(SECRET_CANARY)).is_err());
}
fn prepared_session_binding() -> PreparedSessionBindingV1 {
PreparedSessionBindingV1 {
session_id: "prepared-session-1".to_string(),
workspace_fingerprint: "blake3:workspace-a".to_string(),
runtime: provenance(),
consumer: SecretConsumerBinding {
kind: SecretConsumerKind::Provider,
id: "prepared-run".to_string(),
environment_name: None,
},
}
}
fn prepared_runtime_attachment() -> PreparedRuntimeAttachment {
PreparedRuntimeAttachment {
session_id: "prepared-session-1".to_string(),
workspace_fingerprint: "blake3:workspace-a".to_string(),
runtime: provenance(),
consumer: SecretConsumerBinding {
kind: SecretConsumerKind::Provider,
id: "prepared-run".to_string(),
environment_name: None,
},
}
}
#[tokio::test(flavor = "current_thread")]
async fn local_turn_reuses_attached_authority_and_identity_across_an_await() {
let mut identity = identity_requirement();
identity.binding.consumer = prepared_session_binding().consumer;
let mut run_intent = intent();
run_intent.identity_brokers = vec![identity.clone()];
let mut host = host_facts();
host.identity_brokers
.insert(identity.broker_id.clone(), identity_facts(&identity));
let mut brokers = IdentityBrokerRegistry::default();
brokers.insert(
identity.broker_id.clone(),
Arc::new(FixtureIdentityBroker {
requirement: identity.clone(),
}),
);
let receipts = Arc::new(MemoryAuthorityReceiptSink::default());
let session = PreparedSession::new(
PreparedRun::with_clock((), receipts.clone(), Arc::new(|| NOW_MS))
.with_identity_brokers(brokers, prepared_session_binding().consumer),
Arc::new(MemoryPreparedSessionLeaseStore::default()),
);
let batch = match session.prepare(prepared_session_binding(), run_intent, host.clone()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("local turn requires the same approval, got {other:?}"),
};
let lease = match session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: batch.request_id,
batch_fingerprint: batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("approved local turn must become ready, got {other:?}"),
};
let active = session
.attach(lease, host, prepared_runtime_attachment())
.expect("attach the approved session");
let reached = std::rc::Rc::new(std::cell::Cell::new(false));
let turn_reached = reached.clone();
let material_len = session
.run_turn_with(&active, async {
tokio::task::yield_now().await;
for requirement in executor_requirements() {
active.authorize(&requirement).unwrap();
}
assert!(active
.authorize(&AuthorityRequirement::Network(network(
"undeclared.example.test"
)))
.is_err());
let len = consume_provider_identity(
&identity.binding.provider,
&identity.binding.audience,
identity.binding.tenant.as_deref(),
|material| Ok(material.as_ref().len()),
)
.await
.expect("identity scope survives the local await")
.expect("the brokered identity was actually consumed");
turn_reached.set(true);
len
})
.await;
assert!(reached.get());
assert_eq!(material_len, SECRET_CANARY.len());
let receipt = match session.finish(active, true).unwrap() {
PreparedSessionUpdate::Terminal { receipt, .. } => receipt,
other => panic!("local turn must retain terminal accounting, got {other:?}"),
};
assert!(receipt.executor_invoked);
assert_eq!(receipt.status, AuthorityReceiptStatus::Completed);
assert!(!receipt.used.is_empty());
assert!(!receipt.denied.is_empty());
assert_eq!(receipts.receipts().last(), Some(&receipt));
assert!(!serde_json::to_string(&receipt)
.unwrap()
.contains(SECRET_CANARY));
}
#[tokio::test]
async fn bridge_approval_grants_only_a_canonical_selected_allow_before_execution() {
for (response, approved) in [
(
serde_json::json!({"outcome":{"outcome":"selected","optionId":"allow"}}),
true,
),
(
serde_json::json!({"outcome":{"outcome":"selected","optionId":"reject"}}),
false,
),
(serde_json::json!({}), false),
(serde_json::json!({"outcome":"approved"}), false),
(
serde_json::json!({"outcome":{"outcome":"cancelled"}}),
false,
),
(
serde_json::json!({
"outcome":{"outcome":"selected","optionId":"allow"},
"_meta":{"harn":{"permissionDecision":{}}}
}),
false,
),
] {
let model_calls = Arc::new(AtomicUsize::new(0));
let receipts = Arc::new(MemoryAuthorityReceiptSink::default());
let session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: executor_requirements(),
model_calls: model_calls.clone(),
},
receipts.clone(),
Arc::new(|| NOW_MS),
),
Arc::new(MemoryPreparedSessionLeaseStore::default()),
);
let batch = match session.prepare(prepared_session_binding(), intent(), host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected grouped approval, got {other:?}"),
};
let pending = Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::<
u64,
tokio::sync::oneshot::Sender<serde_json::Value>,
>::new()));
let responses = pending.clone();
let fingerprint = batch.batch_fingerprint.clone();
let approval_calls = Arc::new(AtomicUsize::new(0));
let observed_calls = approval_calls.clone();
let writer = Arc::new(move |line: &str| {
let request: serde_json::Value = serde_json::from_str(line).unwrap();
assert_eq!(request["method"], "session/request_permission");
assert_eq!(
request["params"]["toolCall"]["rawInput"]["batch_fingerprint"],
fingerprint
);
observed_calls.fetch_add(1, Ordering::SeqCst);
let id = request["id"].as_u64().unwrap();
responses
.try_lock()
.unwrap()
.remove(&id)
.unwrap()
.send(serde_json::json!({"id":id,"result":response.clone()}))
.map_err(|_| "approval receiver closed".to_string())
});
let bridge = crate::bridge::HostBridge::from_parts_with_writer(
pending,
Arc::new(std::sync::atomic::AtomicBool::new(false)),
writer,
1,
);
bridge.set_session_id("prepared-session-1");
let update = session
.pending_approval(&bridge, "prepared-session-1", &batch)
.unwrap()
.wait()
.await
.unwrap();
assert_eq!(approval_calls.load(Ordering::SeqCst), 1);
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
match update {
PreparedSessionUpdate::Ready { lease, .. } if approved => {
let active = session
.attach(*lease, host_facts(), prepared_runtime_attachment())
.unwrap();
assert_eq!(session.run_turn(&active).await.unwrap(), "completed");
assert_eq!(model_calls.load(Ordering::SeqCst), 1);
match session.finish(active, true).unwrap() {
PreparedSessionUpdate::Terminal { receipt, .. } => {
assert_eq!(receipt.status, AuthorityReceiptStatus::Completed);
assert_eq!(receipt.used.len(), executor_requirements().len());
}
other => panic!("expected terminal accounting, got {other:?}"),
}
}
PreparedSessionUpdate::Blocked { receipt, .. } if !approved => {
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
let receipt = receipt.expect("grouped rejection preserves its decision receipt");
assert_eq!(receipt.status, AuthorityReceiptStatus::Blocked);
assert_eq!(receipt.stage, AuthorityReceiptStage::ApprovalDecision);
assert!(receipt
.diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "prepared_session_approval_denied"));
}
other => panic!("approval result disagrees with canonical answer: {other:?}"),
}
assert!(receipts
.receipts()
.iter()
.any(|receipt| receipt.stage == AuthorityReceiptStage::ApprovalDecision));
}
}
#[tokio::test]
async fn prepared_session_persists_one_approval_reuses_the_envelope_and_rejects_replay() {
let model_calls = Arc::new(AtomicUsize::new(0));
let receipts = Arc::new(MemoryAuthorityReceiptSink::default());
let claims = Arc::new(MemoryPreparedSessionLeaseStore::default());
let host_session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: executor_requirements(),
model_calls: model_calls.clone(),
},
receipts.clone(),
Arc::new(|| NOW_MS),
),
claims.clone(),
);
let mut earlier_intent = intent();
earlier_intent.budget.turns = Some(7);
let earlier_batch =
match host_session.prepare(prepared_session_binding(), earlier_intent, host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected earlier grouped approval, got {other:?}"),
};
let batch = match host_session.prepare(prepared_session_binding(), intent(), host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("interactive session must present one grouped batch, got {other:?}"),
};
assert_ne!(earlier_batch.batch_fingerprint, batch.batch_fingerprint);
match host_session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: earlier_batch.request_id,
batch_fingerprint: earlier_batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Blocked { diagnostics, .. } => assert!(diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "prepared_session_approval_binding")),
other => panic!("stale decision must refuse without consuming its successor: {other:?}"),
}
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
let decision = PreparedSessionApprovalDecision {
request_id: batch.request_id,
batch_fingerprint: batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
};
let lease = match host_session.decide("prepared-session-1", decision) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("approved session must become ready, got {other:?}"),
};
assert_eq!(lease.schema, PREPARED_SESSION_SCHEMA);
let schema: serde_json::Value =
serde_json::from_str(PREPARED_SESSION_V1_SCHEMA_JSON).expect("prepared-session schema");
let validator = jsonschema::validator_for(&schema).expect("compile prepared-session schema");
let lease_value = serde_json::to_value(&lease).expect("serialize prepared-session lease");
let errors = validator
.iter_errors(&lease_value)
.map(|error| error.to_string())
.collect::<Vec<_>>();
assert!(
errors.is_empty(),
"prepared-session lease schema violations: {errors:#?}"
);
assert!(receipts
.receipts()
.iter()
.any(|receipt| receipt.stage == AuthorityReceiptStage::ApprovalDecision));
let server_session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: executor_requirements(),
model_calls: model_calls.clone(),
},
receipts.clone(),
Arc::new(|| NOW_MS),
),
claims,
);
let active = server_session
.attach(lease.clone(), host_facts(), prepared_runtime_attachment())
.expect("external server accepts the exact prepared lease");
assert_eq!(server_session.run_turn(&active).await.unwrap(), "completed");
assert_eq!(server_session.run_turn(&active).await.unwrap(), "completed");
assert_eq!(model_calls.load(Ordering::SeqCst), 2);
let replay = server_session.attach(lease, host_facts(), prepared_runtime_attachment());
match replay {
Err(PreparedSessionUpdate::Blocked { diagnostics, .. }) => assert!(diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "prepared_session_replay")),
_ => panic!("a claimed prepared-session lease must reject replay"),
}
match server_session
.finish(active, true)
.expect("terminal receipt")
{
PreparedSessionUpdate::Terminal { receipt, .. } => {
assert_eq!(receipt.status, AuthorityReceiptStatus::Completed);
assert_eq!(receipt.used.len(), executor_requirements().len());
assert!(receipt.unused.len() < receipt.granted.len());
}
other => panic!("finished session must emit terminal accounting, got {other:?}"),
}
}
#[tokio::test]
async fn accepted_stop_and_pivot_persist_stopped_authority_and_retire_the_lease() {
for pivot in [false, true] {
let directory = tempfile::tempdir().expect("receipt directory");
let receipt_path = directory.path().join("authority.ndjson");
let model_calls = Arc::new(AtomicUsize::new(0));
let mut requested = intent();
requested.receipt_uri = receipt_path.to_string_lossy().into_owned();
let session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: executor_requirements(),
model_calls: model_calls.clone(),
},
Arc::new(NdjsonAuthorityReceiptSink::new(&receipt_path)),
Arc::new(|| NOW_MS),
),
Arc::new(MemoryPreparedSessionLeaseStore::default()),
);
let batch = match session.prepare(prepared_session_binding(), requested, host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected approval batch, got {other:?}"),
};
let lease = match session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: batch.request_id,
batch_fingerprint: batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("expected ready lease, got {other:?}"),
};
let active = session
.attach(lease.clone(), host_facts(), prepared_runtime_attachment())
.expect("attach approved lease");
assert_eq!(session.run_turn(&active).await.unwrap(), "completed");
let stopped = session
.stop(active, pivot)
.expect("persist accepted control");
let receipt = match stopped {
PreparedSessionUpdate::Stopped { receipt, .. } if !pivot => receipt,
PreparedSessionUpdate::Pivoted { receipt, .. } if pivot => receipt,
other => panic!("accepted control must preserve its kind, got {other:?}"),
};
assert_eq!(receipt.stage, AuthorityReceiptStage::Stopped);
assert_eq!(receipt.status, AuthorityReceiptStatus::Stopped);
assert_eq!(receipt.used.len(), executor_requirements().len());
assert_eq!(
receipt.used.len() + receipt.unused.len(),
receipt.granted.len()
);
assert!(receipt.executor_invoked);
let persisted = std::fs::read_to_string(&receipt_path).expect("read durable receipts");
assert!(!persisted.contains(SECRET_CANARY));
let terminal: RunAuthorityReceipt =
serde_json::from_str(persisted.lines().last().expect("terminal row"))
.expect("decode terminal receipt");
assert_eq!(terminal, receipt);
let wire = serde_json::to_value(&terminal).expect("receipt wire shape");
assert_eq!(wire["stage"], "stopped");
assert_eq!(wire["status"], "stopped");
assert!(session
.attach(lease, host_facts(), prepared_runtime_attachment())
.is_err());
assert_eq!(model_calls.load(Ordering::SeqCst), 1);
}
}
#[tokio::test]
async fn prepared_session_rejects_stale_runtime_and_cross_workspace_attach_before_turns() {
let model_calls = Arc::new(AtomicUsize::new(0));
let session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: executor_requirements(),
model_calls: model_calls.clone(),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
),
Arc::new(MemoryPreparedSessionLeaseStore::default()),
);
let mut approved_host = host_facts();
let batch = match session.prepare(prepared_session_binding(), intent(), approved_host.clone()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected approval batch, got {other:?}"),
};
approved_host
.approved_batches
.insert(batch.batch_fingerprint.clone(), AuthorityDecider::Person);
let lease = match session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: batch.request_id,
batch_fingerprint: batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("expected ready lease, got {other:?}"),
};
for drift in ["workspace", "runtime", "consumer"] {
let mut attachment = prepared_runtime_attachment();
if drift == "workspace" {
attachment.workspace_fingerprint = "blake3:workspace-b".to_string();
} else if drift == "runtime" {
attachment.runtime.runtime_digest = "blake3:stale-runtime".to_string();
} else {
attachment.consumer.id = "wrong-provider".to_string();
}
match session.attach(lease.clone(), approved_host.clone(), attachment) {
Err(PreparedSessionUpdate::Blocked { diagnostics, .. }) => assert!(diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "prepared_session_attachment_binding")),
_ => panic!("{drift} drift must block before engine startup"),
}
}
assert_eq!(model_calls.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn prepared_session_widening_is_one_semantic_delta_batch() {
let session = PreparedSession::new(
PreparedRun::with_clock(
FixtureExecutor {
requirements: Vec::new(),
model_calls: Arc::new(AtomicUsize::new(0)),
},
Arc::new(MemoryAuthorityReceiptSink::default()),
Arc::new(|| NOW_MS),
),
Arc::new(MemoryPreparedSessionLeaseStore::default()),
);
let mut earlier_intent = intent();
earlier_intent.budget.turns = Some(7);
let earlier_session_batch =
match session.prepare(prepared_session_binding(), earlier_intent, host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected earlier session approval, got {other:?}"),
};
let earlier_lease = match session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: earlier_session_batch.request_id,
batch_fingerprint: earlier_session_batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("expected earlier ready session, got {other:?}"),
};
let earlier_active = session
.attach(earlier_lease, host_facts(), prepared_runtime_attachment())
.expect("attach earlier same-id session");
let batch = match session.prepare(prepared_session_binding(), intent(), host_facts()) {
PreparedSessionUpdate::NeedsApproval { batch, .. } => batch,
other => panic!("expected initial approval batch, got {other:?}"),
};
let lease = match session.decide(
"prepared-session-1",
PreparedSessionApprovalDecision {
request_id: batch.request_id,
batch_fingerprint: batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Ready { lease, .. } => *lease,
other => panic!("expected ready lease, got {other:?}"),
};
let active = session
.attach(lease, host_facts(), prepared_runtime_attachment())
.expect("attach prepared session");
let widened = AuthorityRequirement::Tool {
pattern: "deploy".to_string(),
};
let earlier_widening = AuthorityRequirement::Tool {
pattern: "publish".to_string(),
};
let earlier_batch = match session.request_delta(&active, earlier_widening.clone()) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::NeedsApproval { batch },
..
} => batch,
other => panic!("expected earlier delta approval, got {other:?}"),
};
let delta_batch = match session.request_delta(&active, widened.clone()) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::NeedsApproval { batch },
..
} => batch,
other => panic!("widening must yield one approval batch, got {other:?}"),
};
assert_eq!(delta_batch.groups.len(), 1);
assert_eq!(delta_batch.groups[0].semantic_group, "host_capabilities");
assert_ne!(
earlier_active.lease().plan_fingerprint,
active.lease().plan_fingerprint,
);
match session.decide_delta(
&earlier_active,
PreparedSessionApprovalDecision {
request_id: delta_batch.request_id,
batch_fingerprint: delta_batch.batch_fingerprint.clone(),
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::Blocked { diagnostic },
..
} => assert_eq!(diagnostic.code, "prepared_session_delta_binding"),
other => panic!("stale active lease must not consume or grant a newer delta: {other:?}"),
}
assert!(earlier_active.authorize(&widened).is_err());
assert_ne!(
earlier_batch.batch_fingerprint,
delta_batch.batch_fingerprint
);
match session.decide_delta(
&active,
PreparedSessionApprovalDecision {
request_id: earlier_batch.request_id,
batch_fingerprint: earlier_batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::Blocked { diagnostic },
..
} => assert_eq!(diagnostic.code, "prepared_session_delta_binding"),
other => panic!("stale delta must not consume its successor: {other:?}"),
}
assert!(active.authorize(&earlier_widening).is_err());
let repeated = AuthorityRequirement::Tool {
pattern: "repeat-deploy".to_string(),
};
for approved in [false, true] {
let mut batches = Vec::new();
for _ in 0..2 {
match session.request_delta(&active, repeated.clone()) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::NeedsApproval { batch },
..
} => batches.push(batch),
other => panic!("expected identical pending delta: {other:?}"),
}
}
assert_eq!(batches[0].batch_fingerprint, batches[1].batch_fingerprint);
assert_ne!(batches[0].request_id, batches[1].request_id);
match session.decide_delta(
&active,
PreparedSessionApprovalDecision {
request_id: batches[0].request_id,
batch_fingerprint: batches[0].batch_fingerprint.clone(),
approved,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::Blocked { diagnostic },
..
} => assert_eq!(diagnostic.code, "prepared_session_delta_binding"),
other => panic!("stale answer must not consume replacement delta: {other:?}"),
}
assert!(active.authorize(&repeated).is_err());
match session.decide_delta(
&active,
PreparedSessionApprovalDecision {
request_id: batches[1].request_id,
batch_fingerprint: batches[1].batch_fingerprint.clone(),
approved: false,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::Blocked { diagnostic },
..
} => assert_eq!(diagnostic.code, "prepared_session_delta_denied"),
other => panic!("current delta denial must consume its own request: {other:?}"),
}
}
assert!(active.authorize(&widened).is_err());
let delta_batch = match session.request_delta(&active, widened.clone()) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::NeedsApproval { batch },
..
} => batch,
other => panic!("expected fresh widening approval: {other:?}"),
};
match session.decide_delta(
&active,
PreparedSessionApprovalDecision {
request_id: delta_batch.request_id,
batch_fingerprint: delta_batch.batch_fingerprint,
approved: true,
decider: AuthorityDecider::Person,
},
) {
PreparedSessionUpdate::Delta {
outcome: PreparedSessionDelta::Granted { .. },
..
} => {}
other => panic!("approved delta must join the active envelope, got {other:?}"),
}
active
.authorize(&widened)
.expect("approved widening is live without re-preparation");
assert!(active.authorize(&earlier_widening).is_err());
assert!(earlier_active.authorize(&widened).is_err());
}