bamboo-storage 2026.10.21

//! Mode control-plane resets share the real Actor writer/claim boundary.
use super::*;
use bamboo_domain::{
    ActorActivationClaim, ActorActivationFinish, ActorDirectoryPort, ConversationSummary,
    ModelContextResetReason, ModelContextState, RootActorRuntimeWrite, SessionAuthorityConflict,
};

fn root() -> Session {
    let mut root = Session::new("activated-mode-root", "model");
    root.model_context_state = Some(ModelContextState {
        state_revision: 7,
        prefix_epoch: 3,
        cache_scope_sha256: Some("a".repeat(64)),
        transcript_item_sha256: vec!["b".repeat(64)],
        ..Default::default()
    });
    root.conversation_summary = Some(ConversationSummary::new("preserved summary", 1, 10));
    root.add_message(bamboo_domain::Message::user("preserved history"));
    root.activate_provider_transcript_route(
        bamboo_domain::session::provider_transcript::ProviderFamily::OpenAi,
        bamboo_domain::session::provider_transcript::ProviderProtocol::OpenAiResponsesV1,
        &"c".repeat(64),
    )
    .unwrap();
    root
}

fn request(root: &Session, epoch: u64, enabled: bool) -> RootModeOperationRequest {
    RootModeOperationRequest {
        session_id: root.id.clone(),
        operation_id: format!("{epoch}:{}", Uuid::new_v4()),
        birth_token: root.root_mode_birth_token(),
        expected_epoch: epoch,
        requested_enabled: enabled,
        action: RootModeOperationAction::Select,
    }
}

async fn activate(store: &SessionStoreV2, root: &Session) -> RootActorRuntimeWrite {
    let now = Utc::now();
    let activation = store
        .claim_activation(&ActorActivationClaim {
            actor_id: root.id.clone(),
            run_id: Uuid::new_v4().to_string(),
            lease_owner: "fixture-host".into(),
            lease_expires_at: now + chrono::Duration::minutes(5),
            inbox_generation: 0,
            placement_ref: None,
            now,
        })
        .await
        .unwrap();
    store
        .start_activation(&activation.fence(), now)
        .await
        .unwrap();
    RootActorRuntimeWrite {
        fence: activation.fence(),
        created_at: root.created_at,
        conversation_generation: root.conversation_generation_observation(),
    }
}

fn assert_conflict(error: io::Error) {
    assert!(
        error
            .get_ref()
            .is_some_and(|cause| cause.is::<SessionAuthorityConflict>()),
        "{error:?}"
    );
}

fn canonical(directory: &Path) -> Vec<Vec<u8>> {
    [
        "session.json",
        RUNTIME_SIDECAR_FILE,
        root_context::ROOT_TOOL_AUTHORITY_PROOF_FILE,
    ]
    .map(|name| std::fs::read(directory.join(name)).unwrap())
    .into()
}

#[tokio::test]
async fn root_mode_activated_idle_resets_context_both_directions_across_restart() {
    let home = tempfile::tempdir().unwrap();
    let first = SessionStoreV2::new(home.path().into()).await.unwrap();
    let original = root();
    first.save_session(&original).await.unwrap();
    let old_owner = activate(&first, &original).await;
    first
        .finish_activation(
            &old_owner.fence,
            Utc::now(),
            ActorActivationFinish::Succeeded,
        )
        .await
        .unwrap();
    let actor_before = first.inspect_actor(&original.id).await.unwrap();
    assert!(actor_before.actor.current_attempt > 0);
    assert!(actor_before.activation.is_some());
    let select = request(&original, 0, true);
    // Reproduce the original mode port's ordinary-save path on the exact
    // activated fixture before exercising the authorized control-plane path.
    let mut unfenced_mode = first.load_session(&original.id).await.unwrap().unwrap();
    unfenced_mode.set_root_orchestration_only(true).unwrap();
    unfenced_mode
        .record_root_mode_operation(RootModeOperationReceipt {
            operation_id: select.operation_id.clone(),
            expected_epoch: 0,
            resulting_epoch: 0,
            requested_enabled: true,
            enabled_at_completion: false,
            tool_authority_revision: 0,
            outcome: RootModeOperationOutcome::Committed,
        })
        .unwrap();
    let before = canonical(&first.sessions_dir.join(&original.id));
    let rejected = first.save_session(&unfenced_mode).await.unwrap_err();
    assert_eq!(rejected.kind(), io::ErrorKind::Unsupported);
    assert_conflict(rejected);
    assert_eq!(canonical(&first.sessions_dir.join(&original.id)), before);
    let committed = first.root_mode_operation(&select).await.unwrap();
    first.flush_search_index().await;
    drop(first);

    let restarted = SessionStoreV2::new(home.path().into()).await.unwrap();
    assert_eq!(
        restarted.root_mode_operation(&select).await.unwrap(),
        committed
    );
    assert_eq!(
        restarted
            .root_mode_operation(&RootModeOperationRequest {
                action: RootModeOperationAction::Recover,
                ..select.clone()
            })
            .await
            .unwrap(),
        committed
    );
    let ultra = restarted.load_session(&original.id).await.unwrap().unwrap();
    assert!(ultra.root_orchestration_only_enabled());
    assert_eq!(ultra.root_tool_authority_revision, 1);
    assert_eq!(ultra.root_mode_transition_epoch, 1);
    let state = ultra.model_context_state.as_ref().unwrap();
    assert_eq!(state.prefix_epoch, 4);
    assert_eq!(state.state_revision, 8);
    assert!(state.cache_scope_sha256.is_none());
    assert!(state.transcript_item_sha256.is_empty());
    assert_eq!(
        state.last_reset_reason,
        Some(ModelContextResetReason::CacheScopeChanged)
    );
    assert_eq!(
        serde_json::to_value(&ultra.messages).unwrap(),
        serde_json::to_value(&original.messages).unwrap()
    );
    assert_eq!(
        serde_json::to_value(&ultra.conversation_summary).unwrap(),
        serde_json::to_value(&original.conversation_summary).unwrap()
    );
    assert_eq!(
        restarted.inspect_actor(&original.id).await.unwrap(),
        actor_before
    );

    let standard = request(&original, 1, false);
    let second = restarted.root_mode_operation(&standard).await.unwrap();
    restarted.flush_search_index().await;
    drop(restarted);
    let final_store = SessionStoreV2::new(home.path().into()).await.unwrap();
    let durable = final_store
        .load_session(&original.id)
        .await
        .unwrap()
        .unwrap();
    assert!(!durable.root_orchestration_only_enabled());
    assert_eq!(durable.root_tool_authority_revision, 2);
    assert_eq!(durable.root_mode_transition_epoch, 2);
    assert_eq!(
        durable.model_context_state.as_ref().unwrap().prefix_epoch,
        4 // Both selections share the existing pending, unseeded reset.
    );
    assert_eq!(
        durable.model_context_state.as_ref().unwrap().state_revision,
        8
    );
    assert_eq!(durable.root_mode_operations.len(), 2);
    assert_eq!(
        durable.provider_transcript.epoch(),
        original.provider_transcript.epoch() + 2
    );
    assert_eq!(
        final_store.root_mode_operation(&standard).await.unwrap(),
        second
    );
    assert_eq!(
        final_store.inspect_actor(&original.id).await.unwrap(),
        actor_before
    );
    let before = canonical(&final_store.sessions_dir.join(&original.id));
    let mut unfenced = durable.clone();
    unfenced.model_context_state.as_mut().unwrap().prefix_epoch += 1;
    for runtime in [false, true] {
        let error = if runtime {
            final_store.save_runtime_state(&unfenced).await
        } else {
            final_store.save_session(&unfenced).await
        }
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::Unsupported);
        assert_conflict(error);
    }
    assert_conflict(
        final_store
            .save_root_actor_runtime(&old_owner, &durable, false, Arc::new(|_| {}))
            .await
            .unwrap_err(),
    );
    assert_eq!(
        canonical(&final_store.sessions_dir.join(&original.id)),
        before
    );
    final_store.flush_search_index().await;
}

#[tokio::test]
async fn root_mode_running_owner_cannot_restore_pre_transition_context() {
    let home = tempfile::tempdir().unwrap();
    let store = SessionStoreV2::new(home.path().into()).await.unwrap();
    let original = root();
    store.save_session(&original).await.unwrap();
    let owner = activate(&store, &original).await;
    let actor_before = store.inspect_actor(&original.id).await.unwrap();
    store
        .root_mode_operation(&request(&original, 0, true))
        .await
        .unwrap();
    let directory = store.sessions_dir.join(&original.id);
    let before = canonical(&directory);
    for runtime in [false, true] {
        assert_conflict(
            store
                .save_root_actor_runtime(&owner, &original, runtime, Arc::new(|_| {}))
                .await
                .unwrap_err(),
        );
        assert_eq!(canonical(&directory), before);
    }
    // The control plane does not mint, replace or revoke the actual owner.
    assert_eq!(
        store.inspect_actor(&original.id).await.unwrap(),
        actor_before
    );
    let mut current = store.load_session(&original.id).await.unwrap().unwrap();
    current.conversation_summary.as_mut().unwrap().content = "current owner context".into();
    store
        .save_root_actor_runtime(&owner, &current, true, Arc::new(|_| {}))
        .await
        .unwrap();
    let loaded = store.load_session(&original.id).await.unwrap().unwrap();
    assert_eq!(loaded.model_context_state, current.model_context_state);
    assert_eq!(
        serde_json::to_value(&loaded.conversation_summary).unwrap(),
        serde_json::to_value(&current.conversation_summary).unwrap()
    );
    store.flush_search_index().await;
}

#[tokio::test]
async fn root_mode_supervisor_retains_its_existing_context_writer_authority() {
    let home = tempfile::tempdir().unwrap();
    let store = SessionStoreV2::new(home.path().into()).await.unwrap();
    let bootstrap = store
        .get_or_create_default_supervisor("model")
        .await
        .unwrap();
    let mut original = store
        .load_session(&bootstrap.session_id)
        .await
        .unwrap()
        .unwrap();
    assert!(!original.authority_identity.is_ordinary());
    original.model_context_state = root().model_context_state;
    original.provider_transcript = root().provider_transcript;
    store.save_session(&original).await.unwrap();
    let original = store.load_session(&original.id).await.unwrap().unwrap();
    let select = request(&original, 0, true);
    let record = |request: &RootModeOperationRequest| RootModeOperationReceipt {
        operation_id: request.operation_id.clone(),
        expected_epoch: request.expected_epoch,
        resulting_epoch: 0,
        requested_enabled: request.requested_enabled,
        enabled_at_completion: false,
        tool_authority_revision: 0,
        outcome: RootModeOperationOutcome::Committed,
    };

    // This is the mode port's original full-writer path, using a Supervisor
    // minted by its bootstrap authority rather than a fabricated identity.
    let mut baseline = original.clone();
    baseline.set_root_orchestration_only(true).unwrap();
    baseline
        .record_root_mode_operation(record(&select))
        .unwrap();
    store.save_session(&baseline).await.unwrap();
    assert_eq!(
        store.root_mode_operation(&select).await.unwrap(),
        RootModeOperationDecision::Terminal(
            baseline
                .root_mode_operation(&select.operation_id)
                .unwrap()
                .clone()
        )
    );
    let standard = request(&original, 1, false);
    let mut expected = store.load_session(&original.id).await.unwrap().unwrap();
    expected.set_root_orchestration_only(false).unwrap();
    expected
        .record_root_mode_operation(record(&standard))
        .unwrap();
    let result = store.root_mode_operation(&standard).await.unwrap();
    assert_eq!(
        result,
        RootModeOperationDecision::Terminal(
            expected
                .root_mode_operation(&standard.operation_id)
                .unwrap()
                .clone()
        )
    );
    store.flush_search_index().await;
    drop(store);

    let restarted = SessionStoreV2::new(home.path().into()).await.unwrap();
    let durable = restarted.load_session(&original.id).await.unwrap().unwrap();
    let authority = restarted
        .load_root_authority(&original.id)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(durable.authority_identity, original.authority_identity);
    assert_eq!(authority.authority_identity, original.authority_identity);
    assert!(!durable.root_orchestration_only_enabled());
    assert_eq!(durable.root_tool_authority_revision, 2);
    assert_eq!(durable.root_mode_transition_epoch, 2);
    assert_eq!(durable.model_context_state, expected.model_context_state);
    assert_eq!(authority.model_context_state, expected.model_context_state);
    assert_eq!(durable.provider_transcript, expected.provider_transcript);
    assert_eq!(durable.root_mode_operations, expected.root_mode_operations);
    assert_eq!(
        authority.root_mode_operations,
        expected.root_mode_operations
    );
    assert_eq!(
        restarted.root_mode_operation(&standard).await.unwrap(),
        result
    );
    restarted.flush_search_index().await;
}

#[tokio::test]
async fn root_mode_actor_replacement_waits_for_committed_reset_and_old_owner_stays_fenced() {
    let home = tempfile::tempdir().unwrap();
    let first = SessionStoreV2::new(home.path().into()).await.unwrap();
    let original = root();
    first.save_session(&original).await.unwrap();
    let old_owner = activate(&first, &original).await;
    first
        .finish_activation(
            &old_owner.fence,
            Utc::now(),
            ActorActivationFinish::Succeeded,
        )
        .await
        .unwrap();
    let second = SessionStoreV2::new(home.path().into()).await.unwrap();
    let (reached, release) = first.pause_full_save_before_filesystem_commit_for_test(&original.id);
    let select = request(&original, 0, true);
    let (committed, owner) = tokio::join!(first.root_mode_operation(&select), async {
        reached.wait().await;
        let claim = activate(&second, &original);
        tokio::pin!(claim);
        assert!(tokio::time::timeout(Duration::from_millis(100), &mut claim)
            .await
            .is_err());
        release.wait().await;
        claim.await
    });
    assert!(matches!(
        committed.unwrap(),
        RootModeOperationDecision::Terminal(_)
    ));
    assert!(owner.fence.attempt > old_owner.fence.attempt);
    let mut current = second.load_session(&original.id).await.unwrap().unwrap();
    assert_eq!(
        current.model_context_state.as_ref().unwrap().prefix_epoch,
        4
    );
    assert_conflict(
        first
            .save_root_actor_runtime(&old_owner, &current, true, Arc::new(|_| {}))
            .await
            .unwrap_err(),
    );
    current.conversation_summary.as_mut().unwrap().content = "replacement owner".into();
    second
        .save_root_actor_runtime(&owner, &current, true, Arc::new(|_| {}))
        .await
        .unwrap();
    assert_eq!(
        first
            .load_session(&original.id)
            .await
            .unwrap()
            .unwrap()
            .conversation_summary
            .unwrap()
            .content,
        current.conversation_summary.unwrap().content
    );
    first.flush_search_index().await;
    second.flush_search_index().await;
}

#[tokio::test]
async fn root_mode_running_snapshot_adopts_missed_round_trip_before_owned_checkpoint() {
    let home = tempfile::tempdir().unwrap();
    let store = SessionStoreV2::new(home.path().into()).await.unwrap();
    let mut running = root();
    store.save_session(&running).await.unwrap();
    let owner = activate(&store, &running).await;
    store
        .root_mode_operation(&request(&running, 0, true))
        .await
        .unwrap();
    store
        .root_mode_operation(&request(&running, 1, false))
        .await
        .unwrap();
    let durable = store.load_session(&running.id).await.unwrap().unwrap();
    assert_eq!(running.root_thinking_mode(), durable.root_thinking_mode());
    running.adopt_root_tool_authority_from(&durable).unwrap();
    assert_eq!(running.model_context_state, durable.model_context_state);
    assert_eq!(running.provider_transcript, durable.provider_transcript);
    store
        .save_root_actor_runtime(&owner, &running, false, Arc::new(|_| {}))
        .await
        .unwrap();
    let after = store.load_session(&running.id).await.unwrap().unwrap();
    assert_eq!(after.model_context_state, durable.model_context_state);
    assert_eq!(after.root_mode_transition_epoch, 2);
    assert_eq!(after.root_mode_operations, durable.root_mode_operations);
    store.flush_search_index().await;
}

#[tokio::test]
async fn root_mode_serializes_running_checkpoint_and_finish_without_context_rollback() {
    for finish in [false, true] {
        let home = tempfile::tempdir().unwrap();
        let first = SessionStoreV2::new(home.path().into()).await.unwrap();
        let original = root();
        first.save_session(&original).await.unwrap();
        let owner = activate(&first, &original).await;
        let second = SessionStoreV2::new(home.path().into()).await.unwrap();
        let (reached, release) =
            first.pause_full_save_before_filesystem_commit_for_test(&original.id);
        let select = request(&original, 0, true);
        let (committed, _) = tokio::join!(first.root_mode_operation(&select), async {
            reached.wait().await;
            let competing = async {
                if finish {
                    second
                        .finish_activation(
                            &owner.fence,
                            Utc::now(),
                            ActorActivationFinish::Succeeded,
                        )
                        .await
                        .unwrap();
                } else {
                    assert_conflict(
                        second
                            .save_root_actor_runtime(&owner, &original, false, Arc::new(|_| {}))
                            .await
                            .unwrap_err(),
                    );
                }
            };
            tokio::pin!(competing);
            assert!(
                tokio::time::timeout(Duration::from_millis(100), &mut competing)
                    .await
                    .is_err()
            );
            release.wait().await;
            competing.await
        });
        assert!(matches!(
            committed.unwrap(),
            RootModeOperationDecision::Terminal(_)
        ));
        let current = second.load_session(&original.id).await.unwrap().unwrap();
        assert!(current.root_orchestration_only_enabled());
        assert_eq!(current.root_mode_transition_epoch, 1);
        assert_eq!(
            current.model_context_state.as_ref().unwrap().prefix_epoch,
            4
        );
        assert_eq!(
            serde_json::to_value(&current.messages).unwrap(),
            serde_json::to_value(&original.messages).unwrap()
        );
        first.flush_search_index().await;
        second.flush_search_index().await;
    }
}

#[tokio::test]
async fn root_mode_unknown_actor_witness_rejects_reset_but_recovery_still_fences() {
    for witness in ["actor-authority.json", "actor-authority.initialized.json"] {
        let home = tempfile::tempdir().unwrap();
        let store = SessionStoreV2::new(home.path().into()).await.unwrap();
        let original = root();
        store.save_session(&original).await.unwrap();
        activate(&store, &original).await;
        let directory = store.sessions_dir.join(&original.id);
        std::fs::remove_file(directory.join(witness)).unwrap();
        let before = canonical(&directory);
        let select = request(&original, 0, true);
        assert_conflict(store.root_mode_operation(&select).await.unwrap_err());
        assert_eq!(canonical(&directory), before);
        let recovered = store
            .root_mode_operation(&RootModeOperationRequest {
                action: RootModeOperationAction::Recover,
                ..select.clone()
            })
            .await
            .unwrap();
        assert!(matches!(
            recovered,
            RootModeOperationDecision::Terminal(RootModeOperationReceipt {
                outcome: RootModeOperationOutcome::Fenced,
                ..
            })
        ));
        let current = store.load_session(&original.id).await.unwrap().unwrap();
        assert_eq!(current.model_context_state, original.model_context_state);
        assert!(!current.root_orchestration_only_enabled());
        assert_eq!(store.root_mode_operation(&select).await.unwrap(), recovered);
        store.flush_search_index().await;
    }
}

#[tokio::test]
async fn root_mode_context_capability_rejects_unrelated_or_fabricated_changes() {
    let home = tempfile::tempdir().unwrap();
    let store = SessionStoreV2::new(home.path().into()).await.unwrap();
    let original = root();
    store.save_session(&original).await.unwrap();
    let original = store.load_session(&original.id).await.unwrap().unwrap();
    activate(&store, &original).await;
    store
        .root_mode_operation(&request(&original, 0, true))
        .await
        .unwrap();
    let candidate = store.load_session(&original.id).await.unwrap().unwrap();
    let directory = store.sessions_dir.join(&original.id);
    let permit =
        root_context::RootModeContextWrite::capture(directory.clone(), &original, &candidate)
            .unwrap();
    assert!(permit.validate_candidate(&candidate, false).is_err());
    for mutation in ["summary", "history", "ledger", "model"] {
        let mut extra = candidate.clone();
        match mutation {
            "summary" => extra.conversation_summary.as_mut().unwrap().content = "forged".into(),
            "history" => extra.add_message(bamboo_domain::Message::user("forged")),
            "ledger" => extra.model_context_state.as_mut().unwrap().prefix_epoch += 1,
            "model" => extra.model = "forged".into(),
            _ => unreachable!(),
        }
        assert!(
            root_context::RootModeContextWrite::capture(directory.clone(), &original, &extra)
                .is_err()
        );
        assert!(permit.validate_candidate(&extra, true).is_err());
    }
    store.flush_search_index().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn root_mode_rechecks_actor_witness_at_actual_filesystem_publication() {
    let home = tempfile::tempdir().unwrap();
    let store = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let original = root();
    store.save_session(&original).await.unwrap();
    activate(&store, &original).await;
    let directory = store.sessions_dir.join(&original.id);
    let before = canonical(&directory);
    let hook = default_actor_context_tests::DefaultWriteHook::install(
        &store,
        root_context::ROOT_TOOL_AUTHORITY_PROOF_FILE,
        DurableWritePhase::BeforeReplace,
        false,
    );
    let select = request(&original, 0, true);
    let writer = store.clone();
    let pending = tokio::spawn(async move { writer.root_mode_operation(&select).await });
    let wait = hook.clone();
    tokio::task::spawn_blocking(move || wait.wait())
        .await
        .unwrap();
    std::fs::write(directory.join("actor-authority.json"), "{}").unwrap();
    hook.release();
    assert_conflict(pending.await.unwrap().unwrap_err());
    assert_eq!(canonical(&directory), before);
    store.flush_search_index().await;
}

fn scheduled_startup_root(id: &str) -> Session {
    let mut session = Session::new(id, "model-a");
    session
        .metadata
        .insert("created_by_schedule_id".into(), "schedule-a".into());
    session.model_ref = Some(bamboo_domain::ProviderModelRef::new("account-a", "model-a"));
    session.set_provider_name("account-a");
    session.add_message(bamboo_domain::Message::user("scheduled work"));
    session
}

#[tokio::test]
async fn scheduled_startup_profile_conflict_does_not_write_or_publish() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    let home = tempfile::tempdir().unwrap();
    let first = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let root = scheduled_startup_root("scheduled-startup-conflict");
    first.save_session(&root).await.unwrap();
    let original = first.load_session(&root.id).await.unwrap().unwrap();
    let owner = activate(&first, &original).await;
    let expected = first.load_session(&root.id).await.unwrap().unwrap();
    let second = SessionStoreV2::new(home.path().into()).await.unwrap();
    let mut changed = second.load_session(&root.id).await.unwrap().unwrap();
    changed.model = "model-b".into();
    changed.model_ref = Some(bamboo_domain::ProviderModelRef::new("account-b", "model-b"));
    changed.set_provider_name("account-b");
    changed.metadata_version += 1;
    second.save_session(&changed).await.unwrap();

    let directory = first.sessions_dir.join(&root.id);
    let files_before: Vec<_> = ["session.json", RUNTIME_SIDECAR_FILE]
        .into_iter()
        .map(|name| std::fs::read(directory.join(name)).unwrap())
        .collect();
    let durable_before = second.load_session(&root.id).await.unwrap().unwrap();
    let root_publications = Arc::new(AtomicUsize::new(0));
    let root_count = root_publications.clone();
    let publisher: bamboo_domain::RootActorRuntimePublisher = Arc::new(move |_| {
        root_count.fetch_add(1, Ordering::SeqCst);
    });
    let storage: Arc<dyn Storage> = first.clone();
    let bound = crate::session_merge::LockedSessionStore::new(storage)
        .bind_root_actor_writer(owner, publisher)
        .unwrap();
    let mut candidate = expected.clone();
    candidate.set_last_run_status("running");
    let candidate_before = serde_json::to_value(&candidate).unwrap();
    let local_publications = Arc::new(AtomicUsize::new(0));
    let local_count = local_publications.clone();
    let error = bound
        .merge_save_bound_startup_and_publish(&mut candidate, &expected, move |_, _| {
            local_count.fetch_add(1, Ordering::SeqCst);
        })
        .await
        .unwrap_err();
    assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
    assert!(error
        .get_ref()
        .is_some_and(|cause| cause.is::<SessionAuthorityConflict>()));
    assert!(error
        .to_string()
        .contains("bound startup execution profile changed"));
    assert_eq!(serde_json::to_value(&candidate).unwrap(), candidate_before);
    assert_eq!(root_publications.load(Ordering::SeqCst), 0);
    assert_eq!(local_publications.load(Ordering::SeqCst), 0);
    let durable_after = second.load_session(&root.id).await.unwrap().unwrap();
    assert_eq!(
        serde_json::to_value(durable_after).unwrap(),
        serde_json::to_value(durable_before).unwrap()
    );
    let files_after: Vec<_> = ["session.json", RUNTIME_SIDECAR_FILE]
        .into_iter()
        .map(|name| std::fs::read(directory.join(name)).unwrap())
        .collect();
    assert_eq!(files_after, files_before);
}

#[tokio::test]
async fn scheduled_startup_commits_and_publishes_with_inherited_wait() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    let home = tempfile::tempdir().unwrap();
    let store = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let mut root = scheduled_startup_root("scheduled-startup-success");
    let mut runtime = bamboo_domain::AgentRuntimeState::new("previous-run");
    runtime.waiting_for_children = Some(bamboo_domain::WaitingForChildrenState::for_children(
        vec!["ordinary-child".into()],
        bamboo_domain::ChildWaitPolicy::All,
        Utc::now(),
    ));
    root.agent_runtime_state = Some(runtime);
    store.save_session(&root).await.unwrap();
    let original = store.load_session(&root.id).await.unwrap().unwrap();
    let owner = activate(&store, &original).await;
    let expected = store.load_session(&root.id).await.unwrap().unwrap();
    let inherited = bamboo_domain::InheritedChildWait::capture(&expected).unwrap();
    let root_publications = Arc::new(AtomicUsize::new(0));
    let root_count = root_publications.clone();
    let publisher: bamboo_domain::RootActorRuntimePublisher = Arc::new(move |_| {
        root_count.fetch_add(1, Ordering::SeqCst);
    });
    let storage: Arc<dyn Storage> = store.clone();
    let bound = crate::session_merge::LockedSessionStore::new(storage)
        .bind_root_actor_writer(owner, publisher)
        .unwrap()
        .bind_inherited_child_wait(inherited);
    assert!(bamboo_domain::RuntimeSessionPersistence::supports_bound_startup_session(&bound));
    let mut candidate = expected.clone();
    candidate.set_last_run_status("running");
    let local_publications = Arc::new(AtomicUsize::new(0));
    let local_count = local_publications.clone();
    bound
        .merge_save_bound_startup_and_publish(&mut candidate, &expected, move |_, committed| {
            assert!(committed);
            local_count.fetch_add(1, Ordering::SeqCst);
        })
        .await
        .unwrap();
    let saved = store.load_session(&root.id).await.unwrap().unwrap();
    assert_eq!(saved.last_run_status().as_deref(), Some("running"));
    assert_eq!(saved.model_ref, expected.model_ref);
    assert_eq!(saved.provider_name(), expected.provider_name());
    assert_eq!(
        saved
            .agent_runtime_state
            .as_ref()
            .unwrap()
            .waiting_for_children,
        expected
            .agent_runtime_state
            .as_ref()
            .unwrap()
            .waiting_for_children
    );
    assert_eq!(root_publications.load(Ordering::SeqCst), 1);
    assert_eq!(local_publications.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn scheduled_startup_without_root_writer_fails_closed() {
    let home = tempfile::tempdir().unwrap();
    let store = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let root = scheduled_startup_root("scheduled-startup-unbound");
    store.save_session(&root).await.unwrap();
    let expected = store.load_session(&root.id).await.unwrap().unwrap();
    let mut candidate = expected.clone();
    candidate.set_last_run_status("running");
    let storage: Arc<dyn Storage> = store.clone();
    let unbound = crate::session_merge::LockedSessionStore::new(storage);
    assert!(!bamboo_domain::RuntimeSessionPersistence::supports_bound_startup_session(&unbound));
    let error = unbound
        .merge_save_bound_startup(&mut candidate, &expected)
        .await
        .unwrap_err();
    assert_eq!(error.kind(), io::ErrorKind::Unsupported);
    assert!(error
        .get_ref()
        .is_some_and(|cause| cause.is::<SessionAuthorityConflict>()));
    let saved = store.load_session(&root.id).await.unwrap().unwrap();
    assert_eq!(saved.last_run_status(), expected.last_run_status());
    assert_eq!(saved.model_ref, expected.model_ref);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bound_response_does_not_overwrite_concurrent_profile_change() {
    use std::sync::atomic::{AtomicUsize, Ordering};

    for change in ["model", "question", "response", "consumed"] {
        let home = tempfile::tempdir().unwrap();
        let first = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
        let mut root = scheduled_startup_root("response-profile-race");
        root.metadata.remove("created_by_schedule_id");
        root.metadata
            .insert("created_by_connect_key".into(), "fake:chat:user".into());
        root.add_message(bamboo_domain::Message::tool_result(
            "call-1",
            "initial response occurrence",
        ));
        root.set_pending_question(
            "call-1".into(),
            "question".into(),
            "Continue?".into(),
            vec!["Yes".into()],
            false,
        );
        first.save_session(&root).await.unwrap();
        let original = first.load_session(&root.id).await.unwrap().unwrap();
        let owner = activate(&first, &original).await;
        let other_owner = owner.clone();
        let second = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
        let root_publications = Arc::new(AtomicUsize::new(0));
        let count = root_publications.clone();
        let publisher: bamboo_domain::RootActorRuntimePublisher = Arc::new(move |_| {
            count.fetch_add(1, Ordering::SeqCst);
        });
        let storage: Arc<dyn Storage> = first.clone();
        let bound = Arc::new(
            crate::session_merge::LockedSessionStore::new(storage)
                .bind_root_actor_writer(owner, publisher)
                .unwrap(),
        );
        let reached = Arc::new(tokio::sync::Notify::new());
        let entered = reached.clone();
        let (release, wait) = std::sync::mpsc::sync_channel::<()>(0);
        let local_publications = Arc::new(AtomicUsize::new(0));
        let local_count = local_publications.clone();
        let id = root.id.clone();
        let response = tokio::spawn(async move {
            bound
                .mutate_runtime_session_and_publish(
                    &id,
                    || None,
                    move |session| {
                        entered.notify_one();
                        wait.recv_timeout(std::time::Duration::from_secs(10))
                            .expect("release response mutation");
                        session.clear_pending_question();
                        session.add_message(bamboo_domain::Message::user("accepted answer"));
                        Ok::<(), ()>(())
                    },
                    move |_| {
                        local_count.fetch_add(1, Ordering::SeqCst);
                    },
                )
                .await
        });
        tokio::time::timeout(std::time::Duration::from_secs(5), reached.notified())
            .await
            .expect("response must reach the controlled mutation");
        let mut changed = second.load_session(&root.id).await.unwrap().unwrap();
        match change {
            "model" => {
                changed.model = "model-b".into();
                changed.model_ref =
                    Some(bamboo_domain::ProviderModelRef::new("account-b", "model-b"));
                changed.set_provider_name("account-b");
                changed.metadata_version += 1;
            }
            "question" => changed.set_pending_question(
                "call-1".into(),
                "question".into(),
                "Different?".into(),
                vec!["No".into()],
                false,
            ),
            "response" => {
                changed
                    .messages
                    .iter_mut()
                    .find(|message| message.tool_call_id.as_deref() == Some("call-1"))
                    .unwrap()
                    .content = "changed response occurrence".into()
            }
            "consumed" => {
                changed.metadata.insert(
                    bamboo_domain::CONSUMED_RESPONSE_OCCURRENCES_KEY.into(),
                    "[]".into(),
                );
            }
            _ => unreachable!(),
        }
        if change == "response" {
            second
                .save_root_actor_runtime(&other_owner, &changed, false, Arc::new(|_| {}))
                .await
                .unwrap();
        } else {
            second.save_session(&changed).await.unwrap();
        }
        release.send(()).unwrap();
        let error = response.await.unwrap().unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
        assert!(error
            .get_ref()
            .is_some_and(|cause| cause.is::<SessionAuthorityConflict>()));
        let durable = second.load_session(&root.id).await.unwrap().unwrap();
        assert_eq!(durable.model_ref, changed.model_ref);
        assert_eq!(durable.pending_question, changed.pending_question);
        assert_eq!(
            durable
                .metadata
                .get(bamboo_domain::CONSUMED_RESPONSE_OCCURRENCES_KEY),
            changed
                .metadata
                .get(bamboo_domain::CONSUMED_RESPONSE_OCCURRENCES_KEY)
        );
        assert!(!durable
            .messages
            .iter()
            .any(|message| message.content == "accepted answer"));
        assert_eq!(root_publications.load(Ordering::SeqCst), 0);
        assert_eq!(local_publications.load(Ordering::SeqCst), 0);
    }
}

#[tokio::test]
async fn basic_root_writer_without_profile_cas_reports_unsupported() {
    struct BasicOnly;
    #[async_trait::async_trait]
    impl Storage for BasicOnly {
        fn supports_root_actor_runtime_write(&self) -> bool {
            true
        }
        async fn save_session(&self, _: &Session) -> io::Result<()> {
            unreachable!()
        }
        async fn load_session(&self, _: &str) -> io::Result<Option<Session>> {
            unreachable!()
        }
        async fn delete_session(&self, _: &str) -> io::Result<bool> {
            unreachable!()
        }
    }
    let home = tempfile::tempdir().unwrap();
    let store = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let root = scheduled_startup_root("unsupported-profile-cas");
    store.save_session(&root).await.unwrap();
    let original = store.load_session(&root.id).await.unwrap().unwrap();
    let owner = activate(&store, &original).await;
    let storage: Arc<dyn Storage> = Arc::new(BasicOnly);
    let writer = crate::session_merge::LockedSessionStore::new(storage)
        .bind_root_actor_writer(owner, Arc::new(|_| {}))
        .unwrap();
    assert!(!bamboo_domain::RuntimeSessionPersistence::supports_bound_startup_session(&writer));
    let view: Arc<dyn bamboo_domain::RuntimeSessionPersistence> = Arc::new(writer);
    assert!(!view.supports_bound_startup_session());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bound_response_preserves_cross_instance_inherited_wait_completion() {
    let home = tempfile::tempdir().unwrap();
    let first = Arc::new(SessionStoreV2::new(home.path().into()).await.unwrap());
    let mut root = scheduled_startup_root("response-inherited-wait-race");
    let mut runtime = bamboo_domain::AgentRuntimeState::new("previous-run");
    runtime.waiting_for_children = Some(bamboo_domain::WaitingForChildrenState::for_children(
        vec!["ordinary-child".into()],
        bamboo_domain::ChildWaitPolicy::All,
        Utc::now(),
    ));
    root.agent_runtime_state = Some(runtime);
    root.set_pending_question(
        "call-1".into(),
        "question".into(),
        "Continue?".into(),
        vec!["Yes".into()],
        false,
    );
    first.save_session(&root).await.unwrap();
    let original = first.load_session(&root.id).await.unwrap().unwrap();
    let inherited = bamboo_domain::InheritedChildWait::capture(&original).unwrap();
    let owner = activate(&first, &original).await;
    let second = SessionStoreV2::new(home.path().into()).await.unwrap();
    let storage: Arc<dyn Storage> = first.clone();
    let bound = Arc::new(
        crate::session_merge::LockedSessionStore::new(storage)
            .bind_root_actor_writer(owner, Arc::new(|_| {}))
            .unwrap()
            .bind_inherited_child_wait(inherited),
    );
    let reached = Arc::new(tokio::sync::Notify::new());
    let entered = reached.clone();
    let (release, wait) = std::sync::mpsc::sync_channel::<()>(0);
    let id = root.id.clone();
    let response = tokio::spawn(async move {
        bound
            .mutate_runtime_session_and_publish(
                &id,
                || None,
                move |session| {
                    entered.notify_one();
                    wait.recv_timeout(std::time::Duration::from_secs(10))
                        .expect("release mutation");
                    session.clear_pending_question();
                    session.add_message(bamboo_domain::Message::user("accepted answer"));
                    Ok::<(), ()>(())
                },
                |_| {},
            )
            .await
    });
    tokio::time::timeout(std::time::Duration::from_secs(5), reached.notified())
        .await
        .expect("response must reach the controlled mutation");
    let mut completed = second.load_session(&root.id).await.unwrap().unwrap();
    completed
        .agent_runtime_state
        .as_mut()
        .unwrap()
        .waiting_for_children = None;
    second.save_session(&completed).await.unwrap();
    release.send(()).unwrap();
    let saved = response.await.unwrap().unwrap().unwrap().unwrap();
    assert!(saved
        .agent_runtime_state
        .as_ref()
        .unwrap()
        .waiting_for_children
        .is_none());
    let durable = second.load_session(&root.id).await.unwrap().unwrap();
    assert!(durable
        .agent_runtime_state
        .as_ref()
        .unwrap()
        .waiting_for_children
        .is_none());
    assert!(durable.pending_question.is_none());
    assert!(durable
        .messages
        .iter()
        .any(|message| message.content == "accepted answer"));
}