use super::root_actor_runtime::{conflict, RootActorWriteProof};
use super::*;
use crate::session_inbox::{FileSessionInbox, InboxAuthority, OwnedFilesystem};
use bamboo_domain::{
ScopedLoopControlBinding, ScopedLoopControlCheckpoint, ScopedLoopControlClaim,
SessionActivationPort,
};
#[derive(Clone)]
pub(super) struct ScopedLoopWriteProof {
binding: ScopedLoopControlBinding,
activation: Arc<dyn SessionActivationPort>,
native_census: Option<super::scoped_transport_expiry::ScopedTransportCensus>,
}
impl ScopedLoopWriteProof {
pub(super) fn validate_retained(
&self,
guard: &Arc<dyn bamboo_domain::SessionRunDeliveryGuard>,
) -> io::Result<()> {
if self
.native_census
.as_ref()
.is_none_or(|census| census.check_current_live(&self.binding).is_ok())
&& guard.matches(
&self.binding.target_session_id,
self.binding.target_created_at,
&self.binding.owner_run_id,
)
{
Ok(())
} else {
Err(conflict("scoped Loop owner stopped before publication"))
}
}
pub(super) fn retain(&self) -> io::Result<Arc<dyn bamboo_domain::SessionRunDeliveryGuard>> {
let guard = self
.activation
.retain_supervisor_loop_blocking(&self.binding)
.map_err(conflict)?;
let guard = match self.native_census.as_ref() {
Some(census) => census.wrap_live_guard(&self.binding, guard)?,
None => guard,
};
if !guard.matches(
&self.binding.target_session_id,
self.binding.target_created_at,
&self.binding.owner_run_id,
) {
return Err(conflict("scoped Loop retained owner identity mismatch"));
}
Ok(guard)
}
}
#[derive(Clone)]
pub(super) struct ScopedLoopInputProof {
inbox: FileSessionInbox,
directory: PathBuf,
claim: ScopedLoopControlClaim,
root: Option<RootActorWriteProof>,
terminal_consumption: Option<bamboo_domain::ScopedLoopControlConsumption>,
conversation_generation: bamboo_domain::ConversationGenerationObservation,
}
impl ScopedLoopInputProof {
pub(super) fn validate(&self, allow_terminal: bool) -> io::Result<()> {
let allow_terminal = allow_terminal || self.terminal_consumption.is_some();
match &self.claim {
ScopedLoopControlClaim::Owned(claim) => {
if let Some(root) = &self.root {
root.validate_deadline(claim.lease.expires_at)?;
}
self.inbox
.locked_owned_claim(
&self.directory,
&claim.claim.envelope.target_session_id,
claim,
Utc::now(),
allow_terminal,
)
.map(|_| ())
.map_err(conflict)
}
ScopedLoopControlClaim::Legacy(claim) => self
.inbox
.locked_scoped_legacy_claim(&self.directory, claim, allow_terminal)
.map_err(conflict),
}
}
fn validate_consumption(&self, session: &Session) -> io::Result<()> {
self.conversation_generation
.check_canonical(session)
.map_err(conflict)?;
if !bamboo_domain::has_scoped_loop_consumption(session, self.claim.envelope())
|| !session.session_inbox_admission().is_some_and(|cursor| {
cursor.contains(&self.claim.envelope().id)
&& cursor.last_admitted_sequence >= self.claim.generation()
})
{
return Err(conflict(
"scoped Loop checkpoint lacks typed consumption and cursor",
));
}
if let Some(prior) = &self.terminal_consumption {
if !bamboo_domain::scoped_loop_consumptions(session)
.map_err(conflict)?
.contains(prior)
{
return Err(conflict(
"immutable scoped Loop consumption changed on ACK retry",
));
}
}
Ok(())
}
}
fn read_committed(directory: &Path) -> io::Result<Session> {
let main_bytes =
actor_transcript::regular_bytes(&directory.join("session.json")).map_err(conflict)?;
compact_main::validate_full_main(&main_bytes)?;
let main: Session = actor_transcript::decode(&main_bytes).map_err(conflict)?;
let side: Session = actor_transcript::decode(
&actor_transcript::regular_bytes(&directory.join(RUNTIME_SIDECAR_FILE))
.map_err(conflict)?,
)
.map_err(conflict)?;
supervisor::validate_overlay(&main, Some(&side))?;
main.check_conversation_generation(&side)
.map_err(conflict)?;
if main.id != side.id
|| main.created_at != side.created_at
|| main.kind != SessionKind::Root
|| main.parent_session_id.is_some()
|| main.spawn_depth != 0
|| !main.authority_identity.is_ordinary()
{
return Err(conflict(
"scoped Loop canonical checkpoint identity changed",
));
}
Ok(overlay_runtime_sidecar(main, Some(side)))
}
#[cfg(test)]
mod generation_tests {
use super::super::scoped_transport_expiry::generation_tests::{
binding, next_conversation, source,
};
use super::*;
#[test]
fn scoped_loop_cannot_overlay_runtime_from_another_conversation() {
let home = tempfile::tempdir().unwrap();
let main = Session::new("scoped-loop-conversation", "model");
let mut side = runtime_sidecar_snapshot(&main);
side.conversation_reset = Some(bamboo_domain::ConversationResetState {
schema_version: 1,
conversation_revision: 1,
receipts: vec![bamboo_domain::ConversationResetReceipt {
operation_id: format!("0:{}", Uuid::new_v4()),
birth_token: main.root_mode_birth_token(),
expected_snapshot_sha256: "a".repeat(64),
previous_revision: 0,
conversation_revision: 1,
retained_system_messages: 0,
reset_at: main.updated_at,
}],
});
std::fs::write(
home.path().join("session.json"),
compact_main::serialize_main(&main).unwrap(),
)
.unwrap();
std::fs::write(
home.path().join(RUNTIME_SIDECAR_FILE),
serde_json::to_vec(&side).unwrap(),
)
.unwrap();
assert!(read_committed(home.path()).is_err());
}
async fn consumed_input(
nonzero: bool,
) -> (
tempfile::TempDir,
Arc<SessionStoreV2>,
ScopedLoopInputProof,
Session,
) {
let home = tempfile::tempdir().unwrap();
let store = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
let zero = source();
let mut session = if nonzero {
next_conversation(&zero)
} else {
zero
};
let selector = binding(&session);
let envelope = bamboo_domain::ScopedLoopControlInput {
binding: selector.clone(),
command: bamboo_domain::LoopControlRequest::Start {
operation_id: Uuid::new_v4().to_string(),
expected_generation: None,
expected_revision: 0,
objective: "synthetic bounded operation".into(),
max_iterations: 2,
max_total_rounds: 4,
},
}
.envelope()
.unwrap();
let input = bamboo_domain::scoped_loop_control_input(&envelope)
.unwrap()
.unwrap();
let consumption = bamboo_domain::ScopedLoopControlConsumption {
input_id: envelope.id.clone(),
input_sha256: Sha256::digest(bamboo_domain::canonical_json_bytes(
&envelope.idempotency_semantics(),
))
.iter()
.map(|byte| format!("{byte:02x}"))
.collect(),
binding: selector,
operation_id: input.operation_id().to_owned(),
configured: false,
receipt: None,
rejection: Some("synthetic rejection".into()),
};
session.metadata.insert(
bamboo_domain::SCOPED_LOOP_CONSUMPTION_KEY.into(),
serde_json::to_string(&vec![consumption.clone()]).unwrap(),
);
session
.session_inbox_admission_mut()
.record(envelope.id.clone(), 3);
let proof = ScopedLoopInputProof {
inbox: FileSessionInbox::new(
store.clone(),
bamboo_domain::SessionInboxLimits::default(),
),
directory: home.path().to_owned(),
claim: ScopedLoopControlClaim::Legacy(bamboo_domain::SessionInboxClaim {
envelope,
generation: 3,
activation_policy: bamboo_domain::SessionActivationPolicy::InterruptSpecificWait,
claim_id: "synthetic-generation-claim".into(),
}),
root: None,
terminal_consumption: Some(consumption),
conversation_generation: session.conversation_generation_observation(),
};
(home, store, proof, session)
}
#[tokio::test]
async fn identical_consumption_and_cursor_do_not_refresh_original_conversation() {
for same_revision in [false, true] {
let (home, store, proof, original) = consumed_input(same_revision).await;
proof.validate_consumption(&original).unwrap();
let mut current = if same_revision {
original.clone()
} else {
next_conversation(&original)
};
if same_revision {
current.conversation_reset.as_mut().unwrap().receipts[0].expected_snapshot_sha256 =
"b".repeat(64);
}
assert!(bamboo_domain::has_scoped_loop_consumption(
¤t,
proof.claim.envelope()
));
assert_eq!(
current.session_inbox_admission(),
original.session_inbox_admission()
);
let error = proof.validate_consumption(¤t).unwrap_err();
assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
assert!(error
.get_ref()
.is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>()));
drop(proof);
store.flush_search_index().await;
drop(store);
home.close().unwrap();
}
}
#[tokio::test]
async fn current_conversation_still_requires_exact_consumption_and_cursor() {
let (home, store, proof, original) = consumed_input(true).await;
let mut appended = original.clone();
appended.add_message(Message::user("same conversation append"));
proof.validate_consumption(&appended).unwrap();
let mut no_cursor = original.clone();
no_cursor
.runtime_metadata
.as_mut()
.unwrap()
.session_inbox_admission = None;
assert!(proof.validate_consumption(&no_cursor).is_err());
let mut no_consumption = original;
no_consumption
.metadata
.remove(bamboo_domain::SCOPED_LOOP_CONSUMPTION_KEY);
assert!(proof.validate_consumption(&no_consumption).is_err());
drop(proof);
store.flush_search_index().await;
drop(store);
home.close().unwrap();
}
}
impl SessionStoreV2 {
pub(crate) fn scoped_loop_execution_stopped(
session: &Session,
binding: &ScopedLoopControlBinding,
) -> io::Result<bool> {
let Some(raw) = session.metadata.get("runtime.counted_loop") else {
return Ok(false);
};
if raw.len() > 128 * 1024 {
return Err(conflict("scoped Loop state exceeds byte bound"));
}
let state: serde_json::Value = serde_json::from_str(raw).map_err(conflict)?;
let owner = state["owner_execution_id"]
.as_str()
.ok_or_else(|| conflict("scoped Loop execution identity is malformed"))?;
if owner != binding.execution_id.as_str() {
return Ok(false);
}
let Some(control) = state.get("control").filter(|control| !control.is_null()) else {
return Ok(false);
};
control["stop_requested"]
.as_bool()
.ok_or_else(|| conflict("scoped Loop stop proof is malformed"))
}
pub(super) async fn load_scoped_loop_session_impl(
&self,
binding: &ScopedLoopControlBinding,
activation: Arc<dyn SessionActivationPort>,
) -> io::Result<Option<Session>> {
let authority = self.lock_scoped_loop_authority(binding).await?;
let session = self
.load_execution_admission_unchecked(&binding.target_session_id)
.await?;
let proof = ScopedLoopWriteProof {
binding: binding.clone(),
activation,
native_census: authority.scoped_native(),
};
tokio::task::spawn_blocking(move || {
let _authority = authority;
let _runtime = proof.retain()?;
Ok(session)
})
.await
.map_err(io::Error::other)?
}
pub(super) async fn checkpoint_scoped_loop_impl(
&self,
request: ScopedLoopControlCheckpoint<'_>,
) -> io::Result<()> {
let original_generation = request.session.conversation_generation_observation();
if request
.session
.metadata
.contains_key(bamboo_domain::SCOPED_LOOP_UNCERTAIN_KEY)
{
return Err(conflict("scoped Loop checkpoint outcome is uncertain"));
}
let binding = request
.bindings
.first()
.ok_or_else(|| conflict("scoped Loop binding set is empty"))?;
if request.bindings.iter().any(|other| other != binding) {
return Err(conflict("scoped Loop binding set is not one current grant"));
}
if bamboo_domain::scoped_loop_objective_origins(request.session)
.map_err(conflict)?
.iter()
.any(|origin| !request.bindings.contains(origin))
{
return Err(conflict("scoped Loop objective origin is not retained"));
}
let authority = self.lock_scoped_loop_authority(binding).await?;
let root = request.owner.map(|owner| {
RootActorWriteProof::new(
self.sessions_dir.join(&binding.target_session_id),
owner.clone(),
)
});
if request.owner.is_some_and(|owner| {
owner.fence.actor_id != binding.target_session_id
|| owner.fence.run_id != binding.owner_run_id
|| owner.created_at != binding.target_created_at
}) {
return Err(conflict("scoped Loop Root capability identity changed"));
}
let proof = ScopedLoopWriteProof {
binding: binding.clone(),
activation: request.activation,
native_census: authority.scoped_native(),
};
let mut guards = DefaultWriterGuards::scoped_loop(authority, root.clone(), proof.clone());
let input_fs: Option<(
FileSessionInbox,
PathBuf,
OwnedFilesystem,
ScopedLoopControlClaim,
)> = if let Some((inbox, claim)) = request.input {
let input = bamboo_domain::scoped_loop_control_input(claim.envelope())
.map_err(conflict)?
.ok_or_else(|| conflict("scoped Loop input required"))?;
if input.binding != *binding {
return Err(conflict("scoped Loop input binding changed"));
}
let inbox = inbox
.as_any()
.and_then(|any| any.downcast_ref::<FileSessionInbox>())
.ok_or_else(|| conflict("scoped Loop requires the canonical concrete Inbox"))?
.bind_scoped_checkpoint(request.owner, &self.sessions_dir)?;
let (directory, filesystem) = inbox
.filesystem_with_authority(
&binding.target_session_id,
InboxAuthority::Root {
_guard: guards.physical.clone(),
},
)
.await
.map_err(conflict)?;
let canonical = self
.load_execution_admission_unchecked(&binding.target_session_id)
.await?
.ok_or_else(|| conflict("Loop input target disappeared"))?;
let terminal_consumption =
if bamboo_domain::has_scoped_loop_consumption(&canonical, claim.envelope())
&& canonical.session_inbox_admission().is_some_and(|cursor| {
cursor.contains(&claim.envelope().id)
&& cursor.last_admitted_sequence >= claim.generation()
})
{
bamboo_domain::scoped_loop_consumptions(&canonical)
.map_err(conflict)?
.into_iter()
.find(|record| record.input_id == claim.envelope().id)
} else {
None
};
let input = ScopedLoopInputProof {
terminal_consumption,
inbox: inbox.clone(),
directory: directory.clone(),
claim: claim.clone(),
root: root.clone(),
conversation_generation: original_generation,
};
let validate = input.clone();
let target_directory = self.sessions_dir.join(&binding.target_session_id);
let root_check = root.clone();
let retain = proof.clone();
let filesystem = filesystem
.with_runtime_retainer(Arc::new(move || retain.retain()), binding.clone())
.with_check(Arc::new(move || {
validate.validate(true)?;
let durable = match &root_check {
Some(root) => root.validate()?,
None => read_committed(&target_directory)?,
};
validate.validate_consumption(&durable)
}));
let fresh = Arc::get_mut(&mut guards).expect("new scoped writer has no borrowers");
fresh.scoped_input = Some(input);
fresh._input_filesystem = Some(filesystem.clone());
Some((inbox, directory, filesystem, claim))
} else {
None
};
let target = binding.target_session_id.clone();
let result = async {
let canonical = self
.load_execution_admission_unchecked(&target)
.await?
.ok_or_else(|| conflict("scoped Loop target disappeared"))?;
if request.session.id != canonical.id
|| request.session.created_at != canonical.created_at
|| request.session.metadata_version != canonical.metadata_version
|| request.session.authority_identity != canonical.authority_identity
|| request.session.kind != SessionKind::Root
|| request.session.parent_session_id.is_some()
|| request.session.spawn_depth != 0
|| request.session.project_id_meta() != canonical.project_id_meta()
|| request.session.root_tool_authority_revision
!= canonical.root_tool_authority_revision
|| request.session.root_orchestration_only != canonical.root_orchestration_only
|| request
.session
.agent_runtime_state
.as_ref()
.and_then(|runtime| runtime.execution_id.as_deref())
!= Some(binding.execution_id.as_str())
{
return Err(conflict(
"scoped Loop candidate is not the current canonical lifetime/execution",
));
}
bamboo_domain::append_missing_runtime_messages(request.session, &canonical);
bamboo_domain::merge_session_inbox_admission(request.session, &canonical);
if let Some(input) = &guards.scoped_input {
input.validate(false)?;
input.validate_consumption(request.session)?;
}
self.save_session_after_lock(request.session, Instant::now(), &guards, None)
.await?;
let committed = self
.load_execution_admission_unchecked(&target)
.await?
.ok_or_else(|| conflict("scoped Loop committed target disappeared"))?;
let committed_for_publish = committed.clone();
let root_check = root.clone();
let input_check = guards.scoped_input.clone();
let publish = request.publish;
Self::default_writer_job(&guards, move || {
if let Some(root) = root_check {
root.validate()?;
}
if let Some(input) = input_check {
input.validate(false)?;
input.validate_consumption(&committed_for_publish)?;
}
publish(&committed_for_publish);
Ok(())
})
.await?;
*request.session = committed;
if let Some((inbox, directory, filesystem, claim)) = input_fs {
let input = bamboo_domain::scoped_loop_control_input(claim.envelope())
.map_err(conflict)?
.ok_or_else(|| conflict("scoped Loop input disappeared before ACK"))?;
let configured = bamboo_domain::scoped_loop_consumptions(request.session)
.map_err(conflict)?
.iter()
.any(|record| record.input_id == claim.envelope().id && record.configured);
if matches!(
input.command,
bamboo_domain::LoopControlRequest::Stop { .. }
) && configured
&& Self::scoped_loop_execution_stopped(request.session, binding)?
{
inbox
.retire_scoped_loop_tail_with_filesystem(
&directory,
binding,
&claim,
request.session,
&filesystem,
)
.await
.map_err(conflict)?;
}
match claim {
ScopedLoopControlClaim::Owned(claim) => inbox
.ack_owned_with_filesystem(
&directory,
&target,
&claim,
Utc::now(),
filesystem,
)
.await
.map_err(conflict)?,
ScopedLoopControlClaim::Legacy(claim) => inbox
.ack_scoped_legacy_with_filesystem(&directory, &target, &claim, filesystem)
.await
.map_err(conflict)?,
}
}
Ok(())
}
.await;
if result.is_err() {
match self.load_execution_admission_unchecked(&target).await {
Ok(Some(canonical)) => *request.session = canonical,
_ => {
request.session.metadata.insert(
bamboo_domain::SCOPED_LOOP_UNCERTAIN_KEY.into(),
"true".into(),
);
}
}
}
result
}
}
impl SessionStoreV2 {
pub(super) async fn retire_scoped_loop_transport_impl(
&self,
request: bamboo_domain::storage::ScopedLoopTransportRetirement<'_>,
) -> io::Result<()> {
let binding = request.binding;
validate_session_id(&binding.target_session_id)?;
let lifecycle = self.lock_default_writer_lifecycle().await?;
let task = self.lock_runtime_task_sidecar_shared().await?;
let session = self
.acquire_session_write_lock(&binding.target_session_id, SaveKind::Runtime)
.await?;
let directory = self.sessions_dir.join(&binding.target_session_id);
let root = request
.owner
.map(|owner| RootActorWriteProof::new(directory.clone(), owner.clone()));
if request.owner.is_some_and(|owner| {
owner.fence.actor_id != binding.target_session_id
|| owner.created_at != binding.target_created_at
|| owner.fence.run_id != binding.owner_run_id
}) {
return Err(conflict("target retirement Root identity changed"));
}
let guards =
DefaultWriterGuards::shared_with_root_actor(lifecycle, task, session, root.clone());
let census = super::scoped_transport_expiry::ScopedTransportCensus::read(&directory)?
.ok_or_else(|| conflict("target retirement canonical census missing"))?;
let committed = census.session.clone();
let native_inert = census.native_inert(binding)?;
if committed.id != binding.target_session_id
|| committed.created_at != binding.target_created_at
{
return Err(conflict("target retirement canonical lifetime changed"));
}
if !native_inert {
if committed
.agent_runtime_state
.as_ref()
.and_then(|state| state.execution_id.as_deref())
!= Some(binding.execution_id.as_str())
{
return Err(conflict("target retirement canonical execution changed"));
}
if let Some(root) = &root {
root.validate()?;
} else {
self.check_default_actor_clear(&directory).await?;
}
}
let inbox = request
.inbox
.as_any()
.and_then(|value| value.downcast_ref::<FileSessionInbox>())
.ok_or_else(|| conflict("target retirement requires its concrete Inbox"))?;
let inbox = inbox.bind_scoped_checkpoint(request.owner, &self.sessions_dir)?;
let (dir, filesystem) = inbox
.filesystem_with_authority(
&binding.target_session_id,
InboxAuthority::Root {
_guard: guards.physical.clone(),
},
)
.await
.map_err(conflict)?;
let filesystem = filesystem.with_runtime_retainer(request.retain, binding.clone());
let filesystem = if native_inert {
let binding = binding.clone();
filesystem.with_check(Arc::new(move || {
let current = census.require_same_snapshot()?;
if current.native_inert(&binding)? {
Ok(())
} else {
Err(conflict("target transport expiry no longer proved"))
}
}))
} else if let Some(root) = root {
filesystem.with_check(Arc::new(move || root.validate().map(|_| ())))
} else {
filesystem
};
inbox
.retire_closed_loop_with_filesystem(&dir, binding, &committed, &filesystem)
.await
.map_err(conflict)
}
}