agent-workspace-contract 0.7.0

//! Persistence ports and typed metadata deltas for Environment aggregates.

use super::*;
use crate::{CommandSpec, PortUrl, ProviderPortTarget};

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub enum EnvironmentIdempotencyState {
    Pending,
    Completed,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct EnvironmentIdempotencyRecord {
    pub id: String,
    pub scope: ResourceScope,
    pub action: String,
    /// SHA-256 of the caller-provided key; the key itself is never persisted.
    pub key_hash: String,
    pub request_hash: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resource_id: Option<String>,
    pub state: EnvironmentIdempotencyState,
    /// Runtime instance currently allowed to enter Provider I/O.
    pub claim_owner: String,
    /// A dead instance can be taken over only after this lease expires.
    pub claim_expires_at_ms: u64,
    pub attempt: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_json: Option<String>,
    pub started_at_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub completed_at_ms: Option<u64>,
}

/// Durable metadata snapshot for one tenant/project scope.
///
/// Provider bytes and credentials are deliberately absent. Repositories may
/// store each collection in a dedicated table while exposing one CAS unit to
/// the application service so fencing and revision changes commit together.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct EnvironmentMetadataSnapshot {
    #[serde(default)]
    pub user_computers: Vec<crate::UserComputerBinding>,
    #[serde(default)]
    pub conversation_spaces: Vec<crate::ConversationSpaceBinding>,
    #[serde(default)]
    pub workspace_bindings: Vec<crate::WorkspacePolicyBinding>,
    #[serde(default)]
    pub workspace_policies: Vec<crate::WorkspacePolicyRecord>,
    pub version: u64,
    #[serde(default)]
    pub computers: Vec<ComputerRecord>,
    #[serde(default)]
    pub spaces: Vec<SpaceRecord>,
    #[serde(default)]
    pub sandboxes: Vec<SandboxRecord>,
    #[serde(default)]
    pub managed_workloads: Vec<ManagedWorkload>,
    #[serde(default)]
    pub snapshots: Vec<SnapshotRecord>,
    #[serde(default)]
    pub templates: Vec<TemplateVersionRecord>,
    #[serde(default)]
    pub grants: Vec<SpaceGrant>,
    #[serde(default)]
    pub sessions: Vec<SpaceCollaborationSession>,
    #[serde(default)]
    pub tickets: Vec<SpaceAccessTicketRecord>,
    #[serde(default)]
    pub leases: Vec<SpaceWriteLease>,
    #[serde(default)]
    pub fencing_tokens: BTreeMap<String, u64>,
    #[serde(default)]
    pub delegations: Vec<DelegationRecord>,
    #[serde(default)]
    pub change_sets: Vec<SpaceChangeSetRecord>,
    #[serde(default)]
    pub idempotency: Vec<EnvironmentIdempotencyRecord>,
}

/// Incremental metadata changes for one aggregate collection.
///
/// Upserts and deletes are committed under the enclosing scope-version CAS;
/// unchanged aggregates never cross the repository boundary.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnvironmentMetadataCollectionPatch<T> {
    pub upserts: Vec<T>,
    pub deletes: Vec<String>,
}

impl<T> Default for EnvironmentMetadataCollectionPatch<T> {
    fn default() -> Self {
        Self {
            upserts: Vec::new(),
            deletes: Vec::new(),
        }
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct EnvironmentFencingTokenPatch {
    pub upserts: BTreeMap<String, u64>,
    pub deletes: Vec<String>,
}

/// Typed delta for one tenant/project Environment aggregate set.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct EnvironmentMetadataPatch {
    pub user_computers: EnvironmentMetadataCollectionPatch<crate::UserComputerBinding>,
    pub conversation_spaces: EnvironmentMetadataCollectionPatch<crate::ConversationSpaceBinding>,
    pub workspace_bindings: EnvironmentMetadataCollectionPatch<crate::WorkspacePolicyBinding>,
    pub workspace_policies: EnvironmentMetadataCollectionPatch<crate::WorkspacePolicyRecord>,
    pub version: u64,
    pub computers: EnvironmentMetadataCollectionPatch<ComputerRecord>,
    pub spaces: EnvironmentMetadataCollectionPatch<SpaceRecord>,
    pub sandboxes: EnvironmentMetadataCollectionPatch<SandboxRecord>,
    pub managed_workloads: EnvironmentMetadataCollectionPatch<ManagedWorkload>,
    pub snapshots: EnvironmentMetadataCollectionPatch<SnapshotRecord>,
    pub templates: EnvironmentMetadataCollectionPatch<TemplateVersionRecord>,
    pub grants: EnvironmentMetadataCollectionPatch<SpaceGrant>,
    pub sessions: EnvironmentMetadataCollectionPatch<SpaceCollaborationSession>,
    pub tickets: EnvironmentMetadataCollectionPatch<SpaceAccessTicketRecord>,
    pub leases: EnvironmentMetadataCollectionPatch<SpaceWriteLease>,
    pub fencing_tokens: EnvironmentFencingTokenPatch,
    pub delegations: EnvironmentMetadataCollectionPatch<DelegationRecord>,
    pub change_sets: EnvironmentMetadataCollectionPatch<SpaceChangeSetRecord>,
    pub idempotency: EnvironmentMetadataCollectionPatch<EnvironmentIdempotencyRecord>,
}

impl EnvironmentMetadataPatch {
    pub fn between(
        before: &EnvironmentMetadataSnapshot,
        after: &EnvironmentMetadataSnapshot,
    ) -> Self {
        Self {
            version: after.version,
            user_computers: collection_patch(&before.user_computers, &after.user_computers, |v| {
                v.id.clone()
            }),
            conversation_spaces: collection_patch(
                &before.conversation_spaces,
                &after.conversation_spaces,
                |v| v.id.clone(),
            ),
            workspace_bindings: collection_patch(
                &before.workspace_bindings,
                &after.workspace_bindings,
                |value| value.id.clone(),
            ),
            workspace_policies: collection_patch(
                &before.workspace_policies,
                &after.workspace_policies,
                |value| value.id.clone(),
            ),
            computers: collection_patch(&before.computers, &after.computers, |value| {
                value.id.clone()
            }),
            spaces: collection_patch(&before.spaces, &after.spaces, |value| value.id.clone()),
            sandboxes: collection_patch(&before.sandboxes, &after.sandboxes, |value| {
                value.id.clone()
            }),
            managed_workloads: collection_patch(
                &before.managed_workloads,
                &after.managed_workloads,
                |value| value.id.clone(),
            ),
            snapshots: collection_patch(&before.snapshots, &after.snapshots, |value| {
                value.id.clone()
            }),
            templates: collection_patch(&before.templates, &after.templates, |value| {
                format!("{}\0{}", value.template_id, value.version)
            }),
            grants: collection_patch(&before.grants, &after.grants, |value| value.id.clone()),
            sessions: collection_patch(&before.sessions, &after.sessions, |value| value.id.clone()),
            tickets: collection_patch(&before.tickets, &after.tickets, |value| value.id.clone()),
            leases: collection_patch(&before.leases, &after.leases, |value| value.id.clone()),
            fencing_tokens: map_patch(&before.fencing_tokens, &after.fencing_tokens),
            delegations: collection_patch(&before.delegations, &after.delegations, |value| {
                value.id.clone()
            }),
            change_sets: collection_patch(&before.change_sets, &after.change_sets, |value| {
                value.id.clone()
            }),
            idempotency: collection_patch(&before.idempotency, &after.idempotency, |value| {
                value.id.clone()
            }),
        }
    }

    pub fn changed_aggregate_count(&self) -> usize {
        let collections = [
            patch_size(&self.workspace_bindings),
            patch_size(&self.user_computers),
            patch_size(&self.conversation_spaces),
            patch_size(&self.workspace_policies),
            patch_size(&self.computers),
            patch_size(&self.spaces),
            patch_size(&self.sandboxes),
            patch_size(&self.managed_workloads),
            patch_size(&self.snapshots),
            patch_size(&self.templates),
            patch_size(&self.grants),
            patch_size(&self.sessions),
            patch_size(&self.tickets),
            patch_size(&self.leases),
            patch_size(&self.delegations),
            patch_size(&self.change_sets),
            patch_size(&self.idempotency),
        ];
        collections.into_iter().sum::<usize>()
            + self.fencing_tokens.upserts.len()
            + self.fencing_tokens.deletes.len()
    }
}

fn collection_patch<T: Clone + PartialEq>(
    before: &[T],
    after: &[T],
    id: impl Fn(&T) -> String,
) -> EnvironmentMetadataCollectionPatch<T> {
    let previous = before
        .iter()
        .map(|value| (id(value), value))
        .collect::<BTreeMap<_, _>>();
    let current = after
        .iter()
        .map(|value| (id(value), value))
        .collect::<BTreeMap<_, _>>();
    EnvironmentMetadataCollectionPatch {
        upserts: current
            .iter()
            .filter(|(key, value)| previous.get(*key).is_none_or(|old| *old != **value))
            .map(|(_, value)| (*value).clone())
            .collect(),
        deletes: previous
            .keys()
            .filter(|key| !current.contains_key(*key))
            .cloned()
            .collect(),
    }
}

fn map_patch(
    before: &BTreeMap<String, u64>,
    after: &BTreeMap<String, u64>,
) -> EnvironmentFencingTokenPatch {
    EnvironmentFencingTokenPatch {
        upserts: after
            .iter()
            .filter(|(key, value)| before.get(*key).is_none_or(|old| old != *value))
            .map(|(key, value)| (key.clone(), *value))
            .collect(),
        deletes: before
            .keys()
            .filter(|key| !after.contains_key(*key))
            .cloned()
            .collect(),
    }
}

fn patch_size<T>(patch: &EnvironmentMetadataCollectionPatch<T>) -> usize {
    patch.upserts.len() + patch.deletes.len()
}

/// Existence checks reveal no foreign aggregate or owner metadata.
#[derive(Debug, Clone, Copy)]
pub enum EnvironmentResourceKind {
    Computer,
    Space,
}

#[async_trait]
pub trait EnvironmentMetadataRepository: Send + Sync + Debug {
    async fn resource_exists_outside_scope(
        &self,
        kind: EnvironmentResourceKind,
        id: &str,
        scope: &ResourceScope,
    ) -> Result<bool>;
    async fn load(&self, scope: &ResourceScope) -> Result<EnvironmentMetadataSnapshot>;

    /// Stable tenant/project pagination for background lifecycle reconciliation.
    async fn list_scopes(
        &self,
        after: Option<&ResourceScope>,
        limit: usize,
    ) -> Result<Vec<ResourceScope>>;

    /// Atomically replace metadata when the stored scope version equals
    /// `expected_version`. Returns false on a concurrent writer conflict.
    async fn compare_and_swap(
        &self,
        scope: &ResourceScope,
        expected_version: u64,
        snapshot: &EnvironmentMetadataSnapshot,
    ) -> Result<bool>;

    /// Atomically apply only changed aggregates when the stored scope version
    /// equals `expected_version`.
    async fn compare_and_patch(
        &self,
        scope: &ResourceScope,
        expected_version: u64,
        patch: &EnvironmentMetadataPatch,
    ) -> Result<bool>;
}

/// Command metadata and output are stored independently from the Environment
/// aggregate snapshot. This keeps output appends O(chunks) instead of copying
/// every resource in a tenant/project scope.
#[async_trait]
pub trait EnvironmentCommandRepository: Send + Sync + Debug {
    async fn store_computer_command(
        &self,
        request: StoreComputerCommandRequest,
    ) -> Result<StoreComputerCommandOutcome>;

    async fn get_computer_command(
        &self,
        scope: &ResourceScope,
        computer_id: &str,
        command_id: &str,
    ) -> Result<Option<ComputerCommandResource>>;

    async fn transition_computer_command(
        &self,
        request: ComputerCommandTransition,
    ) -> Result<ComputerCommandResource>;

    async fn complete_computer_command(
        &self,
        request: CompleteComputerCommandRequest,
    ) -> Result<ComputerCommandResource>;

    async fn computer_command_output(
        &self,
        scope: &ResourceScope,
        computer_id: &str,
        command_id: &str,
        after: u64,
        limit: usize,
    ) -> Result<ComputerCommandOutputPage>;
}

/// Production Environment persistence owns both aggregate metadata and the
/// independently indexed command log.
pub trait EnvironmentRepository:
    EnvironmentMetadataRepository + EnvironmentCommandRepository
{
}
impl<T> EnvironmentRepository for T where
    T: EnvironmentMetadataRepository + EnvironmentCommandRepository
{
}

#[async_trait]
pub trait ComputerProvider: Send + Sync + Debug {
    fn kind(&self) -> &str;
    fn capabilities(&self) -> ProviderCapabilities;
    fn validate_create(&self, _request: &ProviderComputerRequest) -> Result<()> {
        Ok(())
    }
    async fn create_computer(
        &self,
        scope: &ResourceScope,
        request: &ProviderComputerRequest,
        idempotency_key: &str,
    ) -> Result<ProviderResourceRef>;
    async fn delete_computer(&self, external_id: &str, idempotency_key: &str) -> Result<()>;
    async fn port_url(&self, computer_external_id: &str, port: u16) -> Result<PortUrl> {
        let _ = (computer_external_id, port);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support preview URLs",
        )
        .into())
    }
    async fn port_target(
        &self,
        computer_external_id: &str,
        port: u16,
        path: &str,
        query: Option<&str>,
    ) -> Result<ProviderPortTarget> {
        let _ = (computer_external_id, port, path, query);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support proxied preview traffic",
        )
        .into())
    }
    async fn create_space_directory(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        request: &ProviderSpaceRequest,
        idempotency_key: &str,
    ) -> Result<()> {
        let _ = (
            computer_external_id,
            space_directory,
            request,
            idempotency_key,
        );
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support Space directories",
        )
        .into())
    }
    async fn delete_space_directory(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        idempotency_key: &str,
    ) -> Result<()> {
        let _ = (computer_external_id, space_directory, idempotency_key);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support Space directories",
        )
        .into())
    }
    async fn read_file(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        path: &str,
    ) -> Result<ProviderFileContent> {
        let _ = (computer_external_id, space_directory, path);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support file reads",
        )
        .into())
    }
    async fn write_file(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        path: &str,
        content: &[u8],
        expected_provider_revision: Option<&str>,
        idempotency_key: &str,
    ) -> Result<ProviderWriteResult> {
        let _ = (
            computer_external_id,
            space_directory,
            path,
            content,
            expected_provider_revision,
            idempotency_key,
        );
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support file writes",
        )
        .into())
    }
    async fn delete_file(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        path: &str,
        idempotency_key: &str,
    ) -> Result<()> {
        let _ = (computer_external_id, space_directory, path, idempotency_key);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support file deletes",
        )
        .into())
    }
    async fn list_files(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        path: &str,
    ) -> Result<Vec<ProviderFileEntry>> {
        let _ = (computer_external_id, space_directory, path);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support directory listing",
        )
        .into())
    }
    async fn source_archive(
        &self,
        computer_external_id: &str,
        space_directory: &str,
    ) -> Result<ProviderSourceArchive> {
        let _ = (computer_external_id, space_directory);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support source archives",
        )
        .into())
    }
    async fn create_change_set(
        &self,
        computer_external_id: &str,
        source_space_directory: &str,
        target_space_directory: &str,
        idempotency_key: &str,
    ) -> Result<String> {
        let _ = (
            computer_external_id,
            source_space_directory,
            target_space_directory,
            idempotency_key,
        );
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support change sets",
        )
        .into())
    }
    async fn apply_change_set(
        &self,
        computer_external_id: &str,
        source_space_directory: &str,
        target_space_directory: &str,
        provider_change_ref: &str,
        idempotency_key: &str,
    ) -> Result<ProviderWriteResult> {
        let _ = (
            computer_external_id,
            source_space_directory,
            target_space_directory,
            provider_change_ref,
            idempotency_key,
        );
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support change sets",
        )
        .into())
    }
    async fn exec_command(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        command: &CommandSpec,
    ) -> Result<ProviderCommandOutput> {
        let _ = (computer_external_id, space_directory, command);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "computer does not support managed commands",
        )
        .into())
    }
    /// Set mode/owner/group on a Space path through the Provider file API.
    /// Prefer this over user-level `chmod`/`chown` when Provider file writes create
    /// root-owned files that the sandbox user cannot mutate.
    async fn set_space_path_permissions(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        path: &str,
        mode: &str,
        owner: Option<&str>,
        group: Option<&str>,
    ) -> Result<()> {
        let _ = (
            computer_external_id,
            space_directory,
            path,
            mode,
            owner,
            group,
        );
        Ok(())
    }
    async fn exec_managed_command(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<ProviderCommandOutput> {
        let _ = command_id;
        self.exec_command(computer_external_id, space_directory, command)
            .await
    }
    async fn resume_managed_command(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<Option<ProviderCommandOutput>> {
        let _ = (computer_external_id, space_directory, command_id, command);
        Ok(None)
    }
    async fn retry_managed_command(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<Option<ProviderCommandOutput>> {
        self.resume_managed_command(computer_external_id, space_directory, command_id, command)
            .await
    }
    async fn cancel_command(
        &self,
        computer_external_id: &str,
        space_directory: &str,
        command_id: &str,
    ) -> Result<()> {
        let _ = (computer_external_id, space_directory, command_id);
        Ok(())
    }
}

#[async_trait]
pub trait SandboxProvider: Send + Sync + Debug {
    fn kind(&self) -> &str;
    fn capabilities(&self) -> ProviderCapabilities;
    fn validate_create(&self, _request: &ProviderSandboxRequest) -> Result<()> {
        Ok(())
    }
    async fn create_sandbox(
        &self,
        scope: &ResourceScope,
        request: &ProviderSandboxRequest,
        idempotency_key: &str,
    ) -> Result<ProviderResourceRef>;
    async fn stop_sandbox(&self, external_id: &str, idempotency_key: &str) -> Result<()>;
    async fn write_file(
        &self,
        external_id: &str,
        path: &str,
        content: &[u8],
        expected_provider_revision: Option<&str>,
        idempotency_key: &str,
    ) -> Result<ProviderWriteResult> {
        let _ = (
            external_id,
            path,
            content,
            expected_provider_revision,
            idempotency_key,
        );
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "sandbox does not support file writes",
        )
        .into())
    }
    async fn read_file(&self, external_id: &str, path: &str) -> Result<ProviderFileContent> {
        let _ = (external_id, path);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "sandbox does not support file reads",
        )
        .into())
    }
    async fn exec_command(
        &self,
        external_id: &str,
        command: &CommandSpec,
    ) -> Result<ProviderCommandOutput> {
        let _ = (external_id, command);
        Err(EnvironmentDomainError::new(
            EnvironmentErrorCode::CapabilityUnsupported,
            "sandbox does not support managed commands",
        )
        .into())
    }
    /// Set mode/owner/group on a Sandbox path through the Provider file API.
    async fn set_path_permissions(
        &self,
        external_id: &str,
        path: &str,
        mode: &str,
        owner: Option<&str>,
        group: Option<&str>,
    ) -> Result<()> {
        let _ = (external_id, path, mode, owner, group);
        Ok(())
    }
    async fn exec_managed_command(
        &self,
        external_id: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<ProviderCommandOutput> {
        let _ = command_id;
        self.exec_command(external_id, command).await
    }
    async fn resume_managed_command(
        &self,
        external_id: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<Option<ProviderCommandOutput>> {
        let _ = (external_id, command_id, command);
        Ok(None)
    }
    async fn retry_managed_command(
        &self,
        external_id: &str,
        command_id: &str,
        command: &CommandSpec,
    ) -> Result<Option<ProviderCommandOutput>> {
        self.resume_managed_command(external_id, command_id, command)
            .await
    }
    async fn cancel_command(&self, external_id: &str, command_id: &str) -> Result<()> {
        let _ = (external_id, command_id);
        Ok(())
    }
}

#[async_trait]
pub trait SpaceProvider: Send + Sync + Debug {
    fn kind(&self) -> &str;
    fn capabilities(&self) -> ProviderCapabilities;
    async fn create_space(
        &self,
        scope: &ResourceScope,
        request: &ProviderSpaceRequest,
        idempotency_key: &str,
    ) -> Result<ProviderResourceRef>;
    async fn read_file(&self, external_id: &str, path: &str) -> Result<ProviderFileContent>;
    async fn write_file(
        &self,
        external_id: &str,
        path: &str,
        content: &[u8],
        expected_provider_revision: Option<&str>,
        idempotency_key: &str,
    ) -> Result<ProviderWriteResult>;
    async fn create_change_set(
        &self,
        source_external_id: &str,
        target_external_id: &str,
        idempotency_key: &str,
    ) -> Result<String>;
    async fn apply_change_set(
        &self,
        source_external_id: &str,
        target_external_id: &str,
        provider_change_ref: &str,
        idempotency_key: &str,
    ) -> Result<ProviderWriteResult>;
    async fn delete_space(&self, external_id: &str, idempotency_key: &str) -> Result<()>;
}

#[async_trait]
pub trait SnapshotProviderPort: Send + Sync + Debug {
    fn kind(&self) -> &str;
    async fn capture_snapshot(
        &self,
        source: &ProviderSnapshotSource,
        idempotency_key: &str,
    ) -> Result<ProviderResourceRef>;
    async fn delete_snapshot(
        &self,
        snapshot_external_id: &str,
        idempotency_key: &str,
    ) -> Result<()>;
}

#[async_trait]
pub trait TemplateProviderPort: Send + Sync + Debug {
    fn kind(&self) -> &str;
    async fn inspect_template(&self, external_id: &str) -> Result<ProviderCapabilities>;
}

pub trait EnvironmentProviderRegistry: Send + Sync + Debug {
    fn computer(&self, provider: &str) -> Result<Arc<dyn ComputerProvider>>;
    fn sandbox(&self, provider: &str) -> Result<Arc<dyn SandboxProvider>>;
    fn space(&self, provider: &str) -> Result<Arc<dyn SpaceProvider>>;
    fn snapshot(&self, provider: &str) -> Result<Arc<dyn SnapshotProviderPort>>;
    fn template(&self, provider: &str) -> Result<Arc<dyn TemplateProviderPort>>;
}

#[cfg(test)]
mod patch_tests {
    use super::*;

    fn record(scope: &ResourceScope, id: &str, attempt: u32) -> EnvironmentIdempotencyRecord {
        EnvironmentIdempotencyRecord {
            id: id.into(),
            scope: scope.clone(),
            action: "test".into(),
            key_hash: format!("key-{id}"),
            request_hash: format!("request-{id}"),
            resource_id: None,
            state: EnvironmentIdempotencyState::Pending,
            claim_owner: "worker".into(),
            claim_expires_at_ms: 10,
            attempt,
            result_json: None,
            started_at_ms: 1,
            completed_at_ms: None,
        }
    }

    #[test]
    fn metadata_patch_contains_only_changed_deleted_and_new_aggregates() {
        let scope = ResourceScope {
            tenant_id: "tenant".into(),
            project_id: "project".into(),
        };
        let unchanged = record(&scope, "unchanged", 1);
        let changed = record(&scope, "changed", 1);
        let deleted = record(&scope, "deleted", 1);
        let before = EnvironmentMetadataSnapshot {
            version: 7,
            idempotency: vec![unchanged.clone(), changed, deleted],
            fencing_tokens: BTreeMap::from([
                ("unchanged-space".into(), 1),
                ("changed-space".into(), 1),
                ("deleted-space".into(), 1),
            ]),
            ..EnvironmentMetadataSnapshot::default()
        };
        let changed = record(&scope, "changed", 2);
        let added = record(&scope, "added", 1);
        let after = EnvironmentMetadataSnapshot {
            version: 8,
            idempotency: vec![unchanged, changed.clone(), added.clone()],
            fencing_tokens: BTreeMap::from([
                ("unchanged-space".into(), 1),
                ("changed-space".into(), 2),
                ("added-space".into(), 1),
            ]),
            ..EnvironmentMetadataSnapshot::default()
        };

        let patch = EnvironmentMetadataPatch::between(&before, &after);

        assert_eq!(patch.version, 8);
        assert_eq!(patch.idempotency.upserts, vec![added, changed]);
        assert_eq!(patch.idempotency.deletes, vec!["deleted"]);
        assert_eq!(
            patch.fencing_tokens.upserts,
            BTreeMap::from([("added-space".into(), 1), ("changed-space".into(), 2)])
        );
        assert_eq!(patch.fencing_tokens.deletes, vec!["deleted-space"]);
        assert_eq!(patch.changed_aggregate_count(), 6);
    }
}