gate4agent_node_core/

session_registry.rs

use crate::protocol::{
    AgentId, ManagedSessionRecord, ManagedSessionState, ManagedWorktreeLeaseId,
    ManagedWorktreeSpawnReceipt, NodeId, NodeIncarnationId, OpaqueHostPath,
    ResolvedBundleReceipt, ResolvedContextPackReceipt, ResolvedEnvironmentProfileReceipt,
    SessionAddress, SessionMode,
    SessionRecordId, SpawnIdempotencyKey, WorktreeProfileId,
    WorktreeProfileRevision, WorkspaceId,
    MAX_NODE_TEXT_BYTES, MAX_SESSION_DISPLAY_NAME_BYTES, MAX_WORKSPACE_ROOT_BYTES,
    NODE_STATE_SCHEMA_V1, NODE_STATE_SCHEMA_V2, NODE_STATE_SCHEMA_V3, NODE_STATE_SCHEMA_V4,
    NODE_STATE_SCHEMA_V5,
    NODE_STATE_SCHEMA_V6, NODE_STATE_SCHEMA_V7, NODE_STATE_SCHEMA_V8,
    NODE_STATE_SCHEMA_V9, NODE_STATE_SCHEMA_V10,
};
use crate::managed_worktree::ManagedWorktreeLeaseRecord;
use crate::materialization::{
    opaque_to_path, path_to_opaque, MaterializationId, MaterializationOwner,
    MaterializationOwnershipRecord, MaterializationState, MaterializedPathDeclaration,
    MAX_SESSION_MATERIALIZATIONS,
};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::error::Error as StdError;
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Write};
#[cfg(windows)]
use std::os::windows::fs::OpenOptionsExt;
#[cfg(unix)]
use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

const MAX_STATE_BYTES: u64 = 4 * 1024 * 1024;
const MAX_PERSISTED_WORKSPACES: usize = 256;
const MAX_PERSISTED_MANAGED_WORKTREE_BRANCH_BYTES: usize = 512;
pub const MAX_MANAGED_WORKTREE_SPAWN_REPLAYS: usize = 256;
pub const MAX_MANAGED_SESSION_RECORDS: usize = 4_096;
const BACKUP_ROTATION_WARNING: &str = "durable-state-backup-rotation-failed";
const CORRUPT_PRIMARY_PRESERVED_WARNING: &str = "durable-state-corrupt-primary-preserved";
pub const DURABLE_STATE_COMMIT_FAILED: &str = "durable-state-commit-failed";
pub const DURABLE_STATE_RECOVERY_WARNING: &str = "durable-state-recovery-required";
pub const DURABLE_STATE_WORKSPACE_WARNING: &str =
    "durable-state-workspace-unavailable";
static NEXT_TEMPORARY_FILE: AtomicU64 = AtomicU64::new(1);

pub type ProviderSessionSemanticKey = (String, u8, String);

pub fn provider_session_semantic_key(
    provider: &AgentId,
    identity: &gate4agent_types::ProviderSessionIdentity,
) -> ProviderSessionSemanticKey {
    let key = match identity.key {
        gate4agent_types::ProviderSessionKey::SessionId => 0,
        gate4agent_types::ProviderSessionKey::ConversationId => 1,
    };
    (provider.as_str().to_owned(), key, identity.id.clone())
}

pub fn same_provider_session(
    left_provider: &AgentId,
    left: &gate4agent_types::ProviderSessionIdentity,
    right_provider: &AgentId,
    right: &gate4agent_types::ProviderSessionIdentity,
) -> bool {
    left_provider == right_provider && left.key == right.key && left.id == right.id
}

pub fn sanitized_persistence_summary(message: &str) -> String {
    match message {
        BACKUP_ROTATION_WARNING
        | CORRUPT_PRIMARY_PRESERVED_WARNING
        | DURABLE_STATE_COMMIT_FAILED
        | DURABLE_STATE_RECOVERY_WARNING
        | DURABLE_STATE_WORKSPACE_WARNING => message.to_owned(),
        _ => DURABLE_STATE_COMMIT_FAILED.to_owned(),
    }
}

pub struct StatePathLock {
    file: Option<File>,
    #[cfg(windows)]
    path: PathBuf,
}

impl StatePathLock {
    pub fn acquire(path: Option<&Path>) -> io::Result<Option<Self>> {
        let Some(path) = path else {
            return Ok(None);
        };
        let parent = path.parent().ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
        })?;
        fs::create_dir_all(parent)?;
        #[cfg(unix)]
        secure_state_directory(parent)?;
        let lock_path = sibling_path(path, "lock");
        #[cfg(windows)]
        let file = OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .share_mode(0)
            .open(&lock_path)
            .map_err(|error| {
                if error.raw_os_error() == Some(32) {
                    io::Error::new(
                        io::ErrorKind::WouldBlock,
                        "durable state path is already owned by another node process",
                    )
                } else {
                    error
                }
            })?;
        #[cfg(unix)]
        let file = acquire_unix_state_lock(&lock_path)?;
        Ok(Some(Self {
            file: Some(file),
            #[cfg(windows)]
            path: lock_path,
        }))
    }
}

impl Drop for StatePathLock {
    fn drop(&mut self) {
        #[cfg(unix)]
        if let Some(file) = self.file.as_ref() {
            unsafe {
                libc::flock(file.as_raw_fd(), libc::LOCK_UN);
            }
        }
        drop(self.file.take());
        #[cfg(windows)]
        let _ = fs::remove_file(&self.path);
    }
}

#[cfg(unix)]
use std::os::fd::AsRawFd;

#[cfg(unix)]
fn secure_state_directory(path: &Path) -> io::Result<()> {
    let metadata = fs::symlink_metadata(path)?;
    if !metadata.file_type().is_dir() || metadata.uid() != unsafe { libc::geteuid() } {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "durable state directory is not owned by the current user",
        ));
    }
    fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
    let metadata = fs::symlink_metadata(path)?;
    if metadata.permissions().mode() & 0o7777 != 0o700 {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "durable state directory must have mode 0700",
        ));
    }
    Ok(())
}

#[cfg(unix)]
fn acquire_unix_state_lock(path: &Path) -> io::Result<File> {
    let file = OpenOptions::new()
        .create(true)
        .read(true)
        .write(true)
        .mode(0o600)
        .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW)
        .open(path)?;
    let metadata = file.metadata()?;
    if !metadata.file_type().is_file()
        || metadata.uid() != unsafe { libc::geteuid() }
    {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "durable state lock is not an owner-controlled regular file",
        ));
    }
    file.set_permissions(fs::Permissions::from_mode(0o600))?;
    let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
    if result != 0 {
        let error = io::Error::last_os_error();
        if error.kind() == io::ErrorKind::WouldBlock {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "durable state path is already owned by another node process",
            ));
        }
        return Err(error);
    }
    Ok(file)
}

#[derive(Clone, Eq, PartialEq)]
pub struct LoadedNodeState {
    pub workspaces: BTreeMap<WorkspaceId, String>,
    pub records: Vec<ManagedSessionRecord>,
    pub managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    pub managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    pub materializations: Vec<MaterializationOwnershipRecord>,
    pub managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
    pub warning: Option<String>,
}

impl LoadedNodeState {
    fn empty() -> Self {
        Self {
            workspaces: BTreeMap::new(),
            records: Vec::new(),
            managed_worktrees: Vec::new(),
            managed_worktree_tombstones: Vec::new(),
            materializations: Vec::new(),
            managed_spawn_replays: Vec::new(),
            warning: None,
        }
    }
}

impl fmt::Debug for LoadedNodeState {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("LoadedNodeState")
            .field("workspace_count", &self.workspaces.len())
            .field("record_count", &self.records.len())
            .field("managed_worktree_count", &self.managed_worktrees.len())
            .field("managed_worktree_tombstone_count", &self.managed_worktree_tombstones.len())
            .field("materialization_count", &self.materializations.len())
            .field("managed_spawn_replay_count", &self.managed_spawn_replays.len())
            .field("warning", &self.warning)
            .finish()
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV1 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV1>,
    session_records: Vec<PersistedManagedSessionRecordV1>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV2 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<PersistedManagedSessionRecordV2>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV3 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<PersistedManagedSessionRecordV3>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV4 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<PersistedManagedSessionRecordV3>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV5 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<PersistedManagedSessionRecordV5>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV6 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<PersistedManagedSessionRecordV5>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    materializations: Vec<PersistedMaterializationRecordV6>,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV7 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<ManagedSessionRecord>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    materializations: Vec<PersistedMaterializationRecordV7>,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedNodeStateV8 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<ManagedSessionRecord>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    materializations: Vec<PersistedMaterializationRecordV8>,
}

type PersistedNodeStateV9 = PersistedNodeStateV8;

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct PersistedNodeStateV10 {
    version: u16,
    node_id: NodeId,
    workspaces: Vec<PersistedWorkspaceV2>,
    session_records: Vec<ManagedSessionRecord>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    materializations: Vec<PersistedMaterializationRecordV8>,
    managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ManagedWorktreeSpawnReplayRecordV10 {
    pub idempotency_key: SpawnIdempotencyKey,
    pub request_digest: String,
    pub source_workspace_id: WorkspaceId,
    pub profile_id: WorktreeProfileId,
    pub expected_profile_revision: WorktreeProfileRevision,
    pub state: ManagedWorktreeSpawnReplayStateV10,
    pub created_at_unix_ms: u64,
    pub updated_at_unix_ms: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "kebab-case", deny_unknown_fields)]
pub enum ManagedWorktreeSpawnReplayStateV10 {
    Pending {
        owner_incarnation_id: NodeIncarnationId,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        lease_id: Option<ManagedWorktreeLeaseId>,
    },
    Committed {
        receipt: ManagedWorktreeSpawnReceipt,
    },
}

struct DecodedNodeState {
    node_id: NodeId,
    workspaces: Vec<(WorkspaceId, String)>,
    session_records: Vec<ManagedSessionRecord>,
    managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
    managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
    materializations: Vec<MaterializationOwnershipRecord>,
    managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
}

#[derive(Deserialize)]
struct PersistedNodeStateHeader {
    version: u16,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StateLoadRefusal {
    UnsupportedSchema,
    PathSemanticsUnsupported,
    NodeIdentityMismatch,
}

#[derive(Debug)]
enum StateLoadRefusalError {
    UnsupportedSchema(u16),
    PathSemanticsUnsupported,
    NodeIdentityMismatch,
}

impl StateLoadRefusalError {
    fn refusal(&self) -> StateLoadRefusal {
        match self {
            Self::UnsupportedSchema(_) => StateLoadRefusal::UnsupportedSchema,
            Self::PathSemanticsUnsupported => StateLoadRefusal::PathSemanticsUnsupported,
            Self::NodeIdentityMismatch => StateLoadRefusal::NodeIdentityMismatch,
        }
    }
}

impl fmt::Display for StateLoadRefusalError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedSchema(version) => {
                write!(formatter, "unsupported durable state version {version}")
            }
            Self::PathSemanticsUnsupported => {
                formatter.write_str("durable state path semantics are unsupported on this host")
            }
            Self::NodeIdentityMismatch => {
                formatter.write_str("durable state belongs to a different node")
            }
        }
    }
}

impl StdError for StateLoadRefusalError {}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedWorkspaceV1 {
    workspace_id: WorkspaceId,
    canonical_root: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedWorkspaceV2 {
    workspace_id: WorkspaceId,
    canonical_root: OpaqueHostPath,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedManagedSessionRecordV1 {
    record_id: SessionRecordId,
    display_name: String,
    provider: LegacyAgentProvider,
    mode: SessionMode,
    state: ManagedSessionState,
    workspace_id: WorkspaceId,
    canonical_root: String,
    provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
    active_session: Option<SessionAddress>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
    last_error: Option<String>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedManagedSessionRecordV2 {
    record_id: SessionRecordId,
    display_name: String,
    provider: LegacyAgentProvider,
    mode: SessionMode,
    state: ManagedSessionState,
    workspace_id: WorkspaceId,
    canonical_root: OpaqueHostPath,
    provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
    active_session: Option<SessionAddress>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
    last_error: Option<String>,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "kebab-case")]
enum LegacyAgentProvider {
    Claude,
    Codex,
    Kimi,
}

impl LegacyAgentProvider {
    fn into_agent_id(self) -> AgentId {
        let value = match self {
            Self::Claude => "claude",
            Self::Codex => "codex",
            Self::Kimi => "kimi",
        };
        AgentId::new(value).expect("legacy provider IDs remain valid")
    }

    #[cfg(test)]
    fn from_agent_id(provider: &AgentId) -> Self {
        match provider.as_str() {
            "claude" => Self::Claude,
            "codex" => Self::Codex,
            "kimi" => Self::Kimi,
            value => panic!("{value} is not a legacy provider ID"),
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedManagedSessionRecordV3 {
    record_id: SessionRecordId,
    display_name: String,
    provider: AgentId,
    mode: SessionMode,
    state: ManagedSessionState,
    workspace_id: WorkspaceId,
    canonical_root: OpaqueHostPath,
    provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
    active_session: Option<SessionAddress>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
    last_error: Option<String>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
struct PersistedManagedSessionRecordV5 {
    record_id: SessionRecordId,
    display_name: String,
    provider: AgentId,
    mode: SessionMode,
    state: ManagedSessionState,
    workspace_id: WorkspaceId,
    canonical_root: OpaqueHostPath,
    provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
    active_session: Option<SessionAddress>,
    environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
    last_error: Option<String>,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct PersistedMaterializationRecordV6 {
    materialization_id: MaterializationId,
    environment_profile: ResolvedEnvironmentProfileReceipt,
    owner: MaterializationOwner,
    managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
    state: MaterializationState,
    root: OpaqueHostPath,
    provider_home: OpaqueHostPath,
    declared_paths: Vec<MaterializedPathDeclaration>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct PersistedMaterializationRecordV7 {
    materialization_id: MaterializationId,
    environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
    bundle: Option<ResolvedBundleReceipt>,
    owner: MaterializationOwner,
    managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
    state: MaterializationState,
    root: OpaqueHostPath,
    provider_home: OpaqueHostPath,
    bundle_root: OpaqueHostPath,
    plugin_data: OpaqueHostPath,
    declared_paths: Vec<MaterializedPathDeclaration>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct PersistedMaterializationRecordV8 {
    materialization_id: MaterializationId,
    environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
    bundle: Option<ResolvedBundleReceipt>,
    context: Option<ResolvedContextPackReceipt>,
    owner: MaterializationOwner,
    managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
    state: MaterializationState,
    root: OpaqueHostPath,
    provider_home: OpaqueHostPath,
    bundle_root: OpaqueHostPath,
    plugin_data: OpaqueHostPath,
    declared_paths: Vec<MaterializedPathDeclaration>,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
}

pub fn load(path: Option<&Path>, expected_node_id: &NodeId) -> io::Result<LoadedNodeState> {
    let Some(path) = path else {
        return Ok(LoadedNodeState::empty());
    };
    let backup = sibling_path(path, "bak");
    let pending_backup = sibling_path(path, "pending-backup");
    if !path.exists() {
        if !pending_backup.exists() && !backup.exists() {
            return Ok(LoadedNodeState::empty());
        }
        return recover_from_copies(
            path,
            &pending_backup,
            &backup,
            expected_node_id,
            "is missing".to_owned(),
        );
    }
    match load_one(path, expected_node_id) {
        Ok(loaded) => Ok(loaded),
        Err(primary_error) if is_authoritative_state_error(&primary_error) => {
            Err(primary_error)
        }
        Err(primary_error) => recover_from_copies(
            path,
            &pending_backup,
            &backup,
            expected_node_id,
            format!("is invalid: {primary_error}"),
        ),
    }
}

fn recover_from_copies(
    primary: &Path,
    pending_backup: &Path,
    backup: &Path,
    expected_node_id: &NodeId,
    _primary_problem: String,
) -> io::Result<LoadedNodeState> {
    for candidate in [pending_backup, backup] {
        if !candidate.exists() {
            continue;
        }
        match load_one(candidate, expected_node_id) {
            Ok(mut loaded) => {
                loaded.warning = Some(DURABLE_STATE_RECOVERY_WARNING.to_owned());
                return Ok(loaded);
            }
            Err(error) if is_authoritative_state_error(&error) => {
                return Err(error);
            }
            Err(_) => {}
        }
    }
    let _ = primary;
    Err(invalid_data("durable-state-recovery-failed"))
}

fn load_one(path: &Path, expected_node_id: &NodeId) -> io::Result<LoadedNodeState> {
    let metadata = fs::metadata(path)?;
    if metadata.len() > MAX_STATE_BYTES {
        return Err(invalid_data("durable state exceeds the bounded file size"));
    }
    let mut bytes = Vec::with_capacity(metadata.len() as usize);
    File::open(path)?
        .take(MAX_STATE_BYTES + 1)
        .read_to_end(&mut bytes)?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("durable state exceeds the bounded file size"));
    }
    let state = decode_persisted_state(&bytes)?;
    if &state.node_id != expected_node_id {
        return Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            StateLoadRefusalError::NodeIdentityMismatch,
        ));
    }
    if state.workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("durable state contains too many workspaces"));
    }
    if state.session_records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("durable state contains too many session records"));
    }

    let mut workspaces = BTreeMap::new();
    let mut roots = BTreeSet::new();
    for (workspace_id, canonical_root) in state.workspaces {
        validate_root(&canonical_root)?;
        if workspaces
            .insert(workspace_id.clone(), canonical_root.clone())
            .is_some()
        {
            return Err(invalid_data("durable state contains duplicate workspace IDs"));
        }
        if !roots.insert(crate::platform::root_identity(&canonical_root)) {
            return Err(invalid_data("durable state contains duplicate workspace roots"));
        }
    }

    let mut record_ids = BTreeSet::new();
    let mut provider_sessions = BTreeSet::new();
    let mut records = Vec::with_capacity(state.session_records.len());
    for mut record in state.session_records {
        if let Some(identity) = &mut record.provider_session {
            identity.transcript_path = None;
        }
        validate_record(&record)?;
        record.last_error = record
            .last_error
            .as_deref()
            .map(sanitized_record_error_summary);
        if !record_ids.insert(record.record_id.clone()) {
            return Err(invalid_data("durable state contains duplicate session record IDs"));
        }
        if let Some(identity) = &mut record.provider_session {
            let key = provider_session_semantic_key(&record.provider, identity);
            if !provider_sessions.insert(key) {
                return Err(invalid_data("durable state contains duplicate provider sessions"));
            }
        }
        record.active_session = None;
        record.state = if record.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        records.push(record);
    }
    Ok(LoadedNodeState {
        workspaces,
        records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
        materializations: state.materializations,
        managed_spawn_replays: state.managed_spawn_replays,
        warning: None,
    })
}

fn decode_persisted_state(bytes: &[u8]) -> io::Result<DecodedNodeState> {
    let header: PersistedNodeStateHeader = serde_json::from_slice(bytes)
        .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
    match header.version {
        NODE_STATE_SCHEMA_V1 => {
            let state: PersistedNodeStateV1 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v1_state(state)
        }
        NODE_STATE_SCHEMA_V2 => {
            let state: PersistedNodeStateV2 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v2_state(state)
        }
        NODE_STATE_SCHEMA_V3 => {
            let state: PersistedNodeStateV3 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v3_state(state)
        }
        NODE_STATE_SCHEMA_V4 => {
            let state: PersistedNodeStateV4 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v4_state(state)
        }
        NODE_STATE_SCHEMA_V5 => {
            let state: PersistedNodeStateV5 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v5_state(state)
        }
        NODE_STATE_SCHEMA_V6 => {
            let state: PersistedNodeStateV6 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v6_state(state)
        }
        NODE_STATE_SCHEMA_V7 => {
            let state: PersistedNodeStateV7 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v7_state(state)
        }
        NODE_STATE_SCHEMA_V8 => {
            let state: PersistedNodeStateV8 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v8_state(state)
        }
        NODE_STATE_SCHEMA_V9 => {
            let state: PersistedNodeStateV9 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v9_state(state)
        }
        NODE_STATE_SCHEMA_V10 => {
            let state: PersistedNodeStateV10 = serde_json::from_slice(bytes)
                .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
            decode_v10_state(state)
        }
        version => Err(io::Error::new(
            io::ErrorKind::Unsupported,
            StateLoadRefusalError::UnsupportedSchema(version),
        )),
    }
}

fn decode_v1_state(state: PersistedNodeStateV1) -> io::Result<DecodedNodeState> {
    Ok(DecodedNodeState {
        node_id: state.node_id,
        workspaces: state
            .workspaces
            .into_iter()
            .map(|workspace| (workspace.workspace_id, workspace.canonical_root))
            .collect(),
        session_records: state
            .session_records
            .into_iter()
            .map(|record| {
                Ok(ManagedSessionRecord {
                    record_id: record.record_id,
                    display_name: record.display_name,
                    provider: record.provider.into_agent_id(),
                    mode: record.mode,
                    state: record.state,
                    workspace_id: record.workspace_id,
                    canonical_root: OpaqueHostPath::utf8(record.canonical_root)
                        .map_err(|error| invalid_data(format!(
                            "durable state contains an invalid workspace path: {error}"
                        )))?,
                    provider_session: record.provider_session,
                    active_session: record.active_session,
                    environment_profile: None,
                    bundle: None,
                    context_id: None,
                    context: None,
                    exported_context: None,
                    task_binding: None,
                    created_at_unix_ms: record.created_at_unix_ms,
                    updated_at_unix_ms: record.updated_at_unix_ms,
                    last_error: record.last_error,
                })
            })
            .collect::<io::Result<_>>()?,
        managed_worktrees: Vec::new(),
        managed_worktree_tombstones: Vec::new(),
        materializations: Vec::new(),
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v2_state(state: PersistedNodeStateV2) -> io::Result<DecodedNodeState> {
    Ok(DecodedNodeState {
        node_id: state.node_id,
        workspaces: state
            .workspaces
            .into_iter()
            .map(|workspace| {
                Ok((
                    workspace.workspace_id,
                    require_utf8_state_path(&workspace.canonical_root)?.to_owned(),
                ))
            })
            .collect::<io::Result<_>>()?,
        session_records: state
            .session_records
            .into_iter()
            .map(|record| {
                require_utf8_state_path(&record.canonical_root)?;
                Ok(ManagedSessionRecord {
                    record_id: record.record_id,
                    display_name: record.display_name,
                    provider: record.provider.into_agent_id(),
                    mode: record.mode,
                    state: record.state,
                    workspace_id: record.workspace_id,
                    canonical_root: record.canonical_root,
                    provider_session: record.provider_session,
                    active_session: record.active_session,
                    environment_profile: None,
                    bundle: None,
                    context_id: None,
                    context: None,
                    exported_context: None,
                    task_binding: None,
                    created_at_unix_ms: record.created_at_unix_ms,
                    updated_at_unix_ms: record.updated_at_unix_ms,
                    last_error: record.last_error,
                })
            })
            .collect::<io::Result<_>>()?,
        managed_worktrees: Vec::new(),
        managed_worktree_tombstones: Vec::new(),
        materializations: Vec::new(),
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v3_state(state: PersistedNodeStateV3) -> io::Result<DecodedNodeState> {
    Ok(DecodedNodeState {
        node_id: state.node_id,
        workspaces: state
            .workspaces
            .into_iter()
            .map(|workspace| {
                Ok((
                    workspace.workspace_id,
                    require_utf8_state_path(&workspace.canonical_root)?.to_owned(),
                ))
            })
            .collect::<io::Result<_>>()?,
        session_records: state
            .session_records
            .into_iter()
            .map(|record| {
                require_utf8_state_path(&record.canonical_root)?;
                Ok(ManagedSessionRecord {
                    record_id: record.record_id,
                    display_name: record.display_name,
                    provider: record.provider,
                    mode: record.mode,
                    state: record.state,
                    workspace_id: record.workspace_id,
                    canonical_root: record.canonical_root,
                    provider_session: record.provider_session,
                    active_session: record.active_session,
                    environment_profile: None,
                    bundle: None,
                    context_id: None,
                    context: None,
                    exported_context: None,
                    task_binding: None,
                    created_at_unix_ms: record.created_at_unix_ms,
                    updated_at_unix_ms: record.updated_at_unix_ms,
                    last_error: record.last_error,
                })
            })
            .collect::<io::Result<_>>()?,
        managed_worktrees: Vec::new(),
        managed_worktree_tombstones: Vec::new(),
        materializations: Vec::new(),
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v4_state(state: PersistedNodeStateV4) -> io::Result<DecodedNodeState> {
    let legacy = decode_v3_state(PersistedNodeStateV3 {
        version: NODE_STATE_SCHEMA_V3,
        node_id: state.node_id,
        workspaces: state.workspaces,
        session_records: state.session_records,
    })?;
    validate_managed_worktree_records(&state.managed_worktrees, false)?;
    validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
    validate_managed_worktree_identity_sets(
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    Ok(DecodedNodeState {
        node_id: legacy.node_id,
        workspaces: legacy.workspaces,
        session_records: legacy.session_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
        materializations: Vec::new(),
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v5_state(state: PersistedNodeStateV5) -> io::Result<DecodedNodeState> {
    let mut environment_profiles = Vec::with_capacity(state.session_records.len());
    let legacy_records = state
        .session_records
        .into_iter()
        .map(|record| {
            environment_profiles.push(record.environment_profile);
            PersistedManagedSessionRecordV3 {
                record_id: record.record_id,
                display_name: record.display_name,
                provider: record.provider,
                mode: record.mode,
                state: record.state,
                workspace_id: record.workspace_id,
                canonical_root: record.canonical_root,
                provider_session: record.provider_session,
                active_session: record.active_session,
                created_at_unix_ms: record.created_at_unix_ms,
                updated_at_unix_ms: record.updated_at_unix_ms,
                last_error: record.last_error,
            }
        })
        .collect();
    let mut decoded = decode_v4_state(PersistedNodeStateV4 {
        version: NODE_STATE_SCHEMA_V4,
        node_id: state.node_id,
        workspaces: state.workspaces,
        session_records: legacy_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
    })?;
    for (record, environment_profile) in decoded
        .session_records
        .iter_mut()
        .zip(environment_profiles)
    {
        record.environment_profile = environment_profile;
    }
    Ok(decoded)
}

fn decode_v6_state(state: PersistedNodeStateV6) -> io::Result<DecodedNodeState> {
    let mut decoded = decode_v5_state(PersistedNodeStateV5 {
        version: NODE_STATE_SCHEMA_V5,
        node_id: state.node_id,
        workspaces: state.workspaces,
        session_records: state.session_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
    })?;
    let materializations = state.materializations.into_iter().map(|record| {
        MaterializationOwnershipRecord::from_persisted(
            record.materialization_id,
            Some(record.environment_profile),
            None,
            None,
            record.owner,
            record.managed_lease_id,
            record.state,
            opaque_to_path(&record.root)?,
            opaque_to_path(&record.provider_home)?,
            opaque_to_path(&record.root)?.join("bundle"),
            opaque_to_path(&record.root)?.join("plugin-data"),
            record.declared_paths,
            record.created_at_unix_ms,
            record.updated_at_unix_ms,
        ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
    }).collect::<io::Result<Vec<_>>>()?;
    validate_materialization_records(
        &materializations,
        &decoded.session_records,
        &decoded.managed_worktrees,
        &decoded.managed_worktree_tombstones,
    )?;
    decoded.materializations = materializations;
    Ok(decoded)
}

fn decode_v7_state(state: PersistedNodeStateV7) -> io::Result<DecodedNodeState> {
    let session_records = state.session_records.into_iter().map(|mut record| {
        record.context_id = None;
        record.context = None;
        record.task_binding = None;
        record
    }).collect::<Vec<_>>();
    let materializations = state.materializations.into_iter().map(|record| {
        MaterializationOwnershipRecord::from_persisted(
            record.materialization_id,
            record.environment_profile,
            record.bundle,
            None,
            record.owner,
            record.managed_lease_id,
            record.state,
            opaque_to_path(&record.root)?,
            opaque_to_path(&record.provider_home)?,
            opaque_to_path(&record.bundle_root)?,
            opaque_to_path(&record.plugin_data)?,
            record.declared_paths,
            record.created_at_unix_ms,
            record.updated_at_unix_ms,
        ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
    }).collect::<io::Result<Vec<_>>>()?;
    validate_managed_worktree_records(&state.managed_worktrees, false)?;
    validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
    validate_managed_worktree_identity_sets(
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    validate_materialization_records(
        &materializations,
        &session_records,
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    Ok(DecodedNodeState {
        node_id: state.node_id,
        workspaces: state.workspaces.into_iter().map(|workspace| {
            Ok((workspace.workspace_id, require_utf8_state_path(&workspace.canonical_root)?.to_owned()))
        }).collect::<io::Result<_>>()?,
        session_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
        materializations,
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v8_state(mut state: PersistedNodeStateV8) -> io::Result<DecodedNodeState> {
    for record in &mut state.session_records {
        record.task_binding = None;
    }
    decode_v9_state(state)
}

fn decode_v9_state(state: PersistedNodeStateV9) -> io::Result<DecodedNodeState> {
    let materializations = state.materializations.into_iter().map(|record| {
        MaterializationOwnershipRecord::from_persisted(
            record.materialization_id,
            record.environment_profile,
            record.bundle,
            record.context,
            record.owner,
            record.managed_lease_id,
            record.state,
            opaque_to_path(&record.root)?,
            opaque_to_path(&record.provider_home)?,
            opaque_to_path(&record.bundle_root)?,
            opaque_to_path(&record.plugin_data)?,
            record.declared_paths,
            record.created_at_unix_ms,
            record.updated_at_unix_ms,
        ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
    }).collect::<io::Result<Vec<_>>>()?;
    validate_managed_worktree_records(&state.managed_worktrees, false)?;
    validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
    validate_managed_worktree_identity_sets(
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    validate_materialization_records(
        &materializations,
        &state.session_records,
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    Ok(DecodedNodeState {
        node_id: state.node_id,
        workspaces: state.workspaces.into_iter().map(|workspace| {
            Ok((workspace.workspace_id, require_utf8_state_path(&workspace.canonical_root)?.to_owned()))
        }).collect::<io::Result<_>>()?,
        session_records: state.session_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
        materializations,
        managed_spawn_replays: Vec::new(),
    })
}

fn decode_v10_state(state: PersistedNodeStateV10) -> io::Result<DecodedNodeState> {
    validate_managed_spawn_replays(
        &state.managed_spawn_replays,
        &state.managed_worktrees,
        &state.managed_worktree_tombstones,
    )?;
    let mut decoded = decode_v9_state(PersistedNodeStateV9 {
        version: NODE_STATE_SCHEMA_V9,
        node_id: state.node_id,
        workspaces: state.workspaces,
        session_records: state.session_records,
        managed_worktrees: state.managed_worktrees,
        managed_worktree_tombstones: state.managed_worktree_tombstones,
        materializations: state.materializations,
    })?;
    decoded.managed_spawn_replays = state.managed_spawn_replays;
    Ok(decoded)
}

fn validate_managed_spawn_replays(
    records: &[ManagedWorktreeSpawnReplayRecordV10],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
) -> io::Result<()> {
    if records.len() > MAX_MANAGED_WORKTREE_SPAWN_REPLAYS {
        return Err(invalid_data("durable state contains too many managed spawn replays"));
    }
    let mut keys = BTreeSet::new();
    for record in records {
        let digest = record.request_digest.strip_prefix("sha256:").ok_or_else(|| {
            invalid_data("durable managed spawn replay request digest is invalid")
        })?;
        if digest.len() != 64
            || !digest
                .bytes()
                .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
        {
            return Err(invalid_data(
                "durable managed spawn replay request digest is invalid",
            ));
        }
        if record.updated_at_unix_ms < record.created_at_unix_ms {
            return Err(invalid_data(
                "durable managed spawn replay update precedes creation",
            ));
        }
        let (lease_id, receipt) = match &record.state {
            ManagedWorktreeSpawnReplayStateV10::Pending { lease_id, .. } => {
                (lease_id.as_ref(), None)
            }
            ManagedWorktreeSpawnReplayStateV10::Committed { receipt } => {
                if receipt.spawn.idempotency_key != record.idempotency_key {
                    return Err(invalid_data(
                        "durable managed spawn replay receipt is not linked to its idempotency key",
                    ));
                }
                if receipt.lease.source_workspace_id != record.source_workspace_id
                    || receipt.lease.profile_id != record.profile_id
                    || receipt.lease.profile_revision != record.expected_profile_revision
                {
                    return Err(invalid_data(
                        "durable managed spawn replay receipt is not linked to its reservation",
                    ));
                }
                (Some(&receipt.lease.lease_id), Some(receipt))
            }
        };
        if let Some(lease_id) = lease_id {
            let durable_lease = managed_worktrees
                .iter()
                .chain(managed_worktree_tombstones)
                .find(|lease| &lease.lease_id == lease_id);
            match durable_lease {
                Some(durable_lease)
                    if durable_lease.source_workspace_id == record.source_workspace_id
                        && durable_lease.profile_id == record.profile_id
                        && durable_lease.profile_revision == record.expected_profile_revision
                        && receipt.map_or(true, |receipt| {
                            durable_lease.workspace_id == receipt.lease.workspace_id
                        }) => {}
                None if receipt.is_none() => {}
                _ => {
                    return Err(invalid_data(
                        "durable managed spawn replay lease identity is inconsistent",
                    ));
                }
            }
        }
        if !keys.insert(record.idempotency_key.clone()) {
            return Err(invalid_data(
                "durable state contains duplicate managed spawn replay keys",
            ));
        }
    }
    Ok(())
}

fn validate_materialization_records(
    records: &[MaterializationOwnershipRecord],
    session_records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
) -> io::Result<()> {
    if records.len() > MAX_SESSION_MATERIALIZATIONS {
        return Err(invalid_data("durable state contains too many materializations"));
    }
    let known_records = session_records
        .iter()
        .map(|record| (&record.record_id, record))
        .collect::<BTreeMap<_, _>>();
    let active_leases = managed_worktrees
        .iter()
        .map(|lease| (&lease.lease_id, lease))
        .collect::<BTreeMap<_, _>>();
    let tombstone_ids = managed_worktree_tombstones
        .iter()
        .map(|lease| &lease.lease_id)
        .collect::<BTreeSet<_>>();
    let mut ids = BTreeSet::new();
    let mut roots = BTreeSet::new();
    let mut owners = BTreeSet::new();
    for record in records {
        record.validate().map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))?;
        if record.context().is_some_and(|context| !context.is_valid()) {
            return Err(invalid_data(
                "durable state materialization context receipt is invalid",
            ));
        }
        let (owner_key, record_workspace) = match record.owner() {
            MaterializationOwner::Session { incarnation_id, instance_id, generation } => {
                (
                    format!("session:{incarnation_id}:{}:{}", instance_id.0, generation.0),
                    None,
                )
            }
            MaterializationOwner::Record { record_id } => {
                let managed_record = known_records.get(record_id).ok_or_else(|| {
                    invalid_data(
                        "durable state materialization references an unknown session record",
                    )
                })?;
                if record.environment_profile() != managed_record.environment_profile.as_ref()
                    || record.bundle() != managed_record.bundle.as_ref()
                    || record.context() != managed_record.context.as_ref()
                {
                    return Err(invalid_data(
                        "durable state materialization receipt disagrees with its session record",
                    ));
                }
                (
                    format!("record:{}", record_id.as_str()),
                    Some(&managed_record.workspace_id),
                )
            }
        };
        if let Some(lease_id) = record.managed_lease_id() {
            if tombstone_ids.contains(lease_id) {
                return Err(invalid_data(
                    "durable state materialization references a removed managed worktree lease",
                ));
            }
            let lease = active_leases.get(lease_id).ok_or_else(|| {
                invalid_data(
                    "durable state materialization references an unknown managed worktree lease",
                )
            })?;
            if let Some(record_workspace) = record_workspace {
                if record_workspace != &lease.workspace_id {
                    return Err(invalid_data(
                        "durable state materialization record and managed worktree lease disagree on workspace",
                    ));
                }
            }
        }
        if !ids.insert(record.id().clone())
            || !roots.insert(record.root().to_path_buf())
            || !owners.insert(owner_key)
        {
            return Err(invalid_data("durable state contains duplicate materialization ownership"));
        }
    }
    for record in session_records.iter().filter(|record| {
        record.bundle.is_some() || record.context.is_some()
    }) {
        if !records.iter().any(|materialization| {
            materialization.owner()
                == &MaterializationOwner::Record {
                    record_id: record.record_id.clone(),
                }
                && materialization.bundle() == record.bundle.as_ref()
                && materialization.environment_profile() == record.environment_profile.as_ref()
                && materialization.context() == record.context.as_ref()
        }) {
            return Err(invalid_data(
                "durable bound session record lacks exact materialization ownership",
            ));
        }
    }
    Ok(())
}

fn validate_managed_worktree_identity_sets(
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
) -> io::Result<()> {
    let active_ids = managed_worktrees
        .iter()
        .map(|lease| &lease.lease_id)
        .collect::<BTreeSet<_>>();
    if managed_worktree_tombstones
        .iter()
        .any(|lease| active_ids.contains(&lease.lease_id))
    {
        return Err(invalid_data(
            "durable state contains duplicate managed worktree leases",
        ));
    }
    Ok(())
}

fn validate_managed_worktree_records(
    records: &[ManagedWorktreeLeaseRecord],
    tombstones: bool,
) -> io::Result<()> {
    if records.len() > crate::protocol::MAX_MANAGED_WORKTREE_LEASES {
        return Err(invalid_data("durable state contains too many managed worktree leases"));
    }
    let mut lease_ids = BTreeSet::new();
    let mut workspace_ids = BTreeSet::new();
    let mut roots = BTreeSet::new();
    let mut branches = BTreeSet::new();
    for lease in records {
        validate_root(&lease.target_root)?;
        let mut session_holders = BTreeSet::new();
        let mut record_holders = BTreeSet::new();
        if lease.branch.is_empty()
            || lease.branch.len() > MAX_PERSISTED_MANAGED_WORKTREE_BRANCH_BYTES
            || !valid_git_object_id(&lease.base_commit)
            || lease.expected_head.as_ref().is_some_and(|head| !valid_git_object_id(head))
            || lease.source_workspace_id == lease.workspace_id
            || lease.branch.chars().any(char::is_control)
            || lease.base_commit.chars().any(char::is_control)
            || lease.created_at_unix_ms > lease.updated_at_unix_ms
            || tombstones != (lease.state == crate::protocol::ManagedWorktreeLeaseState::Removed)
            || !lease_ids.insert(lease.lease_id.clone())
            || !workspace_ids.insert(lease.workspace_id.clone())
            || !roots.insert(crate::platform::root_identity(&lease.target_root))
            || !branches.insert(lease.branch.clone())
            || lease.session_holders.len() > gate4agent_types::CONTROL_SESSIONS_MAX
            || lease.record_holders.len() > MAX_MANAGED_SESSION_RECORDS
            || lease.session_holders.iter().any(|holder| !session_holders.insert((
                holder.incarnation_id,
                holder.instance_id,
                holder.generation,
            )))
            || lease.record_holders.iter().any(|holder| !record_holders.insert(holder.clone()))
            || (lease.state == crate::protocol::ManagedWorktreeLeaseState::Removed
                && (!lease.session_holders.is_empty() || !lease.record_holders.is_empty()))
            || (matches!(
                lease.state,
                crate::protocol::ManagedWorktreeLeaseState::Allocating
                    | crate::protocol::ManagedWorktreeLeaseState::Ready
                    | crate::protocol::ManagedWorktreeLeaseState::Retained
                    | crate::protocol::ManagedWorktreeLeaseState::CleanupBlocked
                    | crate::protocol::ManagedWorktreeLeaseState::Removed
            ) && (!lease.session_holders.is_empty() || !lease.record_holders.is_empty()))
            || (lease.state == crate::protocol::ManagedWorktreeLeaseState::InUse
                && lease.session_holders.is_empty() && lease.record_holders.is_empty())
            || (matches!(
                lease.state,
                crate::protocol::ManagedWorktreeLeaseState::CleanupBlocked
                    | crate::protocol::ManagedWorktreeLeaseState::RecoveryRequired
            ) != lease.cleanup_failure.is_some())
        {
            return Err(invalid_data("durable managed worktree record is invalid"));
        }
    }
    Ok(())
}

fn valid_git_object_id(value: &str) -> bool {
    matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
}

fn require_utf8_state_path(path: &OpaqueHostPath) -> io::Result<&str> {
    path.as_utf8().ok_or_else(|| io::Error::new(
        io::ErrorKind::Unsupported,
        StateLoadRefusalError::PathSemanticsUnsupported,
    ))
}

fn is_authoritative_state_error(error: &io::Error) -> bool {
    state_load_refusal(error).is_some()
}

pub fn state_load_refusal(error: &io::Error) -> Option<StateLoadRefusal> {
    error
        .get_ref()
        .and_then(|source| source.downcast_ref::<StateLoadRefusalError>())
        .map(StateLoadRefusalError::refusal)
}

#[cfg(test)]
pub fn save(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
) -> io::Result<Option<String>> {
    let Some(path) = path else {
        return Ok(None);
    };
    if workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("refusing to persist too many workspaces"));
    }
    if records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("refusing to persist too many session records"));
    }
    let mut persisted_records = Vec::with_capacity(records.len());
    let mut provider_sessions = BTreeSet::new();
    for record in records {
        let mut persisted = record.clone();
        persisted.active_session = None;
        if let Some(identity) = &mut persisted.provider_session {
            if !provider_sessions.insert(provider_session_semantic_key(
                &persisted.provider,
                identity,
            )) {
                return Err(invalid_data(
                    "refusing to persist duplicate provider sessions",
                ));
            }
            identity.transcript_path = None;
        }
        persisted.last_error = persisted
            .last_error
            .as_deref()
            .map(sanitized_record_error_summary);
        validate_record(&persisted)?;
        persisted.state = if persisted.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        persisted_records.push(persisted);
    }
    let state = PersistedNodeStateV3 {
        version: NODE_STATE_SCHEMA_V3,
        node_id: node_id.clone(),
        workspaces: workspaces
            .iter()
            .map(|(workspace_id, canonical_root)| {
                Ok(PersistedWorkspaceV2 {
                    workspace_id: workspace_id.clone(),
                    canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
                        invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
                    })?,
                })
            })
            .collect::<io::Result<_>>()?,
        session_records: persisted_records
            .into_iter()
                .map(persisted_v3_record)
            .collect(),
    };
    let bytes = serde_json::to_vec_pretty(&state)
        .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("refusing to persist oversized durable state"));
    }
    let previous_primary_valid = if !path.exists() {
        true
    } else {
        match load_one(path, node_id) {
            Ok(_) => true,
            Err(error) if is_authoritative_state_error(&error) => return Err(error),
            Err(_) => false,
        }
    };
    atomic_write(path, &bytes, previous_primary_valid)
}

#[cfg(test)]
pub fn save_v4(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
) -> io::Result<Option<String>> {
    let Some(path) = path else {
        return Ok(None);
    };
    if workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("refusing to persist too many workspaces"));
    }
    if records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("refusing to persist too many session records"));
    }
    validate_managed_worktree_records(managed_worktrees, false)?;
    validate_managed_worktree_records(managed_worktree_tombstones, true)?;
    let mut persisted_records = Vec::with_capacity(records.len());
    let mut provider_sessions = BTreeSet::new();
    for record in records {
        let mut persisted = record.clone();
        persisted.active_session = None;
        if let Some(identity) = &mut persisted.provider_session {
            if !provider_sessions.insert(provider_session_semantic_key(
                &persisted.provider,
                identity,
            )) {
                return Err(invalid_data("refusing to persist duplicate provider sessions"));
            }
            identity.transcript_path = None;
        }
        persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
        validate_record(&persisted)?;
        persisted.state = if persisted.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        persisted_records.push(persisted_v3_record(persisted));
    }
    let state = PersistedNodeStateV4 {
        version: NODE_STATE_SCHEMA_V4,
        node_id: node_id.clone(),
        workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
            Ok(PersistedWorkspaceV2 {
                workspace_id: workspace_id.clone(),
                canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
                    invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
                })?,
            })
        }).collect::<io::Result<_>>()?,
        session_records: persisted_records,
        managed_worktrees: managed_worktrees.to_vec(),
        managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
    };
    let bytes = serde_json::to_vec_pretty(&state)
        .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("refusing to persist oversized durable state"));
    }
    let previous_primary_valid = if !path.exists() {
        true
    } else {
        match load_one(path, node_id) {
            Ok(_) => true,
            Err(error) if is_authoritative_state_error(&error) => return Err(error),
            Err(_) => false,
        }
    };
    atomic_write(path, &bytes, previous_primary_valid)
}

#[cfg(test)]
pub fn save_v5(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
) -> io::Result<Option<String>> {
    let Some(path) = path else {
        return Ok(None);
    };
    if workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("refusing to persist too many workspaces"));
    }
    if records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("refusing to persist too many session records"));
    }
    validate_managed_worktree_records(managed_worktrees, false)?;
    validate_managed_worktree_records(managed_worktree_tombstones, true)?;
    let mut persisted_records = Vec::with_capacity(records.len());
    let mut provider_sessions = BTreeSet::new();
    for record in records {
        let mut persisted = record.clone();
        persisted.active_session = None;
        if let Some(identity) = &mut persisted.provider_session {
            if !provider_sessions.insert(provider_session_semantic_key(
                &persisted.provider,
                identity,
            )) {
                return Err(invalid_data("refusing to persist duplicate provider sessions"));
            }
            identity.transcript_path = None;
        }
        persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
        validate_record(&persisted)?;
        persisted.state = if persisted.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        persisted_records.push(persisted_v5_record(persisted));
    }
    let state = PersistedNodeStateV5 {
        version: NODE_STATE_SCHEMA_V5,
        node_id: node_id.clone(),
        workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
            Ok(PersistedWorkspaceV2 {
                workspace_id: workspace_id.clone(),
                canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
                    invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
                })?,
            })
        }).collect::<io::Result<_>>()?,
        session_records: persisted_records,
        managed_worktrees: managed_worktrees.to_vec(),
        managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
    };
    let bytes = serde_json::to_vec_pretty(&state)
        .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("refusing to persist oversized durable state"));
    }
    let previous_primary_valid = if !path.exists() {
        true
    } else {
        match load_one(path, node_id) {
            Ok(_) => true,
            Err(error) if is_authoritative_state_error(&error) => return Err(error),
            Err(_) => false,
        }
    };
    atomic_write(path, &bytes, previous_primary_valid)
}

#[cfg(test)]
pub fn save_v7(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
    materializations: &[MaterializationOwnershipRecord],
) -> io::Result<Option<String>> {
    let Some(path) = path else {
        return Ok(None);
    };
    if records.iter().any(|record| record.context_id.is_some() || record.context.is_some())
        || materializations.iter().any(|record| record.context().is_some())
    {
        return Err(invalid_data(
            "refusing to persist context metadata in legacy durable state v7",
        ));
    }
    if workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("refusing to persist too many workspaces"));
    }
    if records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("refusing to persist too many session records"));
    }
    validate_managed_worktree_records(managed_worktrees, false)?;
    validate_managed_worktree_records(managed_worktree_tombstones, true)?;
    validate_managed_worktree_identity_sets(
        managed_worktrees,
        managed_worktree_tombstones,
    )?;
    validate_materialization_records(
        materializations,
        records,
        managed_worktrees,
        managed_worktree_tombstones,
    )?;
    let mut persisted_records = Vec::with_capacity(records.len());
    let mut provider_sessions = BTreeSet::new();
    for record in records {
        let mut persisted = record.clone();
        persisted.active_session = None;
        if let Some(identity) = &mut persisted.provider_session {
            if !provider_sessions.insert(provider_session_semantic_key(&persisted.provider, identity)) {
                return Err(invalid_data("refusing to persist duplicate provider sessions"));
            }
            identity.transcript_path = None;
        }
        persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
        validate_record(&persisted)?;
        persisted.state = if persisted.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        persisted_records.push(persisted);
    }
    let state = PersistedNodeStateV7 {
        version: NODE_STATE_SCHEMA_V7,
        node_id: node_id.clone(),
        workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
            Ok(PersistedWorkspaceV2 {
                workspace_id: workspace_id.clone(),
                canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
                    invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
                })?,
            })
        }).collect::<io::Result<_>>()?,
        session_records: persisted_records,
        managed_worktrees: managed_worktrees.to_vec(),
        managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
        materializations: materializations.iter().map(|record| {
            Ok(PersistedMaterializationRecordV7 {
                materialization_id: record.id().clone(),
                environment_profile: record.environment_profile().cloned(),
                bundle: record.bundle().cloned(),
                owner: record.owner().clone(),
                managed_lease_id: record.managed_lease_id().cloned(),
                state: record.state(),
                root: path_to_opaque(record.root())?,
                provider_home: path_to_opaque(record.provider_home())?,
                bundle_root: path_to_opaque(record.bundle_root())?,
                plugin_data: path_to_opaque(record.plugin_data())?,
                declared_paths: record.declared_paths().to_vec(),
                created_at_unix_ms: record.created_at_unix_ms(),
                updated_at_unix_ms: record.updated_at_unix_ms(),
            })
        }).collect::<io::Result<Vec<_>>>()?,
    };
    let bytes = serde_json::to_vec_pretty(&state)
        .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("refusing to persist oversized durable state"));
    }
    let previous_primary_valid = if !path.exists() {
        true
    } else {
        match load_one(path, node_id) {
            Ok(_) => true,
            Err(error) if is_authoritative_state_error(&error) => return Err(error),
            Err(_) => false,
        }
    };
    atomic_write(path, &bytes, previous_primary_valid)
}

#[cfg(test)]
pub fn save_v8(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
    materializations: &[MaterializationOwnershipRecord],
) -> io::Result<Option<String>> {
    if records.iter().any(|record| record.task_binding.is_some()) {
        return Err(invalid_data(
            "refusing to persist task bindings in legacy durable state v8",
        ));
    }
    save_current(
        path,
        node_id,
        workspaces,
        records,
        managed_worktrees,
        managed_worktree_tombstones,
        materializations,
        &[],
        NODE_STATE_SCHEMA_V8,
    )
}

#[cfg(test)]
pub fn save_v9(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
    materializations: &[MaterializationOwnershipRecord],
) -> io::Result<Option<String>> {
    save_current(
        path,
        node_id,
        workspaces,
        records,
        managed_worktrees,
        managed_worktree_tombstones,
        materializations,
        &[],
        NODE_STATE_SCHEMA_V9,
    )
}

pub fn save_v10(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
    materializations: &[MaterializationOwnershipRecord],
    managed_spawn_replays: &[ManagedWorktreeSpawnReplayRecordV10],
) -> io::Result<Option<String>> {
    save_current(
        path,
        node_id,
        workspaces,
        records,
        managed_worktrees,
        managed_worktree_tombstones,
        materializations,
        managed_spawn_replays,
        NODE_STATE_SCHEMA_V10,
    )
}

fn save_current(
    path: Option<&Path>,
    node_id: &NodeId,
    workspaces: &BTreeMap<WorkspaceId, String>,
    records: &[ManagedSessionRecord],
    managed_worktrees: &[ManagedWorktreeLeaseRecord],
    managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
    materializations: &[MaterializationOwnershipRecord],
    managed_spawn_replays: &[ManagedWorktreeSpawnReplayRecordV10],
    version: u16,
) -> io::Result<Option<String>> {
    let Some(path) = path else {
        return Ok(None);
    };
    if workspaces.len() > MAX_PERSISTED_WORKSPACES {
        return Err(invalid_data("refusing to persist too many workspaces"));
    }
    if records.len() > MAX_MANAGED_SESSION_RECORDS {
        return Err(invalid_data("refusing to persist too many session records"));
    }
    validate_managed_worktree_records(managed_worktrees, false)?;
    validate_managed_worktree_records(managed_worktree_tombstones, true)?;
    validate_managed_worktree_identity_sets(
        managed_worktrees,
        managed_worktree_tombstones,
    )?;
    validate_materialization_records(
        materializations,
        records,
        managed_worktrees,
        managed_worktree_tombstones,
    )?;
    if version == NODE_STATE_SCHEMA_V10 {
        validate_managed_spawn_replays(
            managed_spawn_replays,
            managed_worktrees,
            managed_worktree_tombstones,
        )?;
    } else if !managed_spawn_replays.is_empty() {
        return Err(invalid_data(
            "refusing to persist managed spawn replays in a legacy durable state",
        ));
    }
    let mut persisted_records = Vec::with_capacity(records.len());
    let mut provider_sessions = BTreeSet::new();
    for record in records {
        let mut persisted = record.clone();
        persisted.active_session = None;
        if let Some(identity) = &mut persisted.provider_session {
            if !provider_sessions.insert(provider_session_semantic_key(&persisted.provider, identity)) {
                return Err(invalid_data("refusing to persist duplicate provider sessions"));
            }
            identity.transcript_path = None;
        }
        persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
        validate_record(&persisted)?;
        persisted.state = if persisted.provider_session.is_some() {
            ManagedSessionState::Dormant
        } else {
            ManagedSessionState::Unavailable
        };
        persisted_records.push(persisted);
    }
    let persisted_workspaces = workspaces.iter().map(|(workspace_id, canonical_root)| {
            Ok(PersistedWorkspaceV2 {
                workspace_id: workspace_id.clone(),
                canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
                    invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
                })?,
            })
        }).collect::<io::Result<Vec<_>>>()?;
    let persisted_materializations = materializations.iter().map(|record| {
            Ok(PersistedMaterializationRecordV8 {
                materialization_id: record.id().clone(),
                environment_profile: record.environment_profile().cloned(),
                bundle: record.bundle().cloned(),
                context: record.context().cloned(),
                owner: record.owner().clone(),
                managed_lease_id: record.managed_lease_id().cloned(),
                state: record.state(),
                root: path_to_opaque(record.root())?,
                provider_home: path_to_opaque(record.provider_home())?,
                bundle_root: path_to_opaque(record.bundle_root())?,
                plugin_data: path_to_opaque(record.plugin_data())?,
                declared_paths: record.declared_paths().to_vec(),
                created_at_unix_ms: record.created_at_unix_ms(),
                updated_at_unix_ms: record.updated_at_unix_ms(),
            })
        }).collect::<io::Result<Vec<_>>>()?;
    let bytes = if version == NODE_STATE_SCHEMA_V10 {
        serde_json::to_vec_pretty(&PersistedNodeStateV10 {
            version,
            node_id: node_id.clone(),
            workspaces: persisted_workspaces,
            session_records: persisted_records,
            managed_worktrees: managed_worktrees.to_vec(),
            managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
            materializations: persisted_materializations,
            managed_spawn_replays: managed_spawn_replays.to_vec(),
        })
    } else {
        serde_json::to_vec_pretty(&PersistedNodeStateV9 {
            version,
            node_id: node_id.clone(),
            workspaces: persisted_workspaces,
            session_records: persisted_records,
            managed_worktrees: managed_worktrees.to_vec(),
            managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
            materializations: persisted_materializations,
        })
    };
    let bytes = bytes.map_err(|error| {
        invalid_data(format!("durable state encoding failed: {error}"))
    })?;
    if bytes.len() as u64 > MAX_STATE_BYTES {
        return Err(invalid_data("refusing to persist oversized durable state"));
    }
    let previous_primary_valid = if !path.exists() {
        true
    } else {
        match load_one(path, node_id) {
            Ok(_) => true,
            Err(error) if is_authoritative_state_error(&error) => return Err(error),
            Err(_) => false,
        }
    };
    atomic_write(path, &bytes, previous_primary_valid)
}

#[cfg(test)]
fn persisted_v3_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV3 {
    PersistedManagedSessionRecordV3 {
        record_id: record.record_id,
        display_name: record.display_name,
        provider: record.provider,
        mode: record.mode,
        state: record.state,
        workspace_id: record.workspace_id,
        canonical_root: record.canonical_root,
        provider_session: record.provider_session,
        active_session: record.active_session,
        created_at_unix_ms: record.created_at_unix_ms,
        updated_at_unix_ms: record.updated_at_unix_ms,
        last_error: record.last_error,
    }
}

#[cfg(test)]
fn persisted_v5_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV5 {
    PersistedManagedSessionRecordV5 {
        record_id: record.record_id,
        display_name: record.display_name,
        provider: record.provider,
        mode: record.mode,
        state: record.state,
        workspace_id: record.workspace_id,
        canonical_root: record.canonical_root,
        provider_session: record.provider_session,
        active_session: record.active_session,
        environment_profile: record.environment_profile,
        created_at_unix_ms: record.created_at_unix_ms,
        updated_at_unix_ms: record.updated_at_unix_ms,
        last_error: record.last_error,
    }
}

pub fn validate_display_name(display_name: &str) -> io::Result<()> {
    if display_name.trim().is_empty()
        || display_name.len() > MAX_SESSION_DISPLAY_NAME_BYTES
        || display_name.chars().any(char::is_control)
    {
        return Err(invalid_data(format!(
            "session display name must contain 1..={MAX_SESSION_DISPLAY_NAME_BYTES} bytes and no control characters",
        )));
    }
    Ok(())
}

fn validate_record(record: &ManagedSessionRecord) -> io::Result<()> {
    validate_display_name(&record.display_name)?;
    validate_root(require_utf8_state_path(&record.canonical_root)?)?;
    if record.created_at_unix_ms > record.updated_at_unix_ms {
        return Err(invalid_data("session record timestamps are inconsistent"));
    }
    if !record.context_binding_is_valid() {
        return Err(invalid_data(
            "session record context id and receipt are not correlated",
        ));
    }
    if !record.task_binding_is_valid()
        || record.task_binding.as_ref().is_some_and(|binding| {
            binding.changed_at_unix_ms < record.created_at_unix_ms
                || binding.changed_at_unix_ms > record.updated_at_unix_ms
        })
    {
        return Err(invalid_data("session record task binding is invalid"));
    }
    if let Some(identity) = &record.provider_session {
        identity
            .validate()
            .map_err(|error| invalid_data(format!("provider session identity is invalid: {error}")))?;
    }
    if record
        .last_error
        .as_ref()
        .is_some_and(|message| {
            message.len() > MAX_NODE_TEXT_BYTES || message.chars().any(|character| character == '\0')
        })
    {
        return Err(invalid_data("session record error is invalid"));
    }
    Ok(())
}

pub fn sanitized_record_error_summary(message: &str) -> String {
    let normalized = message.to_ascii_lowercase();
    if matches!(
        normalized.as_str(),
        "workspace-unavailable"
            | "provider-session-scope-conflict"
            | "provider-session-live-conflict"
            | "managed-session-capacity"
            | "provider-session-identity-allocation-failed"
            | "provider-resume-rejected"
            | "provider-authentication-required"
            | "provider-readiness-failed"
            | "provider-process-exited"
            | "provider-resume-failed"
            | "durable-state-commit-failed"
            | "provider-runtime-failed"
            | "environment-profile-unavailable"
            | "bundle-unavailable"
            | "bundle-binding-mismatch"
            | "bundle-materialization-failed"
    ) {
        return normalized;
    }
    if normalized.contains("workspace is unavailable") {
        "workspace-unavailable".to_owned()
    } else if normalized.contains("another workspace or transport") {
        "provider-session-scope-conflict".to_owned()
    } else if normalized.contains("conflicts with another live managed record") {
        "provider-session-live-conflict".to_owned()
    } else if normalized.contains("managed session capacity") {
        "managed-session-capacity".to_owned()
    } else if normalized.contains("allocate a record") {
        "provider-session-identity-allocation-failed".to_owned()
    } else if normalized.contains("resume command was rejected") {
        "provider-resume-rejected".to_owned()
    } else if normalized.contains("authentication") || normalized.contains("sign in") {
        "provider-authentication-required".to_owned()
    } else if normalized.contains("readiness") {
        "provider-readiness-failed".to_owned()
    } else if normalized.contains("exited with code")
        || normalized.contains("no row for root pid")
    {
        "provider-process-exited".to_owned()
    } else if normalized.contains("resume") {
        "provider-resume-failed".to_owned()
    } else if normalized.contains("durable state") || normalized.contains("persistence") {
        "durable-state-commit-failed".to_owned()
    } else {
        "provider-runtime-failed".to_owned()
    }
}

fn validate_root(root: &str) -> io::Result<()> {
    if root.is_empty()
        || root.len() > MAX_WORKSPACE_ROOT_BYTES
        || root.chars().any(char::is_control)
        || !Path::new(root).is_absolute()
    {
        return Err(invalid_data("durable workspace root is invalid"));
    }
    Ok(())
}

#[derive(Clone, Copy, Eq, PartialEq)]
enum AtomicWriteFault {
    None,
    BeforePrimaryCommit,
    BeforeBackupRotation,
}

fn atomic_write(
    path: &Path,
    bytes: &[u8],
    previous_primary_valid: bool,
) -> io::Result<Option<String>> {
    atomic_write_with_policy(
        path,
        bytes,
        AtomicWriteFault::None,
        previous_primary_valid,
    )
}

#[cfg(test)]
fn atomic_write_with_fault(
    path: &Path,
    bytes: &[u8],
    fault: AtomicWriteFault,
) -> io::Result<Option<String>> {
    atomic_write_with_policy(path, bytes, fault, true)
}

fn atomic_write_with_policy(
    path: &Path,
    bytes: &[u8],
    fault: AtomicWriteFault,
    previous_primary_valid: bool,
) -> io::Result<Option<String>> {
    let parent = path.parent().ok_or_else(|| {
        io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
    })?;
    fs::create_dir_all(parent)?;
    #[cfg(unix)]
    secure_state_directory(parent)?;
    let (temporary, mut file) = create_unique_sibling_file(path, "tmp")?;
    let backup = sibling_path(path, "bak");
    let prepared = (|| {
        file.write_all(bytes)?;
        file.flush()?;
        file.sync_all()
    })();
    drop(file);
    if let Err(error) = prepared {
        let _ = fs::remove_file(&temporary);
        return Err(error);
    }
    let pending_backup = sibling_path(path, "pending-backup");
    if pending_backup.exists() {
        let pending_result = if backup.exists() {
            rotate_backup_after_commit(&backup, &pending_backup, AtomicWriteFault::None)
        } else {
            fs::rename(&pending_backup, &backup)
        };
        if let Err(error) = pending_result {
            let _ = fs::remove_file(&temporary);
            return Err(error);
        }
    }
    let displaced = if path.exists() {
        let displaced = if !previous_primary_valid {
            unique_sibling_path(path, "corrupt")?
        } else if backup.exists() {
            pending_backup.clone()
        } else {
            backup.clone()
        };
        if let Err(error) = fs::rename(path, &displaced) {
            let _ = fs::remove_file(&temporary);
            return Err(error);
        }
        Some(displaced)
    } else {
        None
    };
    let result = (|| {
        if fault == AtomicWriteFault::BeforePrimaryCommit {
            return Err(io::Error::new(
                io::ErrorKind::Other,
                "injected failure before durable primary commit",
            ));
        }
        fs::rename(&temporary, path)?;
        sync_parent_directory(path)
    })();
    if let Err(error) = result {
        let _ = fs::remove_file(&temporary);
        if let Some(displaced) = displaced.as_deref() {
            if !path.exists() {
                fs::rename(displaced, path).map_err(|rollback_error| {
                    io::Error::new(
                        rollback_error.kind(),
                        format!(
                            "durable primary commit failed ({error}); restoring the prior primary also failed: {rollback_error}"
                        ),
                    )
                })?;
            }
        }
        return Err(error);
    }

    let backup_warning = if let Some(displaced) = displaced.as_deref() {
        if displaced == pending_backup {
            rotate_backup_after_commit(&backup, displaced, fault)
                .err()
                .map(|_| BACKUP_ROTATION_WARNING.to_owned())
        } else if displaced != backup {
            Some(CORRUPT_PRIMARY_PRESERVED_WARNING.to_owned())
        } else {
            None
        }
    } else {
        None
    };
    Ok(backup_warning)
}

fn rotate_backup_after_commit(
    backup: &Path,
    previous_primary: &Path,
    fault: AtomicWriteFault,
) -> io::Result<()> {
    if fault == AtomicWriteFault::BeforeBackupRotation {
        return Err(io::Error::new(
            io::ErrorKind::Other,
            "injected failure before durable backup rotation",
        ));
    }
    let preserved_backup = unique_sibling_path(backup, "preserved")?;
    replace_file(backup, previous_primary, &preserved_backup)?;
    let _ = fs::remove_file(preserved_backup);
    Ok(())
}

fn replace_file(replaced: &Path, replacement: &Path, preserved: &Path) -> io::Result<()> {
    #[cfg(windows)]
    {
    use std::ffi::c_void;
    use std::os::windows::ffi::OsStrExt;

    #[link(name = "kernel32")]
    extern "system" {
        #[link_name = "ReplaceFileW"]
        fn replace_file_w(
            replaced_file_name: *const u16,
            replacement_file_name: *const u16,
            backup_file_name: *const u16,
            replace_flags: u32,
            exclude: *mut c_void,
            reserved: *mut c_void,
        ) -> i32;
    }

    let replaced = replaced
        .as_os_str()
        .encode_wide()
        .chain(std::iter::once(0))
        .collect::<Vec<_>>();
    let replacement = replacement
        .as_os_str()
        .encode_wide()
        .chain(std::iter::once(0))
        .collect::<Vec<_>>();
    let preserved = preserved
        .as_os_str()
        .encode_wide()
        .chain(std::iter::once(0))
        .collect::<Vec<_>>();
    // SAFETY: all paths are NUL-terminated UTF-16 buffers that remain alive
    // for the call. The reserved pointers are null as required by Win32.
    let replaced = unsafe {
        replace_file_w(
            replaced.as_ptr(),
            replacement.as_ptr(),
            preserved.as_ptr(),
            0,
            std::ptr::null_mut(),
            std::ptr::null_mut(),
        )
    };
    if replaced == 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(())
    }
    }
    #[cfg(unix)]
    {
        fs::rename(replaced, preserved)?;
        match fs::rename(replacement, replaced) {
            Ok(()) => sync_parent_directory(replaced),
            Err(error) => {
                let rollback = fs::rename(preserved, replaced);
                match rollback {
                    Ok(()) => Err(error),
                    Err(rollback_error) => Err(io::Error::new(
                        rollback_error.kind(),
                        format!(
                            "durable backup rotation failed ({error}); restoring the prior backup also failed: {rollback_error}"
                        ),
                    )),
                }
            }
        }
    }
}

#[cfg(windows)]
fn sync_parent_directory(_path: &Path) -> io::Result<()> {
    Ok(())
}

#[cfg(unix)]
fn sync_parent_directory(path: &Path) -> io::Result<()> {
    let parent = path.parent().ok_or_else(|| {
        io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
    })?;
    File::open(parent)?.sync_all()
}

fn create_unique_sibling_file(path: &Path, suffix: &str) -> io::Result<(PathBuf, File)> {
    for _ in 0..32 {
        let candidate = unique_sibling_candidate(path, suffix);
        let mut options = OpenOptions::new();
        options.create_new(true).write(true);
        #[cfg(unix)]
        options.mode(0o600);
        match options.open(&candidate) {
            Ok(file) => return Ok((candidate, file)),
            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
            Err(error) => return Err(error),
        }
    }
    Err(io::Error::new(
        io::ErrorKind::AlreadyExists,
        "could not allocate a unique durable state temporary file",
    ))
}

fn unique_sibling_path(path: &Path, suffix: &str) -> io::Result<PathBuf> {
    for _ in 0..32 {
        let candidate = unique_sibling_candidate(path, suffix);
        if !candidate.exists() {
            return Ok(candidate);
        }
    }
    Err(io::Error::new(
        io::ErrorKind::AlreadyExists,
        "could not allocate a unique durable state sibling path",
    ))
}

fn unique_sibling_candidate(path: &Path, suffix: &str) -> PathBuf {
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("state-v1.json");
    let sequence = NEXT_TEMPORARY_FILE.fetch_add(1, Ordering::Relaxed);
    path.with_file_name(format!(
        ".{name}.{suffix}-{}-{sequence}",
        std::process::id(),
    ))
}

fn sibling_path(path: &Path, suffix: &str) -> PathBuf {
    let name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("state-v1.json");
    path.with_file_name(format!(".{name}.{suffix}"))
}

fn invalid_data(message: impl Into<String>) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, message.into())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::{SessionMode, SessionRecordId};
    use gate4agent_types::{ProviderSessionIdentity, ProviderSessionKey};
    use std::sync::atomic::{AtomicU64, Ordering};

    static NEXT_TEMP: AtomicU64 = AtomicU64::new(1);

    fn agent(value: &str) -> AgentId {
        AgentId::new(value).unwrap()
    }

    fn temp_path(test: &str) -> PathBuf {
        let unique = NEXT_TEMP.fetch_add(1, Ordering::Relaxed);
        std::env::temp_dir()
            .join(format!("gate4agent-node-state-{}-{unique}", std::process::id()))
            .join(format!("{test}.json"))
    }

    fn fixture(path: &Path) -> (NodeId, BTreeMap<WorkspaceId, String>, ManagedSessionRecord) {
        let node_id = NodeId::new("node-a").unwrap();
        let workspace_id = WorkspaceId::new("primary").unwrap();
        let root = path.parent().unwrap().to_string_lossy().into_owned();
        let workspaces = BTreeMap::from([(workspace_id.clone(), root.clone())]);
        let record = ManagedSessionRecord {
            record_id: SessionRecordId::new("session-1").unwrap(),
            display_name: "Claude release check".to_owned(),
            provider: agent("claude"),
            mode: SessionMode::Pty,
            state: ManagedSessionState::Live,
            workspace_id,
            canonical_root: OpaqueHostPath::utf8(root).unwrap(),
            provider_session: Some(ProviderSessionIdentity {
                key: ProviderSessionKey::SessionId,
                id: "provider-session-1".to_owned(),
                transcript_path: Some(r"C:\private\provider-transcript.jsonl".to_owned()),
            }),
            active_session: None,
            environment_profile: None,
            bundle: None,
            context_id: None,
            context: None,
            exported_context: None,
            task_binding: None,
            created_at_unix_ms: 10,
            updated_at_unix_ms: 11,
            last_error: None,
        };
        (node_id, workspaces, record)
    }

    fn managed_lease(path: &Path, id: &str, workspace: &str, branch: &str) -> ManagedWorktreeLeaseRecord {
        ManagedWorktreeLeaseRecord {
            lease_id: crate::protocol::ManagedWorktreeLeaseId::new(id).unwrap(),
            source_workspace_id: WorkspaceId::new("primary").unwrap(),
            workspace_id: WorkspaceId::new(workspace).unwrap(),
            profile_id: crate::protocol::WorktreeProfileId::new("default").unwrap(),
            profile_revision: crate::protocol::WorktreeProfileRevision::new("v1").unwrap(),
            target_root: path.to_string_lossy().into_owned(),
            branch: branch.to_owned(),
            base_commit: "0123456789abcdef0123456789abcdef01234567".to_owned(),
            expected_head: Some("0123456789abcdef0123456789abcdef01234567".to_owned()),
            retention: crate::protocol::ManagedWorktreeRetention::RemoveWhenReleased,
            state: crate::protocol::ManagedWorktreeLeaseState::Ready,
            session_holders: Vec::new(),
            record_holders: Vec::new(),
            cleanup_failure: None,
            created_at_unix_ms: 10,
            updated_at_unix_ms: 11,
        }
    }

    fn materialization_ownership(
        path: &Path,
        id: &str,
        owner: MaterializationOwner,
        managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
    ) -> MaterializationOwnershipRecord {
        let root = path
            .parent()
            .unwrap()
            .join("session-environments")
            .join(id);
        MaterializationOwnershipRecord::from_persisted(
            MaterializationId::new(id).unwrap(),
            Some(ResolvedEnvironmentProfileReceipt {
                profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude")
                    .unwrap(),
                profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1")
                    .unwrap(),
                    network_allowlist: None,
                    browser_profile_id: None,
            }),
            None,
            None,
            owner,
            managed_lease_id,
            MaterializationState::Ready,
            root.clone(),
            root.join("home"),
            root.join("bundle"),
            root.join("plugin-data"),
            Vec::new(),
            10,
            11,
        )
        .unwrap()
    }

    fn persisted_materialization(
        record: &MaterializationOwnershipRecord,
    ) -> PersistedMaterializationRecordV6 {
        PersistedMaterializationRecordV6 {
            materialization_id: record.id().clone(),
            environment_profile: record.environment_profile().cloned().unwrap(),
            owner: record.owner().clone(),
            managed_lease_id: record.managed_lease_id().cloned(),
            state: record.state(),
            root: path_to_opaque(record.root()).unwrap(),
            provider_home: path_to_opaque(record.provider_home()).unwrap(),
            declared_paths: record.declared_paths().to_vec(),
            created_at_unix_ms: record.created_at_unix_ms(),
            updated_at_unix_ms: record.updated_at_unix_ms(),
        }
    }

    fn context_receipt(id: &str, digest_byte: char) -> ResolvedContextPackReceipt {
        ResolvedContextPackReceipt {
            id: crate::protocol::SpawnContextId::new(id).unwrap(),
            digest: crate::protocol::SpawnContextDigest::new(format!(
                "sha256:{}",
                digest_byte.to_string().repeat(64),
            ))
            .unwrap(),
            lineage: crate::protocol::ContextPackLineageReceipt {
                source_node_id: NodeId::new("source-node").unwrap(),
                source_session: SessionAddress {
                    workspace_id: WorkspaceId::new("source-workspace").unwrap(),
                    session: crate::protocol::SessionKey {
                        instance_id: gate4agent_types::AgentInstanceId(7),
                        generation: gate4agent_types::SessionGeneration(3),
                    },
                },
                source_provider: agent("codex"),
            },
            source_message_count: 12,
            retained_message_count: 9,
            byte_len: 4096,
            truncated: true,
        }
    }

    fn context_materialization(
        path: &Path,
        id: &str,
        record_id: SessionRecordId,
        context: ResolvedContextPackReceipt,
    ) -> MaterializationOwnershipRecord {
        let root = path
            .parent()
            .unwrap()
            .join("session-environments")
            .join(id);
        MaterializationOwnershipRecord::from_persisted(
            MaterializationId::new(id).unwrap(),
            None,
            None,
            Some(context),
            MaterializationOwner::Record { record_id },
            None,
            MaterializationState::Ready,
            root.clone(),
            root.join("home"),
            root.join("bundle"),
            root.join("plugin-data"),
            Vec::new(),
            10,
            11,
        )
        .unwrap()
    }

    #[test]
    fn v4_roundtrip_preserves_host_only_managed_lease_identity() {
        let path = temp_path("v4-managed-roundtrip");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        let lease = managed_lease(
            &path.parent().unwrap().join("managed-a"),
            "mw-a",
            "managed-a",
            "gate4agent/mw-a",
        );
        save_v4(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[lease.clone()],
            &[],
        ).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(
            &std::fs::read(&path).unwrap(),
        ).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V4);
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.managed_worktrees, vec![lease]);
        assert!(loaded.managed_worktree_tombstones.is_empty());
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v5_roundtrip_preserves_only_opaque_environment_profile_identity() {
        let path = temp_path("v5-environment-profile-roundtrip");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let environment_profile = ResolvedEnvironmentProfileReceipt {
            profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
            profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new(
                "local-claude-r1",
            )
            .unwrap(),
            network_allowlist: None,
            browser_profile_id: None,
        };
        record.environment_profile = Some(environment_profile.clone());

        save_v5(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
        )
        .unwrap();
        let bytes = std::fs::read(&path).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V5);
        let serialized = String::from_utf8(bytes).unwrap();
        assert!(serialized.contains("local-claude"));
        assert!(!serialized.contains("ANTHROPIC_AUTH_TOKEN"));
        assert!(!serialized.contains("secret"));

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records.len(), 1);
        assert_eq!(
            loaded.records[0].environment_profile.as_ref(),
            Some(&environment_profile),
        );
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v4_to_v5_migration_defaults_environment_profile_to_none() {
        let path = temp_path("v4-to-v5-environment-profile");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        save_v4(Some(&path), &node_id, &workspaces, &[record], &[], &[]).unwrap();

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records.len(), 1);
        assert!(loaded.records[0].environment_profile.is_none());
        save_v5(
            Some(&path),
            &node_id,
            &workspaces,
            &loaded.records,
            &[],
            &[],
        )
        .unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(
            &std::fs::read(&path).unwrap(),
        )
        .unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V5);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v5_to_v6_migration_starts_with_empty_materialization_registry() {
        let path = temp_path("v5-to-v6-materializations");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        save_v5(Some(&path), &node_id, &workspaces, &[record], &[], &[]).unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert!(loaded.materializations.is_empty());
        save_v7(
            Some(&path),
            &node_id,
            &workspaces,
            &loaded.records,
            &[],
            &[],
            &loaded.materializations,
        ).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V7);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_roundtrip_persists_ownership_without_secret_reference_or_value() {
        let path = temp_path("v6-materialization-roundtrip");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, _) = fixture(&path);
        let materialization_root = path.parent().unwrap().join("session-environments").join("m-1");
        let ownership = MaterializationOwnershipRecord::from_persisted(
            MaterializationId::new("m-1").unwrap(),
            Some(ResolvedEnvironmentProfileReceipt {
                profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
                profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
                network_allowlist: None,
                browser_profile_id: None,
            }),
            None,
            None,
            MaterializationOwner::Session {
                incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
                instance_id: gate4agent_types::AgentInstanceId(9),
                generation: gate4agent_types::SessionGeneration(2),
            },
            None,
            MaterializationState::Ready,
            materialization_root.clone(),
            materialization_root.join("home"),
            materialization_root.join("bundle"),
            materialization_root.join("plugin-data"),
            vec![MaterializedPathDeclaration {
                class: crate::materialization::NodeSessionPathClass::Config,
                relative_path: PathBuf::from("provider/settings.json"),
                kind: crate::materialization::MaterializedPathKind::Generated,
            }],
            10,
            11,
        ).unwrap();
        save_v7(Some(&path), &node_id, &workspaces, &[], &[], &[], &[ownership.clone()]).unwrap();
        let bytes = std::fs::read(&path).unwrap();
        let serialized = String::from_utf8(bytes).unwrap();
        assert!(!serialized.contains("fixture-secret-reference"));
        assert!(!serialized.contains("fixture-secret-value"));
        let loaded = load(Some(&path), &node_id).unwrap();
        assert!(loaded.materializations == vec![ownership]);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_save_rejects_unknown_materialization_managed_lease() {
        let path = temp_path("v6-unknown-materialization-lease");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let node_id = NodeId::new("node-a").unwrap();
        let ownership = materialization_ownership(
            &path,
            "m-unknown",
            MaterializationOwner::Session {
                incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
                    3;
                    crate::protocol::NODE_INCARNATION_ID_BYTES
                ]),
                instance_id: gate4agent_types::AgentInstanceId(9),
                generation: gate4agent_types::SessionGeneration(2),
            },
            Some(crate::protocol::ManagedWorktreeLeaseId::new("mw-missing").unwrap()),
        );

        let error = save_v7(
            Some(&path),
            &node_id,
            &BTreeMap::new(),
            &[],
            &[],
            &[],
            &[ownership],
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert!(!path.exists());
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_decode_rejects_tombstoned_materialization_managed_lease() {
        let path = temp_path("v6-tombstoned-materialization-lease");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let node_id = NodeId::new("node-a").unwrap();
        let mut tombstone = managed_lease(
            &path.parent().unwrap().join("managed-a"),
            "mw-a",
            "managed-a",
            "gate4agent/mw-a",
        );
        tombstone.state = crate::protocol::ManagedWorktreeLeaseState::Removed;
        let ownership = materialization_ownership(
            &path,
            "m-tombstone",
            MaterializationOwner::Session {
                incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
                    3;
                    crate::protocol::NODE_INCARNATION_ID_BYTES
                ]),
                instance_id: gate4agent_types::AgentInstanceId(9),
                generation: gate4agent_types::SessionGeneration(2),
            },
            Some(tombstone.lease_id.clone()),
        );
        let state = PersistedNodeStateV6 {
            version: NODE_STATE_SCHEMA_V6,
            node_id: node_id.clone(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            managed_worktree_tombstones: vec![tombstone],
            materializations: vec![persisted_materialization(&ownership)],
        };
        std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();

        assert_eq!(
            load(Some(&path), &node_id).unwrap_err().kind(),
            io::ErrorKind::InvalidData,
        );
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_save_rejects_record_materialization_for_foreign_lease_workspace() {
        let path = temp_path("v6-record-foreign-lease-workspace");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        let lease = managed_lease(
            &path.parent().unwrap().join("managed-a"),
            "mw-a",
            "managed-a",
            "gate4agent/mw-a",
        );
        let ownership = materialization_ownership(
            &path,
            "m-foreign",
            MaterializationOwner::Record {
                record_id: record.record_id.clone(),
            },
            Some(lease.lease_id.clone()),
        );

        let error = save_v7(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[lease],
            &[],
            &[ownership],
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert!(!path.exists());
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_roundtrip_accepts_session_materialization_for_active_lease() {
        let path = temp_path("v6-session-active-materialization-lease");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let node_id = NodeId::new("node-a").unwrap();
        let lease = managed_lease(
            &path.parent().unwrap().join("managed-a"),
            "mw-a",
            "managed-a",
            "gate4agent/mw-a",
        );
        let ownership = materialization_ownership(
            &path,
            "m-active",
            MaterializationOwner::Session {
                incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
                    3;
                    crate::protocol::NODE_INCARNATION_ID_BYTES
                ]),
                instance_id: gate4agent_types::AgentInstanceId(9),
                generation: gate4agent_types::SessionGeneration(2),
            },
            Some(lease.lease_id.clone()),
        );

        save_v7(
            Some(&path),
            &node_id,
            &BTreeMap::new(),
            &[],
            &[lease.clone()],
            &[],
            &[ownership.clone()],
        )
        .unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.managed_worktrees, vec![lease]);
        assert!(loaded.materializations == vec![ownership]);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_rejects_declared_path_prefix_collisions() {
        let path = temp_path("v6-prefix-collision");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let node_id = NodeId::new("node-a").unwrap();
        let materialization_root = path.parent().unwrap().join("session-environments").join("m-1");
        let state = PersistedNodeStateV6 {
            version: NODE_STATE_SCHEMA_V6,
            node_id: node_id.clone(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            managed_worktree_tombstones: Vec::new(),
            materializations: vec![PersistedMaterializationRecordV6 {
                materialization_id: MaterializationId::new("m-1").unwrap(),
                environment_profile: ResolvedEnvironmentProfileReceipt {
                    profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
                    profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
                    network_allowlist: None,
                    browser_profile_id: None,
                },
                owner: MaterializationOwner::Session {
                    incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
                    instance_id: gate4agent_types::AgentInstanceId(9),
                    generation: gate4agent_types::SessionGeneration(2),
                },
                managed_lease_id: None,
                state: MaterializationState::Ready,
                root: path_to_opaque(&materialization_root).unwrap(),
                provider_home: path_to_opaque(&materialization_root.join("home")).unwrap(),
                declared_paths: vec![
                    MaterializedPathDeclaration {
                        class: crate::materialization::NodeSessionPathClass::Config,
                        relative_path: PathBuf::from("provider"),
                        kind: crate::materialization::MaterializedPathKind::Generated,
                    },
                    MaterializedPathDeclaration {
                        class: crate::materialization::NodeSessionPathClass::Config,
                        relative_path: PathBuf::from("provider/settings.json"),
                        kind: crate::materialization::MaterializedPathKind::Generated,
                    },
                ],
                created_at_unix_ms: 10,
                updated_at_unix_ms: 11,
            }],
        };
        std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
        assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn v6_rejects_case_ambiguous_materialization_ids() {
        let path = temp_path("v6-case-ambiguous-materialization-id");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let node_id = NodeId::new("node-a").unwrap();
        let first_root = path.parent().unwrap().join("session-environments").join("a");
        let second_root = path.parent().unwrap().join("session-environments").join("b");
        let first = PersistedMaterializationRecordV6 {
            materialization_id: MaterializationId::new("a").unwrap(),
            environment_profile: ResolvedEnvironmentProfileReceipt {
                profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
                profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
                network_allowlist: None,
                browser_profile_id: None,
            },
            owner: MaterializationOwner::Session {
                incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
                instance_id: gate4agent_types::AgentInstanceId(9),
                generation: gate4agent_types::SessionGeneration(2),
            },
            managed_lease_id: None,
            state: MaterializationState::Ready,
            root: path_to_opaque(&first_root).unwrap(),
            provider_home: path_to_opaque(&first_root.join("home")).unwrap(),
            declared_paths: Vec::new(),
            created_at_unix_ms: 10,
            updated_at_unix_ms: 11,
        };
        let mut second = first.clone();
        second.materialization_id = MaterializationId::new("b").unwrap();
        second.owner = MaterializationOwner::Session {
            incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
            instance_id: gate4agent_types::AgentInstanceId(10),
            generation: gate4agent_types::SessionGeneration(2),
        };
        second.root = path_to_opaque(&second_root).unwrap();
        second.provider_home = path_to_opaque(&second_root.join("home")).unwrap();
        let state = PersistedNodeStateV6 {
            version: NODE_STATE_SCHEMA_V6,
            node_id: node_id.clone(),
            workspaces: Vec::new(),
            session_records: Vec::new(),
            managed_worktrees: Vec::new(),
            managed_worktree_tombstones: Vec::new(),
            materializations: vec![first, second],
        };
        let mut value = serde_json::to_value(state).unwrap();
        value["materializations"][1]["materialization_id"] = serde_json::Value::String("A".to_owned());
        std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
        assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    #[test]
    fn malformed_v4_duplicate_branch_is_rejected_without_startup_panic() {
        let path = temp_path("v4-managed-duplicate-branch");
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        let first = managed_lease(
            &path.parent().unwrap().join("managed-a"),
            "mw-a",
            "managed-a",
            "gate4agent/shared",
        );
        let second = managed_lease(
            &path.parent().unwrap().join("managed-b"),
            "mw-b",
            "managed-b",
            "gate4agent/shared",
        );
        let state = PersistedNodeStateV4 {
            version: NODE_STATE_SCHEMA_V4,
            node_id: node_id.clone(),
            workspaces: workspaces.into_iter().map(|(workspace_id, canonical_root)| {
                PersistedWorkspaceV2 {
                    workspace_id,
                    canonical_root: OpaqueHostPath::utf8(canonical_root).unwrap(),
                }
            }).collect(),
            session_records: vec![persisted_v3_record(record)],
            managed_worktrees: vec![first, second],
            managed_worktree_tombstones: Vec::new(),
        };
        std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
        assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
        std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
    }

    fn persisted_v1_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV1 {
        PersistedManagedSessionRecordV1 {
            record_id: record.record_id,
            display_name: record.display_name,
            provider: LegacyAgentProvider::from_agent_id(&record.provider),
            mode: record.mode,
            state: record.state,
            workspace_id: record.workspace_id,
            canonical_root: record.canonical_root.as_utf8().unwrap().to_owned(),
            provider_session: record.provider_session,
            active_session: record.active_session,
            created_at_unix_ms: record.created_at_unix_ms,
            updated_at_unix_ms: record.updated_at_unix_ms,
            last_error: record.last_error,
        }
    }

    fn persisted_v1_state(
        version: u16,
        node_id: NodeId,
        workspaces: &BTreeMap<WorkspaceId, String>,
        records: Vec<ManagedSessionRecord>,
    ) -> PersistedNodeStateV1 {
        PersistedNodeStateV1 {
            version,
            node_id,
            workspaces: workspaces
                .iter()
                .map(|(workspace_id, canonical_root)| PersistedWorkspaceV1 {
                    workspace_id: workspace_id.clone(),
                    canonical_root: canonical_root.clone(),
                })
                .collect(),
            session_records: records.into_iter().map(persisted_v1_record).collect(),
        }
    }

    fn persisted_v2_state(
        node_id: NodeId,
        workspaces: &BTreeMap<WorkspaceId, String>,
        records: Vec<ManagedSessionRecord>,
    ) -> PersistedNodeStateV2 {
        PersistedNodeStateV2 {
            version: NODE_STATE_SCHEMA_V2,
            node_id,
            workspaces: workspaces
                .iter()
                .map(|(workspace_id, canonical_root)| PersistedWorkspaceV2 {
                    workspace_id: workspace_id.clone(),
                    canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).unwrap(),
                })
                .collect(),
            session_records: records
                .into_iter()
                .map(|record| PersistedManagedSessionRecordV2 {
                    record_id: record.record_id,
                    display_name: record.display_name,
                    provider: LegacyAgentProvider::from_agent_id(&record.provider),
                    mode: record.mode,
                    state: record.state,
                    workspace_id: record.workspace_id,
                    canonical_root: record.canonical_root,
                    provider_session: record.provider_session,
                    active_session: record.active_session,
                    created_at_unix_ms: record.created_at_unix_ms,
                    updated_at_unix_ms: record.updated_at_unix_ms,
                    last_error: record.last_error,
                })
                .collect(),
        }
    }

    #[test]
    fn durable_state_round_trips_without_ephemeral_active_binding() {
        let path = temp_path("round-trip");
        let (node_id, workspaces, mut record) = fixture(&path);
        record.provider_session.as_mut().unwrap().transcript_path =
            Some("C:\\private\\bearer-secret\nprovider-transcript.jsonl".to_owned());
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        let persisted = fs::read_to_string(&path).unwrap();
        assert_eq!(
            serde_json::from_str::<PersistedNodeStateHeader>(&persisted)
                .unwrap()
                .version,
            NODE_STATE_SCHEMA_V3,
        );
        assert!(!persisted.contains("provider-transcript.jsonl"));
        assert!(!persisted.contains(r"C:\private"));
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.workspaces, workspaces);
        assert_eq!(loaded.records.len(), 1);
        assert_eq!(loaded.records[0].record_id, record.record_id);
        assert_eq!(loaded.records[0].state, ManagedSessionState::Dormant);
        assert_eq!(loaded.records[0].active_session, None);
        assert_eq!(
            loaded.records[0]
                .provider_session
                .as_ref()
                .unwrap()
                .transcript_path,
            None,
        );
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v3_state_round_trips_open_provider_id_exactly() {
        let path = temp_path("v3-open-provider-round-trip");
        let (node_id, workspaces, mut record) = fixture(&path);
        record.provider = agent("third-party-agent");
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();

        let persisted: PersistedNodeStateV3 =
            serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
        assert_eq!(persisted.version, NODE_STATE_SCHEMA_V3);
        assert_eq!(persisted.session_records.len(), 1);
        assert_eq!(persisted.session_records[0].provider, agent("third-party-agent"));

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records.len(), 1);
        assert_eq!(loaded.records[0].provider, agent("third-party-agent"));
        assert_eq!(loaded.records[0].record_id, record.record_id);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn semantic_provider_session_key_ignores_only_transcript_path() {
        let without_path = ProviderSessionIdentity {
            key: ProviderSessionKey::SessionId,
            id: "provider-session-1".to_owned(),
            transcript_path: None,
        };
        let with_path = ProviderSessionIdentity {
            transcript_path: Some(r"C:\private\provider-transcript.jsonl".to_owned()),
            ..without_path.clone()
        };
        assert_eq!(
            provider_session_semantic_key(&agent("claude"), &without_path),
            provider_session_semantic_key(&agent("claude"), &with_path),
        );
        assert_ne!(
            provider_session_semantic_key(&agent("claude"), &without_path),
            provider_session_semantic_key(&agent("codex"), &with_path),
        );
        let conversation = ProviderSessionIdentity {
            key: ProviderSessionKey::ConversationId,
            ..with_path
        };
        assert_ne!(
            provider_session_semantic_key(&agent("claude"), &without_path),
            provider_session_semantic_key(&agent("claude"), &conversation),
        );
    }

    #[test]
    fn legacy_state_transcript_path_is_removed_from_loaded_identity() {
        let path = temp_path("legacy-transcript-path");
        let (node_id, workspaces, mut record) = fixture(&path);
        record.provider_session.as_mut().unwrap().transcript_path =
            Some("C:\\private\\bearer-secret\nprovider-transcript.jsonl".to_owned());
        let legacy = persisted_v1_state(
            NODE_STATE_SCHEMA_V1,
            node_id.clone(),
            &workspaces,
            vec![record],
        );
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records.len(), 1);
        assert_eq!(
            loaded.records[0]
                .provider_session
                .as_ref()
                .unwrap()
                .transcript_path,
            None,
        );
        let snapshot = serde_json::to_string(&loaded.records).unwrap();
        assert!(!snapshot.contains("provider-transcript.jsonl"));
        assert!(!snapshot.contains("bearer-secret"));
        assert!(!snapshot.contains(r"C:\private"));
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v1_state_loads_losslessly_and_next_save_migrates_to_v3() {
        let path = temp_path("v1-migration");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, record) = fixture(&path);
        let legacy = persisted_v1_state(
            NODE_STATE_SCHEMA_V1,
            node_id.clone(),
            &workspaces,
            vec![record.clone()],
        );
        let legacy_bytes = serde_json::to_vec_pretty(&legacy).unwrap();
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, &legacy_bytes).unwrap();

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.workspaces, workspaces);
        assert_eq!(loaded.records[0].record_id, record.record_id);
        assert_eq!(
            loaded.records[0].canonical_root,
            record.canonical_root,
        );
        save(Some(&path), &node_id, &loaded.workspaces, &loaded.records).unwrap();

        let migrated = fs::read(&path).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&migrated).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V3);
        assert_eq!(fs::read(&backup).unwrap(), legacy_bytes);
        let reloaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(reloaded.workspaces, workspaces);
        assert_eq!(reloaded.records, loaded.records);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v2_state_loads_losslessly_and_next_save_migrates_to_v3() {
        let path = temp_path("v2-migration");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, record) = fixture(&path);
        let legacy = persisted_v2_state(
            node_id.clone(),
            &workspaces,
            vec![record.clone()],
        );
        let legacy_bytes = serde_json::to_vec_pretty(&legacy).unwrap();
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, &legacy_bytes).unwrap();

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.workspaces, workspaces);
        assert_eq!(loaded.records[0].provider, agent("claude"));
        assert_eq!(loaded.records[0].record_id, record.record_id);
        save(Some(&path), &node_id, &loaded.workspaces, &loaded.records).unwrap();

        let migrated = fs::read(&path).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&migrated).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V3);
        assert_eq!(fs::read(&backup).unwrap(), legacy_bytes);
        assert_eq!(load(Some(&path), &node_id).unwrap().records, loaded.records);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn save_rejects_semantic_provider_session_duplicates_before_commit() {
        let path = temp_path("duplicate-provider-session-save");
        let (node_id, workspaces, record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        let original = fs::read(&path).unwrap();
        let mut duplicate = record.clone();
        duplicate.record_id = SessionRecordId::new("session-2").unwrap();
        duplicate.display_name = "duplicate".to_owned();
        duplicate.provider_session.as_mut().unwrap().transcript_path =
            Some(r"D:\another\transcript.jsonl".to_owned());
        let error = save(
            Some(&path),
            &node_id,
            &workspaces,
            &[record, duplicate],
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert_eq!(fs::read(&path).unwrap(), original);
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records.len(), 1);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn load_rejects_semantic_provider_session_duplicates_with_different_paths() {
        let path = temp_path("duplicate-provider-session-load");
        let (node_id, workspaces, record) = fixture(&path);
        let mut duplicate = record.clone();
        duplicate.record_id = SessionRecordId::new("session-2").unwrap();
        duplicate.display_name = "duplicate".to_owned();
        duplicate.provider_session.as_mut().unwrap().transcript_path =
            Some(r"D:\another\transcript.jsonl".to_owned());
        let legacy = persisted_v1_state(
            NODE_STATE_SCHEMA_V1,
            node_id.clone(),
            &workspaces,
            vec![record, duplicate],
        );
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap();
        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn durable_state_persists_only_allowlisted_error_summaries() {
        let path = temp_path("sanitized-error");
        let (node_id, workspaces, mut record) = fixture(&path);
        record.last_error = Some(
            r"provider failed with bearer-secret-123 at C:\private\session.jsonl".to_owned(),
        );
        save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
        let persisted = fs::read_to_string(&path).unwrap();
        assert!(!persisted.contains("bearer-secret-123"));
        assert!(!persisted.contains(r"C:\private"));
        assert!(persisted.contains("provider-runtime-failed"));
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(
            loaded.records[0].last_error.as_deref(),
            Some("provider-runtime-failed"),
        );
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn corrupt_primary_recovers_valid_backup_without_rewriting_corrupt_bytes() {
        let path = temp_path("recover");
        let (node_id, workspaces, record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        let mut renamed = record;
        renamed.display_name = "renamed".to_owned();
        save(Some(&path), &node_id, &workspaces, &[renamed]).unwrap();
        fs::write(&path, b"{corrupt").unwrap();
        let corrupt = fs::read(&path).unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records[0].display_name, "Claude release check");
        assert_eq!(
            loaded.warning.as_deref(),
            Some(DURABLE_STATE_RECOVERY_WARNING),
        );
        assert!(!loaded.warning.unwrap().contains(&path.to_string_lossy().into_owned()));
        assert_eq!(fs::read(&path).unwrap(), corrupt);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn newer_primary_schema_refuses_without_falling_back_to_valid_v1_backup() {
        let path = temp_path("newer-primary-schema");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        load_one(&backup, &node_id).unwrap();

        let future = persisted_v1_state(
            NODE_STATE_SCHEMA_V10 + 1,
            node_id.clone(),
            &workspaces,
            vec![record.clone()],
        );
        let future_bytes = serde_json::to_vec_pretty(&future).unwrap();
        fs::write(&path, &future_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::Unsupported);
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::UnsupportedSchema),
        );
        assert_eq!(fs::read(&path).unwrap(), future_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
        assert_eq!(save_error.kind(), io::ErrorKind::Unsupported);
        assert_eq!(fs::read(&path).unwrap(), future_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn newer_pending_schema_refuses_without_skipping_to_older_backup() {
        let path = temp_path("newer-pending-schema");
        let backup = sibling_path(&path, "bak");
        let pending_backup = sibling_path(&path, "pending-backup");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        load_one(&backup, &node_id).unwrap();
        fs::remove_file(&path).unwrap();

        let future = persisted_v1_state(
            NODE_STATE_SCHEMA_V10 + 1,
            node_id.clone(),
            &workspaces,
            vec![record],
        );
        let future_bytes = serde_json::to_vec_pretty(&future).unwrap();
        fs::write(&pending_backup, &future_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::Unsupported);
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::UnsupportedSchema),
        );
        assert_eq!(fs::read(&pending_backup).unwrap(), future_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v2_unix_bytes_primary_refuses_without_fallback_or_rewrite() {
        let path = temp_path("unix-bytes-primary");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        let foreign = PersistedNodeStateV2 {
            version: NODE_STATE_SCHEMA_V2,
            node_id: node_id.clone(),
            workspaces: vec![PersistedWorkspaceV2 {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                canonical_root: OpaqueHostPath::unix_bytes(b"/srv/repo".to_vec()).unwrap(),
            }],
            session_records: Vec::new(),
        };
        let foreign_bytes = serde_json::to_vec_pretty(&foreign).unwrap();
        fs::write(&path, &foreign_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::PathSemanticsUnsupported),
        );
        let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
        assert_eq!(
            state_load_refusal(&save_error),
            Some(StateLoadRefusal::PathSemanticsUnsupported),
        );
        assert_eq!(fs::read(&path).unwrap(), foreign_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v2_unix_bytes_pending_refuses_without_skipping_older_backup() {
        let path = temp_path("unix-bytes-pending");
        let backup = sibling_path(&path, "bak");
        let pending_backup = sibling_path(&path, "pending-backup");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
        fs::remove_file(&path).unwrap();
        let foreign = PersistedNodeStateV2 {
            version: NODE_STATE_SCHEMA_V2,
            node_id: node_id.clone(),
            workspaces: vec![PersistedWorkspaceV2 {
                workspace_id: WorkspaceId::new("primary").unwrap(),
                canonical_root: OpaqueHostPath::unix_bytes(b"/srv/repo".to_vec()).unwrap(),
            }],
            session_records: Vec::new(),
        };
        let foreign_bytes = serde_json::to_vec_pretty(&foreign).unwrap();
        fs::write(&pending_backup, &foreign_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::PathSemanticsUnsupported),
        );
        assert_eq!(fs::read(&pending_backup).unwrap(), foreign_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn wrong_node_primary_refuses_without_falling_back_or_rewriting() {
        let path = temp_path("wrong-node-primary");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        load_one(&backup, &node_id).unwrap();

        let wrong_node = persisted_v1_state(
            NODE_STATE_SCHEMA_V1,
            NodeId::new("node-b").unwrap(),
            &workspaces,
            vec![record.clone()],
        );
        let wrong_node_bytes = serde_json::to_vec_pretty(&wrong_node).unwrap();
        fs::write(&path, &wrong_node_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::NodeIdentityMismatch),
        );
        let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
        assert_eq!(save_error.kind(), io::ErrorKind::PermissionDenied);
        assert_eq!(fs::read(&path).unwrap(), wrong_node_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn wrong_node_pending_refuses_without_skipping_to_older_backup() {
        let path = temp_path("wrong-node-pending");
        let backup = sibling_path(&path, "bak");
        let pending_backup = sibling_path(&path, "pending-backup");
        let (node_id, workspaces, mut record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        load_one(&backup, &node_id).unwrap();
        fs::remove_file(&path).unwrap();

        let wrong_node = persisted_v1_state(
            NODE_STATE_SCHEMA_V1,
            NodeId::new("node-b").unwrap(),
            &workspaces,
            vec![record],
        );
        let wrong_node_bytes = serde_json::to_vec_pretty(&wrong_node).unwrap();
        fs::write(&pending_backup, &wrong_node_bytes).unwrap();
        let backup_bytes = fs::read(&backup).unwrap();

        let error = load(Some(&path), &node_id).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
        assert_eq!(
            state_load_refusal(&error),
            Some(StateLoadRefusal::NodeIdentityMismatch),
        );
        assert_eq!(fs::read(&pending_backup).unwrap(), wrong_node_bytes);
        assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn authoritative_refusals_are_typed_not_inferred_from_io_kinds() {
        let generic_permission = io::Error::new(io::ErrorKind::PermissionDenied, "access denied");
        let generic_unsupported = io::Error::new(io::ErrorKind::Unsupported, "operation unsupported");
        assert_eq!(state_load_refusal(&generic_permission), None);
        assert_eq!(state_load_refusal(&generic_unsupported), None);

        let schema = io::Error::new(
            io::ErrorKind::Unsupported,
            StateLoadRefusalError::UnsupportedSchema(NODE_STATE_SCHEMA_V3 + 1),
        );
        let node = io::Error::new(
            io::ErrorKind::PermissionDenied,
            StateLoadRefusalError::NodeIdentityMismatch,
        );
        let path = io::Error::new(
            io::ErrorKind::Unsupported,
            StateLoadRefusalError::PathSemanticsUnsupported,
        );
        assert_eq!(
            state_load_refusal(&schema),
            Some(StateLoadRefusal::UnsupportedSchema),
        );
        assert_eq!(
            state_load_refusal(&node),
            Some(StateLoadRefusal::NodeIdentityMismatch),
        );
        assert_eq!(
            state_load_refusal(&path),
            Some(StateLoadRefusal::PathSemanticsUnsupported),
        );
    }

    #[test]
    fn backup_tracks_the_immediately_previous_generation_across_four_commits() {
        let path = temp_path("backup-generations");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, mut record) = fixture(&path);
        for generation in 1..=4 {
            record.display_name = format!("generation-{generation}");
            record.updated_at_unix_ms = 10 + generation;
            let warning = save(
                Some(&path),
                &node_id,
                &workspaces,
                &[record.clone()],
            )
            .unwrap();
            assert_eq!(warning, None);
            if generation > 1 {
                let loaded_backup = load_one(&backup, &node_id).unwrap();
                assert_eq!(
                    loaded_backup.records[0].display_name,
                    format!("generation-{}", generation - 1),
                );
            }
        }

        fs::write(&path, b"{corrupt").unwrap();
        let recovered = load(Some(&path), &node_id).unwrap();
        assert_eq!(recovered.records[0].display_name, "generation-3");
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn failed_backup_rotation_preserves_old_backup_and_pending_previous_primary() {
        let path = temp_path("backup-rotation-failure");
        let backup = sibling_path(&path, "bak");
        let pending_backup = sibling_path(&path, "pending-backup");
        let (node_id, workspaces, mut record) = fixture(&path);
        record.display_name = "generation-1".to_owned();
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        record.display_name = "generation-2".to_owned();
        record.updated_at_unix_ms += 1;
        save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
        let old_backup = fs::read(&backup).unwrap();
        let previous_primary = fs::read(&path).unwrap();

        let warning = atomic_write_with_fault(
            &path,
            &previous_primary,
            AtomicWriteFault::BeforeBackupRotation,
        )
        .unwrap();
        assert_eq!(warning.as_deref(), Some(BACKUP_ROTATION_WARNING));
        assert_eq!(fs::read(&backup).unwrap(), old_backup);
        assert_eq!(fs::read(&pending_backup).unwrap(), previous_primary);
        fs::write(&path, b"{corrupt").unwrap();
        let recovered = load(Some(&path), &node_id).unwrap();
        assert_eq!(recovered.records[0].display_name, "generation-2");
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn failed_recovery_commit_preserves_the_valid_backup() {
        let path = temp_path("recover-failed-commit");
        let backup = sibling_path(&path, "bak");
        let (node_id, workspaces, record) = fixture(&path);
        save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
        let mut renamed = record;
        renamed.display_name = "renamed".to_owned();
        save(Some(&path), &node_id, &workspaces, &[renamed]).unwrap();
        fs::write(&path, b"{corrupt").unwrap();
        let valid_backup = fs::read(&backup).unwrap();
        let corrupt_primary = fs::read(&path).unwrap();

        let error = atomic_write_with_policy(
            &path,
            br#"{"new":"state"}"#,
            AtomicWriteFault::BeforePrimaryCommit,
            false,
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::Other);
        assert_eq!(fs::read(&backup).unwrap(), valid_backup);
        assert_eq!(fs::read(&path).unwrap(), corrupt_primary);
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records[0].display_name, "Claude release check");
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn state_path_lock_rejects_a_second_owner_until_drop() {
        let path = temp_path("exclusive-owner");
        let first = StatePathLock::acquire(Some(&path)).unwrap().unwrap();
        let error = match StatePathLock::acquire(Some(&path)) {
            Ok(_) => panic!("a second durable state owner was accepted"),
            Err(error) => error,
        };
        assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
        drop(first);
        let second = StatePathLock::acquire(Some(&path)).unwrap().unwrap();
        drop(second);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn corrupt_only_copy_fails_and_is_preserved() {
        let path = temp_path("corrupt-only");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        fs::write(&path, b"not-json").unwrap();
        let error = load(Some(&path), &NodeId::new("node-a").unwrap()).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert_eq!(fs::read(&path).unwrap(), b"not-json");
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v6_to_v7_migration_preserves_environment_receipt_and_adds_no_bundle() {
        let path = temp_path("v6-to-v7-bundle-migration");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let environment_profile = ResolvedEnvironmentProfileReceipt {
            profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
            profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
            network_allowlist: None,
            browser_profile_id: None,
        };
        record.environment_profile = Some(environment_profile.clone());
        let ownership = materialization_ownership(
            &path,
            "legacy-environment",
            MaterializationOwner::Record {
                record_id: record.record_id.clone(),
            },
            None,
        );
        let state = PersistedNodeStateV6 {
            version: NODE_STATE_SCHEMA_V6,
            node_id: node_id.clone(),
            workspaces: workspaces.iter().map(|(workspace_id, root)| PersistedWorkspaceV2 {
                workspace_id: workspace_id.clone(),
                canonical_root: OpaqueHostPath::utf8(root.clone()).unwrap(),
            }).collect(),
            session_records: vec![persisted_v5_record(record)],
            managed_worktrees: Vec::new(),
            managed_worktree_tombstones: Vec::new(),
            materializations: vec![persisted_materialization(&ownership)],
        };
        fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records[0].environment_profile.as_ref(), Some(&environment_profile));
        assert!(loaded.records[0].bundle.is_none());
        assert!(loaded.materializations[0].bundle().is_none());
        save_v7(
            Some(&path),
            &node_id,
            &workspaces,
            &loaded.records,
            &[],
            &[],
            &loaded.materializations,
        )
        .unwrap();
        let header: PersistedNodeStateHeader =
            serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V7);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v7_roundtrip_preserves_opaque_bundle_and_rejects_receipt_mismatch() {
        let path = temp_path("v7-bundle-roundtrip");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let environment_profile = ResolvedEnvironmentProfileReceipt {
            profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
            profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
            network_allowlist: None,
            browser_profile_id: None,
        };
        let bundle = ResolvedBundleReceipt {
            id: crate::protocol::SpawnBundleId::new("review-bundle").unwrap(),
            revision: crate::protocol::SpawnBundleRevision::new("r1").unwrap(),
            digest: crate::protocol::SpawnBundleDigest::new(format!("sha256:{}", "0".repeat(64))).unwrap(),
        };
        record.environment_profile = Some(environment_profile.clone());
        record.bundle = Some(bundle.clone());
        let root = path.parent().unwrap().join("session-environments").join("bundle-record");
        let ownership = MaterializationOwnershipRecord::from_persisted(
            MaterializationId::new("bundle-record").unwrap(),
            Some(environment_profile),
            Some(bundle.clone()),
            None,
            MaterializationOwner::Record { record_id: record.record_id.clone() },
            None,
            MaterializationState::Ready,
            root.clone(),
            root.join("home"),
            root.join("bundle"),
            root.join("plugin-data"),
            Vec::new(),
            10,
            11,
        )
        .unwrap();
        save_v7(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[ownership],
        )
        .unwrap();
        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records[0].bundle.as_ref(), Some(&bundle));

        let mut value: serde_json::Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
        value["session_records"][0]["bundle"]["digest"] =
            serde_json::Value::String(format!("sha256:{}", "1".repeat(64)));
        fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
        assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v7_to_v8_migration_defaults_context_metadata_to_none() {
        let path = temp_path("v7-to-v8-context-migration");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);
        save_v7(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[],
        )
        .unwrap();

        let loaded_v7 = load(Some(&path), &node_id).unwrap();
        assert!(loaded_v7.records[0].context_id.is_none());
        assert!(loaded_v7.records[0].context.is_none());
        assert!(loaded_v7.materializations.is_empty());
        save_v8(
            Some(&path),
            &node_id,
            &workspaces,
            &loaded_v7.records,
            &[],
            &[],
            &loaded_v7.materializations,
        )
        .unwrap();

        let bytes = fs::read(&path).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V8);
        let loaded_v8 = load(Some(&path), &node_id).unwrap();
        assert!(loaded_v8.records[0].context_id.is_none());
        assert!(loaded_v8.records[0].context.is_none());
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v8_roundtrip_preserves_only_opaque_context_metadata() {
        let path = temp_path("v8-context-roundtrip");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let context = context_receipt("ctx-context-roundtrip", 'a');
        record.context_id = Some(context.id.clone());
        record.context = Some(context.clone());
        let ownership = context_materialization(
            &path,
            "context-record",
            record.record_id.clone(),
            context.clone(),
        );

        save_v8(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[ownership.clone()],
        )
        .unwrap();
        let bytes = fs::read(&path).unwrap();
        let serialized: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(serialized["version"], NODE_STATE_SCHEMA_V8);
        assert_eq!(
            serialized["session_records"][0]["context"]["id"],
            "ctx-context-roundtrip",
        );
        assert!(serialized["session_records"][0]["context"].get("messages").is_none());
        assert!(serialized["materializations"][0]["context"].get("messages").is_none());
        assert!(!String::from_utf8(bytes).unwrap().contains("bounded-history-message-body"));

        let loaded = load(Some(&path), &node_id).unwrap();
        assert_eq!(loaded.records[0].context_id.as_ref(), Some(&context.id));
        assert_eq!(loaded.records[0].context.as_ref(), Some(&context));
        assert!(loaded.materializations == vec![ownership]);
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v9_roundtrip_loads_with_an_empty_managed_spawn_replay_ledger() {
        let path = temp_path("v9-managed-spawn-migration");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, record) = fixture(&path);

        save_v9(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[],
        )
        .unwrap();

        let bytes = fs::read(&path).unwrap();
        let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(header.version, NODE_STATE_SCHEMA_V9);
        let loaded = load(Some(&path), &node_id).unwrap();
        assert!(loaded.managed_spawn_replays.is_empty());
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v8_save_rejects_context_id_and_receipt_mismatch() {
        let path = temp_path("v8-context-id-mismatch");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let context = context_receipt("ctx-receipt", 'b');
        record.context_id = Some(crate::protocol::SpawnContextId::new("ctx-other").unwrap());
        record.context = Some(context.clone());
        let ownership = context_materialization(
            &path,
            "context-mismatch",
            record.record_id.clone(),
            context,
        );

        let error = save_v8(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[ownership],
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert!(!path.exists());
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v8_save_rejects_mismatched_context_materialization_reference() {
        let path = temp_path("v8-context-materialization-mismatch");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let record_context = context_receipt("ctx-record", 'd');
        record.context_id = Some(record_context.id.clone());
        record.context = Some(record_context);
        let ownership = context_materialization(
            &path,
            "context-foreign",
            record.record_id.clone(),
            context_receipt("ctx-foreign", 'e'),
        );

        let error = save_v8(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[ownership],
        )
        .unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
        assert!(!path.exists());
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }

    #[test]
    fn v8_load_rejects_invalid_context_receipt() {
        let path = temp_path("v8-context-invalid-receipt");
        fs::create_dir_all(path.parent().unwrap()).unwrap();
        let (node_id, workspaces, mut record) = fixture(&path);
        let context = context_receipt("ctx-valid", 'c');
        record.context_id = Some(context.id.clone());
        record.context = Some(context.clone());
        let ownership = context_materialization(
            &path,
            "context-invalid",
            record.record_id.clone(),
            context,
        );
        save_v8(
            Some(&path),
            &node_id,
            &workspaces,
            &[record],
            &[],
            &[],
            &[ownership],
        )
        .unwrap();

        let mut invalid: serde_json::Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
        invalid["materializations"][0]["context"]["byte_len"] = serde_json::Value::from(0);
        fs::write(&path, serde_json::to_vec(&invalid).unwrap()).unwrap();
        assert_eq!(
            load(Some(&path), &node_id).unwrap_err().kind(),
            io::ErrorKind::InvalidData,
        );
        let _ = fs::remove_dir_all(path.parent().unwrap());
    }
}