#[path = "session_merge_child_actor.rs"]
mod child_actor;
use std::sync::Arc;
use bamboo_domain::session::types::Session;
use bamboo_domain::storage::Storage;
use bamboo_domain::{
latest_response_occurrence, PermissionAuditSeed, PermissionAuditSnapshot, ResponseOccurrence,
RetrievalWindowCheckpointOutcome, RuntimeSessionPersistence, CONSUMED_CLARIFICATION_IDS_KEY,
CONSUMED_RESPONSE_OCCURRENCES_KEY,
};
use dashmap::DashMap;
use tokio::sync::{Mutex, OwnedMutexGuard};
const AUTHORITATIVE_METADATA_KEYS: &[&str] = &[
"gold_config",
"workflow.run_ids.v1",
"ticket.runtime.dispatch.v1",
"ticket.runtime.owned_stop.v1",
];
const ROOT_PROJECT_CONTEXT_KEYS: &[&str] = &[
"workspace_source",
"workspace_binding_status",
"project_context_rendered",
"project_resources_rendered",
"runtime_prompt_snapshot",
];
const RESPONSE_CONTROL_METADATA_KEYS: &[&str] = &[
CONSUMED_CLARIFICATION_IDS_KEY,
CONSUMED_RESPONSE_OCCURRENCES_KEY,
"runtime.suspend_reason",
"clarification_resume_pending",
"conclusion_with_options_resume_pending",
"execute.startup_handoff_at",
"permission.reexecute_tool_call_id",
"permission.reexecute_request_generation",
"retry_resume_pending",
"retry_resume_reason",
"provider_name",
"runtime.child_completion_source_v1",
];
const TASK_CONTROL_PLANE_CONFLICT_PREFIX: &str = "Task control-plane changed while saving session ";
const MAX_TASK_CONTROL_PLANE_REBASE_RETRIES: usize = 3;
const LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY: &str =
"context_management.last_manual_archive_occurrence.v1";
const MANUAL_ARCHIVE_REJECTIONS_KEY: &str = "context_management.manual_archive_rejections.v1";
const MAX_MANUAL_ARCHIVE_REJECTIONS: usize = 64;
const RESPONSES_PREVIOUS_RESPONSE_ID_KEY: &str = "responses.previous_response_id";
const ACTOR_PARENT_QUESTION_HANDOFF_KEY: &str = "runtime.actor_parent_question_handoff";
pub(crate) fn adopt_durable_actor_parent_question_handoff(
session: &mut Session,
durable: &Session,
) -> std::io::Result<()> {
if session
.metadata
.get(ACTOR_PARENT_QUESTION_HANDOFF_KEY)
.map(String::as_str)
!= Some("true")
{
return Ok(());
}
let request = durable
.metadata
.get(bamboo_domain::PARENT_QUESTION_REQUEST_KEY)
.and_then(|value| serde_json::from_str::<bamboo_domain::ParentQuestion>(value).ok());
let pending_question = request.as_ref().is_some_and(|question| {
durable.pending_question.as_ref().is_some_and(|pending| {
pending.source == bamboo_domain::PendingQuestionSource::DirectParent
&& pending.tool_call_id == question.tool_call_id
}) && durable
.metadata
.get("runtime.suspend_reason")
.is_some_and(|reason| reason == "awaiting_clarification")
});
let resolved_question = request.as_ref().is_some_and(|question| {
durable
.metadata
.get(bamboo_domain::PARENT_QUESTION_RESOLUTION_KEY)
.and_then(|value| {
serde_json::from_str::<bamboo_domain::ParentQuestionResolution>(value).ok()
})
.is_some_and(|resolution| resolution.request == *question)
});
if session.kind != bamboo_domain::SessionKind::Child
|| session.id != durable.id
|| session.created_at != durable.created_at
|| session.parent_session_id != durable.parent_session_id
|| session.root_session_id != durable.root_session_id
|| session.spawn_depth != durable.spawn_depth
|| session.project_id_meta() != durable.project_id_meta()
|| !request.as_ref().is_some_and(|question| {
question.child.session_id == durable.id
&& question.child.created_at == durable.created_at
&& question.root_session_id == durable.root_session_id
&& question.project_id == durable.project_id_meta()
})
|| !(pending_question || resolved_question)
{
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"canonical direct-parent question handoff is missing or changed",
));
}
session.messages.clone_from(&durable.messages);
session
.provider_transcript
.clone_from(&durable.provider_transcript);
session
.pending_question
.clone_from(&durable.pending_question);
for key in RESPONSE_CONTROL_METADATA_KEYS.iter().copied().chain([
bamboo_domain::PARENT_QUESTION_REQUEST_KEY,
bamboo_domain::PARENT_QUESTION_RESOLUTION_KEY,
]) {
match durable.metadata.get(key) {
Some(value) => {
session.metadata.insert(key.to_string(), value.clone());
}
None => {
session.metadata.remove(key);
}
}
}
session.model.clone_from(&durable.model);
session.model_ref.clone_from(&durable.model_ref);
session.reasoning_effort = durable.reasoning_effort;
session
.agent_runtime_state
.clone_from(&durable.agent_runtime_state);
session.updated_at = session.updated_at.max(durable.updated_at);
session.metadata.remove(ACTOR_PARENT_QUESTION_HANDOFF_KEY);
Ok(())
}
fn ensure_session_execution_compatible(session: &Session) -> std::io::Result<()> {
session.check_execution_compatibility().map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
bamboo_domain::SessionAuthorityConflict(error.to_string()),
)
})
}
fn ensure_scoped_loop_checkpoint_known(session: &Session) -> std::io::Result<()> {
if session
.metadata
.contains_key(bamboo_domain::SCOPED_LOOP_UNCERTAIN_KEY)
{
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"scoped Loop checkpoint outcome is uncertain; strict canonical reload required",
));
}
Ok(())
}
fn reconcile_runtime_checkpoint(
session: &mut Session,
latest: &Session,
) -> std::io::Result<(usize, bool)> {
ensure_model_context_checkpoint_is_current(session, latest)?;
let appended = bamboo_domain::append_missing_runtime_messages(session, latest);
bamboo_domain::merge_session_inbox_admission(session, latest);
let adopted_response = adopt_durable_consumed_clarification(session, latest);
apply_authoritative_metadata(session, latest);
adopt_fresher_disk_permission_posture(session, latest);
let _ = adopt_durable_tagged_child_wait(session, latest);
Ok((appended, adopted_response))
}
pub(crate) const LIVE_BOOTSTRAP_SESSION_JSON_LIMIT: usize = 4 * 1024 * 1024;
struct CancelLiveBootstrapOnDrop(Option<bamboo_domain::AdmissionGate>);
impl Drop for CancelLiveBootstrapOnDrop {
fn drop(&mut self) {
if let Some(gate) = &self.0 {
gate.cancel_if_pending();
}
}
}
#[derive(Default)]
struct LiveBootstrapSizeCounter(usize);
impl LiveBootstrapSizeCounter {
fn add(&mut self, bytes: usize) -> std::io::Result<()> {
self.0 = self
.0
.checked_add(bytes)
.filter(|total| *total <= LIVE_BOOTSTRAP_SESSION_JSON_LIMIT)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
"live bootstrap Session exceeds recovery proof limit",
)
})?;
Ok(())
}
}
impl std::io::Write for LiveBootstrapSizeCounter {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.add(bytes.len())?;
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn live_bootstrap_skipped_size_counter(
session: &Session,
) -> std::io::Result<LiveBootstrapSizeCounter> {
let mut counter = LiveBootstrapSizeCounter::default();
if let Some(value) = &session.force_manual_compression {
counter.add(value.len())?;
}
if let Some((model, _)) = &session.resolved_token_budget {
counter.add(model.len())?;
}
if let Some(runtime) = &session.agent_runtime_state {
for id in runtime
.children
.active_ids
.iter()
.chain(&runtime.children.completed_ids)
{
counter.add(id.len())?;
counter.add(std::mem::size_of::<String>())?;
}
}
Ok(counter)
}
fn ensure_live_bootstrap_session_size(session: &Session) -> std::io::Result<()> {
let mut compact = live_bootstrap_skipped_size_counter(session)?;
serde_json::to_writer(&mut compact, session)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let mut pretty = live_bootstrap_skipped_size_counter(session)?;
serde_json::to_writer_pretty(&mut pretty, session)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))
}
pub(crate) fn bounded_live_bootstrap_value(
session: &Session,
) -> std::io::Result<serde_json::Value> {
ensure_live_bootstrap_session_size(session)?;
serde_json::to_value(session).map_err(std::io::Error::other)
}
pub(crate) fn validate_live_bootstrap_candidate(
prepared: &Session,
latest: &Session,
) -> std::io::Result<()> {
ensure_scoped_loop_checkpoint_known(prepared)?;
ensure_session_execution_compatible(prepared)?;
let expected = bounded_live_bootstrap_value(prepared)?;
let durable = bounded_live_bootstrap_value(latest)?;
let expected_messages = expected
.get("messages")
.and_then(serde_json::Value::as_array)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
"missing prepared transcript",
)
})?;
let durable_messages = durable
.get("messages")
.and_then(serde_json::Value::as_array)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
"missing canonical transcript",
)
})?;
let invalid_prefix = expected_messages.len() < durable_messages.len()
|| expected_messages[..durable_messages.len()] != durable_messages[..];
let invalid_suffix = prepared
.messages
.iter()
.skip(durable_messages.len())
.any(|message| {
message.role != bamboo_domain::Role::Tool
|| message
.tool_call_id
.as_deref()
.is_none_or(|id| id.trim().is_empty())
});
if invalid_prefix || invalid_suffix || prepared.updated_at < latest.updated_at {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
bamboo_domain::SessionAuthorityConflict(
"live bootstrap transcript is not an exact tool-result extension".into(),
),
));
}
let mut normalized = expected.clone();
let object = normalized.as_object_mut().ok_or_else(|| {
std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid prepared Session")
})?;
object.insert(
"messages".into(),
serde_json::Value::Array(durable_messages.clone()),
);
object.insert(
"updated_at".into(),
durable.get("updated_at").cloned().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidData,
"missing canonical update time",
)
})?,
);
if normalized != durable {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
bamboo_domain::SessionAuthorityConflict(
"live bootstrap canonical Session changed".into(),
),
));
}
let mut reconciled = prepared.clone();
reconcile_runtime_checkpoint(&mut reconciled, latest)?;
let actual = bounded_live_bootstrap_value(&reconciled)?;
if actual != expected {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
bamboo_domain::SessionAuthorityConflict(
"live bootstrap prepared Session changed during reconciliation".into(),
),
));
}
Ok(())
}
fn may_publish_runtime_result(result: &std::io::Result<()>) -> bool {
!result.as_ref().err().is_some_and(|error| {
error.get_ref().is_some_and(|cause| {
cause.is::<bamboo_domain::SessionAuthorityConflict>()
|| cause.is::<crate::v2::DirectParentTerminalConflict>()
|| cause.is::<crate::v2::DirectParentTerminalProofError>()
})
})
}
fn is_direct_parent_terminal_conflict(error: &std::io::Error) -> bool {
error.kind() == std::io::ErrorKind::WouldBlock
&& error
.get_ref()
.is_some_and(|cause| cause.is::<crate::v2::DirectParentTerminalConflict>())
}
fn adopt_durable_tagged_child_wait(
session: &mut Session,
latest: &Session,
) -> Option<chrono::DateTime<chrono::Utc>> {
if session.id != latest.id || session.created_at != latest.created_at {
return None;
}
let incoming = session.agent_runtime_state.as_mut()?;
let wait = incoming.waiting_for_children.as_ref()?;
wait.registered_by_tool_call_id.as_ref()?;
let registered_at = wait.registered_at;
incoming.waiting_for_children = latest
.agent_runtime_state
.as_ref()
.and_then(|state| state.waiting_for_children.clone());
if incoming.waiting_for_children.is_some() {
return None;
}
if session
.metadata
.get("runtime.suspend_reason")
.map(String::as_str)
== Some("waiting_for_children")
{
session.metadata.remove("runtime.suspend_reason");
}
if incoming
.suspension
.as_ref()
.is_some_and(|s| s.reason == "waiting_for_children")
{
incoming.suspension = None;
if incoming.status == bamboo_domain::AgentStatusState::Suspended {
incoming.status = bamboo_domain::AgentStatusState::Idle;
}
}
Some(registered_at)
}
pub(crate) fn adopt_finalized_child_wait(
session: &mut Session,
latest: &Session,
inherited: bool,
) -> Option<chrono::DateTime<chrono::Utc>> {
let incoming = session.agent_runtime_state.as_mut()?;
if !inherited
&& (session
.metadata
.get("runtime.suspend_reason")
.map(String::as_str)
!= Some("waiting_for_children")
|| incoming
.waiting_for_children
.as_ref()
.is_none_or(|wait| wait.registered_by_tool_call_id.is_none()))
{
return None;
}
let durable = latest.agent_runtime_state.as_ref();
if !inherited && durable.is_none() {
return None;
}
let registered_at = incoming
.waiting_for_children
.as_ref()
.map(|wait| wait.registered_at);
incoming.waiting_for_children = durable.and_then(|state| state.waiting_for_children.clone());
let cleared = incoming.waiting_for_children.is_none();
let cleared_without_other_suspension =
cleared && !latest.metadata.contains_key("runtime.suspend_reason");
if cleared {
if let Some(durable) = durable {
incoming.status = durable.status;
incoming.suspension = durable.suspension.clone();
}
match latest.metadata.get("runtime.suspend_reason") {
Some(reason) => {
session
.metadata
.insert("runtime.suspend_reason".to_string(), reason.clone());
}
None => {
session.metadata.remove("runtime.suspend_reason");
}
}
}
if session.metadata.contains_key("agent.runtime.state") {
match serde_json::to_string(incoming) {
Ok(serialized) => {
session
.metadata
.insert("agent.runtime.state".to_string(), serialized);
}
Err(_) => {
session.metadata.remove("agent.runtime.state");
}
}
}
if cleared_without_other_suspension && session.last_run_status().as_deref() == Some("suspended")
{
session.set_last_run_status("completed");
}
if cleared {
registered_at
} else {
None
}
}
pub(crate) fn reconcile_inherited_child_wait(
session: &mut Session,
latest: &Session,
inherited: &bamboo_domain::InheritedChildWait,
) -> std::io::Result<()> {
inherited.validate_session(session)?;
inherited.validate_session(latest)?;
ensure_conversation_generation_is_current(session, latest)?;
let Some(incoming) = session.agent_runtime_state.as_mut() else {
return Ok(());
};
if incoming.waiting_for_children.as_ref() != Some(inherited.wait()) {
return Ok(());
}
incoming.waiting_for_children = latest
.agent_runtime_state
.as_ref()
.and_then(|state| state.waiting_for_children.clone());
let cleared = incoming.waiting_for_children.is_none();
if cleared {
if session
.metadata
.get("runtime.suspend_reason")
.map(String::as_str)
== Some("waiting_for_children")
{
session.metadata.remove("runtime.suspend_reason");
}
if incoming
.suspension
.as_ref()
.is_some_and(|s| s.reason == "waiting_for_children")
{
incoming.suspension = None;
if incoming.status == bamboo_domain::AgentStatusState::Suspended {
incoming.status = bamboo_domain::AgentStatusState::Idle;
}
}
}
if session.metadata.contains_key("agent.runtime.state") {
session.metadata.insert(
"agent.runtime.state".into(),
serde_json::to_string(incoming).map_err(std::io::Error::other)?,
);
}
if cleared {
preserve_finalized_hidden_child_resumes(session, latest, inherited.wait().registered_at);
}
Ok(())
}
fn preserve_finalized_hidden_child_resumes(
session: &mut Session,
latest: &Session,
registered_at: chrono::DateTime<chrono::Utc>,
) {
let existing: std::collections::HashSet<_> = session
.messages
.iter()
.map(|message| message.id.as_str())
.collect();
let missing = latest
.messages
.iter()
.filter(|message| {
!existing.contains(message.id.as_str())
&& message.created_at >= registered_at
&& message
.metadata
.as_ref()
.and_then(|metadata| metadata.get("runtime_kind"))
.and_then(|value| value.as_str())
.is_some_and(|kind| {
matches!(kind, "child_completion_resume" | "guardian_review_resume")
})
})
.cloned()
.collect::<Vec<_>>();
session.messages.extend(missing);
}
fn adopt_durable_consumed_clarification(session: &mut Session, durable: &Session) -> bool {
let Some(incoming_tool_call_id) = session
.pending_question
.as_ref()
.map(|pending| pending.tool_call_id.clone())
else {
return false;
};
let occurrence_ledger = durable
.metadata
.get(CONSUMED_RESPONSE_OCCURRENCES_KEY)
.map(|value| serde_json::from_str::<Vec<ResponseOccurrence>>(value));
let was_consumed = match occurrence_ledger {
Some(Ok(consumed)) => latest_response_occurrence(session, &incoming_tool_call_id)
.is_some_and(|incoming| consumed.iter().any(|entry| entry == &incoming)),
Some(Err(_)) => false,
None => {
let legacy_consumed = durable
.metadata
.get(CONSUMED_CLARIFICATION_IDS_KEY)
.and_then(|value| serde_json::from_str::<Vec<String>>(value).ok())
.unwrap_or_default()
.iter()
.any(|tool_call_id| tool_call_id == &incoming_tool_call_id);
legacy_consumed
&& latest_response_occurrence(durable, &incoming_tool_call_id).is_some_and(
|durable_occurrence| {
latest_response_occurrence(session, &incoming_tool_call_id)
.is_some_and(|incoming| incoming == durable_occurrence)
},
)
}
};
if !was_consumed {
return false;
}
bamboo_domain::append_missing_runtime_messages(session, durable);
session
.pending_question
.clone_from(&durable.pending_question);
for key in RESPONSE_CONTROL_METADATA_KEYS {
if let Some(value) = durable.metadata.get(*key) {
session.metadata.insert((*key).to_string(), value.clone());
} else {
session.metadata.remove(*key);
}
}
session.model.clone_from(&durable.model);
session.model_ref.clone_from(&durable.model_ref);
session.reasoning_effort = durable.reasoning_effort;
session
.agent_runtime_state
.clone_from(&durable.agent_runtime_state);
if let Some(runtime_metadata) = session.runtime_metadata.as_mut() {
runtime_metadata.provider_name = durable
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.provider_name.clone());
} else if durable
.runtime_metadata
.as_ref()
.is_some_and(|metadata| metadata.provider_name.is_some())
{
session.runtime_metadata = durable.runtime_metadata.as_ref().map(|metadata| {
let mut response_metadata = bamboo_domain::SessionRuntimeMetadata::default();
response_metadata
.provider_name
.clone_from(&metadata.provider_name);
response_metadata
});
}
true
}
fn is_task_control_plane_save_conflict(error: &std::io::Error) -> bool {
error.kind() == std::io::ErrorKind::WouldBlock
&& error
.to_string()
.starts_with(TASK_CONTROL_PLANE_CONFLICT_PREFIX)
}
fn adopt_durable_task_control_plane(session: &mut Session, durable: &Session) {
session.task_list = durable.task_list.clone();
session
.metadata
.remove(bamboo_domain::session::runtime_metadata::keys::TASK_LIST_VERSION);
if let Some(runtime_metadata) = session.runtime_metadata.as_mut() {
runtime_metadata.task_list_version = None;
}
if session
.runtime_metadata
.as_ref()
.is_some_and(bamboo_domain::session::SessionRuntimeMetadata::is_empty)
{
session.runtime_metadata = None;
}
if let Some(version) = durable.task_list_version_meta() {
session.set_task_list_version_meta(version);
}
}
fn task_list_snapshot_matches(
session: &Session,
expected_task_list: &bamboo_domain::TaskList,
) -> std::io::Result<bool> {
Ok(serde_json::to_value(&session.task_list)
.map_err(|error| std::io::Error::other(error.to_string()))?
== serde_json::to_value(Some(expected_task_list))
.map_err(|error| std::io::Error::other(error.to_string()))?)
}
fn unconditional_task_patch_would_regress(
durable: &Session,
incoming_task_list: &bamboo_domain::TaskList,
incoming_version: &str,
) -> std::io::Result<bool> {
let same_list = task_list_snapshot_matches(durable, incoming_task_list)?;
let Some(durable_version) = durable.task_list_version_meta() else {
return Ok(false);
};
match (
incoming_version.parse::<u64>(),
durable_version.parse::<u64>(),
) {
(Ok(incoming), Ok(durable)) => {
Ok(incoming < durable || (incoming == durable && !same_list))
}
_ => Ok(incoming_version != durable_version || !same_list),
}
}
pub struct LockedSessionStore {
storage: Arc<dyn Storage>,
inherited_child_wait: Option<bamboo_domain::InheritedChildWait>,
child_actor_writer: Option<(
bamboo_domain::ChildActorRuntimeWrite,
Arc<dyn bamboo_domain::ActorDirectoryPort>,
bamboo_domain::RootActorRuntimePublisher,
)>,
root_actor_writer: Option<(
bamboo_domain::RootActorRuntimeWrite,
bamboo_domain::RootActorRuntimePublisher,
)>,
locks: Arc<DashMap<String, Arc<Mutex<()>>>>,
task_pair_transaction_lock: Arc<Mutex<()>>,
}
pub struct SessionLockGuard {
guard: Option<OwnedMutexGuard<()>>,
locks: Arc<DashMap<String, Arc<Mutex<()>>>>,
session_id: String,
}
impl Drop for SessionLockGuard {
fn drop(&mut self) {
self.guard.take();
self.locks
.remove_if(&self.session_id, |_, arc| Arc::strong_count(arc) == 1);
}
}
impl LockedSessionStore {
pub fn new(storage: Arc<dyn Storage>) -> Self {
Self {
storage,
inherited_child_wait: None,
root_actor_writer: None,
child_actor_writer: None,
locks: Arc::new(DashMap::new()),
task_pair_transaction_lock: Arc::new(Mutex::new(())),
}
}
pub fn bind_root_actor_writer(
&self,
owner: bamboo_domain::RootActorRuntimeWrite,
publish: bamboo_domain::RootActorRuntimePublisher,
) -> std::io::Result<Self> {
if !self.storage.supports_root_actor_runtime_write() {
return Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"storage does not support a Root Actor writer",
));
}
Ok(Self {
storage: self.storage.clone(),
inherited_child_wait: self.inherited_child_wait.clone(),
root_actor_writer: Some((owner, publish)),
child_actor_writer: None,
locks: self.locks.clone(),
task_pair_transaction_lock: self.task_pair_transaction_lock.clone(),
})
}
pub fn bind_inherited_child_wait(&self, inherited: bamboo_domain::InheritedChildWait) -> Self {
Self {
storage: self.storage.clone(),
inherited_child_wait: Some(inherited),
root_actor_writer: self.root_actor_writer.clone(),
child_actor_writer: self.child_actor_writer.clone(),
locks: self.locks.clone(),
task_pair_transaction_lock: self.task_pair_transaction_lock.clone(),
}
}
fn inherited_for_session(
&self,
session: &Session,
) -> Option<&bamboo_domain::InheritedChildWait> {
self.inherited_child_wait
.as_ref()
.filter(|wait| wait.session_id() == session.id)
}
async fn save_runtime_snapshot(
&self,
session: &mut Session,
runtime_only: bool,
) -> std::io::Result<()> {
ensure_scoped_loop_checkpoint_known(session)?;
if self.child_actor_writer.is_some() {
return self
.save_child_snapshot(session, runtime_only, None, None)
.await;
}
if let Some(inherited) = self.inherited_for_session(session) {
self.storage
.save_runtime_with_inherited_child_wait(
session,
inherited,
runtime_only,
self.root_actor_writer.clone(),
None,
)
.await
} else if let Some((owner, publish)) = self.root_actor_writer.as_ref() {
self.storage
.save_root_actor_runtime(owner, session, runtime_only, publish.clone())
.await
} else if runtime_only {
self.storage.save_runtime_state(session).await
} else {
self.storage.save_session(session).await
}
}
pub fn storage(&self) -> &Arc<dyn Storage> {
&self.storage
}
pub async fn acquire_lock(&self, session_id: &str) -> SessionLockGuard {
let lock = self
.locks
.entry(session_id.to_string())
.or_insert_with(|| Arc::new(Mutex::new(())))
.clone();
let mut guard = SessionLockGuard {
guard: None,
locks: self.locks.clone(),
session_id: session_id.to_string(),
};
guard.guard = Some(lock.lock_owned().await);
guard
}
pub fn validate_lock(&self, owner: &SessionLockGuard, session_id: &str) -> std::io::Result<()> {
if owner.guard.is_none()
|| owner.session_id != session_id
|| !Arc::ptr_eq(&owner.locks, &self.locks)
{
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"Session persistence owner does not match this coordinator and session",
));
}
Ok(())
}
async fn save_session_rebasing_task_conflicts(
&self,
session: &mut Session,
) -> std::io::Result<()> {
self.save_session_rebasing_task_conflicts_with_input(session, None, None, None)
.await
}
async fn save_session_rebasing_task_conflicts_with_input(
&self,
session: &mut Session,
input: Option<&(
Arc<dyn bamboo_domain::SessionInboxPort>,
bamboo_domain::SessionInboxOwnedClaim,
)>,
inherited: Option<&bamboo_domain::session::runtime_state::WaitingForChildrenState>,
startup_expected: Option<&Session>,
) -> std::io::Result<()> {
for attempt in 0..=MAX_TASK_CONTROL_PLANE_REBASE_RETRIES {
if let Some(provenance) = self.inherited_for_session(session) {
provenance.validate_session(session)?;
}
let result = if let Some(expected) = startup_expected {
if self.child_actor_writer.is_some() || input.is_some() || inherited.is_some() {
return Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
bamboo_domain::SessionAuthorityConflict(
"bound startup requires its Root execution writer".into(),
),
));
}
let (owner, publish) = self.root_actor_writer.as_ref().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
bamboo_domain::SessionAuthorityConflict(
"bound startup has no bound Root execution writer".into(),
),
)
})?;
let carried_wait = self.inherited_for_session(session).cloned();
self.storage
.save_root_actor_bound_startup(
owner,
session,
expected,
carried_wait.as_ref(),
publish.clone(),
)
.await
} else if self.child_actor_writer.is_some() {
self.save_child_snapshot(session, false, input.cloned(), inherited)
.await
} else if inherited.is_none() && self.inherited_for_session(session).is_some() {
self.storage
.save_runtime_with_inherited_child_wait(
session,
self.inherited_for_session(session)
.expect("bound provenance"),
false,
self.root_actor_writer.clone(),
input.cloned(),
)
.await
} else if let Some((inbox, claim)) = input {
let (owner, publish) = self.root_actor_writer.as_ref().ok_or_else(|| {
std::io::Error::other("Root input checkpoint requires a bound writer")
})?;
self.storage
.save_root_actor_input(owner, session, inbox.clone(), claim, publish.clone())
.await
} else if let Some(inherited) = inherited {
self.storage
.save_inherited_child_wait_finalized(
session,
inherited,
self.root_actor_writer.clone(),
)
.await
} else {
self.save_runtime_snapshot(session, false).await
};
match result {
Ok(()) => return Ok(()),
Err(error)
if is_task_control_plane_save_conflict(&error)
&& attempt < MAX_TASK_CONTROL_PLANE_REBASE_RETRIES =>
{
let Some(durable) =
self.storage.load_runtime_control_plane(&session.id).await?
else {
return Err(error);
};
ensure_conversation_generation_is_current(session, &durable)?;
adopt_durable_task_control_plane(session, &durable);
}
Err(error) => {
if is_task_control_plane_save_conflict(&error) {
if let Some(durable) =
self.storage.load_runtime_control_plane(&session.id).await?
{
ensure_conversation_generation_is_current(session, &durable)?;
adopt_durable_task_control_plane(session, &durable);
}
}
return Err(error);
}
}
}
unreachable!("bounded Task conflict retry loop always returns")
}
async fn save_runtime_state_rebasing_task_conflicts(
&self,
session: &mut Session,
) -> std::io::Result<()> {
for attempt in 0..=MAX_TASK_CONTROL_PLANE_REBASE_RETRIES {
match self.save_runtime_snapshot(session, true).await {
Ok(()) => return Ok(()),
Err(error)
if is_task_control_plane_save_conflict(&error)
&& attempt < MAX_TASK_CONTROL_PLANE_REBASE_RETRIES =>
{
let Some(durable) =
self.storage.load_runtime_control_plane(&session.id).await?
else {
return Err(error);
};
ensure_conversation_generation_is_current(session, &durable)?;
adopt_durable_task_control_plane(session, &durable);
}
Err(error) => {
if is_task_control_plane_save_conflict(&error) {
if let Some(durable) =
self.storage.load_runtime_control_plane(&session.id).await?
{
ensure_conversation_generation_is_current(session, &durable)?;
adopt_durable_task_control_plane(session, &durable);
}
}
return Err(error);
}
}
}
unreachable!("bounded Task conflict retry loop always returns")
}
pub async fn save_runtime_only(&self, session: &mut Session) -> std::io::Result<()> {
self.save_runtime_only_and_publish(session, |_| {}).await
}
pub async fn ensure_child_auto_run_launch_intent(
&self,
expected: &Session,
) -> std::io::Result<Option<Session>> {
let _guard = self.acquire_lock(&expected.id).await;
let Some(mut latest) = self
.storage
.load_runtime_control_plane(&expected.id)
.await?
else {
return Ok(None);
};
let generation = expected.child_launch_generation();
if latest.kind != bamboo_domain::SessionKind::Child
|| latest.created_at != expected.created_at
|| latest.parent_session_id != expected.parent_session_id
|| latest.root_session_id != expected.root_session_id
|| latest.child_launch_generation() != generation
|| latest.is_child_launch_cancelled(generation)
{
return Ok(None);
}
if latest.last_run_status().as_deref() != Some("pending") {
let already_promised = latest
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_auto_run_launch_intent_generation)
== Some(generation);
let already_started = matches!(
latest.last_run_status().as_deref(),
Some("running" | "completed" | "error" | "timeout" | "skipped" | "suspended")
);
return Ok((already_promised && already_started).then_some(latest));
}
if latest.recoverable_child_launch_generation() != Some(generation) {
latest.mark_child_auto_run_launch_intent();
self.save_runtime_state_rebasing_task_conflicts(&mut latest)
.await?;
}
Ok(Some(latest))
}
pub async fn save_runtime_only_and_publish<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session) + Send,
{
self.save_runtime_only_and_publish_inner(session, publish, true)
.await
}
pub async fn save_runtime_only_and_publish_on_success<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session) + Send,
{
self.save_runtime_only_and_publish_inner(session, publish, false)
.await
}
async fn save_runtime_only_and_publish_inner<F>(
&self,
session: &mut Session,
publish: F,
publish_on_failure: bool,
) -> std::io::Result<()>
where
F: FnOnce(&Session) + Send,
{
ensure_session_execution_compatible(session)?;
ensure_scoped_loop_checkpoint_known(session)?;
let _guard = self.acquire_lock(&session.id).await;
let original = session;
let mut candidate = original.clone();
let session = &mut candidate;
if let Some(latest) = self.storage.load_runtime_control_plane(&session.id).await? {
ensure_session_execution_compatible(&latest)?;
ensure_conversation_generation_is_current(session, &latest)?;
apply_authoritative_metadata(session, &latest);
adopt_fresher_disk_permission_posture(session, &latest);
adopt_durable_model_context_state(session, &latest);
}
let result = self
.save_runtime_state_rebasing_task_conflicts(session)
.await;
if may_publish_runtime_result(&result) {
*original = candidate;
if result.is_ok() || publish_on_failure {
publish(original);
}
}
result
}
pub async fn register_child_wait_and_publish(
&self,
expected: &Session,
batch: &[(String, Option<String>)],
policy: bamboo_domain::ChildWaitPolicy,
check_terminal: bool,
publish: bamboo_domain::RootActorRuntimePublisher,
) -> std::io::Result<(Session, usize)> {
let _guard = self.acquire_lock(&expected.id).await;
if self.storage.supports_atomic_child_wait_control_plane() {
return self
.storage
.register_child_wait_control_plane(expected, batch, policy, check_terminal, publish)
.await;
}
use crate::v2::child_wait_control_plane::{
register_pending, terminal, validate_incarnation,
};
let mut latest = self
.storage
.load_session(&expected.id)
.await?
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::NotFound,
"child wait parent disappeared",
)
})?;
validate_incarnation(expected, &latest)?;
let mut pending = Vec::new();
let mut satisfied = false;
for entry in batch {
if !check_terminal {
pending.push(entry.clone());
continue;
}
let child = self.storage.load_runtime_control_plane(&entry.0).await?;
let status = child
.filter(|child| child.parent_session_id.as_deref() == Some(&expected.id))
.and_then(|child| child.last_run_status())
.filter(|status| terminal(status));
if let Some(status) = status {
satisfied |= policy == bamboo_domain::ChildWaitPolicy::Any
|| (policy == bamboo_domain::ChildWaitPolicy::FirstError
&& matches!(status.as_str(), "error" | "timeout" | "cancelled"));
} else {
pending.push(entry.clone());
}
}
if satisfied || pending.is_empty() {
publish(&latest);
return Ok((latest, 0));
}
let count = pending.len();
register_pending(&mut latest, pending, policy)?;
self.save_runtime_state_rebasing_task_conflicts(&mut latest)
.await?;
publish(&latest);
Ok((latest, count))
}
pub async fn compare_exchange_child_wait_and_publish(
&self,
expected: &Session,
updated: &mut Session,
runtime_only: bool,
publish: bamboo_domain::RootActorRuntimePublisher,
) -> std::io::Result<bool> {
let _guard = self.acquire_lock(&expected.id).await;
if self.storage.supports_atomic_child_wait_control_plane() {
return self
.storage
.compare_exchange_child_wait_control_plane(expected, updated, runtime_only, publish)
.await;
}
use crate::v2::child_wait_control_plane::{
apply_transition, is_observation, runtime, validate_incarnation,
};
let mut latest = self
.storage
.load_session(&expected.id)
.await?
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::NotFound,
"child wait parent disappeared",
)
})?;
validate_incarnation(expected, &latest)?;
validate_incarnation(expected, updated)?;
if runtime(&latest).waiting_for_children != runtime(expected).waiting_for_children {
return Ok(false);
}
if is_observation(expected, updated) {
*updated = latest;
return Ok(true);
}
apply_transition(&mut latest, expected, updated, runtime_only)?;
if runtime_only {
self.save_runtime_state_rebasing_task_conflicts(&mut latest)
.await?;
} else {
self.save_session_rebasing_task_conflicts(&mut latest)
.await?;
}
publish(&latest);
*updated = latest;
Ok(true)
}
pub async fn update_task_list_control_plane_and_publish<F>(
&self,
session_id: &str,
task_list: &bamboo_domain::TaskList,
version: &str,
publish: F,
) -> std::io::Result<bool>
where
F: FnOnce(&Session) + Send,
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut latest) = self.storage.load_runtime_control_plane(session_id).await? else {
return Ok(false);
};
if unconditional_task_patch_would_regress(&latest, task_list, version)? {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
format!("Task control-plane changed while patching session {session_id}"),
));
}
let original = latest.clone();
latest.task_list = Some(task_list.clone());
latest.set_task_list_version_meta(version.to_string());
if !self
.storage
.save_task_control_plane_if_matches(&original, &latest)
.await?
{
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
format!("Task control-plane changed while patching session {session_id}"),
));
}
publish(&latest);
Ok(true)
}
pub async fn update_task_list_control_plane_if_version_and_publish<F>(
&self,
session_id: &str,
expected_version: &str,
expected_task_list: &bamboo_domain::TaskList,
task_list: &bamboo_domain::TaskList,
version: &str,
publish: F,
) -> std::io::Result<bool>
where
F: FnOnce(&Session) + Send,
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut latest) = self.storage.load_runtime_control_plane(session_id).await? else {
return Ok(false);
};
if latest.task_list_version_meta().as_deref() != Some(expected_version)
|| !task_list_snapshot_matches(&latest, expected_task_list)?
{
return Ok(false);
}
let original = latest.clone();
latest.task_list = Some(task_list.clone());
latest.set_task_list_version_meta(version.to_string());
if !self
.storage
.save_task_control_plane_if_matches(&original, &latest)
.await?
{
return Ok(false);
}
publish(&latest);
Ok(true)
}
pub async fn update_task_list_control_planes_if_version_and_publish<F>(
&self,
session_id: &str,
shared_session_id: &str,
expected_version: &str,
expected_task_list: &bamboo_domain::TaskList,
task_list: &bamboo_domain::TaskList,
version: &str,
publish: F,
) -> std::io::Result<bool>
where
F: FnOnce(&Session, &Session) + Send,
{
if session_id == shared_session_id {
return self
.update_task_list_control_plane_if_version_and_publish(
session_id,
expected_version,
expected_task_list,
task_list,
version,
|session| publish(session, session),
)
.await;
}
let (first_id, second_id) = if session_id < shared_session_id {
(session_id, shared_session_id)
} else {
(shared_session_id, session_id)
};
let _transaction_guard = self.task_pair_transaction_lock.lock().await;
let _first_guard = self.acquire_lock(first_id).await;
let _second_guard = self.acquire_lock(second_id).await;
self.storage
.recover_task_control_plane_transaction(first_id, second_id)
.await?;
let Some(mut local) = self.storage.load_runtime_control_plane(session_id).await? else {
return Ok(false);
};
let Some(mut shared) = self
.storage
.load_runtime_control_plane(shared_session_id)
.await?
else {
return Ok(false);
};
if local.task_list_version_meta().as_deref() != Some(expected_version)
|| shared.task_list_version_meta().as_deref() != Some(expected_version)
|| !task_list_snapshot_matches(&local, expected_task_list)?
|| !task_list_snapshot_matches(&shared, expected_task_list)?
{
return Ok(false);
}
let local_original = local.clone();
let shared_original = shared.clone();
local.task_list = Some(task_list.clone());
local.set_task_list_version_meta(version.to_string());
shared.task_list = Some(task_list.clone());
shared.set_task_list_version_meta(version.to_string());
let (first_original, first_updated, second_original, second_updated) =
if session_id < shared_session_id {
(&local_original, &local, &shared_original, &shared)
} else {
(&shared_original, &shared, &local_original, &local)
};
let committed = self
.storage
.save_task_control_planes_atomically(
first_original,
first_updated,
second_original,
second_updated,
)
.await?;
if !committed {
return Ok(false);
}
publish(&local, &shared);
Ok(true)
}
pub async fn commit_metadata(&self, session: &Session) -> std::io::Result<()> {
let _guard = self.acquire_lock(&session.id).await;
let mut committed = session.clone();
if let Some(latest) = self.storage.load_runtime_control_plane(&session.id).await? {
ensure_conversation_generation_is_current(&committed, &latest)?;
adopt_durable_model_context_state(&mut committed, &latest);
}
self.save_session_rebasing_task_conflicts(&mut committed)
.await
}
pub async fn merge_save_runtime(&self, session: &mut Session) -> std::io::Result<()> {
self.merge_save_runtime_and_publish(session, |_, _| {})
.await
}
pub async fn merge_save_runtime_and_publish<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
self.merge_save_runtime_inner_and_publish(session, true, false, None, None, publish)
.await
}
pub async fn merge_save_bound_startup(
&self,
session: &mut Session,
expected: &Session,
) -> std::io::Result<()> {
self.merge_save_bound_startup_and_publish(session, expected, |_, _| {})
.await
}
pub async fn merge_save_bound_startup_and_publish<F>(
&self,
session: &mut Session,
expected: &Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
self.merge_save_runtime_inner_and_publish(
session,
true,
false,
None,
Some(expected),
publish,
)
.await
}
pub async fn merge_save_finalized_runtime(&self, session: &mut Session) -> std::io::Result<()> {
self.merge_save_finalized_runtime_and_publish(session, |_, _| {})
.await
}
pub async fn merge_save_finalized_runtime_and_publish<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
self.merge_save_runtime_inner_and_publish(session, true, true, None, None, publish)
.await
}
pub async fn merge_save_inherited_child_wait_and_publish<F>(
&self,
session: &mut Session,
inherited: &bamboo_domain::session::runtime_state::WaitingForChildrenState,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
self.merge_save_runtime_inner_and_publish(
session,
true,
true,
Some(inherited),
None,
publish,
)
.await
}
pub async fn checkpoint_live_bootstrap(
&self,
session: &Session,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
gate: bamboo_domain::AdmissionGate,
) -> std::io::Result<bamboo_domain::LiveBootstrapProof> {
let mut cancel_on_drop = CancelLiveBootstrapOnDrop(Some(gate.clone()));
let (owner, publish) = self.root_actor_writer.as_ref().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
"bound Root writer required",
)
})?;
let _guard = self.acquire_lock(&session.id).await;
let result = self
.storage
.checkpoint_root_live_bootstrap(owner, session, inbox, gate, publish.clone())
.await;
if result.is_ok() {
cancel_on_drop.0 = None;
}
result
}
pub async fn confirm_live_bootstrap(
&self,
proof: &bamboo_domain::LiveBootstrapProof,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
publish: bamboo_domain::RootActorRuntimeEventPublisher,
) -> std::io::Result<()> {
let mut cancel_on_drop = CancelLiveBootstrapOnDrop(Some(proof.storage_gate()));
let (owner, _) = self.root_actor_writer.as_ref().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
"bound Root writer required",
)
})?;
let _guard = self.acquire_lock(&owner.fence.actor_id).await;
let result = self
.storage
.confirm_root_live_bootstrap(owner, proof, inbox, publish)
.await;
if result.is_ok() {
cancel_on_drop.0 = None;
}
result
}
pub async fn checkpoint_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
self.checkpoint_runtime_session_and_publish(session, |_, _| {})
.await
}
pub async fn checkpoint_runtime_session_and_publish<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
self.checkpoint_runtime_session_with_input(session, None, publish)
.await
}
pub async fn checkpoint_root_input(
&self,
session: &mut Session,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
claim: &bamboo_domain::SessionInboxOwnedClaim,
) -> std::io::Result<()> {
self.checkpoint_runtime_session_with_input(session, Some((inbox, claim.clone())), |_, _| {})
.await
}
async fn checkpoint_runtime_session_with_input<F>(
&self,
session: &mut Session,
input: Option<(
Arc<dyn bamboo_domain::SessionInboxPort>,
bamboo_domain::SessionInboxOwnedClaim,
)>,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
ensure_scoped_loop_checkpoint_known(session)?;
let _guard = self.acquire_lock(&session.id).await;
let latest = self.storage.load_session(&session.id).await?;
let input_prompt = match input.as_ref() {
Some((_, claim))
if !latest.as_ref().is_some_and(|durable| {
durable.messages.iter().any(|message| {
bamboo_domain::is_matching_session_message(message, &claim.claim.envelope)
})
}) =>
{
claim
.claim
.envelope
.root_chat_prompt()
.map_err(std::io::Error::other)?
.map(str::to_owned)
}
_ => None,
};
if let Some(latest) = latest.as_ref() {
let incoming_count = session.messages.len();
let durable_count = latest.messages.len();
let (appended, adopted_response) = reconcile_runtime_checkpoint(session, latest)?;
tracing::debug!(
"[{}] append-safe runtime checkpoint: durable={}, incoming={}, appended={}, adopted_response={}, saved={}",
session.id,
durable_count,
incoming_count,
appended,
adopted_response,
session.messages.len(),
);
}
if let Some(prompt) = input_prompt.as_deref() {
session
.messages
.retain(|message| message.role != bamboo_domain::Role::System);
session
.messages
.insert(0, bamboo_domain::Message::system(prompt));
}
let mut result = self
.save_session_rebasing_task_conflicts_with_input(session, input.as_ref(), None, None)
.await;
for _ in 0..MAX_TASK_CONTROL_PLANE_REBASE_RETRIES {
if !result
.as_ref()
.err()
.is_some_and(is_direct_parent_terminal_conflict)
{
break;
}
let Some(durable) = self.storage.load_session(&session.id).await? else {
break;
};
reconcile_runtime_checkpoint(session, &durable)?;
if let Some(prompt) = input_prompt.as_deref() {
session
.messages
.retain(|message| message.role != bamboo_domain::Role::System);
session
.messages
.insert(0, bamboo_domain::Message::system(prompt));
}
result = self
.save_session_rebasing_task_conflicts_with_input(
session,
input.as_ref(),
None,
None,
)
.await;
}
if may_publish_runtime_result(&result) {
publish(session, result.is_ok());
}
result
}
pub async fn checkpoint_retrieval_window_and_publish<F>(
&self,
expected_base: &Session,
staged: &mut Session,
publish: F,
) -> std::io::Result<RetrievalWindowCheckpointOutcome>
where
F: FnOnce(&Session) + Send,
{
if staged
.compression_events
.last()
.is_some_and(|event| event.kind == bamboo_domain::CompressionEventKind::MessageSummary)
{
validate_staged_message_summary_transition(expected_base, staged)?;
} else {
validate_staged_retrieval_window_transition(expected_base, staged)?;
}
let _guard = self.acquire_lock(&staged.id).await;
let latest = self.storage.load_session(&staged.id).await?;
if let Some(latest) = latest.as_ref() {
ensure_conversation_generation_is_current(expected_base, latest)?;
ensure_conversation_generation_is_current(staged, latest)?;
if !retrieval_window_base_matches(expected_base, latest)? {
*staged = rebase_retrieval_window_base(expected_base, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
return Ok(RetrievalWindowCheckpointOutcome::Rebased);
}
ensure_model_context_checkpoint_is_current(staged, latest)?;
bamboo_domain::merge_session_inbox_admission(staged, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
}
self.save_session_rebasing_task_conflicts(staged).await?;
publish(staged);
Ok(RetrievalWindowCheckpointOutcome::Committed)
}
pub async fn checkpoint_prompt_rewrite_and_publish<F>(
&self,
expected_base: &Session,
staged: &mut Session,
publish: F,
) -> std::io::Result<RetrievalWindowCheckpointOutcome>
where
F: FnOnce(&Session) + Send,
{
validate_staged_prompt_rewrite_transition(expected_base, staged)?;
let _guard = self.acquire_lock(&staged.id).await;
let latest = self.storage.load_session(&staged.id).await?;
if let Some(latest) = latest.as_ref() {
ensure_conversation_generation_is_current(expected_base, latest)?;
ensure_conversation_generation_is_current(staged, latest)?;
if !retrieval_window_base_matches(expected_base, latest)? {
*staged = rebase_retrieval_window_base(expected_base, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
return Ok(RetrievalWindowCheckpointOutcome::Rebased);
}
ensure_model_context_checkpoint_is_current(staged, latest)?;
bamboo_domain::merge_session_inbox_admission(staged, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
}
self.save_session_rebasing_task_conflicts(staged).await?;
publish(staged);
Ok(RetrievalWindowCheckpointOutcome::Committed)
}
pub async fn checkpoint_manual_archive_rejection_and_publish<F>(
&self,
expected_base: &Session,
staged: &mut Session,
publish: F,
) -> std::io::Result<RetrievalWindowCheckpointOutcome>
where
F: FnOnce(&Session) + Send,
{
validate_staged_manual_archive_rejection_transition(expected_base, staged)?;
let _guard = self.acquire_lock(&staged.id).await;
let latest = self.storage.load_session(&staged.id).await?;
if let Some(latest) = latest.as_ref() {
ensure_conversation_generation_is_current(expected_base, latest)?;
ensure_conversation_generation_is_current(staged, latest)?;
if !retrieval_window_base_matches(expected_base, latest)? {
*staged = latest.clone();
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
return Ok(RetrievalWindowCheckpointOutcome::Rebased);
}
ensure_model_context_checkpoint_is_current(staged, latest)?;
bamboo_domain::merge_session_inbox_admission(staged, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
}
self.save_session_rebasing_task_conflicts(staged).await?;
publish(staged);
Ok(RetrievalWindowCheckpointOutcome::Committed)
}
pub async fn checkpoint_manual_archive_consumption_and_publish<F>(
&self,
expected_base: &Session,
staged: &mut Session,
publish: F,
) -> std::io::Result<RetrievalWindowCheckpointOutcome>
where
F: FnOnce(&Session) + Send,
{
validate_staged_manual_archive_consumption_transition(expected_base, staged)?;
let _guard = self.acquire_lock(&staged.id).await;
let latest = self.storage.load_session(&staged.id).await?;
if let Some(latest) = latest.as_ref() {
ensure_conversation_generation_is_current(expected_base, latest)?;
ensure_conversation_generation_is_current(staged, latest)?;
if !retrieval_window_base_matches(expected_base, latest)? {
*staged = latest.clone();
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
return Ok(RetrievalWindowCheckpointOutcome::Rebased);
}
ensure_model_context_checkpoint_is_current(staged, latest)?;
bamboo_domain::merge_session_inbox_admission(staged, latest);
apply_authoritative_metadata(staged, latest);
adopt_fresher_disk_permission_posture(staged, latest);
}
self.save_session_rebasing_task_conflicts(staged).await?;
publish(staged);
Ok(RetrievalWindowCheckpointOutcome::Committed)
}
pub async fn save_runtime_authoritative_flags(
&self,
session: &mut Session,
) -> std::io::Result<()> {
self.merge_save_runtime_inner_and_publish(session, false, false, None, None, |_, _| {})
.await
}
async fn merge_save_runtime_inner_and_publish<F>(
&self,
session: &mut Session,
adopt_bypass: bool,
finalize_child_wait: bool,
inherited: Option<&bamboo_domain::session::runtime_state::WaitingForChildrenState>,
startup_expected: Option<&Session>,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
ensure_session_execution_compatible(session)?;
ensure_scoped_loop_checkpoint_known(session)?;
let _guard = self.acquire_lock(&session.id).await;
let original = session;
let mut candidate = original.clone();
let session = &mut candidate;
let latest = self.storage.load_session(&session.id).await?;
if let Some(latest) = latest.as_ref() {
ensure_session_execution_compatible(latest)?;
ensure_conversation_generation_is_current(session, latest)?;
}
let existing_message_count = latest.as_ref().map(|s| s.messages.len());
let incoming_message_count = session.messages.len();
if existing_message_count.is_some_and(|existing| existing > incoming_message_count) {
tracing::warn!(
"[{}] merge_save_runtime SHRINK: disk has {:?} messages, saving {} (last_role={:?}, updated_at={}); a stale writer is reverting a concurrent append",
session.id,
existing_message_count,
incoming_message_count,
session.messages.last().map(|m| format!("{:?}", m.role)),
session.updated_at,
);
} else {
tracing::debug!(
"[{}] merge_save_runtime: disk={:?} messages, saving {} (updated_at={})",
session.id,
existing_message_count,
incoming_message_count,
session.updated_at,
);
}
if let Some(latest) = latest.as_ref() {
adopt_durable_actor_parent_question_handoff(session, latest)?;
if finalize_child_wait {
if let Some(registered_at) = adopt_finalized_child_wait(session, latest, false) {
preserve_finalized_hidden_child_resumes(session, latest, registered_at);
}
}
if let Some(registered_at) = adopt_durable_tagged_child_wait(session, latest) {
preserve_finalized_hidden_child_resumes(session, latest, registered_at);
}
adopt_durable_consumed_clarification(session, latest);
apply_authoritative_metadata(session, latest);
let restored = bamboo_domain::restore_missing_admitted_inbox_messages(session, latest);
if restored > 0 {
tracing::warn!(
session_id = %session.id,
restored,
"restored durable SessionInbox transcript messages into stale runtime save"
);
}
bamboo_domain::merge_session_inbox_admission(session, latest);
if adopt_bypass {
adopt_fresher_disk_permission_posture(session, latest);
}
adopt_fresher_durable_model_context_state(session, latest);
}
let result = self
.save_session_rebasing_task_conflicts_with_input(
session,
None,
inherited,
startup_expected,
)
.await;
if may_publish_runtime_result(&result) {
*original = candidate;
publish(original, result.is_ok());
}
result
}
pub async fn seed_runtime_activation_and_publish<F>(
&self,
session: &mut Session,
publish: F,
) -> std::io::Result<()>
where
F: FnOnce(&Session, bool) + Send,
{
ensure_scoped_loop_checkpoint_known(session)?;
let _guard = self.acquire_lock(&session.id).await;
let mut incoming_audit = PermissionAuditSnapshot::from_metadata(&session.metadata)
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"activation seed requires a complete permission audit record",
)
})?;
if let Some(latest) = self.storage.load_session(&session.id).await? {
ensure_conversation_generation_is_current(session, &latest)?;
if session.kind == bamboo_domain::SessionKind::Child {
if latest.kind != bamboo_domain::SessionKind::Child
|| latest.id != session.id
|| latest.root_session_id != session.root_session_id
|| latest.parent_session_id != session.parent_session_id
|| latest.spawn_depth != session.spawn_depth
|| latest.project_id_meta() != session.project_id_meta()
{
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
bamboo_domain::SessionAuthorityConflict(
"warm Child activation disagrees with durable creation identity or Project"
.to_string(),
),
));
}
session.created_at = latest.created_at;
}
apply_authoritative_metadata(session, &latest);
bamboo_domain::restore_missing_admitted_inbox_messages(session, &latest);
bamboo_domain::merge_session_inbox_admission(session, &latest);
adopt_fresher_durable_model_context_state(session, &latest);
let durable_audit = PermissionAuditSnapshot::from_metadata(&latest.metadata);
let durable_floor = durable_audit
.as_ref()
.map(|snapshot| snapshot.audit_revision)
.unwrap_or_default();
if let Some(durable_audit) = durable_audit {
if durable_audit.resolution == incoming_audit.resolution {
incoming_audit.transitioned_at = durable_audit.transitioned_at;
}
}
incoming_audit.audit_revision = bamboo_domain::next_permission_audit_revision_after(
durable_floor.max(incoming_audit.audit_revision),
)
.map_err(|error| {
std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string())
})?;
}
session
.agent_runtime_state
.get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
.set_permission_mode(incoming_audit.resolution.requested);
incoming_audit.write_to(&mut session.metadata);
let result = self.save_session_rebasing_task_conflicts(session).await;
if may_publish_runtime_result(&result) {
publish(session, result.is_ok());
}
result
}
pub async fn update_authoritative_permission_posture_and_publish<M, P>(
&self,
session_id: &str,
seed: &PermissionAuditSeed,
mutate: M,
publish: P,
) -> std::io::Result<Option<Session>>
where
M: FnOnce(&mut Session),
P: FnOnce(&Session),
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut latest) = self.storage.load_session(session_id).await? else {
return Ok(None);
};
let previous_mode = latest
.agent_runtime_state
.as_ref()
.map(|state| state.effective_permission_mode())
.unwrap_or_default();
let previous_resolution = PermissionAuditSnapshot::from_metadata(&latest.metadata)
.map(|snapshot| snapshot.resolution);
mutate(&mut latest);
latest
.agent_runtime_state
.get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
.set_permission_mode(seed.resolution.requested);
let mode_changed = previous_mode != seed.resolution.requested;
let posture_changed = previous_resolution != Some(seed.resolution);
let transitioned_at = posture_changed.then(|| chrono::Utc::now().to_rfc3339());
bamboo_domain::record_permission_audit(
&mut latest.metadata,
seed,
transitioned_at.as_deref(),
)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string()))?;
if mode_changed {
latest.metadata_version = latest.metadata_version.saturating_add(1);
}
self.save_session_rebasing_task_conflicts(&mut latest)
.await?;
publish(&latest);
Ok(Some(latest))
}
pub async fn record_permission_posture_activation_and_publish<P>(
&self,
session_id: &str,
expected_audit_revision: Option<u64>,
seed: &PermissionAuditSeed,
publish: P,
) -> std::io::Result<Option<Session>>
where
P: FnOnce(&Session),
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut latest) = self.storage.load_session(session_id).await? else {
return Ok(None);
};
let durable_audit = PermissionAuditSnapshot::from_metadata(&latest.metadata);
let durable_revision = durable_audit
.as_ref()
.map(|snapshot| snapshot.audit_revision);
if durable_revision != expected_audit_revision {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"stale permission posture activation: durable audit changed after dispatch",
));
}
let durable_requested = latest
.agent_runtime_state
.as_ref()
.map(|state| state.effective_permission_mode())
.unwrap_or_default();
if durable_requested != seed.resolution.requested || !seed.resolution.is_consistent() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"stale or inconsistent permission posture activation",
));
}
bamboo_domain::record_permission_audit(&mut latest.metadata, seed, None).map_err(
|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error.to_string()),
)?;
self.save_session_rebasing_task_conflicts(&mut latest)
.await?;
publish(&latest);
Ok(Some(latest))
}
pub async fn update_runtime_config<F>(
&self,
session_id: &str,
mutate: F,
) -> std::io::Result<Option<Session>>
where
F: FnOnce(&mut Session),
{
self.update_runtime_config_and_publish(session_id, mutate, |_| {})
.await
}
pub async fn update_runtime_config_and_publish<M, P>(
&self,
session_id: &str,
mutate: M,
publish: P,
) -> std::io::Result<Option<Session>>
where
M: FnOnce(&mut Session),
P: FnOnce(&Session),
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut session) = self.storage.load_session(session_id).await? else {
return Ok(None);
};
mutate(&mut session);
self.save_session_rebasing_task_conflicts(&mut session)
.await?;
publish(&session);
Ok(Some(session))
}
async fn load_response_candidate<C>(
&self,
session_id: &str,
load_cached: C,
) -> std::io::Result<Option<Session>>
where
C: FnOnce() -> Option<Session> + Send,
{
let cached_candidate = load_cached();
let durable = self.storage.load_session(session_id).await?;
if let (Some(cached), Some(durable)) = (cached_candidate.as_ref(), durable.as_ref()) {
ensure_conversation_generation_is_current(cached, durable)?;
}
if let Some(cached) = cached_candidate.as_ref() {
ensure_session_execution_compatible(cached)?;
}
if let Some(durable) = durable.as_ref() {
ensure_session_execution_compatible(durable)?;
}
let Some(mut session) = (match (cached_candidate, durable.as_ref()) {
(Some(cached), Some(durable)) => {
let prefer_durable = durable.updated_at > cached.updated_at
|| (durable.updated_at == cached.updated_at
&& cached.pending_question.is_none()
&& durable.pending_question.is_some());
Some(if prefer_durable {
durable.clone()
} else {
cached
})
}
(Some(cached), None) => Some(cached),
(None, durable) => durable.cloned(),
}) else {
return Ok(None);
};
if let Some(latest) = durable.as_ref() {
adopt_durable_consumed_clarification(&mut session, latest);
bamboo_domain::append_missing_runtime_messages(&mut session, latest);
apply_authoritative_metadata(&mut session, latest);
let restored =
bamboo_domain::restore_missing_admitted_inbox_messages(&mut session, latest);
if restored > 0 {
tracing::warn!(
session_id,
restored,
"restored durable SessionInbox transcript messages into response transaction"
);
}
bamboo_domain::merge_session_inbox_admission(&mut session, latest);
adopt_fresher_disk_permission_posture(&mut session, latest);
adopt_fresher_durable_model_context_state(&mut session, latest);
}
Ok(Some(session))
}
pub async fn inspect_runtime_session_for_response<C>(
&self,
session_id: &str,
load_cached: C,
) -> std::io::Result<Option<Session>>
where
C: FnOnce() -> Option<Session> + Send,
{
let _guard = self.acquire_lock(session_id).await;
self.load_response_candidate(session_id, load_cached).await
}
pub async fn mutate_runtime_session_and_publish<C, M, P, E>(
&self,
session_id: &str,
load_cached: C,
mutate: M,
publish: P,
) -> std::io::Result<Result<Option<Session>, E>>
where
C: FnOnce() -> Option<Session> + Send,
M: FnOnce(&mut Session) -> Result<(), E> + Send,
P: FnOnce(&Session) + Send,
E: Send,
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut session) = self
.load_response_candidate(session_id, load_cached)
.await?
else {
return Ok(Ok(None));
};
let expected = (self.root_actor_writer.is_some()
&& self.storage.supports_root_actor_bound_startup())
.then(|| session.clone());
if let Err(error) = mutate(&mut session) {
return Ok(Err(error));
}
if let Some(expected) = expected.as_ref() {
self.save_session_rebasing_task_conflicts_with_input(
&mut session,
None,
None,
Some(expected),
)
.await?;
} else {
self.save_session_rebasing_task_conflicts(&mut session)
.await?;
}
publish(&session);
Ok(Ok(Some(session)))
}
pub async fn clear_legacy_pending_messages_and_publish<F>(
&self,
session_id: &str,
expected: &[serde_json::Value],
publish: F,
) -> std::io::Result<bool>
where
F: FnOnce(&Session) + Send,
{
let _guard = self.acquire_lock(session_id).await;
let Some(mut latest) = self.storage.load_session(session_id).await? else {
return Ok(false);
};
if latest.pending_injected_messages().as_deref() != Some(expected) {
return Ok(false);
}
latest.clear_pending_injected_messages();
self.save_runtime_state_rebasing_task_conflicts(&mut latest)
.await?;
publish(&latest);
Ok(true)
}
}
#[async_trait::async_trait]
impl RuntimeSessionPersistence for LockedSessionStore {
async fn claim_scoped_loop_controls(
&self,
session: &Session,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
active_run_id: &str,
limit: usize,
) -> std::io::Result<Vec<bamboo_domain::ScopedLoopControlClaim>> {
use bamboo_domain::ScopedLoopControlClaim;
if let Some((owner, _)) = self.root_actor_writer.as_ref() {
let inbox = self.storage.bind_root_actor_inbox(owner, inbox)?;
let request = bamboo_domain::SessionInboxLeaseRequest {
consumer: bamboo_domain::SessionInboxConsumerId::new(),
now: chrono::Utc::now(),
duration: chrono::Duration::seconds(5),
};
return inbox
.claim_owned_scoped_loop_controls(&session.id, limit, Some(active_run_id), &request)
.await
.map(|claims| {
claims
.into_iter()
.map(ScopedLoopControlClaim::Owned)
.collect()
})
.map_err(std::io::Error::other);
}
inbox
.claim_scoped_loop_controls(&session.id, limit, Some(active_run_id))
.await
.map(|claims| {
claims
.into_iter()
.map(ScopedLoopControlClaim::Legacy)
.collect()
})
.map_err(std::io::Error::other)
}
async fn checkpoint_scoped_loop_control(
&self,
session: &mut Session,
bindings: &[bamboo_domain::ScopedLoopControlBinding],
activation: Arc<dyn bamboo_domain::SessionActivationPort>,
input: Option<(
Arc<dyn bamboo_domain::SessionInboxPort>,
bamboo_domain::ScopedLoopControlClaim,
)>,
) -> std::io::Result<()> {
let _guard = self.acquire_lock(&session.id).await;
let publish = self
.root_actor_writer
.as_ref()
.map(|(_, publish)| publish.clone())
.unwrap_or_else(|| Arc::new(|_: &Session| {}));
self.storage
.checkpoint_scoped_loop(bamboo_domain::ScopedLoopControlCheckpoint {
session,
bindings,
activation,
owner: self.root_actor_writer.as_ref().map(|(owner, _)| owner),
input,
publish,
})
.await
}
async fn load_loop_control_checkpoint(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.storage
.load_execution_admission_snapshot(session_id)
.await
}
async fn admit_root_inbox(
&self,
session: &mut Session,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
active_run_id: Option<&str>,
) -> std::io::Result<Option<bamboo_domain::RootInboxAdmission>> {
self.admit_owned_child_inbox(session, inbox, active_run_id)
.await
}
fn inherited_child_wait(&self) -> Option<bamboo_domain::InheritedChildWait> {
self.inherited_child_wait.clone()
}
fn bind_inherited_child_wait(
&self,
inherited: bamboo_domain::InheritedChildWait,
) -> std::io::Result<Arc<dyn RuntimeSessionPersistence>> {
Ok(Arc::new(self.bind_inherited_child_wait(inherited)))
}
async fn save_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
self.merge_save_runtime(session).await
}
fn supports_bound_startup_session(&self) -> bool {
self.root_actor_writer.is_some()
&& self.child_actor_writer.is_none()
&& self.storage.supports_root_actor_bound_startup()
}
async fn save_bound_startup_session(
&self,
session: &mut Session,
expected: &Session,
) -> std::io::Result<()> {
self.merge_save_bound_startup(session, expected).await
}
async fn save_finalized_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
self.merge_save_finalized_runtime(session).await
}
async fn save_finalized_runtime_with_inherited_child_wait(
&self,
session: &mut Session,
inherited: &bamboo_domain::session::runtime_state::WaitingForChildrenState,
) -> std::io::Result<()> {
self.merge_save_inherited_child_wait_and_publish(session, inherited, |_, _| {})
.await
}
async fn seed_runtime_activation(&self, session: &mut Session) -> std::io::Result<()> {
self.seed_runtime_activation_and_publish(session, |_, _| {})
.await
}
async fn record_permission_posture_activation(
&self,
session_id: &str,
expected_audit_revision: Option<u64>,
seed: &PermissionAuditSeed,
) -> std::io::Result<Option<Session>> {
self.record_permission_posture_activation_and_publish(
session_id,
expected_audit_revision,
seed,
|_| {},
)
.await
}
async fn save_runtime_control_plane(&self, session: &mut Session) -> std::io::Result<()> {
self.save_runtime_only(session).await
}
async fn load_runtime_control_plane(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.load_child_runtime_control_plane(session_id).await
}
async fn update_task_list_control_plane(
&self,
session_id: &str,
task_list: &bamboo_domain::TaskList,
version: &str,
) -> std::io::Result<bool> {
self.update_task_list_control_plane_and_publish(session_id, task_list, version, |_| {})
.await
}
async fn update_task_list_control_plane_if_version(
&self,
session_id: &str,
expected_version: &str,
expected_task_list: &bamboo_domain::TaskList,
task_list: &bamboo_domain::TaskList,
version: &str,
) -> std::io::Result<bool> {
self.update_task_list_control_plane_if_version_and_publish(
session_id,
expected_version,
expected_task_list,
task_list,
version,
|_| {},
)
.await
}
async fn update_task_list_control_planes_if_version(
&self,
session_id: &str,
shared_session_id: &str,
expected_version: &str,
expected_task_list: &bamboo_domain::TaskList,
task_list: &bamboo_domain::TaskList,
version: &str,
) -> std::io::Result<bool> {
self.update_task_list_control_planes_if_version_and_publish(
session_id,
shared_session_id,
expected_version,
expected_task_list,
task_list,
version,
|_, _| {},
)
.await
}
async fn checkpoint_live_bootstrap(
&self,
session: &Session,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
gate: bamboo_domain::AdmissionGate,
) -> std::io::Result<bamboo_domain::LiveBootstrapProof> {
LockedSessionStore::checkpoint_live_bootstrap(self, session, inbox, gate).await
}
async fn confirm_live_bootstrap(
&self,
proof: &bamboo_domain::LiveBootstrapProof,
inbox: Arc<dyn bamboo_domain::SessionInboxPort>,
publish: bamboo_domain::RootActorRuntimeEventPublisher,
) -> std::io::Result<()> {
LockedSessionStore::confirm_live_bootstrap(self, proof, inbox, publish).await
}
async fn checkpoint_runtime_session(&self, session: &mut Session) -> std::io::Result<()> {
LockedSessionStore::checkpoint_runtime_session(self, session).await
}
async fn checkpoint_retrieval_window(
&self,
expected_base: &Session,
staged: &mut Session,
) -> std::io::Result<RetrievalWindowCheckpointOutcome> {
self.checkpoint_retrieval_window_and_publish(expected_base, staged, |_| {})
.await
}
async fn checkpoint_prompt_rewrite(
&self,
expected_base: &Session,
staged: &mut Session,
) -> std::io::Result<RetrievalWindowCheckpointOutcome> {
self.checkpoint_prompt_rewrite_and_publish(expected_base, staged, |_| {})
.await
}
async fn checkpoint_manual_archive_rejection(
&self,
expected_base: &Session,
staged: &mut Session,
) -> std::io::Result<RetrievalWindowCheckpointOutcome> {
self.checkpoint_manual_archive_rejection_and_publish(expected_base, staged, |_| {})
.await
}
async fn checkpoint_manual_archive_consumption(
&self,
expected_base: &Session,
staged: &mut Session,
) -> std::io::Result<RetrievalWindowCheckpointOutcome> {
self.checkpoint_manual_archive_consumption_and_publish(expected_base, staged, |_| {})
.await
}
async fn load_runtime_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
self.validate_child_runtime_read(session_id).await?;
let session = self.storage.load_session(session_id).await?;
self.validate_child_runtime_readback(session_id, session.as_ref())
.await?;
Ok(session)
}
async fn load_runtime_admission_snapshot(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.validate_child_runtime_read(session_id).await?;
let session = self
.storage
.load_execution_admission_snapshot(session_id)
.await?;
self.validate_child_runtime_readback(session_id, session.as_ref())
.await?;
Ok(session)
}
async fn clear_legacy_pending_messages(
&self,
session_id: &str,
expected: &[serde_json::Value],
) -> std::io::Result<bool> {
self.clear_legacy_pending_messages_and_publish(session_id, expected, |_| {})
.await
}
}
async fn merge_authoritative_metadata_into_stale(
storage: &Arc<dyn Storage>,
session: &mut Session,
) -> std::io::Result<()> {
if let Some(latest) = storage.load_session(&session.id).await? {
ensure_conversation_generation_is_current(session, &latest)?;
adopt_durable_consumed_clarification(session, &latest);
apply_authoritative_metadata(session, &latest);
bamboo_domain::restore_missing_admitted_inbox_messages(session, &latest);
bamboo_domain::merge_session_inbox_admission(session, &latest);
adopt_fresher_disk_permission_posture(session, &latest);
adopt_fresher_durable_model_context_state(session, &latest);
}
Ok(())
}
fn adopt_durable_model_context_state(session: &mut Session, latest: &Session) {
session
.model_context_state
.clone_from(&latest.model_context_state);
}
fn adopt_fresher_durable_model_context_state(session: &mut Session, latest: &Session) {
let adopt = match (
session.model_context_state.as_ref(),
latest.model_context_state.as_ref(),
) {
(None, Some(_)) => true,
(Some(incoming), Some(durable)) => {
durable.state_revision > incoming.state_revision
|| (durable.state_revision == incoming.state_revision && durable != incoming)
}
_ => false,
};
if adopt {
adopt_durable_model_context_state(session, latest);
}
}
fn ensure_conversation_generation_is_current(
session: &Session,
latest: &Session,
) -> std::io::Result<()> {
session
.check_conversation_generation(latest)
.map_err(|error| {
std::io::Error::new(
std::io::ErrorKind::WouldBlock,
bamboo_domain::SessionAuthorityConflict(error.to_string()),
)
})
}
fn ensure_model_context_checkpoint_is_current(
session: &Session,
latest: &Session,
) -> std::io::Result<()> {
ensure_conversation_generation_is_current(session, latest)?;
let stale_or_conflicting = match (
session.model_context_state.as_ref(),
latest.model_context_state.as_ref(),
) {
(None, Some(_)) => true,
(Some(incoming), Some(durable)) => {
durable.state_revision > incoming.state_revision
|| (durable.state_revision == incoming.state_revision && durable != incoming)
}
_ => false,
};
if stale_or_conflicting {
return Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"stale or conflicting model-context ledger checkpoint",
));
}
Ok(())
}
fn message_matches_retrieval_window_base(
expected: &bamboo_domain::Message,
durable: &bamboo_domain::Message,
) -> std::io::Result<bool> {
if durable.image_ocr.is_some() && durable.image_ocr != expected.image_ocr {
return Ok(false);
}
let mut expected = expected.clone();
let mut durable = durable.clone();
expected.image_ocr = None;
durable.image_ocr = None;
let expected = serde_json::to_vec(&expected)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let durable = serde_json::to_vec(&durable)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
Ok(expected == durable)
}
fn invalid_retrieval_window_checkpoint(message: impl Into<String>) -> std::io::Error {
std::io::Error::new(std::io::ErrorKind::InvalidInput, message.into())
}
fn validate_staged_manual_archive_consumption_transition(
expected: &Session,
staged: &Session,
) -> std::io::Result<()> {
if expected.id != staged.id || expected.messages.len() != staged.messages.len() {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-consumption checkpoint must preserve the Session ID and message array length",
));
}
let occurrence = staged
.metadata
.get(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY)
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-consumption checkpoint is missing its consumed occurrence",
)
})
.and_then(|value| {
serde_json::from_str::<ResponseOccurrence>(value).map_err(|error| {
invalid_retrieval_window_checkpoint(format!(
"manual archive-consumption checkpoint has an invalid consumed occurrence: {error}"
))
})
})?;
let result_index = expected
.messages
.iter()
.position(|message| {
message.id == occurrence.tool_result_message_id
&& message.tool_call_id.as_deref() == Some(occurrence.tool_call_id.as_str())
&& matches!(message.role, bamboo_domain::Role::Tool)
})
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-consumption checkpoint occurrence does not identify a Tool result",
)
})?;
let correlated_archive_call = expected.messages[..result_index]
.iter()
.rev()
.find(|message| !matches!(message.role, bamboo_domain::Role::Tool))
.filter(|message| matches!(message.role, bamboo_domain::Role::Assistant))
.and_then(|message| message.tool_calls.as_ref())
.into_iter()
.flatten()
.any(|call| {
call.id == occurrence.tool_call_id
&& bamboo_domain::canonical_tool_name(&call.function.name) == "archive_context"
});
if !correlated_archive_call {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-consumption checkpoint is not correlated to the current archive_context batch",
));
}
let mut canonical = expected.clone();
canonical.metadata.insert(
LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY.to_string(),
staged
.metadata
.get(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY)
.expect("validated occurrence metadata")
.clone(),
);
let canonical = serde_json::to_vec(&canonical)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let staged = serde_json::to_vec(staged)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if canonical != staged {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-consumption checkpoint contains mutations outside the correlated consumed marker",
));
}
Ok(())
}
fn validate_staged_manual_archive_rejection_transition(
expected: &Session,
staged: &Session,
) -> std::io::Result<()> {
if expected.id != staged.id || expected.messages.len() != staged.messages.len() {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint must preserve the Session ID and message array length",
));
}
let occurrence = staged
.metadata
.get(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY)
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint is missing its consumed occurrence",
)
})
.and_then(|value| {
serde_json::from_str::<ResponseOccurrence>(value).map_err(|error| {
invalid_retrieval_window_checkpoint(format!(
"manual archive-rejection checkpoint has an invalid consumed occurrence: {error}"
))
})
})?;
let result_index = expected
.messages
.iter()
.position(|message| {
message.id == occurrence.tool_result_message_id
&& message.tool_call_id.as_deref() == Some(occurrence.tool_call_id.as_str())
&& matches!(message.role, bamboo_domain::Role::Tool)
})
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint occurrence does not identify a Tool result",
)
})?;
let before = &expected.messages[result_index];
let after = &staged.messages[result_index];
let rejection_reason = after
.content
.strip_prefix("archive_context rejected: ")
.filter(|reason| !reason.trim().is_empty())
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint has an invalid Tool result message",
)
})?;
if after.tool_success != Some(false)
|| (before.content == after.content && before.tool_success == after.tool_success)
{
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint did not reject the Tool result",
));
}
let correlated_archive_call = expected.messages[..result_index]
.iter()
.rev()
.find(|message| !matches!(message.role, bamboo_domain::Role::Tool))
.filter(|message| matches!(message.role, bamboo_domain::Role::Assistant))
.and_then(|message| message.tool_calls.as_ref())
.into_iter()
.flatten()
.any(|call| {
call.id == occurrence.tool_call_id
&& bamboo_domain::canonical_tool_name(&call.function.name) == "archive_context"
});
if !correlated_archive_call {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint is not correlated to the current archive_context batch",
));
}
let staged_rejections = staged
.metadata
.get(MANUAL_ARCHIVE_REJECTIONS_KEY)
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint is missing its rejection ledger",
)
})
.and_then(|value| {
serde_json::from_str::<Vec<serde_json::Value>>(value).map_err(|error| {
invalid_retrieval_window_checkpoint(format!(
"manual archive-rejection checkpoint has an invalid rejection ledger: {error}"
))
})
})?;
let occurrence_value = serde_json::to_value(&occurrence)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let Some(last_rejection) = staged_rejections.last() else {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint has an empty rejection ledger",
));
};
if last_rejection.get("occurrence") != Some(&occurrence_value)
|| last_rejection
.get("reason")
.and_then(serde_json::Value::as_str)
!= Some(rejection_reason)
{
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint ledger does not match its Tool result",
));
}
let mut expected_rejections = expected
.metadata
.get(MANUAL_ARCHIVE_REJECTIONS_KEY)
.and_then(|value| serde_json::from_str::<Vec<serde_json::Value>>(value).ok())
.unwrap_or_default();
expected_rejections.retain(|rejection| rejection.get("occurrence") != Some(&occurrence_value));
expected_rejections.push(last_rejection.clone());
if expected_rejections.len() > MAX_MANUAL_ARCHIVE_REJECTIONS {
expected_rejections.drain(..expected_rejections.len() - MAX_MANUAL_ARCHIVE_REJECTIONS);
}
if expected_rejections != staged_rejections {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint rewrote unrelated rejection-ledger entries",
));
}
let mut canonical = expected.clone();
canonical.messages[result_index]
.content
.clone_from(&after.content);
canonical.messages[result_index].tool_success = Some(false);
canonical.metadata.insert(
LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY.to_string(),
staged
.metadata
.get(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY)
.expect("validated occurrence metadata")
.clone(),
);
canonical.metadata.insert(
MANUAL_ARCHIVE_REJECTIONS_KEY.to_string(),
staged
.metadata
.get(MANUAL_ARCHIVE_REJECTIONS_KEY)
.expect("validated rejection metadata")
.clone(),
);
canonical
.metadata
.remove(RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
canonical
.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite);
let canonical = serde_json::to_vec(&canonical)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let staged = serde_json::to_vec(staged)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if canonical != staged {
return Err(invalid_retrieval_window_checkpoint(
"manual archive-rejection checkpoint contains mutations outside the correlated Tool result, bounded metadata, and provider reset",
));
}
Ok(())
}
fn validate_staged_prompt_rewrite_transition(
expected: &Session,
staged: &Session,
) -> std::io::Result<()> {
if expected.id != staged.id || expected.messages.len() != staged.messages.len() {
return Err(invalid_retrieval_window_checkpoint(
"prompt-rewrite checkpoint must preserve the Session ID and message array length",
));
}
let mut canonical = expected.clone();
let mut rewrote_system_prompt = false;
for ((before, after), canonical_message) in expected
.messages
.iter()
.zip(&staged.messages)
.zip(&mut canonical.messages)
{
if before.id != after.id {
return Err(invalid_retrieval_window_checkpoint(
"prompt-rewrite checkpoint reordered or replaced a message",
));
}
if before.content != after.content {
if !matches!(before.role, bamboo_domain::Role::System)
|| !matches!(after.role, bamboo_domain::Role::System)
{
return Err(invalid_retrieval_window_checkpoint(
"prompt-rewrite checkpoint changed non-System message content",
));
}
canonical_message.content.clone_from(&after.content);
rewrote_system_prompt = true;
}
}
if !rewrote_system_prompt {
return Err(invalid_retrieval_window_checkpoint(
"prompt-rewrite checkpoint did not change System message content",
));
}
canonical.metadata.remove("responses.previous_response_id");
canonical
.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite);
let canonical = serde_json::to_vec(&canonical)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let staged = serde_json::to_vec(staged)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if canonical != staged {
return Err(invalid_retrieval_window_checkpoint(
"prompt-rewrite checkpoint contains mutations outside the System prompt and provider reset",
));
}
Ok(())
}
fn validate_staged_retrieval_window_transition(
expected: &Session,
staged: &Session,
) -> std::io::Result<()> {
if expected.id != staged.id {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint Session IDs differ",
));
}
if expected.conversation_summary.is_some() || staged.conversation_summary.is_some() {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint cannot contain a conversation summary",
));
}
if expected.messages.len() != staged.messages.len() {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint must preserve the message array",
));
}
if staged.compression_events.len() != expected.compression_events.len().saturating_add(1) {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint must append exactly one compression event",
));
}
let expected_events = serde_json::to_vec(&expected.compression_events)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let staged_prefix =
serde_json::to_vec(&staged.compression_events[..expected.compression_events.len()])
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if expected_events != staged_prefix {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint rewrote an existing compression event",
));
}
let event = staged
.compression_events
.last()
.expect("length check guarantees a staged compression event");
if event.kind != bamboo_domain::CompressionEventKind::RetrievalWindow
|| event.id.is_empty()
|| expected
.compression_events
.iter()
.any(|existing| existing.id == event.id)
{
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint has an invalid archive event",
));
}
let mut newly_archived = 0usize;
for (before, after) in expected.messages.iter().zip(&staged.messages) {
if before.id != after.id {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint reordered or replaced a message",
));
}
let is_new_archive = !before.compressed && after.compressed;
if is_new_archive {
if before.compressed_by_event_id.is_some()
|| after.compressed_by_event_id.as_deref() != Some(event.id.as_str())
{
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint has an invalid message correlation",
));
}
newly_archived = newly_archived.saturating_add(1);
} else if before.compressed != after.compressed
|| before.compressed_by_event_id != after.compressed_by_event_id
{
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint contains an unsupported archive mutation",
));
}
let mut normalized_after = after.clone();
normalized_after.compressed = before.compressed;
normalized_after
.compressed_by_event_id
.clone_from(&before.compressed_by_event_id);
let before = serde_json::to_vec(before)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let after = serde_json::to_vec(&normalized_after)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if before != after {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint mutated message content",
));
}
}
if newly_archived == 0 || newly_archived != event.messages_compressed {
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint event count does not match message correlations",
));
}
if staged
.model_context_state
.as_ref()
.and_then(|state| state.last_reset_reason)
!= Some(bamboo_domain::ModelContextResetReason::Compression)
{
return Err(invalid_retrieval_window_checkpoint(
"retrieval-window checkpoint is missing its model-context reset",
));
}
Ok(())
}
fn validate_staged_message_summary_transition(
expected: &Session,
staged: &Session,
) -> std::io::Result<()> {
use bamboo_domain::{CompressionEventKind, ModelContextResetReason, Role};
if expected.id != staged.id || expected.messages.len() != staged.messages.len() {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint must preserve the Session ID and raw message array",
));
}
if staged.compression_events.len() != expected.compression_events.len().saturating_add(1) {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint must append exactly one compression event",
));
}
let event = staged
.compression_events
.last()
.expect("validated event count");
if event.kind != CompressionEventKind::MessageSummary
|| event.id.is_empty()
|| expected
.compression_events
.iter()
.any(|prior| prior.id == event.id)
{
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has an invalid archive event",
));
}
let latest_user_id = expected
.messages
.iter()
.rev()
.find(|message| !message.compressed && message.role == Role::User)
.map(|message| message.id.as_str());
let mut canonical = expected.clone();
let mut newly_archived = 0usize;
let mut calls = std::collections::HashSet::new();
let mut results = std::collections::HashSet::new();
for (before, canonical_message) in expected.messages.iter().zip(&mut canonical.messages) {
if before.compressed {
continue;
}
let has_calls = before
.tool_calls
.as_ref()
.is_some_and(|calls| !calls.is_empty());
let has_tool_protocol =
has_calls || before.tool_call_id.is_some() || before.role == Role::Tool;
let generated_recovery = before.role == Role::Assistant
&& before.content.starts_with("[post-compaction-recovery]\n");
let should_archive = before.role != Role::System
&& Some(before.id.as_str()) != latest_user_id
&& (!before.never_compress || generated_recovery);
if has_tool_protocol && (before.never_compress || !should_archive) {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint cannot retain active tool protocol",
));
}
if has_calls {
if before.role != Role::Assistant {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has tool calls outside an assistant message",
));
}
for call in before.tool_calls.iter().flatten() {
if call.id.is_empty() || !calls.insert(call.id.as_str()) {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has empty or duplicate tool call identifiers",
));
}
}
}
if before.role == Role::Tool {
let id = before
.tool_call_id
.as_deref()
.filter(|id| !id.is_empty())
.ok_or_else(|| {
invalid_retrieval_window_checkpoint(
"message-summary checkpoint has an uncorrelated tool result",
)
})?;
if !results.insert(id) {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has duplicate tool results",
));
}
} else if before.tool_call_id.is_some() {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has a tool result identifier outside a tool message",
));
}
if should_archive {
if before.compressed_by_event_id.is_some() {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has a pre-existing active archive correlation",
));
}
canonical_message.compressed = true;
canonical_message.compressed_by_event_id = Some(event.id.clone());
newly_archived += 1;
}
}
if calls != results || newly_archived == 0 || event.messages_compressed != newly_archived {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint must archive complete tool chains with an exact event count",
));
}
if let Some(summary) = staged.conversation_summary.as_ref() {
if summary.content.trim().is_empty()
|| summary.message_count != newly_archived
|| summary.token_count != event.summary_tokens
{
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint has inconsistent summary metadata",
));
}
}
canonical.compression_events.push(event.clone());
canonical
.conversation_summary
.clone_from(&staged.conversation_summary);
canonical.token_usage.clone_from(&staged.token_usage);
canonical.updated_at = staged.updated_at;
canonical.force_manual_compression = None;
canonical
.metadata
.remove(RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
canonical.reset_model_context_epoch(ModelContextResetReason::Compression);
let canonical = serde_json::to_vec(&canonical)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let staged = serde_json::to_vec(staged)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if canonical != staged {
return Err(invalid_retrieval_window_checkpoint(
"message-summary checkpoint changed raw history, protected anchors, or state outside the compression projection and provider reset",
));
}
Ok(())
}
fn retrieval_window_base_matches(expected: &Session, durable: &Session) -> std::io::Result<bool> {
if expected.id != durable.id || durable.messages.len() > expected.messages.len() {
return Ok(false);
}
for (durable_message, expected_message) in durable.messages.iter().zip(expected.messages.iter())
{
if !message_matches_retrieval_window_base(expected_message, durable_message)? {
return Ok(false);
}
}
if durable.compression_events.len() > expected.compression_events.len() {
return Ok(false);
}
for (durable_event, expected_event) in durable
.compression_events
.iter()
.zip(expected.compression_events.iter())
{
let durable_event = serde_json::to_vec(durable_event)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
let expected_event = serde_json::to_vec(expected_event)
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if durable_event != expected_event {
return Ok(false);
}
}
let summary_matches = serde_json::to_vec(&expected.conversation_summary)
.and_then(|expected| {
serde_json::to_vec(&durable.conversation_summary).map(|durable| expected == durable)
})
.map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?;
if !summary_matches {
return Ok(false);
}
if expected.metadata != durable.metadata
|| expected.runtime_metadata != durable.runtime_metadata
|| expected.metadata_version != durable.metadata_version
|| expected.model != durable.model
|| expected.model_ref != durable.model_ref
|| expected.reasoning_effort != durable.reasoning_effort
{
return Ok(false);
}
Ok(ensure_model_context_checkpoint_is_current(expected, durable).is_ok())
}
fn rebase_retrieval_window_base(expected: &Session, durable: &Session) -> Session {
let mut rebased = expected.clone();
bamboo_domain::append_missing_runtime_messages(&mut rebased, durable);
rebased.metadata.clone_from(&durable.metadata);
rebased
.runtime_metadata
.clone_from(&durable.runtime_metadata);
rebased.model.clone_from(&durable.model);
rebased.model_ref.clone_from(&durable.model_ref);
rebased.reasoning_effort = durable.reasoning_effort;
bamboo_domain::merge_session_inbox_admission(&mut rebased, durable);
rebased
.conversation_summary
.clone_from(&durable.conversation_summary);
rebased
.compression_events
.clone_from(&durable.compression_events);
rebased.token_usage.clone_from(&durable.token_usage);
rebased
.model_context_state
.clone_from(&durable.model_context_state);
if durable.updated_at > rebased.updated_at {
rebased.updated_at = durable.updated_at;
}
rebased
}
fn adopt_fresher_disk_permission_posture(session: &mut Session, latest: &Session) {
let Some(disk_mode) = latest
.agent_runtime_state
.as_ref()
.map(|state| state.effective_permission_mode())
else {
return;
};
let current_mode = session
.agent_runtime_state
.as_ref()
.map(|state| state.effective_permission_mode())
.unwrap_or_default();
let Some(disk_audit) = bamboo_domain::fresher_disk_permission_audit(
current_mode,
&session.metadata,
disk_mode,
&latest.metadata,
) else {
return;
};
match session.agent_runtime_state.as_mut() {
Some(state) => state.set_permission_mode(disk_mode),
None if disk_mode != bamboo_domain::SessionPermissionMode::Default => {
let state = session
.agent_runtime_state
.get_or_insert_with(bamboo_domain::AgentRuntimeState::default);
state.set_permission_mode(disk_mode);
}
None => {}
}
disk_audit.write_to(&mut session.metadata);
}
fn apply_authoritative_metadata(session: &mut Session, latest: &Session) {
if session.kind == bamboo_domain::SessionKind::Child
&& latest.kind == bamboo_domain::SessionKind::Child
&& session.created_at == latest.created_at
{
let latest_launch = latest
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_launch_generation);
let incoming_launch = session
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_launch_generation);
if latest_launch > incoming_launch {
session
.runtime_metadata
.get_or_insert_with(Default::default)
.child_launch_generation = latest_launch;
}
let latest_cancelled = latest
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_cancelled_generation);
let incoming_cancelled = session
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_cancelled_generation);
if latest_cancelled > incoming_cancelled {
session
.runtime_metadata
.get_or_insert_with(Default::default)
.child_cancelled_generation = latest_cancelled;
}
let latest_intent = latest
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_auto_run_launch_intent_generation);
let incoming_intent = session
.runtime_metadata
.as_ref()
.and_then(|metadata| metadata.child_auto_run_launch_intent_generation);
if latest_intent > incoming_intent {
session
.runtime_metadata
.get_or_insert_with(Default::default)
.child_auto_run_launch_intent_generation = latest_intent;
}
}
if session.authority_identity.is_ordinary() && session.created_at == latest.created_at {
session.authority_identity = latest.authority_identity.clone();
}
if session.kind == bamboo_domain::SessionKind::Root
&& latest.kind == bamboo_domain::SessionKind::Root
{
let _ = session.adopt_root_tool_authority_from(latest);
}
if session.kind == bamboo_domain::SessionKind::Root
&& latest.kind == bamboo_domain::SessionKind::Root
&& session.created_at == latest.created_at
&& session.authority_identity == latest.authority_identity
{
session.supervisor_management = latest.supervisor_management.clone();
}
if session.kind == bamboo_domain::SessionKind::Root
&& latest.kind == bamboo_domain::SessionKind::Root
&& session.created_at == latest.created_at
&& latest.metadata_version >= session.metadata_version
&& (latest.metadata_version > session.metadata_version
|| latest.project_id_meta() != session.project_id_meta())
{
match latest.project_id_meta() {
Some(project) => session.set_project_id_meta(project),
None => session.clear_project_id_meta(),
}
match latest.workspace_path_meta() {
Some(workspace) => session.set_workspace_path_meta(workspace),
None => {
session.metadata.remove("workspace_path");
if let Some(metadata) = session.runtime_metadata.as_mut() {
metadata.workspace_path = None;
}
}
}
session.workspace.clone_from(&latest.workspace);
for key in ROOT_PROJECT_CONTEXT_KEYS {
match latest.metadata.get(*key) {
Some(value) => {
session.metadata.insert((*key).to_string(), value.clone());
}
None => {
session.metadata.remove(*key);
}
}
}
session.prompt_snapshot.clone_from(&latest.prompt_snapshot);
}
if latest.metadata_version >= session.metadata_version {
session.title = latest.title.clone();
session.title_version = latest.title_version;
session.title_generated = latest.title_generated;
session.pinned = latest.pinned;
for key in AUTHORITATIVE_METADATA_KEYS {
if let Some(value) = latest.metadata.get(*key) {
session.metadata.insert((*key).to_string(), value.clone());
} else {
session.metadata.remove(*key);
}
}
session.metadata_version = latest.metadata_version;
}
}
pub async fn merge_save_session(
storage: &Arc<dyn Storage>,
session: &mut Session,
) -> std::io::Result<()> {
ensure_scoped_loop_checkpoint_known(session)?;
merge_authoritative_metadata_into_stale(storage, session).await?;
storage.save_session(session).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::v2::{RuntimeTaskTransactionFault, SessionStoreV2};
use bamboo_domain::{session::types::Session, PermissionMode};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
#[tokio::test]
async fn scoped_unknown_checkpoint_cannot_enter_generic_merge_or_publish_callbacks() {
let dir = tempfile::tempdir().unwrap();
let storage: Arc<dyn Storage> = Arc::new(
SessionStoreV2::new(dir.path().join("sessions"))
.await
.unwrap(),
);
let canonical = Session::new("scoped-unknown-merge", "model");
storage.save_session(&canonical).await.unwrap();
let before =
serde_json::to_value(storage.load_session(&canonical.id).await.unwrap().unwrap())
.unwrap();
let store = LockedSessionStore::new(storage.clone());
let published = AtomicUsize::new(0);
for mode in 0..4 {
let mut uncertain = canonical.clone();
uncertain.add_message(bamboo_domain::Message::user(
"Unconfirmed proposal must not be published.",
));
uncertain.metadata.insert(
bamboo_domain::SCOPED_LOOP_UNCERTAIN_KEY.into(),
"strict reload failed".into(),
);
let result = match mode {
0 => {
store
.merge_save_runtime_and_publish(&mut uncertain, |_, _| {
published.fetch_add(1, Ordering::SeqCst);
})
.await
}
1 => {
store
.save_runtime_only_and_publish(&mut uncertain, |_| {
published.fetch_add(1, Ordering::SeqCst);
})
.await
}
2 => {
store
.checkpoint_runtime_session_and_publish(&mut uncertain, |_, _| {
published.fetch_add(1, Ordering::SeqCst);
})
.await
}
_ => {
store
.merge_save_finalized_runtime_and_publish(&mut uncertain, |_, _| {
published.fetch_add(1, Ordering::SeqCst);
})
.await
}
};
assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);
assert!(uncertain
.metadata
.contains_key(bamboo_domain::SCOPED_LOOP_UNCERTAIN_KEY));
assert_eq!(published.load(Ordering::SeqCst), 0);
let after =
serde_json::to_value(storage.load_session(&canonical.id).await.unwrap().unwrap())
.unwrap();
assert_eq!(
after, before,
"mode {mode} wrote or projected an unknown candidate"
);
}
}
#[test]
fn authority_merge_updates_ordinary_cache_but_does_not_hide_stale_incarnation() {
let mut latest = Session::new(bamboo_domain::DEFAULT_SUPERVISOR_SESSION_ID, "model");
latest.authority_identity = bamboo_domain::SessionAuthorityIdentity::Supervisor {
incarnation_id: uuid::Uuid::new_v4(),
};
let mut stale = latest.clone();
stale.authority_identity = bamboo_domain::SessionAuthorityIdentity::Ordinary;
stale.metadata_version = 100;
apply_authoritative_metadata(&mut stale, &latest);
assert_eq!(stale.authority_identity, latest.authority_identity);
let old_identity = bamboo_domain::SessionAuthorityIdentity::Supervisor {
incarnation_id: uuid::Uuid::new_v4(),
};
stale.authority_identity = old_identity.clone();
apply_authoritative_metadata(&mut stale, &latest);
assert_eq!(stale.authority_identity, old_identity);
}
struct AuthoritySavePauseStorage {
inner: Arc<SessionStoreV2>,
calls: std::sync::atomic::AtomicUsize,
reached: tokio::sync::Barrier,
release: tokio::sync::Barrier,
}
#[tokio::test]
async fn root_project_merge_publishes_project_revision_and_workspace_together() {
for runtime_only in [false, true] {
for equal_revision in [false, true] {
let temp = tempfile::tempdir().unwrap();
let storage = Arc::new(SessionStoreV2::new(temp.path().into()).await.unwrap());
let mut stale = Session::new("root-project-merge", "model");
stale.set_project_id_meta("project-a");
stale.set_workspace_path_meta("/project-a");
stale.metadata.insert(
"runtime_prompt_snapshot".into(),
"old Project A prompt".into(),
);
stale.add_message(bamboo_domain::Message::user("Keep transcript"));
storage.save_session(&stale).await.unwrap();
let mut current = stale.clone();
current.metadata_version += 1;
current.set_project_id_meta("project-b");
current.set_workspace_path_meta("/project-b");
current.metadata.remove("runtime_prompt_snapshot");
current
.metadata
.insert("workspace_source".into(), "project_default".into());
current
.metadata
.insert("project_context_rendered".into(), "Project B".into());
storage.save_session(¤t).await.unwrap();
if equal_revision {
stale.metadata_version = current.metadata_version;
}
let locked = LockedSessionStore::new(storage.clone());
let published = AtomicBool::new(false);
let publish = |saved: &Session| {
assert!(!saved.metadata.contains_key("runtime_prompt_snapshot"));
assert_eq!(saved.project_id_meta().as_deref(), Some("project-b"));
assert_eq!(saved.workspace_path_meta().as_deref(), Some("/project-b"));
assert_eq!(saved.metadata_version, current.metadata_version);
assert_eq!(
saved.metadata.get("workspace_source").map(String::as_str),
Some("project_default")
);
assert_eq!(
saved
.metadata
.get("project_context_rendered")
.map(String::as_str),
Some("Project B")
);
published.store(true, Ordering::SeqCst);
};
if runtime_only {
locked
.save_runtime_only_and_publish(&mut stale, publish)
.await
.unwrap();
} else {
locked
.merge_save_runtime_and_publish(&mut stale, |saved, committed| {
assert!(committed);
publish(saved);
})
.await
.unwrap();
}
assert!(published.load(Ordering::SeqCst));
assert_eq!(stale.project_id_meta(), current.project_id_meta());
let loaded = storage.load_session(&stale.id).await.unwrap().unwrap();
assert_eq!(loaded.project_id_meta(), current.project_id_meta());
assert_eq!(loaded.messages.len(), 1);
storage.flush_search_index().await;
}
}
}
#[tokio::test]
async fn root_project_change_after_merge_read_rejects_without_cache_publication() {
for runtime_only in [false, true] {
let temp = tempfile::tempdir().unwrap();
let first = Arc::new(SessionStoreV2::new(temp.path().into()).await.unwrap());
let mut stale = Session::new("root-project-race", "model");
stale.set_project_id_meta("project-a");
stale.add_message(bamboo_domain::Message::user("Keep transcript"));
first.save_session(&stale).await.unwrap();
let second = SessionStoreV2::new(temp.path().into()).await.unwrap();
let mut current = stale.clone();
current.metadata_version += 1;
current.set_project_id_meta("project-b");
let paused = Arc::new(AuthoritySavePauseStorage {
inner: first.clone(),
calls: std::sync::atomic::AtomicUsize::new(0),
reached: tokio::sync::Barrier::new(2),
release: tokio::sync::Barrier::new(2),
});
let locked = LockedSessionStore::new(paused.clone());
let published = AtomicBool::new(false);
let save = async {
if runtime_only {
locked
.save_runtime_only_and_publish(&mut stale, |_| {
published.store(true, Ordering::SeqCst);
})
.await
} else {
locked
.merge_save_runtime_and_publish(&mut stale, |_, _| {
published.store(true, Ordering::SeqCst);
})
.await
}
};
let update = async {
paused.reached.wait().await;
second.save_session(¤t).await.unwrap();
paused.release.wait().await;
};
let (result, ()) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
tokio::join!(save, update)
})
.await
.expect("deterministic Project/save race completes");
assert!(!may_publish_runtime_result(&Err(result.unwrap_err())));
assert!(!published.load(Ordering::SeqCst));
let loaded = first.load_session(¤t.id).await.unwrap().unwrap();
assert_eq!(loaded.project_id_meta(), current.project_id_meta());
assert_eq!(loaded.metadata_version, current.metadata_version);
assert_eq!(loaded.messages.len(), 1);
first.flush_search_index().await;
second.flush_search_index().await;
}
}
#[async_trait::async_trait]
impl Storage for AuthoritySavePauseStorage {
async fn load_session(&self, id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(id).await
}
async fn load_runtime_control_plane(&self, id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_runtime_control_plane(id).await
}
async fn delete_session(&self, id: &str) -> std::io::Result<bool> {
self.inner.delete_session(id).await
}
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.calls.fetch_add(1, Ordering::SeqCst);
self.reached.wait().await;
self.release.wait().await;
self.inner.save_session(session).await
}
async fn save_runtime_state(&self, session: &Session) -> std::io::Result<()> {
self.calls.fetch_add(1, Ordering::SeqCst);
self.reached.wait().await;
self.release.wait().await;
self.inner.save_runtime_state(session).await
}
}
#[tokio::test]
async fn actor_claim_after_merge_read_rejects_once_without_cache_publication() {
use bamboo_domain::{ActorActivationClaim, ActorDirectoryPort};
for runtime_only in [false, true] {
let home = tempfile::tempdir().unwrap();
let first = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
let mut stale = Session::new("actor-cache-race", "model");
stale.add_message(bamboo_domain::Message::user("durable"));
first.save_session(&stale).await.unwrap();
let second = SessionStoreV2::new(home.path().into()).await.unwrap();
let paused = Arc::new(AuthoritySavePauseStorage {
inner: first.clone(),
calls: std::sync::atomic::AtomicUsize::new(0),
reached: tokio::sync::Barrier::new(2),
release: tokio::sync::Barrier::new(2),
});
let locked = LockedSessionStore::new(paused.clone());
let id = stale.id.clone();
let cached = Arc::new(std::sync::Mutex::new(stale.clone()));
let publications = std::sync::atomic::AtomicUsize::new(0);
stale.conversation_summary =
Some(bamboo_domain::ConversationSummary::new("reject", 1, 1));
let save = async {
if runtime_only {
locked
.save_runtime_only_and_publish(&mut stale, |saved| {
publications.fetch_add(1, Ordering::SeqCst);
*cached.lock().unwrap() = saved.clone();
})
.await
} else {
locked
.merge_save_runtime_and_publish(&mut stale, |saved, _| {
publications.fetch_add(1, Ordering::SeqCst);
*cached.lock().unwrap() = saved.clone();
})
.await
}
};
let activate = async {
paused.reached.wait().await;
let now = chrono::Utc::now();
second
.claim_activation(&ActorActivationClaim {
actor_id: id.clone(),
run_id: "claim".into(),
lease_owner: "owner".into(),
lease_expires_at: now + chrono::Duration::minutes(5),
inbox_generation: 0,
placement_ref: None,
now,
})
.await
.unwrap();
paused.release.wait().await;
};
let (result, ()) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
tokio::join!(save, activate)
})
.await
.expect("real final save/activation race terminates");
let error = result.unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::Unsupported);
assert!(!is_task_control_plane_save_conflict(&error));
assert!(!may_publish_runtime_result(&Err(error)));
assert_eq!(paused.calls.load(Ordering::SeqCst), 1);
assert_eq!(publications.load(Ordering::SeqCst), 0);
assert!(cached.lock().unwrap().conversation_summary.is_none());
assert!(first
.load_session(&id)
.await
.unwrap()
.unwrap()
.conversation_summary
.is_none());
}
}
#[tokio::test]
async fn root_deleted_after_merge_read_rejects_without_cache_or_event_publication() {
for runtime_only in [false, true] {
let temp = tempfile::tempdir().unwrap();
let first = Arc::new(SessionStoreV2::new(temp.path().into()).await.unwrap());
let mut stale = Session::new("root-delete-race", "model");
stale.set_project_id_meta("project-a");
stale.add_message(bamboo_domain::Message::user("Old lifetime"));
first.save_session(&stale).await.unwrap();
let id = stale.id.clone();
let second = SessionStoreV2::new(temp.path().into()).await.unwrap();
let paused = Arc::new(AuthoritySavePauseStorage {
inner: first.clone(),
calls: std::sync::atomic::AtomicUsize::new(0),
reached: tokio::sync::Barrier::new(2),
release: tokio::sync::Barrier::new(2),
});
let locked = LockedSessionStore::new(paused.clone());
let published = AtomicBool::new(false);
let save = async {
if runtime_only {
locked
.save_runtime_only_and_publish(&mut stale, |_| {
published.store(true, Ordering::SeqCst);
})
.await
} else {
locked
.merge_save_runtime_and_publish(&mut stale, |_, _| {
published.store(true, Ordering::SeqCst);
})
.await
}
};
let delete = async {
paused.reached.wait().await;
assert!(second.delete_session(&id).await.unwrap());
paused.release.wait().await;
};
let (result, ()) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
tokio::join!(save, delete)
})
.await
.expect("deterministic delete/save race completes");
assert!(!may_publish_runtime_result(&Err(result.unwrap_err())));
assert!(!published.load(Ordering::SeqCst));
assert!(first.load_root_authority(&id).await.unwrap().is_none());
assert!(!first.sessions_root_dir().join(&id).exists());
first.flush_search_index().await;
second.flush_search_index().await;
}
}
#[tokio::test]
async fn supervisor_bootstrap_between_merge_read_and_save_rejects_without_publishing() {
for runtime_only in [false, true] {
let temp = tempfile::tempdir().unwrap();
let first = Arc::new(SessionStoreV2::new(temp.path().into()).await.unwrap());
let second = SessionStoreV2::new(temp.path().into()).await.unwrap();
let paused = Arc::new(AuthoritySavePauseStorage {
inner: first.clone(),
calls: std::sync::atomic::AtomicUsize::new(0),
reached: tokio::sync::Barrier::new(2),
release: tokio::sync::Barrier::new(2),
});
let store = LockedSessionStore::new(paused.clone());
let mut stale = Session::new(bamboo_domain::DEFAULT_SUPERVISOR_SESSION_ID, "stale");
stale.add_message(bamboo_domain::Message::user(
"must not enter new Supervisor",
));
let published = AtomicBool::new(false);
let save = async {
if runtime_only {
store
.save_runtime_only_and_publish(&mut stale, |_| {
published.store(true, Ordering::SeqCst);
})
.await
} else {
store
.merge_save_runtime_and_publish(&mut stale, |_, _| {
published.store(true, Ordering::SeqCst);
})
.await
}
};
let bootstrap = async {
paused.reached.wait().await;
let receipt = second
.get_or_create_default_supervisor("supervisor")
.await
.unwrap();
paused.release.wait().await;
receipt
};
let (result, receipt) =
tokio::time::timeout(std::time::Duration::from_secs(10), async {
tokio::join!(save, bootstrap)
})
.await
.expect("deterministic bootstrap/save race completes");
let error = result.unwrap_err();
assert!(!may_publish_runtime_result(&Err(error)));
assert!(!published.load(Ordering::SeqCst));
assert!(stale.authority_identity.is_ordinary());
let observed = first
.load_root_authority(&receipt.session_id)
.await
.unwrap()
.unwrap();
let durable = second
.load_session(&receipt.session_id)
.await
.unwrap()
.unwrap();
assert_eq!(durable.model, "supervisor");
assert_eq!(observed.authority_identity, durable.authority_identity);
assert!(durable.messages.is_empty());
assert_eq!(
durable.authority_identity,
bamboo_domain::SessionAuthorityIdentity::Supervisor {
incarnation_id: receipt.incarnation_id,
}
);
}
}
#[tokio::test]
async fn supervisor_merge_publishes_adopted_identity_but_never_a_rejected_incarnation() {
for runtime_only in [false, true] {
let temp = tempfile::tempdir().unwrap();
let storage = Arc::new(SessionStoreV2::new(temp.path().into()).await.unwrap());
let receipt = storage
.get_or_create_default_supervisor("model")
.await
.unwrap();
let baseline = storage
.load_session(&receipt.session_id)
.await
.unwrap()
.unwrap();
let expected = baseline.authority_identity.clone();
let store = LockedSessionStore::new(storage.clone());
for case in 0..3 {
let rejected = case != 0;
let mut snapshot = baseline.clone();
snapshot.authority_identity = if case == 1 {
bamboo_domain::SessionAuthorityIdentity::Supervisor {
incarnation_id: uuid::Uuid::new_v4(),
}
} else {
bamboo_domain::SessionAuthorityIdentity::Ordinary
};
if case == 2 {
snapshot.created_at -= chrono::Duration::seconds(1);
snapshot.model = "stale Ordinary instance".into();
snapshot.add_message(bamboo_domain::Message::user("must not be rebound"));
}
let published = AtomicBool::new(false);
let callback = |saved: &Session| {
assert_eq!(saved.authority_identity, expected);
published.store(true, Ordering::SeqCst);
};
let result = if runtime_only {
store
.save_runtime_only_and_publish(&mut snapshot, callback)
.await
} else {
store
.merge_save_runtime_and_publish(&mut snapshot, |saved, committed| {
assert!(committed);
callback(saved);
})
.await
};
assert_eq!(result.is_err(), rejected);
assert_eq!(published.load(Ordering::SeqCst), !rejected);
if !rejected {
assert_eq!(snapshot.authority_identity, expected);
}
let durable = storage
.load_session(&receipt.session_id)
.await
.unwrap()
.unwrap();
assert_eq!(durable.authority_identity, expected);
assert_eq!(durable.created_at, baseline.created_at);
assert_eq!(durable.model, baseline.model);
assert!(durable.messages.is_empty());
}
}
}
struct CountingControlPlaneStorage {
inner: Arc<SessionStoreV2>,
control_plane_loads: AtomicUsize,
full_saves: AtomicUsize,
runtime_state_saves: AtomicUsize,
}
struct PairCommitBarrierStorage {
inner: Arc<SessionStoreV2>,
before_commit: Arc<tokio::sync::Barrier>,
}
struct SingleCommitBarrierStorage {
inner: Arc<SessionStoreV2>,
before_commit: Arc<tokio::sync::Barrier>,
}
struct SingleCommitPauseStorage {
inner: Arc<SessionStoreV2>,
commit_reached: Arc<tokio::sync::Barrier>,
release_commit: Arc<tokio::sync::Barrier>,
}
#[async_trait::async_trait]
impl Storage for SingleCommitPauseStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.inner.save_session(session).await
}
async fn load_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(session_id).await
}
async fn delete_session(&self, session_id: &str) -> std::io::Result<bool> {
self.inner.delete_session(session_id).await
}
async fn load_runtime_control_plane(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.inner.load_runtime_control_plane(session_id).await
}
async fn save_task_control_plane_if_matches(
&self,
original: &Session,
updated: &Session,
) -> std::io::Result<bool> {
self.commit_reached.wait().await;
self.release_commit.wait().await;
self.inner
.save_task_control_plane_if_matches(original, updated)
.await
}
async fn save_task_control_planes_atomically(
&self,
first_original: &Session,
first_updated: &Session,
second_original: &Session,
second_updated: &Session,
) -> std::io::Result<bool> {
self.commit_reached.wait().await;
self.release_commit.wait().await;
self.inner
.save_task_control_planes_atomically(
first_original,
first_updated,
second_original,
second_updated,
)
.await
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn supervisor_management_race_rejects_staged_task_callbacks_and_fresh_invocation_succeeds(
) {
use bamboo_domain::{
SupervisorManagementMutation, SupervisorManagementRequest, SupervisorReference,
};
for mode in 0..4 {
let home = tempfile::tempdir().unwrap();
let inner = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
let receipt = inner
.get_or_create_default_supervisor("model")
.await
.unwrap();
let reference = SupervisorReference::from(&receipt);
let mut root = inner
.load_session(&reference.session_id)
.await
.unwrap()
.unwrap();
let list = bamboo_domain::TaskList {
session_id: root.id.clone(),
title: "original".into(),
items: vec![],
created_at: root.created_at,
updated_at: root.created_at,
};
let updated = bamboo_domain::TaskList {
title: "updated".into(),
..list.clone()
};
root.task_list = Some(list.clone());
root.set_task_list_version_meta("1");
inner.save_session(&root).await.unwrap();
let child_id = if mode == 2 { "aaa-child" } else { "zzz-child" };
let mut child = Session::new_child_of(child_id, &root, "model", "child");
child.task_list = Some(list.clone());
child.set_task_list_version_meta("1");
inner.save_session(&child).await.unwrap();
let independent = SessionStoreV2::new(home.path().into()).await.unwrap();
let commit_reached = Arc::new(tokio::sync::Barrier::new(2));
let release_commit = Arc::new(tokio::sync::Barrier::new(2));
let paused = LockedSessionStore::new(Arc::new(SingleCommitPauseStorage {
inner: inner.clone(),
commit_reached: commit_reached.clone(),
release_commit: release_commit.clone(),
}));
let published = AtomicBool::new(false);
let loser = async {
if mode == 0 {
paused
.update_task_list_control_plane_and_publish(&root.id, &updated, "2", |_| {
published.store(true, Ordering::SeqCst)
})
.await
} else if mode == 1 {
paused
.update_task_list_control_plane_if_version_and_publish(
&root.id,
"1",
&list,
&updated,
"2",
|_| published.store(true, Ordering::SeqCst),
)
.await
} else {
paused
.update_task_list_control_planes_if_version_and_publish(
child_id,
&root.id,
"1",
&list,
&updated,
"2",
|_, _| published.store(true, Ordering::SeqCst),
)
.await
}
};
let winner = async {
commit_reached.wait().await;
independent
.mutate_supervisor_management(&SupervisorManagementRequest {
supervisor: reference.clone(),
expected_state_revision: 0,
mutation: SupervisorManagementMutation::ConfigureProjectScope {
allowed_projects: ["project-a".parse().unwrap()].into(),
},
})
.await
.unwrap();
release_commit.wait().await;
};
let (result, ()) = tokio::join!(loser, winner);
if mode == 0 {
assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);
} else {
assert!(!result.unwrap());
}
assert!(!published.load(Ordering::SeqCst));
for id in [&root.id, &child.id] {
assert_eq!(
inner
.load_runtime_control_plane(id)
.await
.unwrap()
.unwrap()
.task_list_version_meta()
.as_deref(),
Some("1")
);
}
let fresh = LockedSessionStore::new(inner.clone());
let check = |saved: &Session| {
assert_eq!(saved.supervisor_management.as_ref().unwrap().revision, 1);
published.store(true, Ordering::SeqCst);
};
let retried = if mode == 0 {
fresh
.update_task_list_control_plane_and_publish(&root.id, &updated, "2", check)
.await
} else if mode == 1 {
fresh
.update_task_list_control_plane_if_version_and_publish(
&root.id, "1", &list, &updated, "2", check,
)
.await
} else {
fresh
.update_task_list_control_planes_if_version_and_publish(
child_id,
&root.id,
"1",
&list,
&updated,
"2",
|_, shared| check(shared),
)
.await
};
assert!(retried.unwrap());
assert!(published.load(Ordering::SeqCst));
assert_eq!(
inner
.load_runtime_control_plane(&root.id)
.await
.unwrap()
.unwrap()
.task_list_version_meta()
.as_deref(),
Some("2")
);
}
}
#[async_trait::async_trait]
impl Storage for SingleCommitBarrierStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.inner.save_session(session).await
}
async fn load_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(session_id).await
}
async fn delete_session(&self, session_id: &str) -> std::io::Result<bool> {
self.inner.delete_session(session_id).await
}
async fn load_runtime_control_plane(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.inner.load_runtime_control_plane(session_id).await
}
async fn save_task_control_plane_if_matches(
&self,
original: &Session,
updated: &Session,
) -> std::io::Result<bool> {
self.before_commit.wait().await;
self.inner
.save_task_control_plane_if_matches(original, updated)
.await
}
}
#[async_trait::async_trait]
impl Storage for PairCommitBarrierStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.inner.save_session(session).await
}
async fn load_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(session_id).await
}
async fn delete_session(&self, session_id: &str) -> std::io::Result<bool> {
self.inner.delete_session(session_id).await
}
async fn save_runtime_state(&self, session: &Session) -> std::io::Result<()> {
self.inner.save_runtime_state(session).await
}
async fn load_runtime_control_plane(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.inner.load_runtime_control_plane(session_id).await
}
async fn recover_task_control_plane_transaction(
&self,
first_session_id: &str,
second_session_id: &str,
) -> std::io::Result<()> {
self.inner
.recover_task_control_plane_transaction(first_session_id, second_session_id)
.await
}
async fn save_task_control_planes_atomically(
&self,
first_original: &Session,
first_updated: &Session,
second_original: &Session,
second_updated: &Session,
) -> std::io::Result<bool> {
self.before_commit.wait().await;
self.inner
.save_task_control_planes_atomically(
first_original,
first_updated,
second_original,
second_updated,
)
.await
}
}
#[async_trait::async_trait]
impl Storage for CountingControlPlaneStorage {
async fn save_session(&self, session: &Session) -> std::io::Result<()> {
self.full_saves.fetch_add(1, Ordering::SeqCst);
self.inner.save_session(session).await
}
async fn load_session(&self, session_id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(session_id).await
}
async fn delete_session(&self, session_id: &str) -> std::io::Result<bool> {
self.inner.delete_session(session_id).await
}
async fn save_runtime_state(&self, session: &Session) -> std::io::Result<()> {
self.runtime_state_saves.fetch_add(1, Ordering::SeqCst);
self.inner.save_runtime_state(session).await
}
async fn load_runtime_control_plane(
&self,
session_id: &str,
) -> std::io::Result<Option<Session>> {
self.control_plane_loads.fetch_add(1, Ordering::SeqCst);
self.inner.load_runtime_control_plane(session_id).await
}
async fn recover_task_control_plane_transaction(
&self,
first_session_id: &str,
second_session_id: &str,
) -> std::io::Result<()> {
self.inner
.recover_task_control_plane_transaction(first_session_id, second_session_id)
.await
}
async fn save_task_control_plane_if_matches(
&self,
original: &Session,
updated: &Session,
) -> std::io::Result<bool> {
let committed = self
.inner
.save_task_control_plane_if_matches(original, updated)
.await?;
if committed {
self.runtime_state_saves.fetch_add(1, Ordering::SeqCst);
}
Ok(committed)
}
async fn save_task_control_planes_atomically(
&self,
first_original: &Session,
first_updated: &Session,
second_original: &Session,
second_updated: &Session,
) -> std::io::Result<bool> {
let committed = self
.inner
.save_task_control_planes_atomically(
first_original,
first_updated,
second_original,
second_updated,
)
.await?;
if committed {
self.runtime_state_saves.fetch_add(2, Ordering::SeqCst);
}
Ok(committed)
}
}
async fn make_storage() -> (tempfile::TempDir, Arc<dyn Storage>) {
let temp = tempfile::tempdir().unwrap();
let storage = SessionStoreV2::new(temp.path().to_path_buf())
.await
.expect("storage init");
(temp, Arc::new(storage) as Arc<dyn Storage>)
}
fn fresh(id: &str) -> Session {
Session::new(id.to_string(), "test-model".to_string())
}
fn stage_retrieval_window_archive(expected: &Session, message_index: usize) -> Session {
let mut staged = expected.clone();
let mut event = bamboo_domain::CompressionEvent::new(
1,
1,
80.0,
60.0,
0,
bamboo_domain::CompressionTriggerType::Auto,
0.0,
None,
0,
);
event.kind = bamboo_domain::CompressionEventKind::RetrievalWindow;
let event_id = event.id.clone();
staged.messages[message_index].compressed = true;
staged.messages[message_index].compressed_by_event_id = Some(event_id);
staged.compression_events.push(event);
staged.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::Compression);
staged
}
fn message_summary_fixture(id: &str) -> (Session, Session) {
use bamboo_domain::{
CompressionEvent, CompressionEventKind, CompressionTriggerType, ConversationSummary,
FunctionCall, Message, ModelContextResetReason, ToolCall,
};
let mut expected = fresh(id);
expected.add_message(Message::system("Keep system policy"));
expected.add_message(Message::user("Historical request"));
let mut assistant = Message::assistant("Historical assistant prose", None);
assistant.tool_calls = Some(vec![ToolCall {
id: "summary-tool-call".into(),
tool_type: "function".into(),
function: FunctionCall {
name: "skills_read".into(),
arguments: "{\"path\":\"skill.md\"}".into(),
},
}]);
expected.add_message(assistant);
expected.add_message(Message::tool_result("summary-tool-call", "Raw tool output"));
let mut recovery = Message::assistant(
"[post-compaction-recovery]\nOld tool-derived recovery",
None,
);
recovery.never_compress = true;
expected.add_message(recovery);
let mut anchor = Message::assistant("Protected prose anchor", None);
anchor.never_compress = true;
expected.add_message(anchor);
expected.add_message(Message::user("Current user request"));
expected.force_manual_compression = Some("Keep the user intent".into());
expected.metadata.insert(
RESPONSES_PREVIOUS_RESPONSE_ID_KEY.into(),
"response-before-summary".into(),
);
let boundary =
bamboo_domain::provider_transcript_boundary_sha256(Some("test"), Some("openai"))
.unwrap();
expected
.activate_provider_transcript_route(
bamboo_domain::ProviderFamily::OpenAi,
bamboo_domain::ProviderProtocol::OpenAiResponsesV1,
&boundary,
)
.unwrap();
let mut staged = expected.clone();
let mut event = CompressionEvent::new(
4,
1,
85.0,
40.0,
20,
CompressionTriggerType::Manual,
2.0,
None,
0,
);
event.kind = CompressionEventKind::MessageSummary;
for message in &mut staged.messages[1..5] {
message.compressed = true;
message.compressed_by_event_id = Some(event.id.clone());
}
staged.compression_events.push(event);
staged.conversation_summary = Some(ConversationSummary::new(
"User and assistant prose summary",
4,
20,
));
staged.force_manual_compression = None;
staged.metadata.remove(RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
staged.reset_model_context_epoch(ModelContextResetReason::Compression);
(expected, staged)
}
struct FailingMessageSummaryStorage {
inner: Arc<dyn Storage>,
attempted_saves: AtomicUsize,
}
#[async_trait::async_trait]
impl Storage for FailingMessageSummaryStorage {
async fn save_session(&self, _: &Session) -> std::io::Result<()> {
self.attempted_saves.fetch_add(1, Ordering::SeqCst);
Err(std::io::Error::other("injected archive save failure"))
}
async fn load_session(&self, id: &str) -> std::io::Result<Option<Session>> {
self.inner.load_session(id).await
}
async fn delete_session(&self, id: &str) -> std::io::Result<bool> {
self.inner.delete_session(id).await
}
}
fn typed_permission_result(
tool_call_id: &str,
message_id: &str,
generation: &str,
content: &str,
) -> bamboo_domain::session::types::Message {
let mut message =
bamboo_domain::session::types::Message::tool_result(tool_call_id, content);
message.id = message_id.to_string();
message.metadata = Some(serde_json::json!({
"permission_request": {
"request_generation": generation,
}
}));
message
}
fn ledger_state(state_revision: u64, marker: &str) -> bamboo_domain::ModelContextState {
bamboo_domain::ModelContextState {
state_revision,
prefix_epoch: state_revision,
cache_scope_sha256: Some("scope".to_string()),
transcript_item_sha256: vec![marker.to_string()],
..bamboo_domain::ModelContextState::default()
}
}
fn set_permission_audit(
session: &mut Session,
requested: bamboo_domain::SessionPermissionMode,
policy_revision: u64,
mapping: &str,
transitioned_at: &str,
) -> u64 {
let resolution = bamboo_domain::resolve_permission_mode(requested, PermissionMode::Default);
session
.agent_runtime_state
.get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
.set_permission_mode(requested);
bamboo_domain::record_permission_audit(
&mut session.metadata,
&PermissionAuditSeed::new(policy_revision, resolution, mapping),
Some(transitioned_at),
)
.unwrap()
}
#[tokio::test]
async fn checked_runtime_mutation_persists_a_cache_only_pending_session() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let mut cached = fresh("cache-only-response");
cached.set_pending_question(
"tool-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
let saved = store
.mutate_runtime_session_and_publish(
&cached.id.clone(),
move || Some(cached),
|session| {
assert!(session.pending_question.is_some());
session.clear_pending_question();
Ok::<_, ()>(())
},
|_| {},
)
.await
.unwrap()
.unwrap()
.expect("cache-only session should be created durably");
assert!(saved.pending_question.is_none());
assert!(storage
.load_session("cache-only-response")
.await
.unwrap()
.unwrap()
.pending_question
.is_none());
}
#[tokio::test]
async fn checked_runtime_mutation_preserves_durable_authorities_for_newer_cache() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "cached-response-authorities";
let mut durable = fresh(session_id);
durable.title = "Durable title".to_string();
durable.title_version = 4;
durable.title_generated = true;
durable.metadata_version = 9;
set_permission_audit(
&mut durable,
bamboo_domain::SessionPermissionMode::Auto,
7,
"bamboo_runtime:durable-auto",
"2026-08-10T09:00:00Z",
);
storage.save_session(&durable).await.unwrap();
let mut cached = fresh(session_id);
cached.created_at = durable.created_at;
cached.title = "Stale cached title".to_string();
cached.updated_at = durable.updated_at + chrono::Duration::seconds(1);
cached.set_pending_question(
"tool-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
store
.mutate_runtime_session_and_publish(
session_id,
move || Some(cached),
|session| {
session.clear_pending_question();
Ok::<_, ()>(())
},
|_| {},
)
.await
.unwrap()
.unwrap()
.expect("session should exist");
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(saved.title, "Durable title");
assert_eq!(saved.title_version, 4);
assert_eq!(saved.metadata_version, 9);
assert_eq!(
saved
.agent_runtime_state
.as_ref()
.unwrap()
.effective_permission_mode(),
bamboo_domain::SessionPermissionMode::Auto
);
let audit = bamboo_domain::PermissionAuditSnapshot::from_metadata(&saved.metadata).unwrap();
assert_eq!(audit.policy_revision, 7);
assert_eq!(audit.executor_mapping, "bamboo_runtime:durable-auto");
}
#[tokio::test]
async fn checked_runtime_mutation_cannot_resurrect_consumed_ask_from_newer_cache() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "cached-consumed-response";
let mut stale_cached = fresh(session_id);
stale_cached.add_message(Message::tool_result("call-1", "waiting"));
stale_cached.set_pending_question(
"call-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
let mut durable = stale_cached.clone();
durable.clear_pending_question();
durable.metadata.insert(
CONSUMED_CLARIFICATION_IDS_KEY.to_string(),
r#"["call-1"]"#.to_string(),
);
durable.messages[0].content = "Selected response: A".to_string();
durable.add_message(Message::user("durable concurrent message"));
storage.save_session(&durable).await.unwrap();
stale_cached.updated_at = durable.updated_at + chrono::Duration::seconds(1);
let saved = store
.mutate_runtime_session_and_publish(
session_id,
move || Some(stale_cached),
|session| {
assert!(session.pending_question.is_none());
Ok::<_, ()>(())
},
|_| {},
)
.await
.unwrap()
.unwrap()
.unwrap();
assert!(saved.pending_question.is_none());
assert_eq!(saved.messages.len(), 2);
assert_eq!(saved.messages[0].content, "Selected response: A");
assert_eq!(saved.messages[1].content, "durable concurrent message");
}
#[tokio::test]
async fn response_inspection_adopts_durable_consumption_without_writing() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "inspect-consumed-response";
let mut stale_cached = fresh(session_id);
stale_cached.add_message(bamboo_domain::session::types::Message::tool_result(
"call-1", "waiting",
));
stale_cached.set_pending_question(
"call-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
let mut durable = stale_cached.clone();
durable.clear_pending_question();
durable.metadata.insert(
CONSUMED_CLARIFICATION_IDS_KEY.to_string(),
r#"["call-1"]"#.to_string(),
);
storage.save_session(&durable).await.unwrap();
stale_cached.updated_at = durable.updated_at + chrono::Duration::seconds(1);
let inspected = store
.inspect_runtime_session_for_response(session_id, move || Some(stale_cached))
.await
.unwrap()
.expect("session should be inspectable");
assert!(inspected.pending_question.is_none());
let unchanged = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(unchanged.updated_at, durable.updated_at);
assert!(unchanged.pending_question.is_none());
}
#[tokio::test]
async fn same_mode_newer_run_start_audit_survives_every_runtime_save_path() {
for path in ["merge", "checkpoint", "control-plane"] {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = format!("same-mode-newer-{path}");
let mut durable = fresh(&session_id);
set_permission_audit(
&mut durable,
bamboo_domain::SessionPermissionMode::Default,
1,
"bamboo_runtime:old-policy",
"2026-07-31T12:00:00Z",
);
storage.save_session(&durable).await.unwrap();
let mut run_start = durable.clone();
let old_revision = PermissionAuditSnapshot::from_metadata(&durable.metadata)
.unwrap()
.audit_revision;
let new_revision = set_permission_audit(
&mut run_start,
bamboo_domain::SessionPermissionMode::Default,
2,
"bamboo_runtime:new-policy",
"2026-07-31T12:00:00Z",
);
assert!(new_revision > old_revision);
match path {
"merge" => store.merge_save_runtime(&mut run_start).await.unwrap(),
"checkpoint" => store
.checkpoint_runtime_session(&mut run_start)
.await
.unwrap(),
"control-plane" => store.save_runtime_only(&mut run_start).await.unwrap(),
_ => unreachable!(),
}
let saved = storage.load_session(&session_id).await.unwrap().unwrap();
let audit = PermissionAuditSnapshot::from_metadata(&saved.metadata).unwrap();
assert_eq!(audit.audit_revision, new_revision, "path={path}");
assert_eq!(audit.policy_revision, 2, "path={path}");
assert_eq!(audit.executor_mapping, "bamboo_runtime:new-policy");
}
}
#[tokio::test]
async fn newer_disk_transition_wins_after_mode_cycles_back() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "permission-cycle-back";
let mut baseline = fresh(session_id);
let stale_revision = set_permission_audit(
&mut baseline,
bamboo_domain::SessionPermissionMode::Default,
1,
"bamboo_runtime:initial",
"2026-07-31T12:00:00Z",
);
storage.save_session(&baseline).await.unwrap();
let mut stale_runtime = baseline.clone();
let mut durable = baseline;
set_permission_audit(
&mut durable,
bamboo_domain::SessionPermissionMode::Auto,
2,
"bamboo_runtime:auto",
"2026-07-31T12:01:00Z",
);
let durable_revision = set_permission_audit(
&mut durable,
bamboo_domain::SessionPermissionMode::Default,
3,
"bamboo_runtime:cycled-default",
"2026-07-31T12:02:00Z",
);
assert!(durable_revision > stale_revision);
storage.save_session(&durable).await.unwrap();
store.merge_save_runtime(&mut stale_runtime).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
let audit = PermissionAuditSnapshot::from_metadata(&saved.metadata).unwrap();
assert_eq!(audit.audit_revision, durable_revision);
assert_eq!(audit.policy_revision, 3);
assert_eq!(audit.executor_mapping, "bamboo_runtime:cycled-default");
}
#[tokio::test]
async fn authoritative_activation_seed_replaces_every_warm_worker_posture() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "warm-permission-matrix";
let cases = [
(
bamboo_domain::SessionPermissionMode::Auto,
PermissionMode::Default,
PermissionMode::Auto,
),
(
bamboo_domain::SessionPermissionMode::Default,
PermissionMode::Default,
PermissionMode::Default,
),
(
bamboo_domain::SessionPermissionMode::Auto,
PermissionMode::Default,
PermissionMode::Auto,
),
(
bamboo_domain::SessionPermissionMode::Bypass,
PermissionMode::Auto,
PermissionMode::BypassPermissions,
),
];
let mut previous_revision = 0;
let created_at = fresh(session_id).created_at;
for (index, (requested, configured, expected_effective)) in cases.into_iter().enumerate() {
let mut activation = fresh(session_id);
activation.created_at = created_at;
activation
.agent_runtime_state
.get_or_insert_with(bamboo_domain::AgentRuntimeState::default)
.set_permission_mode(requested);
let resolution = bamboo_domain::resolve_permission_mode(requested, configured);
bamboo_domain::record_permission_audit(
&mut activation.metadata,
&PermissionAuditSeed::new(
index as u64 + 1,
resolution,
format!("bamboo_worker:{}", resolution.effective.as_str()),
),
Some("2026-07-31T12:00:00Z"),
)
.unwrap();
RuntimeSessionPersistence::seed_runtime_activation(&store, &mut activation)
.await
.unwrap();
let durable = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
durable
.agent_runtime_state
.as_ref()
.unwrap()
.effective_permission_mode(),
requested,
"activation {index}"
);
let audit = PermissionAuditSnapshot::from_metadata(&durable.metadata).unwrap();
assert_eq!(audit.resolution.requested, requested);
assert_eq!(audit.resolution.effective, expected_effective);
assert!(audit.audit_revision > previous_revision);
previous_revision = audit.audit_revision;
}
}
#[tokio::test]
async fn resident_reseed_bumps_etag_only_for_typed_transition() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "resident-atomic-permission";
let mut baseline = fresh(session_id);
baseline.metadata_version = 7;
set_permission_audit(
&mut baseline,
bamboo_domain::SessionPermissionMode::Auto,
1,
"bamboo_runtime:auto",
"2026-07-31T12:00:00Z",
);
storage.save_session(&baseline).await.unwrap();
let initial_audit = PermissionAuditSnapshot::from_metadata(&baseline.metadata).unwrap();
let same_mode_seed = PermissionAuditSeed::bamboo_runtime(
2,
bamboo_domain::resolve_permission_mode(
bamboo_domain::SessionPermissionMode::Auto,
PermissionMode::Default,
),
);
let refreshed = store
.update_authoritative_permission_posture_and_publish(
session_id,
&same_mode_seed,
|session| {
session
.metadata
.insert("resident.marker".to_string(), "same-mode".to_string());
},
|_| {},
)
.await
.unwrap()
.unwrap();
let refreshed_audit = PermissionAuditSnapshot::from_metadata(&refreshed.metadata).unwrap();
assert_eq!(refreshed.metadata_version, 7);
assert!(refreshed_audit.audit_revision > initial_audit.audit_revision);
assert_eq!(refreshed_audit.policy_revision, 2);
let transition_seed = PermissionAuditSeed::bamboo_runtime(
3,
bamboo_domain::resolve_permission_mode(
bamboo_domain::SessionPermissionMode::Default,
PermissionMode::Default,
),
);
let transitioned = store
.update_authoritative_permission_posture_and_publish(
session_id,
&transition_seed,
|session| {
session
.metadata
.insert("resident.marker".to_string(), "transition".to_string());
},
|_| {},
)
.await
.unwrap()
.unwrap();
let transitioned_audit =
PermissionAuditSnapshot::from_metadata(&transitioned.metadata).unwrap();
assert_eq!(transitioned.metadata_version, 8, "old ETag must be invalid");
assert_eq!(
transitioned
.agent_runtime_state
.as_ref()
.unwrap()
.effective_permission_mode(),
bamboo_domain::SessionPermissionMode::Default
);
assert_eq!(
transitioned_audit.resolution.requested,
bamboo_domain::SessionPermissionMode::Default
);
assert!(transitioned_audit.audit_revision > refreshed_audit.audit_revision);
assert_eq!(
transitioned
.metadata
.get("resident.marker")
.map(String::as_str),
Some("transition")
);
}
#[tokio::test]
async fn worker_activation_cas_cannot_overwrite_concurrent_permission_patch() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "permission-activation-cas";
let mut baseline = fresh(session_id);
set_permission_audit(
&mut baseline,
bamboo_domain::SessionPermissionMode::Default,
1,
"bamboo_runtime:default",
"2026-07-31T12:00:00Z",
);
storage.save_session(&baseline).await.unwrap();
let dispatched_revision = PermissionAuditSnapshot::from_metadata(&baseline.metadata)
.unwrap()
.audit_revision;
let patched_resolution = bamboo_domain::resolve_permission_mode(
bamboo_domain::SessionPermissionMode::Auto,
PermissionMode::Default,
);
let patched = store
.update_authoritative_permission_posture_and_publish(
session_id,
&PermissionAuditSeed::new(2, patched_resolution, "patch:auto"),
|_| {},
|_| {},
)
.await
.unwrap()
.unwrap();
let patched_audit = PermissionAuditSnapshot::from_metadata(&patched.metadata).unwrap();
assert!(patched_audit.audit_revision > dispatched_revision);
let stale_worker_seed = PermissionAuditSeed::new(
1,
bamboo_domain::resolve_permission_mode(
bamboo_domain::SessionPermissionMode::Default,
PermissionMode::Default,
),
"worker:stale-default",
);
let error = store
.record_permission_posture_activation_and_publish(
session_id,
Some(dispatched_revision),
&stale_worker_seed,
|_| {},
)
.await
.unwrap_err();
assert!(error.to_string().contains("durable audit changed"));
let durable = storage.load_session(session_id).await.unwrap().unwrap();
let durable_audit = PermissionAuditSnapshot::from_metadata(&durable.metadata).unwrap();
assert_eq!(durable_audit, patched_audit);
assert_eq!(durable_audit.executor_mapping, "patch:auto");
}
#[tokio::test]
async fn update_runtime_config_preserves_concurrently_appended_messages() {
use bamboo_domain::session::types::Message;
use bamboo_domain::ReasoningEffort;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "cfg-preserve";
let mut initial = fresh(session_id);
initial.add_message(Message::user("hello"));
initial.add_message(Message::assistant("hi", None));
storage.save_session(&initial).await.unwrap();
let mut after_chat = storage.load_session(session_id).await.unwrap().unwrap();
after_chat.add_message(Message::user("second question"));
storage.save_session(&after_chat).await.unwrap();
assert_eq!(after_chat.messages.len(), 3);
let updated = store
.update_runtime_config(session_id, |s| {
s.reasoning_effort = Some(ReasoningEffort::Max);
})
.await
.unwrap()
.expect("session exists");
assert_eq!(updated.reasoning_effort, Some(ReasoningEffort::Max));
assert_eq!(
updated.messages.len(),
3,
"config patch must not revert a concurrently-appended message"
);
let on_disk = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(on_disk.messages.len(), 3);
assert_eq!(on_disk.reasoning_effort, Some(ReasoningEffort::Max));
}
#[tokio::test]
async fn update_runtime_config_returns_none_for_missing_session() {
use bamboo_domain::ReasoningEffort;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage);
let result = store
.update_runtime_config("does-not-exist", |s| {
s.reasoning_effort = Some(ReasoningEffort::Low);
})
.await
.unwrap();
assert!(result.is_none());
}
#[tokio::test]
async fn merge_save_runtime_overwrites_messages_from_stale_snapshot() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "stale-clobber";
let mut baseline = fresh(session_id);
baseline.add_message(Message::user("hello"));
storage.save_session(&baseline).await.unwrap();
let mut stale_snapshot = storage.load_session(session_id).await.unwrap().unwrap();
let mut after_chat = storage.load_session(session_id).await.unwrap().unwrap();
after_chat.add_message(Message::user("second"));
storage.save_session(&after_chat).await.unwrap();
assert_eq!(
storage
.load_session(session_id)
.await
.unwrap()
.unwrap()
.messages
.len(),
2
);
store.merge_save_runtime(&mut stale_snapshot).await.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
after.messages.len(),
1,
"merge_save_runtime clobbers concurrent appends — this is why config patches must use update_runtime_config"
);
}
#[tokio::test]
async fn stale_runtime_save_cannot_remove_admitted_inbox_transcript() {
use bamboo_domain::session::types::Message;
use bamboo_domain::SessionMessageId;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "stale-inbox-preserve";
let mut baseline = fresh(session_id);
let mut base = Message::user("base");
base.id = "base".to_string();
baseline.add_message(base);
storage.save_session(&baseline).await.unwrap();
let mut stale = baseline.clone();
let mut later_assistant = Message::assistant("runner output", None);
later_assistant.id = "later-assistant".to_string();
stale.add_message(later_assistant);
let mut durable = baseline;
let inbox_id = SessionMessageId::parse("durable-inbox-id").unwrap();
let mut admitted = Message::user("durable inbox message");
admitted.id = inbox_id.as_str().to_string();
durable.add_message(admitted);
durable
.session_inbox_admission_mut()
.record(inbox_id.clone(), 7);
storage.save_session(&durable).await.unwrap();
store.merge_save_runtime(&mut stale).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
let ids = saved
.messages
.iter()
.map(|message| message.id.as_str())
.collect::<Vec<_>>();
assert_eq!(ids, vec!["base", "durable-inbox-id", "later-assistant"]);
assert_eq!(ids.iter().filter(|id| **id == inbox_id.as_str()).count(), 1);
assert!(saved
.session_inbox_admission()
.is_some_and(|state| state.contains(&inbox_id)));
}
#[tokio::test]
async fn stale_runtime_save_preserves_typed_inbox_message_after_cursor_eviction() {
use bamboo_domain::{
SessionMessageEnvelope, SessionMessageId, SESSION_INBOX_ADMITTED_CAPACITY,
};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "evicted-inbox-preserve";
let mut durable = fresh(session_id);
let mut envelope = SessionMessageEnvelope::user_input(session_id, "old durable inbox");
envelope.id = SessionMessageId::parse("old-inbox-id").unwrap();
durable.add_message(envelope.to_provider_message().unwrap());
durable
.session_inbox_admission_mut()
.record(envelope.id.clone(), 1);
for sequence in 2..=(SESSION_INBOX_ADMITTED_CAPACITY as u64 + 1) {
durable.session_inbox_admission_mut().record(
SessionMessageId::parse(format!("newer-{sequence}")).unwrap(),
sequence,
);
}
assert!(!durable
.session_inbox_admission()
.unwrap()
.contains(&envelope.id));
storage.save_session(&durable).await.unwrap();
let mut stale = fresh(session_id);
stale.created_at = durable.created_at;
store.merge_save_runtime(&mut stale).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
saved
.messages
.iter()
.filter(|message| message.id == envelope.id.as_str())
.count(),
1
);
}
#[tokio::test]
async fn runtime_final_save_cannot_resurrect_a_consumed_clarification() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "consumed-clarification-final-save";
let mut suspended = fresh(session_id);
suspended.add_message(Message::tool_result(
"call-1",
r#"{"status":"awaiting_clarification"}"#,
));
suspended.set_pending_question(
"call-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
suspended.metadata.insert(
"runtime.suspend_reason".to_string(),
"awaiting_clarification".to_string(),
);
storage.save_session(&suspended).await.unwrap();
let mut stale_runner = suspended.clone();
let mut answered = suspended;
answered.clear_pending_question();
answered.metadata.remove("runtime.suspend_reason");
answered.metadata.insert(
CONSUMED_CLARIFICATION_IDS_KEY.to_string(),
r#"["call-1"]"#.to_string(),
);
answered.metadata.insert(
"clarification_resume_pending".to_string(),
"true".to_string(),
);
answered.metadata.insert(
"conclusion_with_options_resume_pending".to_string(),
"true".to_string(),
);
answered.metadata.insert(
"execute.startup_handoff_at".to_string(),
"2026-08-10T09:00:00.000Z".to_string(),
);
let answer = answered
.messages
.iter_mut()
.find(|message| message.tool_call_id.as_deref() == Some("call-1"))
.unwrap();
answer.content = "Selected response: A".to_string();
storage.save_session(&answered).await.unwrap();
store.merge_save_runtime(&mut stale_runner).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert!(saved.pending_question.is_none());
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
assert_eq!(
saved
.metadata
.get("clarification_resume_pending")
.map(String::as_str),
Some("true")
);
assert_eq!(
saved
.metadata
.get("execute.startup_handoff_at")
.map(String::as_str),
Some("2026-08-10T09:00:00.000Z")
);
let answers = saved
.messages
.iter()
.filter(|message| message.tool_call_id.as_deref() == Some("call-1"))
.collect::<Vec<_>>();
assert_eq!(answers.len(), 1);
assert_eq!(answers[0].content, "Selected response: A");
}
#[tokio::test]
async fn actor_parent_question_handoff_adopts_canonical_question_and_discards_terminal_text() {
use bamboo_domain::{
FunctionCall, Message, ParentQuestion, PendingQuestionSource, ToolCall,
PARENT_QUESTION_REQUEST_KEY,
};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let parent = fresh("question-handoff-parent");
storage.save_session(&parent).await.unwrap();
let mut child = Session::new_child_of("question-handoff-child", &parent, "model", "child");
child.add_message(Message::user("assignment"));
storage.save_session(&child).await.unwrap();
let mut old_run = child.clone();
old_run
.metadata
.insert(ACTOR_PARENT_QUESTION_HANDOFF_KEY.into(), "true".into());
old_run.set_last_run_status("suspended");
old_run.add_message(Message::assistant("old terminal text", None));
child.add_message(Message::assistant(
"",
Some(vec![ToolCall {
id: "ask-call".into(),
tool_type: "function".into(),
function: FunctionCall {
name: "AskUserQuestion".into(),
arguments: "{}".into(),
},
}]),
));
child.add_message(Message::tool_result_with_status(
"ask-call",
"Clarification needed: Choose?",
true,
));
child.set_pending_question_with_source(
"ask-call".into(),
"AskUserQuestion".into(),
"Choose?".into(),
vec!["A".into(), "B".into()],
true,
PendingQuestionSource::DirectParent,
);
child.metadata.insert(
"runtime.suspend_reason".into(),
"awaiting_clarification".into(),
);
let question = ParentQuestion::issue_at(&parent, &child, chrono::Utc::now()).unwrap();
child.metadata.insert(
PARENT_QUESTION_REQUEST_KEY.into(),
serde_json::to_string(&question).unwrap(),
);
storage.save_session(&child).await.unwrap();
store.merge_save_runtime(&mut old_run).await.unwrap();
let saved = storage.load_session(&child.id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved.messages).unwrap(),
serde_json::to_value(&child.messages).unwrap()
);
assert_eq!(
serde_json::to_value(&saved.pending_question).unwrap(),
serde_json::to_value(&child.pending_question).unwrap()
);
assert_eq!(
saved.metadata.get(PARENT_QUESTION_REQUEST_KEY),
child.metadata.get(PARENT_QUESTION_REQUEST_KEY)
);
assert_eq!(saved.last_run_status().as_deref(), Some("suspended"));
assert!(!saved
.metadata
.contains_key(ACTOR_PARENT_QUESTION_HANDOFF_KEY));
}
#[tokio::test]
async fn actor_parent_question_handoff_cannot_undo_fast_parent_answer() {
use bamboo_domain::{
FunctionCall, Message, ParentQuestion, ParentQuestionResolution, PendingQuestionSource,
ToolCall, PARENT_QUESTION_REQUEST_KEY, PARENT_QUESTION_RESOLUTION_KEY,
};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let parent = fresh("answered-handoff-parent");
storage.save_session(&parent).await.unwrap();
let mut child = Session::new_child_of("answered-handoff-child", &parent, "model", "child");
child.add_message(Message::user("assignment"));
storage.save_session(&child).await.unwrap();
let mut old_run = child.clone();
old_run
.metadata
.insert(ACTOR_PARENT_QUESTION_HANDOFF_KEY.into(), "true".into());
old_run.set_last_run_status("suspended");
child.add_message(Message::assistant(
"",
Some(vec![ToolCall {
id: "ask-call".into(),
tool_type: "function".into(),
function: FunctionCall {
name: "AskUserQuestion".into(),
arguments: "{}".into(),
},
}]),
));
child.add_message(Message::tool_result_with_status(
"ask-call",
"Clarification needed: Choose?",
true,
));
child.set_pending_question_with_source(
"ask-call".into(),
"AskUserQuestion".into(),
"Choose?".into(),
vec!["A".into(), "B".into()],
true,
PendingQuestionSource::DirectParent,
);
child.metadata.insert(
"runtime.suspend_reason".into(),
"awaiting_clarification".into(),
);
let issued_at = chrono::Utc::now();
let question = ParentQuestion::issue_at(&parent, &child, issued_at).unwrap();
child.metadata.insert(
PARENT_QUESTION_REQUEST_KEY.into(),
serde_json::to_string(&question).unwrap(),
);
storage.save_session(&child).await.unwrap();
let mut answered = child;
answered.clear_pending_question();
answered.metadata.remove("runtime.suspend_reason");
let paired = answered
.messages
.iter_mut()
.find(|message| message.id == question.tool_result_message_id)
.unwrap();
paired.content = "A".into();
paired.tool_success = Some(true);
let resolution = ParentQuestionResolution::answered(
&question,
issued_at + chrono::Duration::milliseconds(1),
"A",
)
.unwrap();
answered.metadata.insert(
PARENT_QUESTION_RESOLUTION_KEY.into(),
serde_json::to_string(&resolution).unwrap(),
);
let source_key = "runtime.child_completion_source_v1";
answered
.metadata
.insert(source_key.into(), "new-run-terminal-source".into());
storage.save_session(&answered).await.unwrap();
store.merge_save_runtime(&mut old_run).await.unwrap();
let saved = storage.load_session(&answered.id).await.unwrap().unwrap();
assert_eq!(
saved.metadata.get(source_key),
answered.metadata.get(source_key),
"an old question handoff must preserve the source of the adopted canonical transcript"
);
assert!(saved.pending_question.is_none());
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
assert_eq!(
saved.metadata.get(PARENT_QUESTION_RESOLUTION_KEY),
answered.metadata.get(PARENT_QUESTION_RESOLUTION_KEY)
);
assert_eq!(
serde_json::to_value(&saved.messages).unwrap(),
serde_json::to_value(&answered.messages).unwrap()
);
assert_eq!(saved.last_run_status().as_deref(), Some("suspended"));
assert!(!saved
.metadata
.contains_key(ACTOR_PARENT_QUESTION_HANDOFF_KEY));
}
#[tokio::test]
async fn runtime_checkpoint_cannot_resurrect_a_consumed_clarification() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "consumed-clarification-checkpoint";
let mut stale_runner = fresh(session_id);
stale_runner.add_message(Message::tool_result("call-1", "waiting"));
stale_runner.set_pending_question(
"call-1".to_string(),
"ConclusionWithOptions".to_string(),
"Choose".to_string(),
vec!["A".to_string()],
false,
);
stale_runner.metadata.insert(
"runtime.suspend_reason".to_string(),
"awaiting_clarification".to_string(),
);
let mut answered = stale_runner.clone();
answered.clear_pending_question();
answered.metadata.remove("runtime.suspend_reason");
answered.metadata.insert(
CONSUMED_CLARIFICATION_IDS_KEY.to_string(),
r#"["call-1"]"#.to_string(),
);
answered.metadata.insert(
"clarification_resume_pending".to_string(),
"true".to_string(),
);
answered.messages[0].content = "Selected response: A".to_string();
storage.save_session(&answered).await.unwrap();
store
.checkpoint_runtime_session(&mut stale_runner)
.await
.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert!(saved.pending_question.is_none());
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
assert_eq!(saved.messages.len(), 1);
assert_eq!(saved.messages[0].content, "Selected response: A");
}
#[tokio::test]
async fn runtime_final_save_does_not_consume_a_new_reused_permission_occurrence() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "reused-permission-final-save";
let mut durable = fresh(session_id);
durable.add_message(typed_permission_result(
"reused-call",
"old-result",
"generation-old",
"Selected response: Approve",
));
durable.metadata.insert(
CONSUMED_RESPONSE_OCCURRENCES_KEY.to_string(),
serde_json::to_string(&vec![ResponseOccurrence {
tool_call_id: "reused-call".to_string(),
tool_result_message_id: "old-result".to_string(),
permission_generation: Some("generation-old".to_string()),
}])
.unwrap(),
);
storage.save_session(&durable).await.unwrap();
let mut new_runner = durable;
new_runner.add_message(typed_permission_result(
"reused-call",
"new-result",
"generation-new",
"waiting for the new decision",
));
new_runner.set_pending_question(
"reused-call".to_string(),
"Permission".to_string(),
"Approve new operation?".to_string(),
vec!["Approve".to_string(), "Deny".to_string()],
false,
);
new_runner.metadata.insert(
"runtime.suspend_reason".to_string(),
"awaiting_permission_approval".to_string(),
);
store.merge_save_runtime(&mut new_runner).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
saved
.pending_question
.as_ref()
.map(|pending| pending.tool_call_id.as_str()),
Some("reused-call")
);
assert_eq!(
saved.messages.last().map(|message| message.id.as_str()),
Some("new-result")
);
assert_eq!(
saved.messages.last().unwrap().content,
"waiting for the new decision"
);
assert_eq!(
latest_response_occurrence(&saved, "reused-call")
.and_then(|occurrence| occurrence.permission_generation),
Some("generation-new".to_string())
);
}
#[tokio::test]
async fn legacy_consumed_id_does_not_consume_a_new_reused_occurrence_after_upgrade() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "legacy-reused-permission-final-save";
let mut durable = fresh(session_id);
durable.add_message(typed_permission_result(
"reused-call",
"old-result",
"generation-old",
"Selected response: Approve",
));
durable.metadata.insert(
CONSUMED_CLARIFICATION_IDS_KEY.to_string(),
r#"["reused-call"]"#.to_string(),
);
storage.save_session(&durable).await.unwrap();
let mut new_runner = durable;
new_runner.add_message(typed_permission_result(
"reused-call",
"new-result",
"generation-new",
"waiting for the new decision",
));
new_runner.set_pending_question(
"reused-call".to_string(),
"Permission".to_string(),
"Approve new operation?".to_string(),
vec!["Approve".to_string(), "Deny".to_string()],
false,
);
store.merge_save_runtime(&mut new_runner).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
saved
.pending_question
.as_ref()
.map(|pending| pending.tool_call_id.as_str()),
Some("reused-call")
);
assert_eq!(
saved.messages.last().map(|message| message.id.as_str()),
Some("new-result")
);
assert_eq!(
latest_response_occurrence(&saved, "reused-call")
.and_then(|occurrence| occurrence.permission_generation),
Some("generation-new".to_string())
);
}
#[tokio::test]
async fn runtime_checkpoint_does_not_consume_a_new_reused_permission_occurrence() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "reused-permission-checkpoint";
let mut durable = fresh(session_id);
durable.add_message(typed_permission_result(
"reused-call",
"old-result",
"generation-old",
"Selected response: Deny",
));
durable.metadata.insert(
CONSUMED_RESPONSE_OCCURRENCES_KEY.to_string(),
serde_json::to_string(&vec![ResponseOccurrence {
tool_call_id: "reused-call".to_string(),
tool_result_message_id: "old-result".to_string(),
permission_generation: Some("generation-old".to_string()),
}])
.unwrap(),
);
storage.save_session(&durable).await.unwrap();
let mut new_runner = durable;
new_runner.add_message(typed_permission_result(
"reused-call",
"new-result",
"generation-new",
"waiting for the new decision",
));
new_runner.set_pending_question(
"reused-call".to_string(),
"Permission".to_string(),
"Approve new operation?".to_string(),
vec!["Approve".to_string(), "Deny".to_string()],
false,
);
store
.checkpoint_runtime_session(&mut new_runner)
.await
.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
saved
.pending_question
.as_ref()
.map(|pending| pending.tool_call_id.as_str()),
Some("reused-call")
);
assert_eq!(
saved.messages.last().map(|message| message.id.as_str()),
Some("new-result")
);
assert_eq!(
latest_response_occurrence(&saved, "reused-call")
.and_then(|occurrence| occurrence.permission_generation),
Some("generation-new".to_string())
);
}
#[tokio::test]
async fn consumed_permission_adoption_keeps_reexecute_id_and_generation_paired() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "consumed-permission-control-pair";
let mut stale_runner = fresh(session_id);
stale_runner.add_message(typed_permission_result(
"call-1",
"result-1",
"generation-1",
"waiting",
));
stale_runner.set_pending_question(
"call-1".to_string(),
"Permission".to_string(),
"Approve?".to_string(),
vec!["Approve".to_string(), "Deny".to_string()],
false,
);
let mut answered = stale_runner.clone();
answered.clear_pending_question();
answered.messages[0].content = "Selected response: Approve".to_string();
answered.metadata.insert(
CONSUMED_RESPONSE_OCCURRENCES_KEY.to_string(),
serde_json::to_string(&vec![ResponseOccurrence {
tool_call_id: "call-1".to_string(),
tool_result_message_id: "result-1".to_string(),
permission_generation: Some("generation-1".to_string()),
}])
.unwrap(),
);
answered.metadata.insert(
"permission.reexecute_tool_call_id".to_string(),
"call-1".to_string(),
);
answered.metadata.insert(
"permission.reexecute_request_generation".to_string(),
"generation-1".to_string(),
);
storage.save_session(&answered).await.unwrap();
store.merge_save_runtime(&mut stale_runner).await.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert!(saved.pending_question.is_none());
assert_eq!(
saved
.metadata
.get("permission.reexecute_tool_call_id")
.map(String::as_str),
Some("call-1")
);
assert_eq!(
saved
.metadata
.get("permission.reexecute_request_generation")
.map(String::as_str),
Some("generation-1")
);
}
#[tokio::test]
async fn checkpoint_runtime_session_preserves_disk_suffix_and_appends_live_messages() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "checkpoint-no-shrink";
let mut baseline = fresh(session_id);
baseline.add_message(Message::user("base"));
storage.save_session(&baseline).await.unwrap();
let mut runner_snapshot = baseline.clone();
let mut durable = baseline;
let mut disk_only = Message::user("concurrent injected message");
disk_only.id = "disk-only".to_string();
durable.add_message(disk_only);
storage.save_session(&durable).await.unwrap();
let mut live_only = Message::assistant("partial runner output", None);
live_only.id = "live-only".to_string();
runner_snapshot.add_message(live_only);
store
.checkpoint_runtime_session(&mut runner_snapshot)
.await
.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
let ids = saved
.messages
.iter()
.map(|message| message.id.as_str())
.collect::<Vec<_>>();
assert_eq!(
ids,
vec![durable.messages[0].id.as_str(), "disk-only", "live-only"]
);
assert_eq!(runner_snapshot.messages.len(), saved.messages.len());
assert_eq!(runner_snapshot.messages[1].id, saved.messages[1].id);
assert_eq!(runner_snapshot.messages[2].id, saved.messages[2].id);
assert_eq!(saved.messages[1].content, "concurrent injected message");
assert_eq!(saved.messages[2].content, "partial runner output");
}
fn direct_parent_proof_pair(
parent: &Session,
) -> (
bamboo_domain::SessionMessageEnvelope,
bamboo_domain::SessionMessageEnvelope,
) {
use bamboo_domain::{
ParentRequest, ParentRequestActivation, ParentRequestActor, ParentRequestDelegation,
ParentRequestKind, ParentRequestOption, ParentResolution, SessionMessageBody,
SessionMessageContent, SessionMessageEnvelope, SessionMessageId, SessionMessageKind,
SessionMessageSource, SessionProviderMessage, SessionRuntimeInstruction,
PARENT_REQUEST_VERSION,
};
use serde_json::json;
let generation = uuid::Uuid::new_v4().to_string();
let child_birth = parent.created_at + chrono::Duration::milliseconds(1);
let created_at = child_birth;
let deadline = created_at + chrono::Duration::seconds(120);
let activation = ParentRequestActivation {
host_scope: uuid::Uuid::new_v4().to_string(),
attempt: 1,
run: "run-1".into(),
epoch: 1,
reply: "reply-1".into(),
};
let request_id = SessionMessageId::stable(
"direct-parent-forced-approval-v1",
&json!({
"child": "approval-child",
"birth": child_birth,
"parent": parent.id,
"parent_birth": parent.created_at,
"generation": generation,
}),
);
let typed = ParentRequest {
version: PARENT_REQUEST_VERSION,
id: request_id.clone(),
generation: generation.clone(),
child: ParentRequestActor {
session_id: "approval-child".into(),
created_at: child_birth,
},
parent: ParentRequestActor {
session_id: parent.id.clone(),
created_at: parent.created_at,
},
root_session_id: parent.root_session_id.clone(),
project_id: parent.project_id_meta(),
activation: activation.clone(),
deadline,
kind: ParentRequestKind::ForcedPermission {
operation_digest: format!("stable-{}", "a".repeat(64)),
policy_revision: 1,
maximum_delegation: ParentRequestDelegation::ExactOperationOnce,
options: vec![ParentRequestOption::Deny, ParentRequestOption::ApproveOnce],
},
};
let live = json!({
"host_scope": activation.host_scope,
"attempt": activation.attempt,
"run": activation.run,
"epoch": activation.epoch,
"reply": activation.reply,
"deadline": deadline,
});
let request_text = "Parent permission request";
let request = SessionMessageEnvelope {
id: request_id.clone(),
source: SessionMessageSource::Runtime {
subsystem: "direct_parent_permission_review".into(),
},
target_session_id: parent.id.clone(),
kind: SessionMessageKind::RuntimeInstruction,
body: SessionMessageBody::RuntimeInstruction(SessionRuntimeInstruction {
instruction: "direct_parent_forced_permission_request_v1".into(),
content: Some(SessionMessageContent::text(request_text)),
data: Some(json!({
"parent_request": typed,
"parent_created_at": parent.created_at,
"child_session_id": "approval-child",
"child_created_at": child_birth,
"root_session_id": parent.root_session_id,
"project_id": parent.project_id_meta(),
"request_generation": generation,
"operation_digest": format!("stable-{}", "a".repeat(64)),
"policy_revision": 1,
"live": live,
})),
provider_message: Some(SessionProviderMessage {
content: SessionMessageContent::text(request_text),
metadata: Default::default(),
never_compress: true,
}),
}),
created_at,
thread_id: None,
in_reply_to: None,
attempt: None,
correlation_id: None,
};
assert!(ParentRequest::from_forced_permission_envelope(&request).is_some());
let resolved_at = created_at + chrono::Duration::seconds(1);
let resolution = ParentResolution::for_forced_permission(&typed, resolved_at, false);
let terminal_text =
"Live parent permission review recorded Denied. This record is not a permission grant.";
let mut terminal = request.clone();
terminal.id = resolution.id.clone();
terminal.created_at = resolved_at;
terminal.in_reply_to = Some(request_id.clone());
terminal.body = SessionMessageBody::RuntimeInstruction(SessionRuntimeInstruction {
instruction: "direct_parent_forced_permission_terminal_v1".into(),
content: Some(SessionMessageContent::text(terminal_text)),
data: Some(json!({
"request": request_id,
"live": live,
"approved": false,
"parent_resolution": resolution,
})),
provider_message: Some(SessionProviderMessage {
content: SessionMessageContent::text(terminal_text),
metadata: Default::default(),
never_compress: true,
}),
});
assert!(ParentResolution::from_forced_permission_terminal(&request, &terminal).is_some());
(request, terminal)
}
#[tokio::test]
async fn checkpoint_rebases_direct_parent_terminal_committed_by_another_store() {
let home = tempfile::tempdir().unwrap();
let first = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
let mut parent = fresh("direct-parent-terminal-race");
let (request, terminal) = direct_parent_proof_pair(&parent);
parent.add_message(request.to_provider_message().unwrap());
first.save_session(&parent).await.unwrap();
let second = SessionStoreV2::new(home.path().into()).await.unwrap();
let paused = Arc::new(AuthoritySavePauseStorage {
inner: first.clone(),
calls: AtomicUsize::new(0),
reached: tokio::sync::Barrier::new(2),
release: tokio::sync::Barrier::new(2),
});
let checkpoint_store = LockedSessionStore::new(paused.clone());
let mut stale = parent.clone();
stale.add_message(bamboo_domain::Message::assistant("runner output", None));
let checkpoint = tokio::spawn(async move {
checkpoint_store
.checkpoint_runtime_session(&mut stale)
.await?;
Ok::<_, std::io::Error>(stale)
});
paused.reached.wait().await;
let mut decided = second.load_session(&parent.id).await.unwrap().unwrap();
decided.add_message(terminal.to_provider_message().unwrap());
second.save_session(&decided).await.unwrap();
let stale_write = first.save_session(&parent).await.unwrap_err();
assert!(is_direct_parent_terminal_conflict(&stale_write));
assert!(!may_publish_runtime_result(&Err(stale_write)));
let mut forged = decided.clone();
forged.messages.last_mut().unwrap().content = "forged decision".into();
let forged_write = second.save_session(&forged).await.unwrap_err();
assert_eq!(forged_write.kind(), std::io::ErrorKind::InvalidData);
assert!(!may_publish_runtime_result(&Err(forged_write)));
paused.release.wait().await;
paused.reached.wait().await;
paused.release.wait().await;
let reconciled = checkpoint.await.unwrap().unwrap();
let saved = second.load_session(&parent.id).await.unwrap().unwrap();
assert_eq!(paused.calls.load(Ordering::SeqCst), 2);
assert_eq!(saved.messages.len(), 3);
assert_eq!(saved.messages[0].id, request.id.as_str());
assert_eq!(saved.messages[1].id, terminal.id.as_str());
assert_eq!(saved.messages[2].content, "runner output");
assert_eq!(reconciled.messages[1].id, terminal.id.as_str());
}
#[tokio::test]
async fn prompt_rewrite_checkpoint_commits_without_an_archive_event() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "prompt-rewrite-checkpoint";
let mut expected = fresh(session_id);
expected.add_message(Message::system("Base\n\nDEGRADABLE TOOL GUIDE"));
expected.add_message(Message::user("continue"));
expected.metadata.insert(
"responses.previous_response_id".to_string(),
"response-before-rewrite".to_string(),
);
storage.save_session(&expected).await.unwrap();
let mut staged = expected.clone();
staged.messages[0].content = "Base".to_string();
staged.metadata.remove("responses.previous_response_id");
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
let outcome = store
.checkpoint_prompt_rewrite_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(saved.messages[0].content, "Base");
assert!(saved.compression_events.is_empty());
assert!(!saved
.metadata
.contains_key("responses.previous_response_id"));
assert_eq!(
saved
.model_context_state
.as_ref()
.and_then(|state| state.last_reset_reason),
Some(bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite)
);
assert_eq!(
serde_json::to_value(&staged).unwrap(),
serde_json::to_value(&saved).unwrap()
);
}
#[tokio::test]
async fn prompt_rewrite_checkpoint_rebases_a_concurrent_suffix_without_writing() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "prompt-rewrite-rebase";
let mut expected = fresh(session_id);
expected.add_message(Message::system("Base\n\nDEGRADABLE TOOL GUIDE"));
storage.save_session(&expected).await.unwrap();
let mut staged = expected.clone();
staged.messages[0].content = "Base".to_string();
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
let mut durable = expected.clone();
let mut concurrent = Message::user("concurrent durable suffix");
concurrent.id = "prompt-rewrite-concurrent-suffix".to_string();
durable.add_message(concurrent);
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_prompt_rewrite_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap()
);
assert_eq!(staged.messages[0].content, expected.messages[0].content);
assert_eq!(staged.messages[1].id, "prompt-rewrite-concurrent-suffix");
assert!(staged.model_context_state.is_none());
}
#[tokio::test]
async fn prompt_rewrite_checkpoint_rebases_a_concurrent_pending_injection() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "prompt-rewrite-pending-injection";
let mut expected = fresh(session_id);
expected.add_message(Message::system("Base\n\nDEGRADABLE TOOL GUIDE"));
storage.save_session(&expected).await.unwrap();
let mut staged = expected.clone();
staged.messages[0].content = "Base".to_string();
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
let mut durable = expected.clone();
durable.set_pending_injected_messages(vec![serde_json::json!({
"content": "background shell completed",
})]);
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_prompt_rewrite_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap()
);
assert_eq!(staged.messages[0].content, expected.messages[0].content);
assert_eq!(
staged.pending_injected_messages(),
durable.pending_injected_messages()
);
assert!(staged.model_context_state.is_none());
}
#[tokio::test]
async fn prompt_rewrite_checkpoint_rebases_concurrent_open_runtime_metadata() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "prompt-rewrite-open-runtime-metadata";
let mut expected = fresh(session_id);
expected.add_message(Message::system("Base\n\nDEGRADABLE TOOL GUIDE"));
storage.save_session(&expected).await.unwrap();
let mut staged = expected.clone();
staged.messages[0].content = "Base".to_string();
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
let mut durable = expected.clone();
durable.metadata.insert(
"concurrent.runtime.marker".to_string(),
"latest".to_string(),
);
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_prompt_rewrite_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert_eq!(
staged
.metadata
.get("concurrent.runtime.marker")
.map(String::as_str),
Some("latest")
);
assert_eq!(staged.messages[0].content, expected.messages[0].content);
assert!(staged.model_context_state.is_none());
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap()
);
}
#[tokio::test]
async fn prompt_rewrite_checkpoint_rebases_concurrent_reasoning_update() {
use bamboo_domain::{reasoning::ReasoningEffort, session::types::Message};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "prompt-rewrite-reasoning-update";
let mut expected = fresh(session_id);
expected.add_message(Message::system("Base\n\nDEGRADABLE TOOL GUIDE"));
storage.save_session(&expected).await.unwrap();
let mut staged = expected.clone();
staged.messages[0].content = "Base".to_string();
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
let mut durable = expected.clone();
durable.reasoning_effort = Some(ReasoningEffort::High);
durable.metadata_version = 1;
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_prompt_rewrite_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert_eq!(staged.reasoning_effort, Some(ReasoningEffort::High));
assert_eq!(staged.metadata_version, 1);
assert_eq!(staged.messages[0].content, expected.messages[0].content);
assert!(staged.model_context_state.is_none());
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap()
);
}
#[tokio::test]
async fn manual_archive_rejection_checkpoint_persists_the_rewritten_tool_result() {
use bamboo_domain::{FunctionCall, Message, ToolCall};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "manual-archive-rejection-checkpoint";
let mut expected = fresh(session_id);
expected.add_message(Message::user("archive older context"));
let mut assistant = Message::assistant("", None);
assistant.tool_calls = Some(vec![ToolCall {
id: "archive-call".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "archive_context".to_string(),
arguments: "{}".to_string(),
},
}]);
expected.add_message(assistant);
let mut result = Message::tool_result("archive-call", "Retrieval-window archive requested");
result.id = "archive-result".to_string();
result.tool_success = Some(true);
expected.add_message(result);
expected.metadata.insert(
RESPONSES_PREVIOUS_RESPONSE_ID_KEY.to_string(),
"response-before-rejection".to_string(),
);
storage.save_session(&expected).await.unwrap();
let occurrence = ResponseOccurrence {
tool_call_id: "archive-call".to_string(),
tool_result_message_id: "archive-result".to_string(),
permission_generation: None,
};
let reason = "no eligible active logical group can be archived";
let stage_rejection = |base: &Session| {
let mut staged = base.clone();
let staged_result = staged
.messages
.iter_mut()
.find(|message| message.id == "archive-result")
.unwrap();
staged_result.tool_success = Some(false);
staged_result.content = format!("archive_context rejected: {reason}");
staged.metadata.insert(
LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY.to_string(),
serde_json::to_string(&occurrence).unwrap(),
);
staged.metadata.insert(
MANUAL_ARCHIVE_REJECTIONS_KEY.to_string(),
serde_json::to_string(&vec![serde_json::json!({
"occurrence": occurrence.clone(),
"reason": reason,
})])
.unwrap(),
);
staged.metadata.remove(RESPONSES_PREVIOUS_RESPONSE_ID_KEY);
staged.reset_model_context_epoch(
bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite,
);
staged
};
let mut staged = stage_rejection(&expected);
let mut durable = expected.clone();
durable.set_pending_injected_messages(vec![serde_json::json!({
"content": "background shell completed",
})]);
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_manual_archive_rejection_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert_eq!(
staged.pending_injected_messages(),
durable.pending_injected_messages()
);
let unchanged = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&unchanged).unwrap(),
serde_json::to_value(&durable).unwrap()
);
expected = staged;
let mut staged = stage_rejection(&expected);
let outcome = store
.checkpoint_manual_archive_rejection_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
let saved_result = saved
.messages
.iter()
.find(|message| message.id == "archive-result")
.unwrap();
assert_eq!(saved_result.tool_success, Some(false));
assert_eq!(
saved_result.content,
"archive_context rejected: no eligible active logical group can be archived"
);
assert!(saved
.metadata
.contains_key(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY));
assert!(saved.metadata.contains_key(MANUAL_ARCHIVE_REJECTIONS_KEY));
assert!(!saved
.metadata
.contains_key(RESPONSES_PREVIOUS_RESPONSE_ID_KEY));
assert_eq!(
saved.pending_injected_messages(),
durable.pending_injected_messages()
);
assert_eq!(
saved
.model_context_state
.as_ref()
.and_then(|state| state.last_reset_reason),
Some(bamboo_domain::ModelContextResetReason::ExplicitHistoryRewrite)
);
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&staged).unwrap()
);
}
#[tokio::test]
async fn manual_archive_consumption_checkpoint_rebases_concurrent_open_metadata() {
use bamboo_domain::{FunctionCall, Message, ToolCall};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "manual-archive-consumption-checkpoint";
let mut expected = fresh(session_id);
expected.add_message(Message::user("archive older context"));
let mut assistant = Message::assistant("", None);
assistant.tool_calls = Some(vec![ToolCall {
id: "archive-call".to_string(),
tool_type: "function".to_string(),
function: FunctionCall {
name: "archive_context".to_string(),
arguments: "{}".to_string(),
},
}]);
expected.add_message(assistant);
let mut result = Message::tool_result("archive-call", "Retrieval-window archive requested");
result.id = "archive-result".to_string();
result.tool_success = Some(true);
expected.add_message(result);
storage.save_session(&expected).await.unwrap();
let occurrence = ResponseOccurrence {
tool_call_id: "archive-call".to_string(),
tool_result_message_id: "archive-result".to_string(),
permission_generation: None,
};
let stage_consumption = |base: &Session| {
let mut staged = base.clone();
staged.metadata.insert(
LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY.to_string(),
serde_json::to_string(&occurrence).unwrap(),
);
staged
};
let mut staged = stage_consumption(&expected);
let mut durable = expected.clone();
durable.metadata.insert(
"concurrent.workflow_index".to_string(),
"workflow-42".to_string(),
);
durable.set_pending_injected_messages(vec![serde_json::json!({
"id": "bash-1",
"status": "completed",
})]);
storage.save_session(&durable).await.unwrap();
let outcome = store
.checkpoint_manual_archive_consumption_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert_eq!(
serde_json::to_value(&staged).unwrap(),
serde_json::to_value(&durable).unwrap()
);
let unchanged = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&unchanged).unwrap(),
serde_json::to_value(&durable).unwrap()
);
expected = staged;
let mut staged = stage_consumption(&expected);
let outcome = store
.checkpoint_manual_archive_consumption_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
saved.metadata.get("concurrent.workflow_index"),
Some(&"workflow-42".to_string())
);
assert_eq!(
saved.pending_injected_messages(),
Some(vec![serde_json::json!({
"id": "bash-1",
"status": "completed",
})])
);
assert_eq!(
saved.metadata.get(LAST_MANUAL_ARCHIVE_OCCURRENCE_KEY),
Some(&serde_json::to_string(&occurrence).unwrap())
);
}
#[tokio::test]
async fn message_summary_checkpoint_commits_raw_history_and_publishes_only_after_save() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let (expected, mut staged) = message_summary_fixture("message-summary-checkpoint");
storage.save_session(&expected).await.unwrap();
let published = std::sync::Mutex::new(None);
let outcome = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, |saved| {
*published.lock().unwrap() = Some(saved.clone());
})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(&expected.id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(published.lock().unwrap().as_ref().unwrap()).unwrap()
);
for (before, after) in expected.messages.iter().zip(&saved.messages) {
let mut original = after.clone();
original.compressed = before.compressed;
original
.compressed_by_event_id
.clone_from(&before.compressed_by_event_id);
assert_eq!(
serde_json::to_value(before).unwrap(),
serde_json::to_value(&original).unwrap()
);
}
assert!(saved.messages[1..5]
.iter()
.all(|message| message.compressed));
assert!(
!saved.messages[5].compressed,
"explicit prose anchor stays active"
);
assert!(
!saved.messages[6].compressed,
"current user request stays active"
);
assert_eq!(
saved.provider_transcript.last_reset_reason(),
Some(bamboo_domain::ProviderTranscriptResetReason::Compression)
);
assert!(!saved
.metadata
.contains_key(RESPONSES_PREVIOUS_RESPONSE_ID_KEY));
}
#[tokio::test]
async fn message_summary_checkpoint_accepts_an_unpersisted_midturn_suffix() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let (expected, mut staged) = message_summary_fixture("message-summary-local-suffix");
let mut durable_prefix = expected.clone();
durable_prefix.messages.truncate(2);
storage.save_session(&durable_prefix).await.unwrap();
let outcome = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, |_| {})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(&expected.id).await.unwrap().unwrap();
assert_eq!(saved.messages.len(), expected.messages.len());
assert!(saved.messages[1..5]
.iter()
.all(|message| message.compressed));
assert_eq!(saved.messages[6].content, "Current user request");
assert!(!saved.messages[6].compressed);
}
#[tokio::test]
async fn message_summary_checkpoint_failed_save_keeps_durable_history_and_cached_projection() {
let (_temp, storage) = make_storage().await;
let (expected, mut staged) = message_summary_fixture("message-summary-failed-save");
storage.save_session(&expected).await.unwrap();
let failing = Arc::new(FailingMessageSummaryStorage {
inner: storage.clone(),
attempted_saves: AtomicUsize::new(0),
});
let store = LockedSessionStore::new(failing.clone());
let cached = std::sync::Mutex::new(expected.clone());
let error = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, |saved| {
*cached.lock().unwrap() = saved.clone();
})
.await
.unwrap_err();
assert_eq!(error.to_string(), "injected archive save failure");
assert_eq!(failing.attempted_saves.load(Ordering::SeqCst), 1);
assert_eq!(
serde_json::to_value(&*cached.lock().unwrap()).unwrap(),
serde_json::to_value(&expected).unwrap()
);
let saved = storage.load_session(&expected.id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&expected).unwrap()
);
}
#[tokio::test]
async fn message_summary_checkpoint_rebases_concurrent_durable_changes_without_writing_or_publishing(
) {
for append in [false, true] {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let (expected, mut staged) = message_summary_fixture("message-summary-concurrent-base");
storage.save_session(&expected).await.unwrap();
let mut durable = expected.clone();
if append {
durable.add_message(bamboo_domain::Message::user("Concurrent current request"));
} else {
durable.messages[3].content = "Concurrent corrected tool output".into();
}
storage.save_session(&durable).await.unwrap();
let published = AtomicBool::new(false);
let outcome = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, |_| {
published.store(true, Ordering::SeqCst);
})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert!(!published.load(Ordering::SeqCst));
assert_eq!(
serde_json::to_value(&staged).unwrap(),
serde_json::to_value(&durable).unwrap()
);
let saved = storage.load_session(&expected.id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap()
);
}
}
#[test]
fn message_summary_checkpoint_rejects_raw_rewrites_and_incomplete_or_retained_tool_protocol() {
let (expected, valid) = message_summary_fixture("message-summary-invalid-projection");
validate_staged_message_summary_transition(&expected, &valid).unwrap();
for mutation in 0..9 {
let mut staged = valid.clone();
match mutation {
0 => staged.messages[3].content = "Rewritten raw tool output".into(),
1 => {
staged.messages[3].compressed = false;
staged.messages[3].compressed_by_event_id = None;
}
2 | 3 => {
let index = if mutation == 2 { 5 } else { 6 };
staged.messages[index].compressed = true;
staged.messages[index].compressed_by_event_id =
Some(staged.compression_events[0].id.clone());
staged.compression_events[0].messages_compressed += 1;
}
4 => staged
.messages
.push(bamboo_domain::Message::assistant("Injected recovery", None)),
5 => {
staged.metadata.insert(
RESPONSES_PREVIOUS_RESPONSE_ID_KEY.into(),
"stale-response".into(),
);
}
6 => staged.provider_transcript = expected.provider_transcript.clone(),
7 => staged.compression_events[0].messages_compressed = 3,
8 => staged.messages[0].content = "Rewritten system policy".into(),
_ => unreachable!(),
}
assert!(
validate_staged_message_summary_transition(&expected, &staged).is_err(),
"accepted forbidden projection mutation {mutation}"
);
}
let mut incomplete = expected.clone();
incomplete.messages[3].tool_call_id = Some("orphan-result".into());
let mut staged = valid.clone();
staged.messages[3].tool_call_id = Some("orphan-result".into());
assert!(validate_staged_message_summary_transition(&incomplete, &staged).is_err());
let mut protected = expected.clone();
protected.messages[3].never_compress = true;
let mut staged = valid;
staged.messages[3].never_compress = true;
assert!(validate_staged_message_summary_transition(&protected, &staged).is_err());
}
#[tokio::test]
async fn retrieval_window_checkpoint_preserves_staged_archive_flags() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "retrieval-checkpoint-archive-flags";
let mut expected = fresh(session_id);
expected.add_message(Message::user("archive me"));
expected.add_message(Message::assistant("retain me", None));
storage.save_session(&expected).await.unwrap();
let mut staged = stage_retrieval_window_archive(&expected, 0);
let event_id = staged.compression_events[0].id.clone();
let published = Arc::new(std::sync::Mutex::new(None));
let published_clone = Arc::clone(&published);
let outcome = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, move |saved| {
*published_clone.lock().unwrap() = Some(saved.clone());
})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Committed);
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert!(saved.messages[0].compressed);
assert_eq!(
saved.messages[0].compressed_by_event_id.as_deref(),
Some(event_id.as_str())
);
assert!(!saved.messages[1].compressed);
assert_eq!(saved.compression_events.len(), 1);
assert_eq!(
saved.compression_events[0].kind,
bamboo_domain::CompressionEventKind::RetrievalWindow
);
assert_eq!(
serde_json::to_value(published.lock().unwrap().as_ref().unwrap()).unwrap(),
serde_json::to_value(&saved).unwrap()
);
assert_eq!(
serde_json::to_value(&staged).unwrap(),
serde_json::to_value(&saved).unwrap()
);
}
#[tokio::test]
async fn retrieval_window_checkpoint_rebases_concurrent_suffix_without_writing() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "retrieval-checkpoint-concurrent-suffix";
let mut expected = fresh(session_id);
expected.add_message(Message::user("archive candidate"));
storage.save_session(&expected).await.unwrap();
let mut staged = stage_retrieval_window_archive(&expected, 0);
let mut durable = expected.clone();
let mut concurrent = Message::user("concurrent durable suffix");
concurrent.id = "concurrent-durable-suffix".to_string();
durable.add_message(concurrent);
storage.save_session(&durable).await.unwrap();
let published = Arc::new(AtomicBool::new(false));
let published_clone = Arc::clone(&published);
let outcome = store
.checkpoint_retrieval_window_and_publish(&expected, &mut staged, move |_| {
published_clone.store(true, Ordering::SeqCst);
})
.await
.unwrap();
assert_eq!(outcome, RetrievalWindowCheckpointOutcome::Rebased);
assert!(!published.load(Ordering::SeqCst));
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
serde_json::to_value(&saved).unwrap(),
serde_json::to_value(&durable).unwrap(),
"a conflict must not write the staged archive"
);
assert_eq!(staged.messages.len(), 2);
assert_eq!(staged.messages[1].id, "concurrent-durable-suffix");
assert!(staged.messages.iter().all(|message| !message.compressed));
assert!(staged.compression_events.is_empty());
assert!(staged.model_context_state.is_none());
}
#[tokio::test]
async fn runtime_only_save_preserves_checkpointed_ledger_and_publishes_merged_state() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-only-ledger-race";
let baseline = fresh(session_id);
storage.save_session(&baseline).await.unwrap();
let mut stale_control = storage.load_session(session_id).await.unwrap().unwrap();
let mut runner = baseline;
runner.model_context_state = Some(ledger_state(1, "runner-l1"));
store.checkpoint_runtime_session(&mut runner).await.unwrap();
stale_control.metadata.insert(
"runtime.suspend_reason".to_string(),
"waiting_for_children".to_string(),
);
let published = std::sync::Arc::new(std::sync::Mutex::new(None));
let published_clone = published.clone();
store
.save_runtime_only_and_publish(&mut stale_control, move |saved| {
*published_clone.lock().unwrap() = Some(saved.clone());
})
.await
.unwrap();
let expected = runner.model_context_state.clone();
assert_eq!(stale_control.model_context_state, expected);
assert_eq!(
published
.lock()
.unwrap()
.as_ref()
.unwrap()
.model_context_state,
expected
);
let sidecar = storage
.load_runtime_control_plane(session_id)
.await
.unwrap()
.unwrap();
assert_eq!(sidecar.model_context_state, expected);
assert_eq!(
sidecar
.metadata
.get("runtime.suspend_reason")
.map(String::as_str),
Some("waiting_for_children")
);
let reloaded = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(reloaded.model_context_state, expected);
}
#[tokio::test]
async fn full_runtime_save_preserves_newer_ledger_but_commits_control_mutation() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "full-save-ledger-race";
let baseline = fresh(session_id);
storage.save_session(&baseline).await.unwrap();
let mut stale = storage.load_session(session_id).await.unwrap().unwrap();
let mut runner = baseline;
runner.model_context_state = Some(ledger_state(1, "runner-l1"));
store.checkpoint_runtime_session(&mut runner).await.unwrap();
stale
.metadata
.insert("activated_tools".to_string(), "[\"search\"]".to_string());
let published = std::sync::Arc::new(std::sync::Mutex::new(None));
let published_clone = published.clone();
store
.merge_save_runtime_and_publish(&mut stale, move |saved, committed| {
assert!(committed);
*published_clone.lock().unwrap() = Some(saved.clone());
})
.await
.unwrap();
let expected = runner.model_context_state.clone();
assert_eq!(stale.model_context_state, expected);
assert_eq!(
published
.lock()
.unwrap()
.as_ref()
.unwrap()
.model_context_state,
expected
);
let sidecar = storage
.load_runtime_control_plane(session_id)
.await
.unwrap()
.unwrap();
assert_eq!(sidecar.model_context_state, expected);
let reloaded = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(reloaded.model_context_state, expected);
assert_eq!(
reloaded.metadata.get("activated_tools").map(String::as_str),
Some("[\"search\"]")
);
}
#[tokio::test]
async fn newer_explicit_epoch_reset_wins_an_ordinary_full_runtime_save() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "full-save-ledger-reset";
let mut durable = fresh(session_id);
durable.model_context_state = Some(ledger_state(1, "runner-l1"));
storage.save_session(&durable).await.unwrap();
let mut compression = storage.load_session(session_id).await.unwrap().unwrap();
compression.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::Compression);
let reset = compression.model_context_state.clone();
assert_eq!(reset.as_ref().unwrap().state_revision, 2);
store.merge_save_runtime(&mut compression).await.unwrap();
let reloaded = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(reloaded.model_context_state, reset);
assert_eq!(
reloaded
.model_context_state
.as_ref()
.and_then(|state| state.last_reset_reason),
Some(bamboo_domain::ModelContextResetReason::Compression)
);
}
#[tokio::test]
async fn checkpoint_rejects_equal_revision_divergence_without_overwriting_disk() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "ledger-checkpoint-cas";
let mut baseline = fresh(session_id);
baseline.model_context_state = Some(ledger_state(1, "runner-l1"));
storage.save_session(&baseline).await.unwrap();
let mut first = baseline.clone();
first.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::Compression);
let mut conflicting = baseline;
conflicting.reset_model_context_epoch(bamboo_domain::ModelContextResetReason::Rollback);
assert_eq!(
first.model_context_state.as_ref().unwrap().state_revision,
conflicting
.model_context_state
.as_ref()
.unwrap()
.state_revision
);
store.checkpoint_runtime_session(&mut first).await.unwrap();
let error = store
.checkpoint_runtime_session(&mut conflicting)
.await
.unwrap_err();
assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
let reloaded = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(reloaded.model_context_state, first.model_context_state);
}
#[tokio::test]
async fn activation_checkpoint_clears_presentation_without_shrinking_concurrent_turn() {
use bamboo_domain::session::runtime_state::{
AgentRuntimeState, AgentStatusState, WaitingForChildrenState,
};
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "activation-no-shrink";
let mut baseline = fresh(session_id);
baseline.add_message(Message::user("base"));
let mut state = AgentRuntimeState::new("activation-run");
state.status = AgentStatusState::Suspended;
state.waiting_for_children = Some(WaitingForChildrenState::for_children(
vec!["child-1".to_string()],
bamboo_domain::session::runtime_state::ChildWaitPolicy::All,
chrono::Utc::now(),
));
baseline.agent_runtime_state = Some(state);
baseline.metadata.insert(
"runtime.suspend_reason".to_string(),
"waiting_for_children".to_string(),
);
storage.save_session(&baseline).await.unwrap();
let mut activation_snapshot = baseline.clone();
let mut concurrent = baseline;
let mut normal = Message::assistant("normal concurrent answer", None);
normal.id = "normal-concurrent".to_string();
concurrent.add_message(normal);
storage.save_session(&concurrent).await.unwrap();
let state = activation_snapshot.agent_runtime_state.as_mut().unwrap();
state.status = AgentStatusState::Idle;
state.suspension = None;
activation_snapshot
.metadata
.remove("runtime.suspend_reason");
store
.checkpoint_runtime_session(&mut activation_snapshot)
.await
.unwrap();
let saved = storage.load_session(session_id).await.unwrap().unwrap();
assert!(saved
.messages
.iter()
.any(|message| message.id == "normal-concurrent"));
let state = saved.agent_runtime_state.unwrap();
assert_eq!(state.status, AgentStatusState::Idle);
assert!(state.waiting_for_children.is_some());
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
}
#[tokio::test]
async fn ordinary_and_checkpoint_tagged_wait_adoption_preserves_current_run_and_clear_winner() {
use bamboo_domain::session::runtime_state::SuspensionState;
use bamboo_domain::session::types::Message;
use bamboo_domain::{
AgentRuntimeState, AgentStatusState, ChildWaitPolicy, WaitingForChildrenState,
};
for checkpoint in [false, true] {
for status in [AgentStatusState::Initializing, AgentStatusState::Running] {
for current_wait in 0..3 {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let mut parent = fresh("mid-run-tagged-wait");
let mut wait = WaitingForChildrenState::for_children(
vec!["child-b".into(), "child-a".into()],
ChildWaitPolicy::Any,
chrono::Utc::now() - chrono::Duration::minutes(5),
);
wait.registered_by_tool_call_id = Some("original-call".into());
let mut previous = AgentRuntimeState::new("old-run");
previous.status = AgentStatusState::Suspended;
previous.waiting_for_children = Some(wait.clone());
parent.agent_runtime_state = Some(previous);
parent.add_message(Message::user("original task"));
storage.save_session(&parent).await.unwrap();
let mut incoming = parent.clone();
let runtime = incoming.agent_runtime_state.as_mut().unwrap();
runtime.run_id = "reasoning-run".into();
runtime.status = status;
incoming.metadata.remove("runtime.suspend_reason");
incoming.add_message(Message::user("new admitted reasoning"));
let expected = match current_wait {
0 => Some(wait),
1 => None,
_ => {
let mut new_wait = wait;
new_wait.wait_for = ChildWaitPolicy::FirstError;
new_wait.registered_by_tool_call_id = Some("new-tool-call".into());
Some(new_wait)
}
};
let current = expected.clone();
store
.update_runtime_config(&parent.id, move |latest| {
let runtime = latest.agent_runtime_state.as_mut().unwrap();
runtime.waiting_for_children = current;
if runtime.waiting_for_children.is_none() {
runtime.status = AgentStatusState::Idle;
runtime.suspension = None;
latest.metadata.remove("runtime.suspend_reason");
}
})
.await
.unwrap()
.unwrap();
if expected.is_none() {
let mut completed =
storage.load_session(&parent.id).await.unwrap().unwrap();
for (content, created_at) in [
(
"old consumed outcome",
chrono::Utc::now() - chrono::Duration::hours(1),
),
("current completed outcome", chrono::Utc::now()),
] {
let mut message = Message::user(content);
message.created_at = created_at;
message.metadata =
Some(serde_json::json!({"runtime_kind":"child_completion_resume"}));
completed.add_message(message);
}
store.merge_save_runtime(&mut completed).await.unwrap();
}
if checkpoint {
store
.checkpoint_runtime_session(&mut incoming)
.await
.unwrap();
} else {
store.merge_save_runtime(&mut incoming).await.unwrap();
}
for saved in [
incoming.clone(),
storage.load_session(&parent.id).await.unwrap().unwrap(),
] {
let runtime = saved.agent_runtime_state.unwrap();
assert_eq!(runtime.waiting_for_children, expected);
assert_eq!(runtime.status, status);
assert_eq!(runtime.run_id, "reasoning-run");
assert!(runtime.suspension.is_none());
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
assert_eq!(
saved
.messages
.iter()
.filter(|m| m.content == "new admitted reasoning")
.count(),
1
);
assert!(saved.messages.iter().any(|m| m.content == "original task"));
if expected.is_none() {
assert_eq!(
saved
.messages
.iter()
.filter(|m| m.content == "current completed outcome")
.count(),
1
);
assert_eq!(
saved
.messages
.iter()
.any(|m| m.content == "old consumed outcome"),
checkpoint
);
}
}
if expected.is_none() {
incoming.agent_runtime_state.as_mut().unwrap().status =
AgentStatusState::Completed;
store
.merge_save_finalized_runtime(&mut incoming)
.await
.unwrap();
let finalized = storage.load_session(&parent.id).await.unwrap().unwrap();
assert!(finalized
.agent_runtime_state
.as_ref()
.unwrap()
.waiting_for_children
.is_none());
assert_eq!(
finalized
.messages
.iter()
.filter(|m| m.content == "current completed outcome")
.count(),
1
);
}
if expected.is_none() {
let mut unrelated = parent;
let runtime = unrelated.agent_runtime_state.as_mut().unwrap();
runtime.status = AgentStatusState::Suspended;
runtime.suspension = Some(SuspensionState {
reason: "awaiting_clarification".into(),
suspended_at: chrono::Utc::now(),
resumable: true,
hook_point: Some("AfterToolExecution".into()),
});
let suspension = runtime.suspension.clone();
unrelated.metadata.insert(
"runtime.suspend_reason".into(),
"awaiting_clarification".into(),
);
store
.checkpoint_runtime_session(&mut unrelated)
.await
.unwrap();
let runtime = unrelated.agent_runtime_state.as_ref().unwrap();
assert!(runtime.waiting_for_children.is_none());
assert_eq!(runtime.status, AgentStatusState::Suspended);
assert_eq!(runtime.suspension, suspension);
assert_eq!(
unrelated
.metadata
.get("runtime.suspend_reason")
.map(String::as_str),
Some("awaiting_clarification")
);
}
}
}
}
}
#[tokio::test]
async fn tagged_control_plane_first_arm_and_untagged_runner_first_save_remain_owned() {
use bamboo_domain::{AgentRuntimeState, ChildWaitPolicy, WaitingForChildrenState};
for tagged in [true, false] {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let mut parent = fresh("first-wait-publication");
parent.agent_runtime_state = Some(AgentRuntimeState::new("parent-run"));
storage.save_session(&parent).await.unwrap();
let mut wait = WaitingForChildrenState::for_children(
vec!["new-child".into()],
ChildWaitPolicy::All,
chrono::Utc::now(),
);
if tagged {
wait.registered_by_tool_call_id = Some("first-arm-call".into());
}
parent
.agent_runtime_state
.as_mut()
.unwrap()
.waiting_for_children = Some(wait.clone());
if tagged {
store.save_runtime_only(&mut parent).await.unwrap();
} else {
store.merge_save_runtime(&mut parent).await.unwrap();
}
let saved = storage.load_session(&parent.id).await.unwrap().unwrap();
assert_eq!(
saved
.agent_runtime_state
.as_ref()
.unwrap()
.waiting_for_children
.as_ref(),
Some(&wait)
);
}
}
#[tokio::test]
async fn finalized_save_adopts_child_completion_after_pipeline_wait_read() {
use bamboo_domain::session::runtime_state::{
AgentRuntimeState, AgentStatusState, ChildWaitPolicy, SuspensionState,
WaitingForChildrenState,
};
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let mut parent = fresh("finalize-fast-child");
let mut runtime = AgentRuntimeState::new("parent-run");
runtime.status = AgentStatusState::Suspended;
runtime.suspension = Some(SuspensionState {
reason: "waiting_for_children".to_string(),
suspended_at: chrono::Utc::now(),
resumable: true,
hook_point: Some("AfterToolExecution".to_string()),
});
let mut wait = WaitingForChildrenState::for_children(
vec!["fast-child".to_string()],
ChildWaitPolicy::All,
chrono::Utc::now(),
);
wait.registered_by_tool_call_id = Some("spawn-call".to_string());
let registered_at = wait.registered_at;
runtime.waiting_for_children = Some(wait);
parent.agent_runtime_state = Some(runtime);
parent.metadata.insert(
"agent.runtime.state".to_string(),
serde_json::to_string(parent.agent_runtime_state.as_ref().unwrap()).unwrap(),
);
parent.metadata.insert(
"runtime.suspend_reason".to_string(),
"waiting_for_children".to_string(),
);
parent.set_last_run_status("suspended");
storage.save_session(&parent).await.unwrap();
let mut stale_runner = parent.clone();
let mut completed = parent;
let completed_runtime = completed.agent_runtime_state.as_mut().unwrap();
completed_runtime.waiting_for_children = None;
completed_runtime.status = AgentStatusState::Idle;
completed_runtime.suspension = None;
completed.metadata.remove("runtime.suspend_reason");
let mut old_resume = Message::user("old consumed child outcome");
old_resume.created_at = registered_at - chrono::Duration::seconds(1);
old_resume.metadata = Some(serde_json::json!({
"runtime_kind": "child_completion_resume"
}));
completed.add_message(old_resume);
let mut outcome = Message::user("fast child outcome");
outcome.metadata = Some(serde_json::json!({
"runtime_kind": "child_completion_resume"
}));
completed.add_message(outcome);
storage.save_session(&completed).await.unwrap();
RuntimeSessionPersistence::save_finalized_runtime_session(&store, &mut stale_runner)
.await
.unwrap();
let saved = storage
.load_session("finalize-fast-child")
.await
.unwrap()
.unwrap();
let state = saved.agent_runtime_state.as_ref().unwrap();
assert!(state.waiting_for_children.is_none());
assert!(state.suspension.is_none());
assert_eq!(state.status, AgentStatusState::Idle);
assert!(!saved.metadata.contains_key("runtime.suspend_reason"));
assert_eq!(saved.last_run_status().as_deref(), Some("completed"));
let mirrored: AgentRuntimeState =
serde_json::from_str(saved.metadata.get("agent.runtime.state").unwrap()).unwrap();
assert!(mirrored.waiting_for_children.is_none());
assert_eq!(mirrored.status, AgentStatusState::Idle);
assert!(saved
.messages
.iter()
.any(|message| message.content == "fast child outcome"));
assert!(!saved
.messages
.iter()
.any(|message| message.content == "old consumed child outcome"));
}
#[tokio::test]
async fn finalized_save_retains_unpersisted_untagged_safety_net_wait() {
use bamboo_domain::session::runtime_state::{
AgentRuntimeState, AgentStatusState, ChildWaitPolicy, WaitingForChildrenState,
};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let mut parent = fresh("finalize-safety-net");
parent.agent_runtime_state = Some(AgentRuntimeState::new("parent-run"));
storage.save_session(&parent).await.unwrap();
let mut runner = parent;
let runtime = runner.agent_runtime_state.as_mut().unwrap();
runtime.status = AgentStatusState::Suspended;
runtime.waiting_for_children = Some(WaitingForChildrenState::for_children(
vec!["active-child".to_string()],
ChildWaitPolicy::All,
chrono::Utc::now(),
));
runner.metadata.insert(
"runtime.suspend_reason".to_string(),
"waiting_for_children".to_string(),
);
runner.set_last_run_status("suspended");
RuntimeSessionPersistence::save_finalized_runtime_session(&store, &mut runner)
.await
.unwrap();
let saved = storage
.load_session("finalize-safety-net")
.await
.unwrap()
.unwrap();
assert!(saved
.agent_runtime_state
.as_ref()
.unwrap()
.waiting_for_children
.is_some());
assert_eq!(
saved
.metadata
.get("runtime.suspend_reason")
.map(String::as_str),
Some("waiting_for_children")
);
assert_eq!(saved.last_run_status().as_deref(), Some("suspended"));
}
#[tokio::test]
async fn merge_save_runtime_preserves_disk_authoritative_metadata_with_single_load() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-merge-meta";
let mut baseline = fresh(session_id);
baseline.title = "Auto Title".to_string();
baseline.metadata_version = 0;
storage.save_session(&baseline).await.unwrap();
let mut stale_snapshot = storage.load_session(session_id).await.unwrap().unwrap();
let mut renamed = storage.load_session(session_id).await.unwrap().unwrap();
renamed.title = "User Renamed".to_string();
renamed.title_version = 1;
renamed.pinned = true;
renamed.metadata_version = 1;
store.commit_metadata(&renamed).await.unwrap();
stale_snapshot.title = "Auto Title".to_string();
store.merge_save_runtime(&mut stale_snapshot).await.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(after.title, "User Renamed");
assert!(after.pinned);
assert_eq!(after.metadata_version, 1);
assert_eq!(stale_snapshot.title, "User Renamed");
assert_eq!(stale_snapshot.metadata_version, 1);
}
#[tokio::test]
async fn merge_save_runtime_preserves_durable_workflow_run_index_from_stale_runner() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-workflow-run-index";
let baseline = fresh(session_id);
storage.save_session(&baseline).await.unwrap();
let mut stale_runner = storage.load_session(session_id).await.unwrap().unwrap();
store
.update_runtime_config(session_id, |session| {
session.metadata.insert(
"workflow.run_ids.v1".to_string(),
r#"["http-started-run"]"#.to_string(),
);
})
.await
.unwrap()
.expect("session exists");
store.merge_save_runtime(&mut stale_runner).await.unwrap();
assert_eq!(
stale_runner
.metadata
.get("workflow.run_ids.v1")
.map(String::as_str),
Some(r#"["http-started-run"]"#)
);
let durable = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(
durable
.metadata
.get("workflow.run_ids.v1")
.map(String::as_str),
Some(r#"["http-started-run"]"#)
);
}
#[tokio::test]
async fn merge_save_runtime_adopts_disk_bypass_permissions() {
use bamboo_domain::AgentRuntimeState;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-bypass";
let baseline = fresh(session_id);
storage.save_session(&baseline).await.unwrap();
let mut loop_snapshot = storage.load_session(session_id).await.unwrap().unwrap();
loop_snapshot.agent_runtime_state = Some(AgentRuntimeState::default());
store
.update_runtime_config(session_id, |s| {
s.agent_runtime_state
.get_or_insert_with(AgentRuntimeState::default)
.bypass_permissions = true;
})
.await
.unwrap()
.expect("session exists");
store.merge_save_runtime(&mut loop_snapshot).await.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert!(
after
.agent_runtime_state
.as_ref()
.is_some_and(|s| s.bypass_permissions),
"disk bypass=ON must survive a stale runtime save (#540)"
);
assert!(loop_snapshot
.agent_runtime_state
.as_ref()
.is_some_and(|s| s.bypass_permissions));
}
#[tokio::test]
async fn merge_save_runtime_adopts_disk_auto_permission_mode() {
use bamboo_domain::{AgentRuntimeState, SessionPermissionMode};
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-auto";
storage.save_session(&fresh(session_id)).await.unwrap();
let mut loop_snapshot = storage.load_session(session_id).await.unwrap().unwrap();
loop_snapshot.agent_runtime_state = Some(AgentRuntimeState::default());
loop_snapshot.metadata.insert(
"permission.requested_mode".to_string(),
"default".to_string(),
);
loop_snapshot.metadata.insert(
"permission.effective_mode".to_string(),
"default".to_string(),
);
loop_snapshot.metadata.insert(
"permission.executor_mapping".to_string(),
"bamboo_runtime:default".to_string(),
);
store
.update_runtime_config(session_id, |session| {
session
.agent_runtime_state
.get_or_insert_with(AgentRuntimeState::default)
.set_permission_mode(SessionPermissionMode::Auto);
session
.metadata
.insert("permission.policy_revision".to_string(), "12".to_string());
session
.metadata
.insert("permission.requested_mode".to_string(), "auto".to_string());
session
.metadata
.insert("permission.effective_mode".to_string(), "auto".to_string());
session.metadata.insert(
"permission.executor_mapping".to_string(),
"bamboo_runtime:auto".to_string(),
);
session.metadata.insert(
"permission.transitioned_at".to_string(),
"2026-07-31T12:00:00Z".to_string(),
);
session.metadata_version = session.metadata_version.saturating_add(1);
})
.await
.unwrap()
.expect("session exists");
store.merge_save_runtime(&mut loop_snapshot).await.unwrap();
let durable = storage.load_session(session_id).await.unwrap().unwrap();
for state in [
durable.agent_runtime_state.as_ref(),
loop_snapshot.agent_runtime_state.as_ref(),
] {
assert_eq!(
state.map(AgentRuntimeState::effective_permission_mode),
Some(SessionPermissionMode::Auto)
);
}
for session in [&durable, &loop_snapshot] {
assert_eq!(
session.metadata.get("permission.policy_revision"),
Some(&"12".to_string())
);
assert_eq!(
session.metadata.get("permission.requested_mode"),
Some(&"auto".to_string())
);
assert_eq!(
session.metadata.get("permission.effective_mode"),
Some(&"auto".to_string())
);
assert_eq!(
session.metadata.get("permission.executor_mapping"),
Some(&"bamboo_runtime:auto".to_string())
);
assert_eq!(
session.metadata.get("permission.transitioned_at"),
Some(&"2026-07-31T12:00:00Z".to_string())
);
}
}
#[tokio::test]
async fn merge_save_runtime_adopts_disk_bypass_off() {
use bamboo_domain::AgentRuntimeState;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-bypass-off";
let mut baseline = fresh(session_id);
let on_state = AgentRuntimeState {
bypass_permissions: true,
..AgentRuntimeState::default()
};
baseline.agent_runtime_state = Some(on_state);
storage.save_session(&baseline).await.unwrap();
let mut loop_snapshot = storage.load_session(session_id).await.unwrap().unwrap();
store
.update_runtime_config(session_id, |s| {
s.agent_runtime_state
.get_or_insert_with(AgentRuntimeState::default)
.bypass_permissions = false;
})
.await
.unwrap()
.expect("session exists");
store.merge_save_runtime(&mut loop_snapshot).await.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert!(
!after
.agent_runtime_state
.as_ref()
.is_some_and(|s| s.bypass_permissions),
"disk bypass=OFF must survive a stale runtime save (#540)"
);
}
#[tokio::test]
async fn save_runtime_authoritative_flags_persists_in_memory_posture_and_audit() {
use bamboo_domain::AgentRuntimeState;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "child-reseed";
let mut baseline = fresh(session_id);
let on_state = AgentRuntimeState {
bypass_permissions: true,
..AgentRuntimeState::default()
};
baseline.agent_runtime_state = Some(on_state);
for (key, value) in [
("permission.policy_revision", "12"),
("permission.requested_mode", "bypass"),
("permission.effective_mode", "bypass"),
("permission.executor_mapping", "bamboo_runtime:bypass"),
("permission.transitioned_at", "2026-07-31T12:00:00Z"),
] {
baseline.metadata.insert(key.to_string(), value.to_string());
}
storage.save_session(&baseline).await.unwrap();
let mut child = storage.load_session(session_id).await.unwrap().unwrap();
child
.agent_runtime_state
.get_or_insert_with(AgentRuntimeState::default)
.bypass_permissions = false;
for (key, value) in [
("permission.policy_revision", "13"),
("permission.requested_mode", "default"),
("permission.effective_mode", "default"),
("permission.executor_mapping", "bamboo_runtime:default"),
("permission.transitioned_at", "2026-07-31T12:01:00Z"),
] {
child.metadata.insert(key.to_string(), value.to_string());
}
store
.save_runtime_authoritative_flags(&mut child)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert!(
!after
.agent_runtime_state
.as_ref()
.is_some_and(|s| s.bypass_permissions),
"authoritative re-seed of bypass=OFF must persist, not be reverted (#540/#74)"
);
for (key, value) in [
("permission.policy_revision", "13"),
("permission.requested_mode", "default"),
("permission.effective_mode", "default"),
("permission.executor_mapping", "bamboo_runtime:default"),
("permission.transitioned_at", "2026-07-31T12:01:00Z"),
] {
assert_eq!(after.metadata.get(key).map(String::as_str), Some(value));
}
}
#[tokio::test]
async fn merge_save_runtime_leaves_bypass_when_disk_has_no_runtime_state() {
use bamboo_domain::AgentRuntimeState;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "no-runtime-state";
let baseline = fresh(session_id);
assert!(baseline.agent_runtime_state.is_none());
storage.save_session(&baseline).await.unwrap();
let mut running = storage.load_session(session_id).await.unwrap().unwrap();
let on_state = AgentRuntimeState {
bypass_permissions: true,
..AgentRuntimeState::default()
};
running.agent_runtime_state = Some(on_state);
store.merge_save_runtime(&mut running).await.unwrap();
assert!(
running
.agent_runtime_state
.as_ref()
.is_some_and(|s| s.bypass_permissions),
"a runtime-state-less disk copy must not force bypass OFF (#540)"
);
}
#[tokio::test]
async fn merge_preserves_disk_title_when_versions_equal() {
let (_temp, storage) = make_storage().await;
let session_id = "merge-equal";
let mut on_disk = fresh(session_id);
on_disk.title = "User Set This".to_string();
on_disk.title_version = 0;
on_disk.title_generated = true;
on_disk.metadata_version = 0;
storage.save_session(&on_disk).await.unwrap();
let mut runtime_copy = fresh(session_id);
runtime_copy.created_at = on_disk.created_at;
runtime_copy.title = "Stale Default".to_string();
runtime_copy.title_version = 0;
runtime_copy.title_generated = false;
runtime_copy.metadata_version = 0;
runtime_copy.messages = vec![];
merge_save_session(&storage, &mut runtime_copy)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(after.title, "User Set This");
assert_eq!(after.title_version, 0);
assert!(after.title_generated);
assert_eq!(runtime_copy.title, "User Set This");
assert!(runtime_copy.title_generated);
}
#[tokio::test]
async fn merge_preserves_disk_when_disk_version_higher() {
let (_temp, storage) = make_storage().await;
let session_id = "merge-higher";
let mut on_disk = fresh(session_id);
on_disk.title = "User Title v3".to_string();
on_disk.title_version = 3;
on_disk.metadata_version = 5;
storage.save_session(&on_disk).await.unwrap();
let mut runtime_copy = fresh(session_id);
runtime_copy.created_at = on_disk.created_at;
runtime_copy.title = "Stale".to_string();
runtime_copy.title_version = 1;
runtime_copy.metadata_version = 0;
merge_save_session(&storage, &mut runtime_copy)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(after.title, "User Title v3");
assert_eq!(after.title_version, 3);
assert_eq!(after.metadata_version, 5);
}
#[tokio::test]
async fn merge_now_preserves_disk_pinned_in_metadata_group() {
let (_temp, storage) = make_storage().await;
let session_id = "pinned-merge";
let mut on_disk = fresh(session_id);
on_disk.pinned = true;
on_disk.metadata_version = 2;
storage.save_session(&on_disk).await.unwrap();
let mut runtime_copy = fresh(session_id);
runtime_copy.created_at = on_disk.created_at;
runtime_copy.pinned = false;
runtime_copy.metadata_version = 0;
merge_save_session(&storage, &mut runtime_copy)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert!(
after.pinned,
"disk pinned=true should win over runtime false"
);
assert_eq!(after.metadata_version, 2);
}
#[tokio::test]
async fn merge_keeps_in_memory_when_session_version_higher() {
let (_temp, storage) = make_storage().await;
let session_id = "merge-bumped";
let mut on_disk = fresh(session_id);
on_disk.title = "Old".to_string();
on_disk.title_version = 1;
on_disk.metadata_version = 3;
storage.save_session(&on_disk).await.unwrap();
let mut authoritative_copy = fresh(session_id);
authoritative_copy.created_at = on_disk.created_at;
authoritative_copy.title = "New Authoritative".to_string();
authoritative_copy.title_version = 2;
authoritative_copy.metadata_version = 4;
authoritative_copy.pinned = true;
merge_save_session(&storage, &mut authoritative_copy)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(after.title, "New Authoritative");
assert_eq!(after.title_version, 2);
assert_eq!(after.metadata_version, 4);
assert!(after.pinned);
}
#[tokio::test]
async fn merge_keeps_runtime_messages_when_disk_only_changed_metadata() {
let (_temp, storage) = make_storage().await;
let session_id = "merge-messages";
let mut on_disk = fresh(session_id);
on_disk.title = "Fresh Title".to_string();
on_disk.title_version = 2;
on_disk.metadata_version = 5;
storage.save_session(&on_disk).await.unwrap();
let mut runtime_copy = fresh(session_id);
runtime_copy.created_at = on_disk.created_at;
runtime_copy.title = "Stale".to_string();
runtime_copy.metadata_version = 0;
runtime_copy.messages = vec![bamboo_domain::session::types::Message {
role: bamboo_domain::session::types::Role::User,
content: "keep me".to_string(),
id: "msg-1".to_string(),
created_at: chrono::Utc::now(),
reasoning: None,
reasoning_signature: None,
content_parts: None,
image_ocr: None,
phase: None,
tool_calls: None,
tool_call_id: None,
tool_success: None,
compressed: false,
compressed_by_event_id: None,
never_compress: false,
compression_level: 0,
metadata: None,
}];
merge_save_session(&storage, &mut runtime_copy)
.await
.unwrap();
let after = storage.load_session(session_id).await.unwrap().unwrap();
assert_eq!(after.title, "Fresh Title");
assert_eq!(after.metadata_version, 5);
assert_eq!(after.messages.len(), 1);
assert_eq!(after.messages[0].content, "keep me");
}
#[tokio::test]
async fn runtime_control_plane_port_uses_sidecar_without_rewriting_messages() {
use bamboo_domain::session::types::Message;
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "runtime-control-plane";
let mut durable = fresh(session_id);
durable.add_message(Message::user("durable transcript"));
storage.save_session(&durable).await.unwrap();
let mut runtime = durable.clone();
runtime.model = "updated-control-plane-model".to_string();
runtime.add_message(Message::assistant("uncheckpointed runtime message", None));
RuntimeSessionPersistence::save_runtime_control_plane(&store, &mut runtime)
.await
.unwrap();
let control_plane =
RuntimeSessionPersistence::load_runtime_control_plane(&store, session_id)
.await
.unwrap()
.expect("control-plane exists");
assert!(
control_plane.messages.is_empty(),
"LockedSessionStore must expose its message-free sidecar"
);
assert_eq!(control_plane.model, "updated-control-plane-model");
let reloaded = storage
.load_session(session_id)
.await
.unwrap()
.expect("session exists");
assert_eq!(reloaded.model, "updated-control-plane-model");
assert_eq!(
reloaded.messages.len(),
1,
"control-plane save must not write the uncheckpointed message"
);
assert_eq!(reloaded.messages[0].content, "durable transcript");
}
#[tokio::test]
async fn atomic_task_patch_loads_inside_lock_and_preserves_interleaved_runtime_state() {
let temp = tempfile::tempdir().unwrap();
let inner = Arc::new(
SessionStoreV2::new(temp.path().to_path_buf())
.await
.expect("storage init"),
);
let session_id = "atomic-task-patch";
inner
.save_session(&fresh(session_id))
.await
.expect("seed session");
let counted = Arc::new(CountingControlPlaneStorage {
inner: inner.clone(),
control_plane_loads: AtomicUsize::new(0),
full_saves: AtomicUsize::new(0),
runtime_state_saves: AtomicUsize::new(0),
});
let storage: Arc<dyn Storage> = counted.clone();
let store = Arc::new(LockedSessionStore::new(storage));
let guard = store.acquire_lock(session_id).await;
let now = chrono::Utc::now();
let task_list = bamboo_domain::TaskList {
session_id: session_id.to_string(),
title: "Atomic Task patch".to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let patch_store = store.clone();
let patch = tokio::spawn(async move {
let _ = started_tx.send(());
RuntimeSessionPersistence::update_task_list_control_plane(
patch_store.as_ref(),
session_id,
&task_list,
"9",
)
.await
});
started_rx.await.expect("patch task started");
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
assert_eq!(
counted.control_plane_loads.load(Ordering::SeqCst),
0,
"Task patch must acquire the session lock before loading its snapshot"
);
let mut latest = inner
.load_runtime_control_plane(session_id)
.await
.expect("load latest control-plane")
.expect("control-plane exists");
latest.agent_runtime_state = Some(bamboo_domain::AgentRuntimeState::new("latest-run"));
latest
.metadata
.insert("concurrent.runtime".to_string(), "preserve".to_string());
inner
.save_runtime_state(&latest)
.await
.expect("publish concurrent runtime transition");
drop(guard);
assert!(
patch.await.expect("patch join").expect("patch succeeds"),
"existing root must be patched"
);
assert_eq!(counted.control_plane_loads.load(Ordering::SeqCst), 1);
let reloaded = inner
.load_session(session_id)
.await
.expect("reload")
.expect("session exists");
assert_eq!(
reloaded
.agent_runtime_state
.as_ref()
.map(|state| state.run_id.as_str()),
Some("latest-run")
);
assert_eq!(
reloaded
.metadata
.get("concurrent.runtime")
.map(String::as_str),
Some("preserve")
);
assert_eq!(reloaded.task_list_version_meta().as_deref(), Some("9"));
assert_eq!(
reloaded.task_list.as_ref().map(|list| list.title.as_str()),
Some("Atomic Task patch")
);
}
#[tokio::test]
async fn paired_task_cas_conflict_cannot_overwrite_newer_root_or_child_state() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let root_id = "task-cas-root";
let child_id = "task-cas-child";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.set_task_list(task_list("newer root"));
root.set_task_list_version_meta("2");
storage.save_session(&root).await.expect("seed root");
let mut child = Session::new_child(child_id, root_id, "model", "child");
child.set_task_list(task_list("current child"));
child.set_task_list_version_meta("1");
storage.save_session(&child).await.expect("seed child");
let updated = RuntimeSessionPersistence::update_task_list_control_planes_if_version(
&store,
child_id,
root_id,
"1",
&task_list("current child"),
&task_list("stale evaluator"),
"3",
)
.await
.expect("CAS returns clean conflict");
assert!(
!updated,
"mismatched root generation must reject both writes"
);
let durable_root = storage
.load_session(root_id)
.await
.expect("load root")
.expect("root exists");
let durable_child = storage
.load_session(child_id)
.await
.expect("load child")
.expect("child exists");
assert_eq!(durable_root.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
durable_root
.task_list
.as_ref()
.map(|list| list.title.as_str()),
Some("newer root")
);
assert_eq!(durable_child.task_list_version_meta().as_deref(), Some("1"));
assert_eq!(
durable_child
.task_list
.as_ref()
.map(|list| list.title.as_str()),
Some("current child")
);
}
#[tokio::test]
async fn paired_task_cas_success_uses_only_targeted_saves_and_preserves_both_transcripts() {
use bamboo_domain::session::types::Message;
let temp = tempfile::tempdir().unwrap();
let inner = Arc::new(
SessionStoreV2::new(temp.path().to_path_buf())
.await
.expect("storage init"),
);
let root_id = "task-cas-success-root";
let child_id = "task-cas-success-child";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.add_message(Message::user("root transcript"));
root.metadata
.insert("unrelated.root".to_string(), "preserve".to_string());
root.agent_runtime_state = Some(bamboo_domain::AgentRuntimeState::new("root-run"));
root.set_task_list(task_list("old shared"));
root.set_task_list_version_meta("1");
inner.save_session(&root).await.expect("seed root");
let mut child = Session::new_child(child_id, root_id, "model", "child");
child.add_message(Message::user("child transcript"));
child
.metadata
.insert("unrelated.child".to_string(), "preserve".to_string());
child.agent_runtime_state = Some(bamboo_domain::AgentRuntimeState::new("child-run"));
child.set_task_list(task_list("old shared"));
child.set_task_list_version_meta("1");
inner.save_session(&child).await.expect("seed child");
let counted = Arc::new(CountingControlPlaneStorage {
inner: inner.clone(),
control_plane_loads: AtomicUsize::new(0),
full_saves: AtomicUsize::new(0),
runtime_state_saves: AtomicUsize::new(0),
});
let storage: Arc<dyn Storage> = counted.clone();
let store = LockedSessionStore::new(storage);
assert!(
RuntimeSessionPersistence::update_task_list_control_planes_if_version(
&store,
child_id,
root_id,
"1",
&task_list("old shared"),
&task_list("evaluated"),
"2",
)
.await
.expect("paired CAS succeeds")
);
assert_eq!(counted.control_plane_loads.load(Ordering::SeqCst), 2);
assert_eq!(counted.runtime_state_saves.load(Ordering::SeqCst), 2);
assert_eq!(
counted.full_saves.load(Ordering::SeqCst),
0,
"evaluation CAS must not call full save_session for child or root"
);
let durable_root = inner.load_session(root_id).await.unwrap().unwrap();
let durable_child = inner.load_session(child_id).await.unwrap().unwrap();
for (session, transcript, metadata_key, run_id) in [
(
&durable_root,
"root transcript",
"unrelated.root",
"root-run",
),
(
&durable_child,
"child transcript",
"unrelated.child",
"child-run",
),
] {
assert_eq!(session.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
session.task_list.as_ref().map(|list| list.title.as_str()),
Some("evaluated")
);
assert_eq!(session.messages.len(), 1);
assert_eq!(session.messages[0].content, transcript);
assert_eq!(
session.metadata.get(metadata_key).map(String::as_str),
Some("preserve")
);
assert_eq!(
session
.agent_runtime_state
.as_ref()
.map(|state| state.run_id.as_str()),
Some(run_id)
);
}
}
#[tokio::test]
async fn locked_runtime_and_full_saves_adopt_task_conflicts_before_publish() {
let temp = tempfile::tempdir().unwrap();
let home = temp.path().to_path_buf();
let first_storage = Arc::new(
SessionStoreV2::new(home.clone())
.await
.expect("first storage init"),
);
let second_storage = Arc::new(
SessionStoreV2::new(home)
.await
.expect("second storage init"),
);
let now = chrono::Utc::now();
let task_list = |session_id: &str, title: &str| bamboo_domain::TaskList {
session_id: session_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let runtime_id = "ordinary-task-retry-runtime";
let full_id = "ordinary-task-retry-full";
let mut runtime_initial = fresh(runtime_id);
runtime_initial.set_task_list(task_list(runtime_id, "runtime v1"));
runtime_initial.set_task_list_version_meta("1");
first_storage
.save_session(&runtime_initial)
.await
.expect("seed runtime session");
let mut full_initial = fresh(full_id);
full_initial.set_task_list(task_list(full_id, "full v1"));
full_initial.set_task_list_version_meta("1");
first_storage
.save_session(&full_initial)
.await
.expect("seed full session");
let mut runtime_advanced = runtime_initial.clone();
runtime_advanced.set_task_list(task_list(runtime_id, "runtime v2"));
runtime_advanced.set_task_list_version_meta("2");
second_storage
.save_runtime_state(&runtime_advanced)
.await
.expect("advance runtime Task generation");
let mut full_advanced = full_initial.clone();
full_advanced.set_task_list(task_list(full_id, "full v2"));
full_advanced.set_task_list_version_meta("2");
second_storage
.save_runtime_state(&full_advanced)
.await
.expect("advance full Task generation");
let storage: Arc<dyn Storage> = first_storage.clone();
let store = LockedSessionStore::new(storage);
let runtime_published = Arc::new(std::sync::Mutex::new(None));
let runtime_callback = runtime_published.clone();
let mut runtime_stale = runtime_initial;
runtime_stale
.metadata
.insert("runtime.non-task".to_string(), "preserved".to_string());
store
.save_runtime_only_and_publish(&mut runtime_stale, move |saved| {
*runtime_callback.lock().expect("runtime publish lock") = Some(saved.clone());
})
.await
.expect("locked runtime save rebases and retries");
let full_published = Arc::new(std::sync::Mutex::new(None));
let full_callback = full_published.clone();
let mut full_stale = full_initial;
full_stale
.metadata
.insert("full.non-task".to_string(), "preserved".to_string());
store
.merge_save_runtime_and_publish(&mut full_stale, move |saved, committed| {
assert!(committed);
*full_callback.lock().expect("full publish lock") = Some(saved.clone());
})
.await
.expect("locked full save rebases and retries");
let durable_runtime = first_storage
.load_session(runtime_id)
.await
.unwrap()
.expect("durable runtime session");
let durable_full = first_storage
.load_session(full_id)
.await
.unwrap()
.expect("durable full session");
let published_runtime = runtime_published
.lock()
.expect("runtime publish lock")
.clone()
.expect("runtime published snapshot");
let published_full = full_published
.lock()
.expect("full publish lock")
.clone()
.expect("full published snapshot");
for (session, expected_title, metadata_key) in [
(&runtime_stale, "runtime v2", "runtime.non-task"),
(&durable_runtime, "runtime v2", "runtime.non-task"),
(&published_runtime, "runtime v2", "runtime.non-task"),
(&full_stale, "full v2", "full.non-task"),
(&durable_full, "full v2", "full.non-task"),
(&published_full, "full v2", "full.non-task"),
] {
assert_eq!(session.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
session.task_list.as_ref().map(|list| list.title.as_str()),
Some(expected_title)
);
assert_eq!(
session.metadata.get(metadata_key).map(String::as_str),
Some("preserved")
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn single_task_cas_rejects_same_version_divergent_snapshot_without_callback() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let session_id = "single-task-exact-snapshot";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: session_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let durable_winner = task_list("durable winner");
let stale_snapshot = task_list("stale same-version snapshot");
let mut session = fresh(session_id);
session.set_task_list(durable_winner.clone());
session.set_task_list_version_meta("1");
storage.save_session(&session).await.expect("seed session");
let published = Arc::new(AtomicBool::new(false));
let callback = published.clone();
assert!(!store
.update_task_list_control_plane_if_version_and_publish(
session_id,
"1",
&stale_snapshot,
&task_list("stale evaluation"),
"2",
move |_| callback.store(true, Ordering::SeqCst),
)
.await
.expect("same-version divergence is a clean stale result"));
assert!(!published.load(Ordering::SeqCst));
let durable = storage
.load_session(session_id)
.await
.unwrap()
.expect("session remains");
assert_eq!(durable.task_list_version_meta().as_deref(), Some("1"));
assert_eq!(
serde_json::to_value(&durable.task_list).expect("serialize durable Task list"),
serde_json::to_value(Some(&durable_winner)).expect("serialize expected Task list")
);
}
#[tokio::test]
async fn paired_task_cas_rejects_same_version_divergent_snapshot_without_callback() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let root_id = "paired-task-exact-snapshot-root";
let child_id = "paired-task-exact-snapshot-child";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let durable_winner = task_list("durable winner");
let stale_snapshot = task_list("stale same-version snapshot");
let mut root = fresh(root_id);
root.set_task_list(durable_winner.clone());
root.set_task_list_version_meta("1");
storage.save_session(&root).await.expect("seed root");
let mut child = Session::new_child(child_id, root_id, "model", "child");
child.set_task_list(durable_winner.clone());
child.set_task_list_version_meta("1");
storage.save_session(&child).await.expect("seed child");
let published = Arc::new(AtomicBool::new(false));
let callback = published.clone();
assert!(!store
.update_task_list_control_planes_if_version_and_publish(
child_id,
root_id,
"1",
&stale_snapshot,
&task_list("stale evaluation"),
"2",
move |_, _| callback.store(true, Ordering::SeqCst),
)
.await
.expect("same-version divergence is a clean stale result"));
assert!(!published.load(Ordering::SeqCst));
for id in [child_id, root_id] {
let durable = storage
.load_session(id)
.await
.unwrap()
.expect("session remains");
assert_eq!(durable.task_list_version_meta().as_deref(), Some("1"));
assert_eq!(
serde_json::to_value(&durable.task_list).expect("serialize durable Task list"),
serde_json::to_value(Some(&durable_winner)).expect("serialize expected Task list")
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn unconditional_root_task_patch_reports_final_cas_conflict_without_publishing() {
let temp = tempfile::tempdir().unwrap();
let home = temp.path().to_path_buf();
let first_inner = Arc::new(
SessionStoreV2::new(home.clone())
.await
.expect("first storage init"),
);
let root_id = "single-task-unconditional-loser";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.task_list = Some(task_list("original"));
root.set_task_list_version_meta("1");
first_inner.save_session(&root).await.expect("seed root");
let second_inner = Arc::new(
SessionStoreV2::new(home)
.await
.expect("second storage init"),
);
let commit_reached = Arc::new(tokio::sync::Barrier::new(2));
let release_commit = Arc::new(tokio::sync::Barrier::new(2));
let storage: Arc<dyn Storage> = Arc::new(SingleCommitPauseStorage {
inner: first_inner.clone(),
commit_reached: commit_reached.clone(),
release_commit: release_commit.clone(),
});
let store = LockedSessionStore::new(storage);
let published = Arc::new(AtomicBool::new(false));
let callback = published.clone();
let loser_candidate = task_list("loser");
let loser = store.update_task_list_control_plane_and_publish(
root_id,
&loser_candidate,
"2",
move |_| callback.store(true, Ordering::SeqCst),
);
let winner = async {
commit_reached.wait().await;
let original = second_inner
.load_runtime_control_plane(root_id)
.await
.expect("load winner original")
.expect("winner original exists");
let mut updated = original.clone();
updated.task_list = Some(task_list("winner"));
updated.set_task_list_version_meta("2");
assert!(second_inner
.save_task_control_plane_if_matches(&original, &updated)
.await
.expect("commit winner"));
release_commit.wait().await;
};
let (loser_result, ()) = tokio::join!(loser, winner);
let error = loser_result.expect_err("unconditional loser must be an explicit conflict");
assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
assert!(!published.load(Ordering::SeqCst));
let durable = first_inner
.load_session(root_id)
.await
.unwrap()
.expect("durable root");
assert_eq!(durable.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
durable.task_list.as_ref().map(|list| list.title.as_str()),
Some("winner")
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn independent_root_task_patches_have_one_final_cas_winner() {
let temp = tempfile::tempdir().unwrap();
let home = temp.path().to_path_buf();
let first_inner = Arc::new(
SessionStoreV2::new(home.clone())
.await
.expect("first storage init"),
);
let root_id = "single-task-cas-race-root";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.set_task_list(task_list("original"));
root.set_task_list_version_meta("1");
first_inner.save_session(&root).await.expect("seed root");
let second_inner = Arc::new(
SessionStoreV2::new(home)
.await
.expect("second storage init"),
);
let before_commit = Arc::new(tokio::sync::Barrier::new(2));
let first_storage: Arc<dyn Storage> = Arc::new(SingleCommitBarrierStorage {
inner: first_inner.clone(),
before_commit: before_commit.clone(),
});
let second_storage: Arc<dyn Storage> = Arc::new(SingleCommitBarrierStorage {
inner: second_inner.clone(),
before_commit,
});
let first_store = LockedSessionStore::new(first_storage);
let second_store = LockedSessionStore::new(second_storage);
let first_published = Arc::new(AtomicBool::new(false));
let second_published = Arc::new(AtomicBool::new(false));
let first_callback = first_published.clone();
let second_callback = second_published.clone();
let expected = task_list("original");
let first_candidate = task_list("candidate one");
let second_candidate = task_list("candidate two");
let first = first_store.update_task_list_control_plane_if_version_and_publish(
root_id,
"1",
&expected,
&first_candidate,
"2",
move |_| first_callback.store(true, Ordering::SeqCst),
);
let second = second_store.update_task_list_control_plane_if_version_and_publish(
root_id,
"1",
&expected,
&second_candidate,
"2",
move |_| second_callback.store(true, Ordering::SeqCst),
);
let (first_result, second_result) = tokio::join!(first, second);
let first_won = first_result.expect("first root result");
let second_won = second_result.expect("second root result");
assert_ne!(first_won, second_won, "exactly one root candidate wins");
assert_eq!(first_published.load(Ordering::SeqCst), first_won);
assert_eq!(second_published.load(Ordering::SeqCst), second_won);
let durable = first_inner
.load_session(root_id)
.await
.unwrap()
.expect("root");
assert_eq!(durable.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
durable.task_list.as_ref().map(|list| list.title.as_str()),
Some(if first_won {
"candidate one"
} else {
"candidate two"
})
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn independent_locked_stores_revalidate_pair_cas_at_storage_commit_point() {
let temp = tempfile::tempdir().unwrap();
let home = temp.path().to_path_buf();
let first_inner = Arc::new(
SessionStoreV2::new(home.clone())
.await
.expect("first storage init"),
);
let root_id = "task-cas-race-root";
let child_id = "task-cas-race-child";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.set_task_list(task_list("original shared"));
root.set_task_list_version_meta("1");
first_inner.save_session(&root).await.expect("seed root");
let mut child = Session::new_child(child_id, root_id, "model", "child");
child.set_task_list(task_list("original shared"));
child.set_task_list_version_meta("1");
first_inner.save_session(&child).await.expect("seed child");
let second_inner = Arc::new(
SessionStoreV2::new(home)
.await
.expect("second storage init"),
);
let before_commit = Arc::new(tokio::sync::Barrier::new(2));
let first_storage: Arc<dyn Storage> = Arc::new(PairCommitBarrierStorage {
inner: first_inner.clone(),
before_commit: before_commit.clone(),
});
let second_storage: Arc<dyn Storage> = Arc::new(PairCommitBarrierStorage {
inner: second_inner.clone(),
before_commit,
});
let first_store = LockedSessionStore::new(first_storage);
let second_store = LockedSessionStore::new(second_storage);
let first_published = Arc::new(AtomicBool::new(false));
let second_published = Arc::new(AtomicBool::new(false));
let first_published_callback = first_published.clone();
let second_published_callback = second_published.clone();
let expected = task_list("original shared");
let first_candidate = task_list("candidate one");
let second_candidate = task_list("candidate two");
let first = first_store.update_task_list_control_planes_if_version_and_publish(
child_id,
root_id,
"1",
&expected,
&first_candidate,
"2",
move |_, _| first_published_callback.store(true, Ordering::SeqCst),
);
let second = second_store.update_task_list_control_planes_if_version_and_publish(
child_id,
root_id,
"1",
&expected,
&second_candidate,
"2",
move |_, _| second_published_callback.store(true, Ordering::SeqCst),
);
let (first_result, second_result) = tokio::join!(first, second);
let first_won = first_result.expect("first CAS result");
let second_won = second_result.expect("second CAS result");
assert_ne!(first_won, second_won, "exactly one staged v1 CAS may win");
assert_eq!(first_published.load(Ordering::SeqCst), first_won);
assert_eq!(second_published.load(Ordering::SeqCst), second_won);
let expected_title = if first_won {
"candidate one"
} else {
"candidate two"
};
let durable_child = first_inner
.load_session(child_id)
.await
.unwrap()
.expect("child");
let durable_root = second_inner
.load_session(root_id)
.await
.unwrap()
.expect("root");
for session in [&durable_child, &durable_root] {
assert_eq!(session.task_list_version_meta().as_deref(), Some("2"));
assert_eq!(
session.task_list.as_ref().map(|list| list.title.as_str()),
Some(expected_title)
);
}
}
#[tokio::test]
async fn paired_task_second_write_failure_rolls_back_and_skips_publish_callback() {
use bamboo_domain::session::types::Message;
let temp = tempfile::tempdir().unwrap();
let inner = Arc::new(
SessionStoreV2::new(temp.path().to_path_buf())
.await
.expect("storage init"),
);
let root_id = "task-cas-failure-root";
let child_id = "task-cas-failure-child";
let now = chrono::Utc::now();
let task_list = |title: &str| bamboo_domain::TaskList {
session_id: root_id.to_string(),
title: title.to_string(),
items: Vec::new(),
created_at: now,
updated_at: now,
};
let mut root = fresh(root_id);
root.add_message(Message::user("root transcript"));
root.metadata
.insert("unrelated.root".to_string(), "preserve".to_string());
root.set_task_list(task_list("old shared"));
root.set_task_list_version_meta("1");
inner.save_session(&root).await.expect("seed root");
let mut child = Session::new_child(child_id, root_id, "model", "child");
child.add_message(Message::user("child transcript"));
child
.metadata
.insert("unrelated.child".to_string(), "preserve".to_string());
child.set_task_list(task_list("old shared"));
child.set_task_list_version_meta("1");
inner.save_session(&child).await.expect("seed child");
inner
.inject_runtime_task_transaction_fault(RuntimeTaskTransactionFault::SecondUpdatedWrite);
let storage: Arc<dyn Storage> = inner.clone();
let store = LockedSessionStore::new(storage);
let published = Arc::new(AtomicBool::new(false));
let published_for_callback = published.clone();
let error = store
.update_task_list_control_planes_if_version_and_publish(
child_id,
root_id,
"1",
&task_list("old shared"),
&task_list("must roll back"),
"2",
move |_, _| published_for_callback.store(true, Ordering::SeqCst),
)
.await
.expect_err("injected second write fails");
assert!(error.to_string().contains("rolled back"), "{error}");
assert!(
!published.load(Ordering::SeqCst),
"durable failure must not publish either cache snapshot"
);
let durable_root = inner.load_session(root_id).await.unwrap().unwrap();
let durable_child = inner.load_session(child_id).await.unwrap().unwrap();
for (session, title, transcript, metadata_key) in [
(
&durable_root,
"old shared",
"root transcript",
"unrelated.root",
),
(
&durable_child,
"old shared",
"child transcript",
"unrelated.child",
),
] {
assert_eq!(session.task_list_version_meta().as_deref(), Some("1"));
assert_eq!(
session.task_list.as_ref().map(|list| list.title.as_str()),
Some(title)
);
assert_eq!(session.messages[0].content, transcript);
assert_eq!(
session.metadata.get(metadata_key).map(String::as_str),
Some("preserve")
);
}
}
#[tokio::test]
async fn locked_merge_save_runtime_serialises_concurrent_writes() {
let (_temp, storage) = make_storage().await;
let store = Arc::new(LockedSessionStore::new(storage));
let session_id = "lock-serial".to_string();
let base = fresh(&session_id);
store.storage().save_session(&base).await.unwrap();
let store_a = store.clone();
let store_b = store.clone();
let sid_a = session_id.clone();
let sid_b = session_id.clone();
let a = tokio::spawn(async move {
let _guard = store_a.acquire_lock(&sid_a).await;
let mut s = store_a
.storage()
.load_session(&sid_a)
.await
.unwrap()
.unwrap();
s.title = "Writer A".to_string();
s.title_version = s.title_version.saturating_add(1);
s.metadata_version = s.metadata_version.saturating_add(1);
s.updated_at = chrono::Utc::now();
store_a.storage().save_session(&s).await.unwrap();
s.title_version
});
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
let b = tokio::spawn(async move {
let _guard = store_b.acquire_lock(&sid_b).await;
let mut s = store_b
.storage()
.load_session(&sid_b)
.await
.unwrap()
.unwrap();
s.title = "Writer B".to_string();
s.title_version = s.title_version.saturating_add(1);
s.metadata_version = s.metadata_version.saturating_add(1);
s.updated_at = chrono::Utc::now();
store_b.storage().save_session(&s).await.unwrap();
s.title_version
});
let (ver_a, ver_b) = tokio::join!(a, b);
let final_s = store
.storage()
.load_session(&session_id)
.await
.unwrap()
.unwrap();
assert!(
ver_a.unwrap() != ver_b.unwrap(),
"concurrent writers must produce distinct versions"
);
assert_eq!(final_s.metadata_version, 2);
}
#[tokio::test]
async fn commit_metadata_is_plain_save_inside_lock() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage);
let session_id = "commit-plain";
let mut s = fresh(session_id);
s.title = "Committed".to_string();
s.metadata_version = 1;
s.title_version = 2;
store.commit_metadata(&s).await.unwrap();
let after = store
.storage()
.load_session(session_id)
.await
.unwrap()
.unwrap();
assert_eq!(after.title, "Committed");
assert_eq!(after.metadata_version, 1);
assert_eq!(after.title_version, 2);
}
#[tokio::test]
async fn validate_lock_requires_active_same_session_same_map_and_accepts_bound_view() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage.clone());
let independent = LockedSessionStore::new(storage);
let owner = store.acquire_lock("owned").await;
assert!(store.validate_lock(&owner, "owned").is_ok());
assert!(store.validate_lock(&owner, "other").is_err());
assert!(independent.validate_lock(&owner, "owned").is_err());
let mut session = fresh("owned");
let mut runtime = bamboo_domain::AgentRuntimeState::new("owned");
runtime.waiting_for_children = Some(bamboo_domain::WaitingForChildrenState::for_children(
vec!["child".into()],
Default::default(),
chrono::Utc::now(),
));
session.agent_runtime_state = Some(runtime);
let bound = store.bind_inherited_child_wait(
bamboo_domain::InheritedChildWait::capture(&session).unwrap(),
);
assert!(bound.validate_lock(&owner, "owned").is_ok());
let inactive = SessionLockGuard {
guard: None,
locks: store.locks.clone(),
session_id: "owned".into(),
};
assert!(store.validate_lock(&inactive, "owned").is_err());
drop(inactive);
let mut waiter = Box::pin(bound.acquire_lock("owned"));
assert!(
tokio::time::timeout(std::time::Duration::from_millis(30), &mut waiter)
.await
.is_err()
);
drop(waiter);
drop(owner);
assert_eq!(store.locks.len(), 0);
let successor = tokio::time::timeout(
std::time::Duration::from_secs(1),
bound.acquire_lock("owned"),
)
.await
.unwrap();
assert!(store.validate_lock(&successor, "owned").is_ok());
}
#[tokio::test]
async fn acquire_lock_self_evicts_when_no_other_holder() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage);
{
let _guard = store.acquire_lock("solo").await;
assert_eq!(store.locks.len(), 1, "entry present while the lock is held");
}
assert_eq!(
store.locks.len(),
0,
"lock entry must be evicted once released with no other holder"
);
}
#[tokio::test]
async fn acquire_lock_many_distinct_ids_do_not_accumulate() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage);
for i in 0..100 {
let _guard = store.acquire_lock(&format!("sess-{i}")).await;
}
assert_eq!(
store.locks.len(),
0,
"acquiring locks for many distinct ids must not grow the map"
);
}
#[tokio::test]
async fn cancelled_last_waiter_reclaims_hundreds_of_session_locks() {
let (_temp, storage) = make_storage().await;
let store = LockedSessionStore::new(storage);
for index in 0..512 {
let id = format!("cancelled-child-{index}");
let held = store.acquire_lock(&id).await;
let mut waiter = Box::pin(store.acquire_lock(&id));
assert!(
std::future::poll_fn(|cx| std::task::Poll::Ready(
std::future::Future::poll(waiter.as_mut(), cx).is_pending()
))
.await
);
drop(held);
drop(waiter);
}
assert!(store.locks.is_empty());
}
#[tokio::test]
async fn acquire_lock_concurrent_waiter_keeps_valid_lock_and_map_drains() {
use std::sync::atomic::{AtomicUsize, Ordering};
let (_temp, storage) = make_storage().await;
let store = Arc::new(LockedSessionStore::new(storage));
let active = Arc::new(AtomicUsize::new(0));
let max_seen = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for _ in 0..8 {
let store = store.clone();
let active = active.clone();
let max_seen = max_seen.clone();
handles.push(tokio::spawn(async move {
let _guard = store.acquire_lock("contended").await;
let now = active.fetch_add(1, Ordering::SeqCst) + 1;
max_seen.fetch_max(now, Ordering::SeqCst);
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
active.fetch_sub(1, Ordering::SeqCst);
}));
}
for h in handles {
h.await.unwrap();
}
assert_eq!(
max_seen.load(Ordering::SeqCst),
1,
"at most one holder of a given session lock at a time"
);
assert_eq!(
store.locks.len(),
0,
"after all holders release, the contended entry must be fully evicted"
);
}
}